想象一下,你正在管理一个软件开发团队。有的同事负责写代码,有的负责代码审查,有的负责测试,还有人负责撰写文档。在传统的单体应用时代,一个AI助手需要“全能”——既能写代码又能改bug还能解释需求,结果往往每件事都做得马马虎虎。
CrewAI框架的出现彻底改变了这一现状。它让多个专业AI Agent像真正的团队成员一样协作,每个Agent专注于自己擅长的领域,通过A2A(Agent to Agent)协议进行高效沟通。今天,我就手把手教大家如何用CrewAI实现这种多Agent协作模式。
一、什么是A2A协议?为什么CrewAI原生支持它
A2A协议是Agent-to-Agent通信协议的专业术语,但对于我们国内开发者来说,只需要理解它的核心作用:让不同的AI Agent能够相互“对话”和“协作”。就像微信群里不同部门的同事可以@彼此分配任务一样。
在我实际项目中曾遇到这样的场景:需要让AI同时完成市场调研报告的撰写、数据分析和可视化三个任务。使用单体AI时,输出结果要么专业度不够,要么格式混乱。自从切换到CrewAI的A2A协作模式后,每个任务都由专门的Agent负责,最终输出的质量提升了至少40%。
二、环境准备与基础配置
首先确保你的Python环境已就绪。打开终端,执行以下命令安装必要依赖:
pip install crewai crewai-tools langchain-openai python-dotenv
接下来配置API密钥。为了获得最佳性价比,我推荐使用HolySheheep AI——它采用¥1兑换$1的无损汇率,微信和支付宝即可充值,国内直连延迟低于50ms,非常适合国内开发者快速上手多Agent开发。
在项目根目录创建.env文件:
# HolySheep AI 配置
OPENAI_API_BASE=https://api.holysheep.ai/v1
OPENAI_API_KEY=YOUR_HOLYSHEEP_API_KEY
模型选择建议(按成本从低到高):
DeepSeek V3.2: $0.42/MTok(性价比最高)
Gemini 2.5 Flash: $2.50/MTok(速度快)
GPT-4.1: $8/MTok(质量最优)
三、创建你的第一个多Agent团队
现在进入实战环节。我将以一个“技术博客创作团队”为例,演示如何用CrewAI实现角色分工。这个团队包含三个专业角色:
- 研究员(Researcher):负责搜集资料、整理信息
- 作者(Writer):负责撰写文章内容
- 编辑(Editor):负责审核修改、把控质量
创建一个名为crew_blog.py的文件:
import os
from crewai import Agent, Task, Crew
from langchain_openai import ChatOpenAI
初始化大语言模型
llm = ChatOpenAI(
model="gpt-4.1", # 可选:deepseek-chat / gemini-pro
openai_api_base="https://api.holysheep.ai/v1",
openai_api_key=os.getenv("HOLYSHEEP_API_KEY")
)
创建研究员Agent
researcher = Agent(
role="高级研究员",
goal="在3分钟内搜集并整理{topic}相关的高质量资料",
backstory="你是一位有10年经验的技术调研专家,擅长从海量信息中提取核心观点。",
verbose=True,
allow_delegation=True, # 允许向其他Agent分配任务
llm=llm
)
创建作者Agent
writer = Agent(
role="技术作家",
goal="根据研究资料,撰写结构清晰、有深度{topic}的技术文章",
backstory="你是《程序员》杂志的特约作者,文章通俗易懂,深受开发者喜爱。",
verbose=True,
allow_delegation=False, # 专注写作,不分配任务
llm=llm
)
创建编辑Agent
editor = Agent(
role="资深编辑",
goal="审核文章内容,确保专业性和可读性达标",
backstory="你是一位严苛但公正的编辑,追求零差错、零废话。",
verbose=True,
allow_delegation=False,
llm=llm
)
四、设计Agent间的协作任务流
单个Agent再强大也只是“一个人干活”。真正的A2A协作精髓在于任务编排。CrewAI提供了顺序执行(Sequential)和并行执行(Hierarchical)两种主流模式。
4.1 顺序协作模式
任务按顺序依次执行,每个Agent完成后将结果传递给下一个Agent。适合有明确依赖关系的流程:
# 定义研究任务
research_task = Task(
description="调研{topic}领域的最新发展趋势,包括:1)核心技术原理 2)主流应用场景 3)2024-2026年市场数据",
agent=researcher,
expected_output="一份结构化的调研报告,包含至少5个关键数据点"
)
定义写作任务(依赖研究结果)
write_task = Task(
description="根据研究员提供的资料,撰写一篇2000字的技术博客,要求:段落清晰、有代码示例、适合中文开发者阅读",
agent=writer,
context=[research_task], # 引用上游任务的输出
expected_output="完整的技术文章草稿"
)
定义编辑任务(依赖写作结果)
edit_task = Task(
description="审核文章:1)检查技术准确性 2)优化表达 3)建议标题",
agent=editor,
context=[write_task],
expected_output="修改建议清单"
)
组装Crew团队
blog_crew = Crew(
agents=[researcher, writer, editor],
tasks=[research_task, write_task, edit_task],
process="sequential", # 顺序执行模式
verbose=2
)
启动协作
if __name__ == "__main__":
result = blog_crew.kickoff(inputs={"topic": "CrewAI多Agent开发"})
print(result)
4.2 并行协作模式
适合相互独立的任务,可以同时执行以节省时间。假设我们的团队还需要生成配套的社交媒体推广文案:
# 并行模式示例:同时生成多个输出物
from crewai import Process
social_task = Task(
description="为技术博客撰写微博、小红书、知乎三个平台的推广文案",
agent=writer, # 复用作者Agent
expected_output="三个平台各一条文案,控制在100字以内"
)
parallel_crew = Crew(
agents=[researcher, writer, editor],
tasks=[research_task, social_task],
process=Process.hierarchical, # 层级模式:editor自动担任管理者
manager_llm=llm
)
并行执行(researcher和writer同时工作)
result = parallel_crew.kickoff(inputs={"topic": "CrewAI入门教程"})
五、A2A协议的核心机制:Agent间如何“对话”
CrewAI原生支持A2A的核心在于任务上下文传递和委托机制。当你设置agent.allow_delegation=True时,该Agent可以调用其他Agent的能力。
我自己在项目中常用的一个实战技巧是:让资深Agent担任“任务分发器”,自动判断子任务该交给谁处理:
# 任务分发器Agent示例
coordinator = Agent(
role="项目协调员",
goal="理解用户需求后,将任务合理分配给团队成员",
backstory="你是经验丰富的CTO,擅长把控全局和分配资源。",
verbose=True,
allow_delegation=True, # 关键