Agency Swarm - 完整学习教程

Agency Swarm - 完整学习教程

教程级别: 从零到一 预计学习时间: 8-10 小时 前置知识: Python 编程(中级)、OpenAI API 基础、Pydantic 基础

环境搭建指南

系统要求

  • 操作系统:macOS 12+、Ubuntu 20.04+、Windows 10+(需 WSL2)
  • Python 版本:3.10 或更高
  • OpenAI API 密钥:需要一个有效的 OpenAI API Key(GPT-4o 或 GPT-5 访问权限)

安装步骤

# 1. 创建项目目录
mkdir my-agency && cd my-agency

# 2. 创建 Python 虚拟环境
python3 -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 3. 安装 Agency Swarm
pip install agency-swarm

# 4. 创建 .env 文件并配置 API Key
cat > .env << 'EOF'
OPENAI_API_KEY=sk-your-api-key-here
EOF

# 5. (可选)安装 LiteLLM 以支持非 OpenAI 模型
pip install litellm

验证安装

python3 -c "
from agency_swarm import Agent, Agency
print('Agency Swarm 安装成功!')
"

执行结果:

Agency Swarm 安装成功!

第一部分:入门篇

1.1 创建你的第一个 Agent(智能体)

概念讲解:

Agent(智能体)是 Agency Swarm 的基本构建单元,就像公司里的一位员工。每个 Agent 有三个核心属性:

  • name:名称,用于识别和通信
  • description:简短职责描述,帮助其他智能体决定是否联系它
  • instructions:详细的系统指令,定义行为规范

代码示例:

# first_agent.py
from agency_swarm import Agent, Agency

# 创建一个简单的智能体
assistant = Agent(
    name="Assistant",
    description="一个通用的 AI 助手。",
    instructions=(
        "你是一个有帮助的 AI 助手。\n"
        "# 角色\n"
        "你负责回答用户的问题,提供准确、简洁的信息。\n"
        "# 规则\n"
        "1. 使用中文回答\n"
        "2. 回答要简洁明了\n"
        "3. 如果不确定,请诚实说明"
    ),
)

# 创建 Agency(仅包含一个智能体)
agency = Agency(assistant)

# 同步运行获取响应
result = agency.get_response_sync("请用一句话介绍什么是多智能体系统")
print(result.final_output)

执行结果:

多智能体系统是由多个 AI 智能体通过协作和通信共同完成复杂任务的系统。

练习题: 1. 修改 instructions,让智能体只回答编程相关问题,测试发送非编程问题时的行为。 2. 将智能体的 name 改为 "CodeExpert",观察行为变化。


1.2 定义自定义工具(Tool)

概念讲解:

工具(Tool)是扩展智能体能力的关键机制。Agency Swarm v1.x 提供两种定义方式:

  1. BaseTool 类:继承 BaseTool,适合复杂工具,支持 Pydantic 字段验证
  2. @function_tool 装饰器:适合简单工具,代码更简洁

工具的文档字符串(docstring)非常重要——智能体根据它决定何时调用工具。

代码示例:

# custom_tools.py
from agency_swarm.tools import BaseTool
from agency_swarm import function_tool
from pydantic import Field
from agency_swarm import Agent, Agency


# === 方式一:BaseTool 类 ===
class CalculatorTool(BaseTool):
    """执行基本的数学计算。当需要进行加减乘除运算时使用此工具。"""

    expression: str = Field(
        ...,
        description="要计算的数学表达式,例如 '2 + 3' 或 '10 * 5'"
    )

    async def run(self):
        """执行数学计算并返回结果"""
        try:
            allowed_chars = set("0123456789+-*/(). ")
            if not all(c in allowed_chars for c in self.expression):
                return "错误:表达式包含不允许的字符"
            result = eval(self.expression)
            return f"{self.expression} = {result}"
        except Exception as e:
            return f"计算错误:{str(e)}"


# === 方式二:@function_tool 装饰器 ===
@function_tool
def get_word_count(text: str) -> str:
    """计算文本的字符数和词数。当需要分析文本长度时使用此工具。

    Args:
        text: 要分析的文本内容
    Returns:
        包含字符数和词数的结果字符串
    """
    char_count = len(text)
    word_count = len(text.split())
    return f"字符数:{char_count},词数:{word_count}"


# === 将工具分配给 Agent ===
writer = Agent(
    name="Writer",
    description="负责撰写和分析文本内容。",
    instructions=(
        "你是一个文本分析师。\n"
        "当用户给你文本时,使用 get_word_count 工具分析文本长度。\n"
        "当用户需要计算时,使用 CalculatorTool 进行数学运算。\n"
        "使用中文回答。"
    ),
    tools=[CalculatorTool, get_word_count],
)

agency = Agency(writer)

result = agency.get_response_sync("请帮我计算一下 125 * 48 + 37 的结果")
print(result.final_output)

执行结果:

125 * 48 + 37 = 6037

练习题: 1. 创建一个 TemperatureConverter 工具(继承 BaseTool),实现摄氏度和华氏度互转。 2. 用 @function_tool 装饰器创建一个 get_current_date 工具,返回当前日期。


1.3 理解 Agency 和 Communication Flows(通信流)

概念讲解:

Agency 是管理多个 Agent 的顶层容器。通信流(Communication Flows)定义了智能体之间的通信权限,使用方向性元组 (agent_a, agent_b) 表示 A 可以联系 B,但 B 不能反过来联系 A。

第一个传入 Agency 的智能体是入口智能体,用户请求从这里开始。

代码示例:

# communication_flows.py
from agency_swarm import Agent, Agency, function_tool

@function_tool
def search_web(query: str) -> str:
    """搜索互联网获取信息。

    Args:
        query: 搜索关键词
    Returns:
        搜索结果摘要
    """
    return f"关于 '{query}' 的搜索结果:这是模拟的搜索结果内容..."

# 定义三个角色
ceo = Agent(
    name="CEO",
    description="首席执行官,负责理解用户需求并分配任务。",
    instructions=(
        "你是 CEO。\n"
        "你的职责是理解用户的需求,然后将任务分配给合适的团队成员。\n"
        "- 如果需要搜索信息,联系 Researcher\n"
        "- 如果需要写代码,联系 Developer\n"
        "使用中文沟通。"
    ),
)

researcher = Agent(
    name="Researcher",
    description="研究员,负责搜索和整理信息。",
    instructions="你是研究员。使用 search_web 工具搜索信息并整理成报告。使用中文。",
    tools=[search_web],
)

developer = Agent(
    name="Developer",
    description="开发者,负责编写代码。",
    instructions="你是开发者。负责编写代码和解决技术问题。使用中文。",
)

# 创建 Agency,定义通信流
agency = Agency(
    ceo,  # 入口智能体
    communication_flows=[
        (ceo, researcher),    # CEO 可以联系 Researcher
        (ceo, developer),     # CEO 可以联系 Developer
    ],
    shared_instructions="所有成员使用中文沟通,回复要简洁。"
)

result = agency.get_response_sync("请搜索一下什么是多智能体系统")
print(result.final_output)

执行结果:

根据搜索结果,多智能体系统是由多个自主智能体组成的系统,
这些智能体通过协作来共同完成复杂任务。

练习题: 1. 添加 (researcher, developer) 通信流,让研究员可直接将技术问题转给开发者。 2. 移除 (ceo, researcher) 通信流,观察 CEO 还能否分配搜索任务。


第二部分:进阶篇

2.1 Agent 项目结构化

详细讲解:

在实际项目中,推荐为每个 Agent 创建独立的文件夹。这种结构的好处是关注点分离、可复用、可测试。

代码示例:

项目目录结构:

# my_agency/
# ├── ceo/
# │   ├── __init__.py
# │   ├── ceo.py
# │   └── instructions.md
# ├── researcher/
# │   ├── __init__.py
# │   ├── researcher.py
# │   ├── instructions.md
# │   └── tools/
# │       ├── __init__.py
# │       └── search_tool.py
# ├── agency.py
# ├── agency_manifesto.md
# └── .env

ceo/ceo.py:

from agency_swarm import Agent

ceo = Agent(
    name="CEO",
    description="首席执行官,负责理解用户需求并分配任务。",
    instructions="./ceo/instructions.md",
)

ceo/instructions.md:

你是 CEO,负责理解用户的需求并分配任务。

# 角色
你是团队的管理者,负责协调 Researcher 和 Developer 的工作。

# 决策规则
- 信息搜索需求 → 联系 Researcher
- 代码开发需求 → 联系 Developer

researcher/tools/search_tool.py:

from agency_swarm.tools import BaseTool
from pydantic import Field

class SearchTool(BaseTool):
    """搜索互联网获取信息。"""

    query: str = Field(..., description="搜索关键词")

    async def run(self):
        return f"搜索结果:关于 '{self.query}' 的信息..."

researcher/researcher.py:

from agency_swarm import Agent
from .tools.search_tool import SearchTool

researcher = Agent(
    name="Researcher",
    description="研究员,负责搜索和整理信息。",
    instructions="./researcher/instructions.md",
    tools=[SearchTool],
)

agency.py(主入口):

from agency_swarm import Agency
from ceo import ceo
from researcher import researcher

agency = Agency(
    ceo,
    communication_flows=[(ceo, researcher)],
    shared_instructions="./agency_manifesto.md",
)

if __name__ == "__main__":
    agency.terminal_demo()

注意事项: - instructions 路径相对于运行脚本的工作目录,不是相对于 Agent 文件本身。 - 每个 Agent 文件夹需要 __init__.py 才能作为 Python 包导入。

练习题: 1. 添加 Developer 角色文件夹,实现 CEO → Developer 通信流。 2. 为 Developer 创建一个 CodeRunnerTool


2.2 Handoff 模式与控制权转移

详细讲解:

标准通信中,SendMessage 保持调用者控制权。Handoff 模式(SendMessageHandoff)将控制权完全转移给接收者,适用于升级处理和专业转接场景。

代码示例:

# handoff_pattern.py
from agency_swarm import Agent, Agency, function_tool
from agency_swarm.tools import SendMessageHandoff

@function_tool
def query_knowledge_base(question: str) -> str:
    """从知识库查询常见问题。

    Args:
        question: 用户的问题
    Returns:
        匹配答案或 '未找到'
    """
    kb = {
        "退款": "退款可在订单页面申请,3-5 个工作日到账。",
        "配送": "标准配送 3-7 天,加急配送 1-2 天。",
        "退货": "收到商品 7 天内可申请退货。",
    }
    for key, answer in kb.items():
        if key in question:
            return answer
    return "未找到匹配的答案"

tier1 = Agent(
    name="Tier1Support",
    description="一线客服,处理常见问题。",
    instructions=(
        "你是一线客服。\n"
        "1. 首先使用 query_knowledge_base 查找答案\n"
        "2. 如果返回 '未找到',用 SendMessageHandoff 升级到 Tier2\n"
        "3. 使用中文回答"
    ),
    tools=[query_knowledge_base, SendMessageHandoff(recipient="Tier2Support")],
)

tier2 = Agent(
    name="Tier2Support",
    description="高级客服,处理复杂问题。",
    instructions="你是高级客服,处理一线无法解决的问题。使用中文回答。",
)

agency = Agency(
    tier1,
    communication_flows=[(tier1, tier2)],
)

# 常见问题(一线直接回答)
result1 = agency.get_response_sync("我想退款,怎么操作?")
print("常见问题:", result1.final_output)

# 复杂问题(升级到高级客服)
result2 = agency.get_response_sync("商品超过7天了但有问题怎么办?")
print("复杂问题:", result2.final_output)

执行结果:

常见问题: 退款可在订单页面申请,3-5 个工作日到账。
复杂问题: 您好,关于超过7天的情况,我建议您联系售后...

注意事项: - SendMessageHandoff 必须显式指定 recipient 参数 - Handoff 后控制权完全转移,原智能体不再参与后续对话

练习题: 1. 添加 Tier3Support,实现三级升级链路。 2. 让 Tier2 完成后通过 Handoff 将控制权交回 Tier1。


2.3 状态持久化与线程管理

详细讲解:

默认对话历史仅存在内存中。通过 load_threads_callbacksave_threads_callback 回调实现持久化。

代码示例:

# state_persistence.py
import json
import os
from agency_swarm import Agent, Agency

STORAGE_DIR = "./thread_storage"

def save_threads(agent_name, threads_data):
    """保存线程到文件"""
    os.makedirs(STORAGE_DIR, exist_ok=True)
    filepath = os.path.join(STORAGE_DIR, f"{agent_name}.json")
    with open(filepath, "w", encoding="utf-8") as f:
        json.dump(threads_data, f, ensure_ascii=False, indent=2)
    print(f"[持久化] 已保存 {agent_name}")

def load_threads(agent_name):
    """从文件加载线程"""
    filepath = os.path.join(STORAGE_DIR, f"{agent_name}.json")
    if os.path.exists(filepath):
        with open(filepath, "r", encoding="utf-8") as f:
            data = json.load(f)
        print(f"[持久化] 已加载 {agent_name}")
        return data
    print(f"[持久化] {agent_name} 无历史数据")
    return None

assistant = Agent(
    name="PersistentAssistant",
    description="有记忆的 AI 助手。",
    instructions="你是有记忆的助手,能记住之前的对话。使用中文回答。",
)

agency = Agency(
    assistant,
    load_threads_callback=load_threads,
    save_threads_callback=save_threads,
)

# 第一次对话
print("=== 第一次对话 ===")
result1 = agency.get_response_sync("我叫小明,我喜欢编程")
print(result1.final_output)

# 第二次对话(记住上下文)
print("\n=== 第二次对话 ===")
result2 = agency.get_response_sync("你还记得我叫什么名字吗?")
print(result2.final_output)

执行结果:

=== 第一次对话 ===
[持久化] PersistentAssistant 无历史数据
[持久化] 已保存 PersistentAssistant
你好小明!很高兴认识你。编程是一个很有趣的领域...

=== 第二次对话 ===
[持久化] 已加载 PersistentAssistant
[持久化] 已保存 PersistentAssistant
你叫小明!你之前告诉我你喜欢编程。

注意事项: - 回调签名:load_callback(name) -> datasave_callback(name, data) -> None - 生产环境建议使用数据库(PostgreSQL、Redis)代替文件

练习题: 1. 将文件存储替换为 SQLite 存储。 2. 添加线程过期机制:超过 7 天的线程自动清理。


第三部分:高级篇

3.1 FastAPI 集成与流式输出

代码示例:

# fastapi_integration.py
from agency_swarm import Agent, Agency
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

ceo = Agent(
    name="CEO",
    description="首席执行官。",
    instructions="你是 CEO,回答用户问题。使用中文。",
)

agency = Agency(ceo)
app = FastAPI(title="Agency Swarm API")

@app.get("/chat")
async def chat(message: str):
    """同步方式:等待完整响应"""
    result = await agency.get_response(message)
    return {"response": result.final_output}

@app.get("/chat/stream")
async def chat_stream(message: str):
    """流式方式:逐 token 返回"""
    async def generate():
        async for event in await agency.get_response_stream(message):
            if hasattr(event, "delta") and event.delta:
                yield f"data: {event.delta}\n\n"
        yield "data: [DONE]\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream")
# 启动服务
uvicorn fastapi_integration:app --host 0.0.0.0 --port 8000 --reload

执行结果:

# GET /chat?message=你好
{"response":"你好!有什么可以帮助你的吗?"}

# GET /chat/stream?message=你好(逐 token)
data: 你
data: 好
...
data: [DONE]

3.2 Guardrails(护栏)机制

代码示例:

# guardrails.py
from agency_swarm import Agent, Agency

def validate_input(message: str) -> str | None:
    """返回 None 通过,返回字符串拒绝"""
    forbidden = ["暴力", "违法"]
    for word in forbidden:
        if word in message:
            return f"输入包含不允许的内容:{word}"
    return None

def validate_output(response: str) -> str | None:
    if len(response) > 500:
        return f"输出超长({len(response)} 字符)"
    return None

agent = Agent(
    name="SafeAgent",
    description="带护栏的助手。",
    instructions="你是助手。使用中文回答,每次不超过200字。",
    input_guardrails=[validate_input],
    output_guardrails=[validate_output],
)

agency = Agency(agent)

# 正常请求
r1 = agency.get_response_sync("什么是 Python?")
print("正常:", r1.final_output)

# 触发护栏
r2 = agency.get_response_sync("请告诉我关于暴力的内容")
print("护栏:", r2.final_output)

执行结果:

正常: Python 是一种广泛使用的高级编程语言...
护栏: 抱歉,您的请求包含不允许的内容。

3.3 性能优化

  • 优化策略一:使用异步 API
import asyncio

async def batch_queries(queries):
    """并发处理多个查询(注意 API 速率限制)"""
    tasks = [agency.get_response(q) for q in queries]
    results = await asyncio.gather(*tasks)
    return [r.final_output for r in results]

results = asyncio.run(batch_queries(["问题1", "问题2"]))
  • 优化策略二:控制模型参数
from agency_swarm import Agent, ModelSettings

agent = Agent(
    name="EfficientAgent",
    instructions="简洁回答。",
    model_settings=ModelSettings(
        temperature=0.1,
        max_tokens=500,
        model="gpt-4o-mini",
    ),
)

3.4 最佳实践

  1. 指令要具体:包含角色定义、决策规则和输出格式要求。
  2. 通信流要最小化:只定义必要路径,避免全连接。
  3. 工具 docstring 要清晰:智能体根据 docstring 决定是否使用。
  4. 设置合理的 max_tokens:控制成本和输出质量。
  5. 使用 shared_instructions:为所有智能体提供通用准则。

第四部分:实战项目

项目需求

构建一个智能客服系统,由三个智能体组成:

  1. Router(路由器):理解用户问题,分配给合适的客服
  2. FAQBot(常见问题机器人):回答常见问题
  3. TechSupport(技术支持):处理复杂技术问题

支持持久化和输出护栏。

项目设计

用户 → Router → FAQBot(常见问题)
                → TechSupport(技术问题)

运用的知识点: - Agent 定义与工具(1.1、1.2 节) - 通信流(1.3 节) - Handoff 模式(2.2 节) - Guardrails 护栏(3.2 节)

完整实现代码

# support_team.py
import json
import os
from agency_swarm import Agent, Agency, function_tool
from agency_swarm.tools import BaseTool, SendMessageHandoff
from pydantic import Field


# ==========================================
# 工具定义
# ==========================================

class FAQSearchTool(BaseTool):
    """搜索常见问题知识库。当用户的问题可能是常见问题时使用。"""

    question: str = Field(..., description="用户的原始问题")

    async def run(self):
        faq = {
            "密码": "请点击登录页面的'忘记密码'链接,按提示重置密码。",
            "收费": "基础版免费,专业版 ¥99/月,企业版联系销售。",
            "导出": "在项目页面点击'导出'按钮,支持 CSV 和 Excel 格式。",
            "API": "API 文档在 docs.example.com,需要先获取 API Key。",
        }
        for key, answer in faq.items():
            if key in self.question:
                return answer
        return "NOT_FOUND"


@function_tool
def create_ticket(title: str, description: str) -> str:
    """创建技术支持工单。

    Args:
        title: 工单标题
        description: 问题描述
    Returns:
        工单编号
    """
    ticket_id = f"TK-{hash(title) % 10000:04d}"
    return f"工单已创建:{ticket_id},标题:{title}"


# ==========================================
# 护栏
# ==========================================

def validate_length(response: str) -> str | None:
    if len(response) > 800:
        return f"回复过长({len(response)} 字),请精简。"
    return None


# ==========================================
# Agent 定义
# ==========================================

router = Agent(
    name="Router",
    description="客服路由器,分配用户问题。",
    instructions=(
        "你是客服路由器。\n"
        "# 职责\n"
        "1. 理解用户问题\n"
        "2. 常见问题(密码、收费、导出、API)→ 联系 FAQBot\n"
        "3. 技术问题或 FAQBot 无法解决的 → 联系 TechSupport\n"
        "# 规则\n"
        "- 使用中文\n"
        "- 不要直接回答,分配给合适的团队\n"
    ),
)

faq_bot = Agent(
    name="FAQBot",
    description="常见问题机器人,回答常见问题。",
    instructions=(
        "你是 FAQ 机器人。\n"
        "1. 使用 FAQSearchTool 搜索知识库\n"
        "2. 如果找到答案,直接回复\n"
        "3. 如果返回 NOT_FOUND,使用 SendMessageHandoff 转给 TechSupport\n"
        "使用中文回答。"
    ),
    tools=[FAQSearchTool, SendMessageHandoff(recipient="TechSupport")],
)

tech_support = Agent(
    name="TechSupport",
    description="技术支持,处理复杂技术问题。",
    instructions=(
        "你是技术支持工程师。\n"
        "# 职责\n"
        "处理复杂技术问题。如果需要跟进,使用 create_ticket 创建工单。\n"
        "# 规则\n"
        "- 使用中文\n"
        "- 回复要专业但易懂\n"
    ),
    tools=[create_ticket],
    output_guardrails=[validate_length],
)


# ==========================================
# Agency 创建
# ==========================================

STORAGE_DIR = "./support_threads"

def save_threads(name, data):
    os.makedirs(STORAGE_DIR, exist_ok=True)
    with open(os.path.join(STORAGE_DIR, f"{name}.json"), "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)

def load_threads(name):
    path = os.path.join(STORAGE_DIR, f"{name}.json")
    if os.path.exists(path):
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    return None

agency = Agency(
    router,
    communication_flows=[
        (router, faq_bot),
        (router, tech_support),
        (faq_bot, tech_support),  # FAQBot 可升级到 TechSupport
    ],
    shared_instructions="所有成员使用中文沟通。回复要专业。",
    load_threads_callback=load_threads,
    save_threads_callback=save_threads,
)


# ==========================================
# 运行
# ==========================================

if __name__ == "__main__":
    print("=== 智能客服系统 ===")
    print("输入 'quit' 退出\n")

    while True:
        user_input = input("用户: ")
        if user_input.lower() in ("quit", "exit", "q"):
            break
        result = agency.get_response_sync(user_input)
        print(f"\n客服: {result.final_output}\n")

执行结果:

=== 智能客服系统 ===
输入 'quit' 退出

用户: 我忘记了密码怎么办

客服: 请点击登录页面的'忘记密码'链接,按提示重置密码。

用户: 我在用你们的 API 时遇到了 500 错误

客服: 您遇到了 API 500 错误,我已经为您创建了工单 TK-3847,
      技术团队会尽快排查。同时建议您检查 API Key 是否正确配置。

代码解析

  1. Agent 定义(1.1 节知识点):Router、FAQBot、TechSupport 各自定义了角色和职责。Router 是入口智能体。

  2. 自定义工具(1.2 节知识点)FAQSearchTool 使用 BaseTool 类定义知识库搜索;create_ticket 使用 @function_tool 定义工单创建。

  3. 通信流(1.3 节知识点):Router → FAQBot、Router → TechSupport、FAQBot → TechSupport 三条路径。

  4. Handoff 模式(2.2 节知识点):FAQBot 配置了 SendMessageHandoff(recipient="TechSupport"),当知识库找不到答案时自动升级。

  5. Guardrails(3.2 节知识点):TechSupport 配置了 output_guardrails=[validate_length],确保回复不超过 800 字。

扩展挑战

  1. 接入真实搜索:将 FAQSearchTool 接入 Elasticsearch 或向量数据库。
  2. 添加满意度评价:在每次对话结束后询问用户满意度并记录。
  3. 多语言支持:添加 LanguageDetector Agent,根据用户语言分派到对应客服。

第五部分:常见问题与排查指南

常见错误及解决方案

错误信息 原因 解决方案
openai.AuthenticationError: Invalid API Key API Key 无效或未配置 检查 .envOPENAI_API_KEY,确保无多余空格
openai.RateLimitError: Rate limit reached API 调用频率超限 添加重试逻辑或降低并发数
ImportError: cannot import name 'BaseTool' 版本过旧 pip install --upgrade agency-swarm
TypeError: Agent() got unexpected keyword argument v0.x 到 v1.x API 变更 参考迁移指南
Agent did not use the tool instructions 不够明确 明确说明何时使用哪个工具,优化 docstring
Communication flow not found 通信路径未定义 检查 communication_flows 是否包含对应元组
Model not found: gpt-4o API Key 无权访问 使用 gpt-4o-mini 或确认权限
线程持久化回调报错 回调签名不正确 确保签名正确:load(name)->datasave(name, data)->None

调试技巧

  1. 使用 terminal_demo 交互调试agency.terminal_demo() 提供交互模式,可逐步观察工具调用行为。

  2. 添加决策日志:在 instructions 末尾添加"在联系其他智能体之前先说明原因",可观察决策逻辑。

  3. 分层测试:先单独测试每个 Agent(工具正常),再测试通信,最后组装完整 Agency。


第六部分:学习路线推荐

官方文档推荐阅读顺序

  1. From Scratch - 搭建第一个 Agency
  2. Custom Tools - 工具定义两种方式
  3. Agents - Agent 配置参数
  4. Agencies - 通信流编排
  5. Guardrails - 输入输出验证
  6. Streaming - 流式输出
  7. FastAPI Integration - Web 服务部署
  8. Deployment to Production - 生产部署

推荐进阶资源