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

验证安装

# 在 Python 交互环境中验证
python3 -c "
from agency_swarm import Agent, Agency
print('Agency Swarm 安装成功！')
print('版本：', __import__('agency_swarm').__version__)
"

执行结果：

Agency Swarm 安装成功！
版本：1.8.0

第一部分：入门篇

1.1 创建你的第一个 Agent（智能体）

概念讲解：

Agent（智能体）是 Agency Swarm 的基本构建单元。每个 Agent 代表一个具有特定角色和能力的 AI 实体，就像公司里的一个员工。定义一个 Agent 需要三个核心要素：

• name：智能体的名称，用于识别和通信
• description：简短描述智能体的职责，帮助其他智能体了解何时应该与其通信
• instructions：详细的系统指令，定义智能体的行为规范和决策逻辑

此外，Agent 还可以配置工具（tools）、模型（model）和模型参数（model_settings）。

代码示例：

# 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）是扩展智能体能力的关键机制。没有工具的智能体只能进行文本对话；有了工具，智能体可以搜索网页、操作文件、调用 API、执行代码等。

Agency Swarm v1.x 提供两种定义工具的方式： 1. 类方式（BaseTool）：继承 BaseTool，适合复杂工具，支持 Pydantic 字段验证 2. 函数方式（@function_tool）：使用装饰器，适合简单工具，代码更简洁

工具的定义会被自动转换为 OpenAI Function Calling 格式，智能体根据工具的文档字符串（docstring）决定何时调用。

代码示例：

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

# === 方式一：类方式（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 ===
from agency_swarm import Agent, Agency

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 工具，实现摄氏度和华氏度之间的互转。 2. 用 @function_tool 装饰器创建一个 get_current_date 工具，返回当前日期和时间。

1.3 理解 Agency 和 Communication Flows（通信流）

概念讲解：

Agency 是管理多个 Agent 实例并编排它们之间通信的顶层容器。通信流（Communication Flows）是 Agency 的核心特性，它定义了智能体之间的通信权限。

通信流使用方向性元组定义：(agent_a, agent_b) 表示 agent_a 可以主动联系 agent_b，但 agent_b 不能反过来联系 agent_a。这模拟了真实组织中的汇报关系——员工可以向经理汇报，但经理也可以向员工分配任务。

第一个传入 Agency 的智能体是入口智能体（Entry Agent），用户的所有请求都从它开始。

代码示例：

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

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

    Args:
        query: 搜索关键词
    Returns:
        搜索结果摘要
    """
    # 模拟搜索结果（实际项目中应接入真实搜索 API）
    return f"关于 '{query}' 的搜索结果：这是模拟的搜索结果内容..."

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

researcher = Agent(
    name="Researcher",
    description="研究员，负责搜索和整理信息。",
    instructions=(
        "你是研究员。\n"
        "使用 search_web 工具搜索信息，然后整理成简洁的报告返回。\n"
        "使用中文回答。"
    ),
    tools=[search_web],
)

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

# 创建 Agency，定义通信流
agency = Agency(
    ceo,  # 入口智能体
    communication_flows=[
        (ceo, researcher),    # CEO 可以联系 Researcher
        (ceo, developer),     # CEO 可以联系 Developer
        # 注意：Researcher 和 Developer 之间没有通信流，他们不能直接联系
    ],
    shared_instructions="所有成员使用中文沟通，回复要简洁。"
)

# 测试：CEO 会自动将搜索任务分配给 Researcher
result = agency.get_response_sync("请搜索一下什么是多智能体系统")
print(result.final_output)

执行结果：

根据搜索结果，多智能体系统（Multi-Agent System）是由多个自主智能体组成的系统，
这些智能体通过协作、竞争或协商来共同完成复杂任务。每个智能体具有独立的感知、
决策和行动能力，可以在不确定环境中自主运行。

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

第二部分：进阶篇

2.1 Agent 项目结构化

详细讲解：

在实际项目中，将所有代码放在一个文件中很快就会变得难以维护。Agency Swarm 推荐的项目结构为每个 Agent 创建独立的文件夹，包含工具、指令文件和 Agent 定义。

这种结构的好处： - 关注点分离：每个 Agent 的工具和指令独立管理 - 可复用：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
# ├── requirements.txt
# └── .env

ceo/__init.py：

from .ceo import ceo

__all__ = ["ceo"]

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 调研，再让 Developer 执行

# 沟通规则
- 使用中文
- 汇总成员的回复后统一反馈给用户

researcher/tools/search_tool.py：

from agency_swarm.tools import BaseTool
from pydantic import Field

class SearchTool(BaseTool):
    """搜索互联网获取信息。当需要查找资料或获取最新信息时使用。"""

    query: str = Field(
        ...,
        description="搜索关键词，尽量使用具体和准确的词语"
    )
    num_results: int = Field(
        default=5,
        description="返回结果的数量，默认 5 条"
    )

    async def run(self):
        # 实际项目中替换为真实的搜索 API 调用
        mock_results = [
            f"结果 {i+1}：关于 '{self.query}' 的信息..."
            for i in range(self.num_results)
        ]
        return "\n".join(mock_results)

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 包导入。 - 工具文件夹中也需要 __init__.py。

练习题： 1. 按照上述结构创建一个完整的项目，添加 Developer 角色，并实现 CEO → Developer 的通信流。 2. 为 Developer 创建一个 CodeRunnerTool，使其能够执行 Python 代码片段。

2.2 Handoff 模式与控制权转移

详细讲解：

在标准通信流中，Agent A 通过 SendMessage 联系 Agent B，B 完成任务后将结果返回给 A，控制权始终在 A 手中。但在某些场景下，我们需要完全转移控制权——这就是 Handoff 模式。

Handoff 的典型场景： - 升级处理：低级别智能体遇到无法处理的问题时，将控制权交给高级别智能体 - 专业转接：通用助手将用户转给专业领域的智能体

Agency Swarm 通过 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_support = Agent(
    name="Tier1Support",
    description="一线客服，处理常见问题。无法解决时升级到高级支持。",
    instructions=(
        "你是一线客服。\n"
        "1. 首先使用 query_knowledge_base 工具查找答案\n"
        "2. 如果知识库返回 '未找到匹配的答案'，使用 SendMessageHandoff 将用户升级到 Tier2Support\n"
        "3. 使用中文回答"
    ),
    tools=[query_knowledge_base, SendMessageHandoff(recipient="Tier2Support")],
)

# 高级客服：处理复杂问题
tier2_support = Agent(
    name="Tier2Support",
    description="高级客服，处理一线客服无法解决的复杂问题。",
    instructions=(
        "你是高级客服。\n"
        "你负责处理一线客服无法解决的问题。\n"
        "请仔细分析用户的问题，提供专业的解决方案。\n"
        "使用中文回答。"
    ),
)

agency = Agency(
    tier1_support,
    communication_flows=[
        (tier1_support, tier2_support),
    ],
)

# 测试常见问题（一线客服可直接回答）
result1 = agency.get_response_sync("我想退款，怎么操作？")
print("测试1 - 常见问题：", result1.final_output)

# 测试复杂问题（需要升级到高级客服）
result2 = agency.get_response_sync("我收到的商品和描述不符，但已经超过7天了，怎么办？")
print("\n测试2 - 需要升级：", result2.final_output)

执行结果：

测试1 - 常见问题：退款可在订单页面申请，3-5 个工作日到账。

测试2 - 需要升级：您好，关于商品与描述不符且超过7天的情况，我建议您...

注意事项： - SendMessageHandoff 必须在工具列表中显式声明，指定 recipient 参数 - Handoff 后控制权完全转移，原智能体不再参与后续对话 - 确保 communication_flows 中定义了对应的通信路径，否则 Handoff 会失败

练习题： 1. 添加一个 Tier3Support 智能体，实现三级升级链路：Tier1 → Tier2 → Tier3。 2. 修改 Tier2Support，使其在处理完问题后能通过 Handoff 将控制权交回 Tier1Support。

2.3 状态持久化与线程管理

详细讲解：

默认情况下，Agent 的对话历史仅存在于内存中，程序结束即丢失。Agency Swarm 通过回调机制支持线程（Thread）的持久化：

• load_threads_callback：在 Agency 初始化时调用，加载已保存的线程
• save_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}_threads.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}_threads.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

# 创建带持久化的 Agency
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_threads_callback(agent_name) -> data、save_threads_callback(agent_name, data) -> None - 持久化的数据格式由框架内部定义，建议直接序列化和反序列化，不要修改数据结构 - 在生产环境中，建议使用数据库（如 PostgreSQL、Redis）代替文件存储

练习题： 1. 将文件存储替换为 SQLite 数据库存储，实现 save_threads_sqlite 和 load_threads_sqlite。 2. 添加线程过期机制：超过 7 天的线程自动清理。

第三部分：高级篇

3.1 FastAPI 集成与流式输出

Agency Swarm 内置 FastAPI 集成，可以快速将多智能体系统部署为 Web 服务，并支持流式输出（Streaming），让用户在智能体思考过程中就能看到中间结果。

代码示例：

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

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

agency = Agency(ceo)

# 创建 FastAPI 应用
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

# 测试同步接口
curl "http://localhost:8000/chat?message=你好"

# 测试流式接口
curl -N "http://localhost:8000/chat/stream?message=你好"

执行结果：

# 同步接口返回
{"response":"你好！我是 CEO，有什么可以帮助你的吗？"}

# 流式接口返回（逐 token）
data: 你
data: 好
data: ！
data: 我
data: 是
...
data: [DONE]

3.2 Guardrails（护栏）机制

Guardrails 用于验证智能体的输入和输出，防止不期望的内容通过。可以用于内容安全、格式校验和业务规则检查。

代码示例：

# guardrails.py
from agency_swarm import Agent, Agency

# 定义输入和输出验证函数
def validate_input(message: str) -> str | None:
    """验证用户输入。返回 None 表示通过，返回字符串表示拒绝原因。"""
    forbidden_words = ["暴力", "违法"]
    for word in forbidden_words:
        if word in message:
            return f"输入包含不允许的内容：{word}"
    return None  # 通过验证

def validate_output(response: str) -> str | None:
    """验证智能体输出。返回 None 表示通过，返回字符串表示拒绝原因。"""
    max_length = 500
    if len(response) > max_length:
        return f"输出超过最大长度限制（{max_length} 字符）"
    return None  # 通过验证

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

agency = Agency(assistant)

# 正常请求
result1 = agency.get_response_sync("什么是 Python？")
print("正常请求：", result1.final_output)

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

执行结果：

正常请求： Python 是一种广泛使用的高级编程语言，以简洁清晰的语法著称...
触发护栏： 抱歉，您的请求包含不允许的内容，我无法处理。

3.3 性能优化

• 优化策略一：使用异步 API：在 Web 服务或高并发场景中，使用 await agency.get_response() 替代 agency.get_response_sync()，避免阻塞事件循环。

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", "问题3"]))

• 优化策略二：控制模型参数：通过 ModelSettings 精细控制 Token 使用和响应质量。

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. 指令要具体：每个 Agent 的 instructions 应包含明确的角色定义、决策规则和输出格式要求。模糊的指令会导致智能体行为不可预测。
2. 通信流要最小化：只定义必要的通信路径，避免全连接的通信图。过多的通信路径会增加智能体的决策复杂度和 Token 消耗。
3. 工具文档字符串要清晰：智能体根据工具的 docstring 决定是否使用该工具。docstring 应说明工具的功能、适用场景和参数含义。
4. 设置合理的 max_tokens：避免因为输出过长导致 Token 浪费或超出上下文窗口。
5. 使用 shared_instructions：通过 Agency 的 shared_instructions 参数为所有智能体提供通用的行为准则。

第四部分：实战项目

项目需求

构建一个智能技术调研团队，由三个智能体组成：

1. TechLead（技术负责人）：理解用户的调研需求，分配任务给研究员和分析师
2. Researcher（研究员）：搜索指定技术的信息，整理成结构化摘要
3. Analyst（分析师）：基于研究员的搜索结果，生成技术评估报告

该团队需要支持持久化（记住之前的调研历史）和输出护栏（限制报告长度）。

项目设计

架构：

用户 → TechLead → Researcher（搜索信息）
                 → Analyst（分析评估）

通信流： - TechLead → Researcher：分配搜索任务 - TechLead → Analyst：分配分析任务 - Researcher 和 Analyst 之间不直接通信（通过 TechLead 中转）

运用的知识点： - Agent 定义与角色配置（1.1 节） - 自定义工具（1.2 节） - 通信流（1.3 节） - Handoff 模式（2.2 节） - Guardrails 护栏（3.2 节）

完整实现代码

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


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

class TechSearchTool(BaseTool):
    """搜索指定技术的相关信息。当需要了解某项技术的基本信息时使用此工具。"""

    tech_name: str = Field(
        ...,
        description="要搜索的技术名称"
    )
    aspect: str = Field(
        default="overview",
        description="要了解的方面：overview(概览)/features(特性)/comparison(对比)"
    )

    async def run(self):
        # 模拟搜索结果（实际项目接入真实搜索 API）
        database = {
            "React": {
                "overview": "React 是 Meta 开发的 JavaScript UI 库，用于构建用户界面。",
                "features": "组件化、虚拟 DOM、Hooks、JSX 语法、丰富的生态系统。",
                "comparison": "vs Vue：React 更灵活，Vue 更易上手。vs Angular：React 更轻量，Angular 更全面。",
            },
            "FastAPI": {
                "overview": "FastAPI 是一个高性能的 Python Web 框架，基于 Starlette 和 Pydantic。",
                "features": "自动生成 OpenAPI 文档、类型提示、异步支持、依赖注入。",
                "comparison": "vs Flask：FastAPI 原生异步，性能更高。vs Django：FastAPI 更轻量，适合 API 服务。",
            },
            "Agency Swarm": {
                "overview": "Agency Swarm 是基于 OpenAI Agents SDK 的多智能体编排框架。",
                "features": "组织结构化通信流、类型安全工具、状态持久化、FastAPI 集成。",
                "comparison": "vs CrewAI：更专注 OpenAI 生态。vs AutoGen：更面向生产部署。",
            },
        }

        tech = database.get(self.tech_name)
        if tech:
            return tech.get(self.aspect, f"未找到 '{self.aspect}' 方面的信息")
        return f"未找到关于 '{self.tech_name}' 的信息"


@function_tool
def generate_report(tech_name: str, search_results: str) -> str:
    """基于搜索结果生成技术评估报告摘要。

    Args:
        tech_name: 技术名称
        search_results: 搜索到的信息内容
    Returns:
        评估报告摘要
    """
    # 这个工具主要用于标记智能体已进入报告生成阶段
    # 实际报告内容由智能体的 LLM 生成
    return f"已收到关于 '{tech_name}' 的信息，准备生成评估报告。"


# ==========================================
# 护栏定义
# ==========================================

def validate_report_length(response: str) -> str | None:
    """验证报告输出不超过 1000 字"""
    if len(response) > 1000:
        return f"报告超过 1000 字限制（当前 {len(response)} 字），请精简内容。"
    return None


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

tech_lead = Agent(
    name="TechLead",
    description="技术负责人，理解调研需求并协调团队成员。",
    instructions=(
        "你是技术团队的负责人。\n"
        "# 职责\n"
        "1. 理解用户的技术调研需求\n"
        "2. 明确要调研的技术名称和关注方面\n"
        "3. 让 Researcher 搜索信息\n"
        "4. 将搜索结果交给 Analyst 进行分析\n"
        "5. 汇总最终报告反馈给用户\n"
        "# 规则\n"
        "- 使用中文沟通\n"
        "- 如果用户没有明确技术名称，先询问\n"
    ),
)

researcher = Agent(
    name="Researcher",
    description="研究员，搜索和整理技术信息。",
    instructions=(
        "你是技术研究员。\n"
        "# 职责\n"
        "1. 使用 TechSearchTool 搜索技术信息\n"
        "2. 搜索 overview、features、comparison 三个方面的信息\n"
        "3. 将搜索结果整理成结构化摘要返回\n"
        "# 输出格式\n"
        "## {技术名称} 调研摘要\n"
        "### 概览\n{内容}\n"
        "### 核心特性\n{内容}\n"
        "### 竞品对比\n{内容}\n"
    ),
    tools=[TechSearchTool],
)

analyst = Agent(
    name="Analyst",
    description="分析师，基于搜索结果生成技术评估报告。",
    instructions=(
        "你是技术分析师。\n"
        "# 职责\n"
        "基于研究员提供的搜索结果，生成简洁的技术评估报告。\n"
        "# 报告格式\n"
        "## 技术评估报告：{技术名称}\n"
        "### 推荐指数\n"
        "⭐⭐⭐⭐⭐（1-5 星）\n"
        "### 优势\n"
        "- {优势1}\n"
        "- {优势2}\n"
        "### 劣势\n"
        "- {劣势1}\n"
        "- {劣势2}\n"
        "### 适用场景\n"
        "- {场景1}\n"
        "### 总结建议\n"
        "{一段话总结}\n"
        "# 规则\n"
        "- 报告不超过 800 字\n"
        "- 使用中文\n"
        "- 评估要客观，优劣势都要提及\n"
    ),
    tools=[generate_report],
    output_guardrails=[validate_report_length],
)


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

STORAGE_DIR = "./research_threads"

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)

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:
            return json.load(f)
    return None

agency = Agency(
    tech_lead,
    communication_flows=[
        (tech_lead, researcher),
        (tech_lead, analyst),
    ],
    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团队回复:\n{result.final_output}\n")

执行结果：

=== 智能技术调研团队 ===
输入 'quit' 退出

你: 帮我调研一下 Agency Swarm

团队回复:
## 技术评估报告：Agency Swarm

### 推荐指数
⭐⭐⭐⭐

### 优势
- 基于 OpenAI Agents SDK，与 OpenAI 生态深度集成
- 组织结构化通信流，直观易懂
- 内置 FastAPI 集成和 Web UI，生产就绪度高
- 类型安全的工具系统，减少运行时错误

### 劣势
- 社区规模较小，第三方资源有限
- 深度绑定 OpenAI 生态，迁移成本较高
- 缺乏内置的长期记忆系统

### 适用场景
- OpenAI 生态项目中的多智能体协作
- 中小规模的智能体团队（3-8 个）
- 需要快速部署的 AI 自动化工作流

### 总结建议
Agency Swarm 是一个定位精准的多智能体编排框架，适合已采用 OpenAI API 的团队。对于 OpenAI 生态深度用户，推荐使用；如果需要多模型或大规模部署，建议评估其他方案。

代码解析

1. Agent 定义（1.1 节知识点）：三个 Agent（TechLead、Researcher、Analyst）各自定义了 name、description 和详细的 instructions。TechLead 是入口智能体。

2. 自定义工具（1.2 节知识点）：TechSearchTool 使用 BaseTool 类方式定义，支持按技术名称和关注方面搜索。generate_report 使用 @function_tool 装饰器方式定义。

3. 通信流（1.3 节知识点）：定义了两条通信路径——TechLead → Researcher 和 TechLead → Analyst。Researcher 和 Analyst 不能直接通信，必须通过 TechLead 中转。

4. Guardrails 护栏（3.2 节知识点）：Analyst 配置了 output_guardrails=[validate_report_length]，确保报告不超过 1000 字。

5. 状态持久化（2.3 节知识点）：通过 load_threads_callback 和 save_threads_callback 实现基于文件的线程持久化。

扩展挑战

1. 添加真实搜索：将 TechSearchTool 接入真实的搜索 API（如 Tavily、SerpAPI），替换模拟数据。
2. 添加 PDF 导出：创建一个 PDFExportTool，让 Analyst 将报告导出为 PDF 文件。
3. 支持多轮调研：实现对话式调研，用户可以针对初步报告提出追问（如"详细对比一下 React 和 Vue"）。

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

常见错误及解决方案

调试技巧

1. 使用 terminal_demo 交互调试：通过 agency.terminal_demo() 进入交互模式，可以逐步观察智能体的决策过程和工具调用行为，比直接调用 get_response_sync 更直观。

2. 检查通信流的实际行为：在 Agent 的 instructions 末尾添加 "在联系其他智能体之前，先说明你为什么要联系他"，这样可以在输出中看到智能体的决策逻辑。

3. 逐步构建，分层测试：先单独测试每个 Agent（确保工具正常），再测试两个 Agent 之间的通信，最后组装完整的 Agency。不要一开始就搭建完整系统。

第六部分：学习路线推荐

官方文档推荐阅读顺序

1. From Scratch - 从零搭建第一个 Agency，掌握基本概念
2. Custom Tools - Step-by-Step - 学习工具定义的两种方式
3. Agents - 深入理解 Agent 的配置参数和生命周期
4. Agencies - 掌握 Agency 的通信流编排和高级配置
5. Communication Flows - 理解 Handoff 和 Orchestrator-Worker 模式
6. Streaming - 实现流式输出，提升用户体验
7. FastAPI Integration - 将 Agency 部署为 Web 服务
8. Guardrails - 实现输入输出验证和安全控制
9. Deployment to Production - 生产环境部署指南
10. API Reference - 完整的 API 参考文档

推荐进阶资源

• Agency Swarm YouTube 频道 - 官方视频教程，包含框架概览和高级特性演示
• Agency Swarm vs AutoGen vs CrewAI vs LangGraph（Medium） - 框架作者撰写的竞品对比文章，有助于理解设计取舍
• OpenAI Agents SDK 官方文档 - Agency Swarm 的底层依赖，理解 Agents SDK 有助于深入掌握 Agency Swarm 的内部机制