Agency Swarm 入门：实用指南

我之前被内容流水线的问题搞得焦头烂额。我的团队每周需要处理几百个产品描述——调研竞品、写文案、做 SEO 检查，还要针对不同平台排版。我试过手动把各种提示词串起来，但上下文经常乱套，错误处理简直是噩梦，而且总是莫名其妙地撞上速率限制。我需要的是一种能把这事儿当成真正的工作流来建模的方案，要有明确的专业分工，而不是简单的一连串 API 调用。

这就是我找到 Agency Swarm 的原因。把 Agent（智能体）像真实的组织架构那样去构建——让“CEO”Agent 把任务分发给专家，专家之间再相互交接——这个思路让我瞬间豁然开朗。以下是我用它搭建第一个 Agency 时踩过的坑和学到的经验。

Agency Swarm 到底是什么？

Agency Swarm 是一个基于 OpenAI Agents SDK 构建的多智能体编排框架。你不需要自己去写底层的 API 调用和手动管理对话状态，只需要定义好具有特定角色、工具和沟通路径的 Agent，剩下的编排工作交给框架就行。你可以把它理解为给 Agent 们发了一张组织架构图，让它们自己琢磨怎么协作。

框架帮你搞定了那些从零开始写很繁琐的东西：智能体间的通信、基于 Pydantic 的工具验证、对话线程管理，还有当一个 Agent 的输出跟下一个 Agent 的预期对不上时的优雅错误处理。

搭建你的第一个 Agency

首先，安装这个包：

pip install agency-swarm

你需要把 OpenAI API 密钥设置为环境变量：

export OPENAI_API_KEY="sk-your-key-here"

我刚开始犯了个错，试图用一个没有正确权限的项目密钥。一定要确保你的密钥能访问你想用的模型——Agency Swarm 默认使用 GPT-4o，如果你的密钥没权限，报的权限错误会让人摸不着头脑。

定义你的 Agent

核心构建模块是 Agent 类。每个 Agent 都有自己的角色、指令，还可以选配工具。下面是我搭建一个简单内容 Agency 的方式：

from agency_swarm import Agent

ceo = Agent(
    name="CEO",
    instructions="""You are the CEO of a content agency. Your job is to:
    1. Receive requests from the user
    2. Break them down into clear tasks
    3. Delegate to the appropriate specialist
    4. Review and synthesize the final output
    
    Always confirm your plan with the user before delegating.""",
    model="gpt-4o"
)

researcher = Agent(
    name="Researcher",
    instructions="""You are a research specialist. When given a topic:
    1. Identify key competitors and their content strategies
    2. Find relevant statistics and data points
    3. Note content gaps in the market
    4. Return a structured research brief
    
    Be thorough but concise. Focus on actionable insights.""",
    model="gpt-4o"
)

copywriter = Agent(
    name="Copywriter",
    instructions="""You are a skilled copywriter. Using the research brief provided:
    1. Write compelling product descriptions
    2. Adapt tone based on the target platform
    3. Include relevant keywords naturally
    4. Provide 2-3 variations when possible
    
    Always ask for the research brief before writing.""",
    model="gpt-4o"
)

有一点让我挺意外的：指令的细节极其重要。我第一次尝试时，给 CEO 的指令很模糊，就写了句“管理团队”，结果它老是想自己把活儿全干了，根本不懂得分配。把分配任务的时机和方式写得明明白白，效果立竿见影。

梳理通信路径

这就是 Agency Swarm 跟简单的 Agent 设置不一样的地方。在创建 Agency 对象时，你要定义通信流——也就是谁可以跟谁说话：

from agency_swarm import Agency

agency = Agency(
    name="ContentAgency",
    agents=[ceo, researcher, copywriter],
    communication_flows=[
        (ceo, researcher),
        (ceo, copywriter),
        (researcher, copywriter),  # Researcher can hand off directly to copywriter
    ],
    shared_instructions="Always be professional. Return responses in markdown format."
)

communication_flows 定义了组织架构图。CEO 可以跟研究员和文案沟通；研究员可以把工作直接交接给文案；但文案不能越过 CEO 去给研究员派活——因为通信流里没这一条。

一开始我把所有通信都设成了双向的，觉得沟通路径越多越好。大错特错！Agent 们全懵了，请求在它们之间来回打乒乓球。限制通信路径反而让系统更稳定了。

添加自定义工具

当 Agent 能真正干活时，它们才更有用。Agency Swarm 用 Pydantic 来做类型安全的工具定义。下面是我给研究员写的用来搜索网页的工具：

from pydantic import Field
from agency_swarm import Tool

class WebSearchTool(Tool):
    """Search the web for information on a given topic."""
    
    query: str = Field(
        ..., 
        description="The search query to look up"
    )
    num_results: int = Field(
        default=5,
        description="Number of results to return"
    )
    
    def run(self):
        # Your actual search implementation here
        # I used a simple SerpAPI integration
        from serpapi import GoogleSearch
        search = GoogleSearch({
            "q": self.query,
            "num": self.num_results,
            "api_key": "your-serp-api-key"
        })
        results = search.get_dict().get("organic_results", [])
        return "\n".join([
            f"- {r['title']}: {r.get('snippet', '')}" 
            for r in results[:self.num_results]
        ])

然后把它挂载到 Agent 上：

researcher = Agent(
    name="Researcher",
    instructions="...",
    model="gpt-4o",
    tools=[WebSearchTool]
)

Pydantic 的验证可没少让我吃苦头。我一开始把 num_results 只定义成了 int，没给默认值，结果 Agent 经常不传这个参数导致运行失败。加上默认值就搞定了。框架会在执行前验证工具的输入，这能帮你避免运行时崩溃——但你设计数据结构时得多留个心眼。

运行 Agency

一切准备就绪，运行任务就很简单了：

response = agency.get_response(
    message="Write a product description for our new wireless earbuds targeting fitness enthusiasts"
)

print(response)

背后的流程是这样的：CEO 收到消息，决定把调研工作分配给研究员；研究员用网页搜索工具查资料，把结果交给文案；文案产出最终草稿；最后 CEO 综合评估并返回响应。

你也可以用流式传输来实时查看进展：

response = agency.get_response(
    message="Write a product description for our new wireless earbuds",
    yield_messages=True
)

for msg in response:
    print(f"[{msg.agent_name}]: {msg.content}")

这对调试来说简直是无价之宝。我能清楚地看到链条在哪里断了，或者哪个 Agent 跑偏了。

一线调试经验

小心死循环。 刚开始，我的研究员老是去找文案要澄清，文案又去找研究员要更多数据，俩人就这么来回踢皮球，直到撞上 token 上限。加上像“不要要求澄清，根据现有信息继续工作”这样明确的指令，才打破了这种死循环。

记全日志。 Agency Swarm 有内置的日志功能，你可以这样开启：

import logging
logging.basicConfig(level=logging.DEBUG)

这会展示每一条智能体间的消息、工具调用和决策。虽然刷屏很烦，但出问题时缺了它绝对不行。

从简开始。 我的第一个 Agency 搞了 6 个 Agent，简直是群魔乱舞。后来我砍到只剩 2 个（CEO + 1 个专家），等跑稳了再一个一个往上加。每加一个都会出点新状况，但至少在可控范围内能修好。

实话实说的局限性

Agency Swarm 不是万能药。以下是我遇到的问题：

• Token 成本飙升得很快。 每一次智能体间的消息传递都要消耗 token。一个本来 500 token 就能搞定的简单任务，只要 Agent 之间互相讨论几句，很容易就膨胀到 5000+ token。千万盯紧你的用量。

• 延迟会叠加。 串行的 Agent 交接意味着串行的 API 调用。我的内容流水线处理一个描述要花 30 到 60 秒。不适合对实时性要求高的应用。

• 错误恢复还得靠手动。 当 Agent 输出垃圾内容时，框架不会自动重试或重新路由。你需要在指令里处理这种情况，或者加一些防护工具。

• 框架还在演进。 它最近刚迁移到基于 OpenAI Agents SDK 的架构，一些老教程已经过时了。如果你是从 v0.x 过来的，记得看看迁移指南。

什么时候值得用？

当你有真正的多步骤工作流，且角色分工明确时，Agency Swarm 就大放异彩了。如果你只是在做简单的提示词链——步骤 A 喂给步骤 B，步骤 B 喂给步骤 C——那用个更简单的框架甚至直接写原生 API 调用可能效率更高。但当你需要条件路由（CEO 决定用哪个专家）、并行工作（研究员和 SEO 分析师同时干活），或者带验证的复杂交接时，这种组织架构的隐喻确实能让事情变得更清爽。

我的内容流水线现在靠 4 个 Agent 跑得稳稳当当，每天处理 50 多个描述，而且模块化的结构意味着我可以随时换一个不同风格的文案，而完全不用动系统里的其他部分。光这一点，就值得我把学习曲线爬一遍了。