想象一下,你正在管理一个软件开发团队。有的同事负责写代码,有的负责代码审查,有的负责测试,还有人负责撰写文档。在传统的单体应用时代,一个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实现角色分工。这个团队包含三个专业角色:

创建一个名为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,  # 关键