前言上一篇我们学习了MCP协议，掌握了AI与工具交互的标准化方法。本文将更进一步，探讨如何让多个AI Agent协同工作——就像组建一个AI开发团队，每个Agent负责不同的角色，通过协作完成复杂任务。—## 一、为什么需要多Agent协作？单Agent在面对复杂任务时有明显局限：| 问题 | 单Agent | 多Agent协作 ||------|---------|------------|| 任务复杂度 | 长任务容易跑偏 | 分而治之，各司其职 || 上下文窗口 | 单个窗口有限 | 各Agent独立上下文 || 专业性 | 通用但不精深 | 各Agent专精一个领域 || 并发性 | 串行执行 | 可并行处理 |典型场景：一个软件开发项目需要产品经理、架构师、开发工程师、测试工程师、代码审查员等角色协作完成。—## 二、主流多Agent框架对比### 2.1 框架对比表| 框架 | 特点 | 适用场景 ||------|------|----------|| CrewAI | 角色驱动，上手简单 | 快速搭建团队协作 || AutoGen | 微软出品，对话式协作 | 研究、对话式任务 || LangGraph | 图状态机，灵活可控 | 复杂工作流编排 || Camel | 角色扮演，多Agent沟通 | 创意生成、辩论 |### 2.2 本文选择CrewAICrewAI以其直观的角色定义和简洁的API成为入门多Agent的最佳选择。—## 三、CrewAI核心概念python"""CrewAI的三个核心概念：1. Agent（智能体）：扮演特定角色，拥有目标、工具和 backstory2. Task（任务）：具体的工作内容，有明确的预期输出3. Crew（团队）：由多个Agent组成，协调完成一系列Task"""# 安装# uv add crewai crewai-tools``````┌──────────────────────────────────────────────┐│ Crew ││ ││ ┌─────────┐ ┌─────────┐ ┌──────────┐ ││ │ 产品经理 │ │ 开发工程师│ │ 测试工程师│ ││ │ Agent │ │ Agent │ │ Agent │ ││ └────┬────┘ └────┬────┘ └────┬─────┘ ││ │ │ │ ││ ┌────▼────┐ ┌────▼────┐ ┌───▼─────┐ ││ │ 需求分析 │ │ 代码实现 │ │ 测试验证 │ ││ │ Task │──│ Task │──│ Task │ ││ └─────────┘ └─────────┘ └─────────┘ │└──────────────────────────────────────────────┘—## 四、实战：搭建AI软件开发团队### 4.1 定义Agent角色python# agents.pyfrom crewai import Agent, Task, Crew, Processfrom crewai_tools import ( FileReadTool, DirectoryReadTool, CodeInterpreterTool,)from langchain_openai import ChatOpenAI # CrewAI需要LLM对象，不能传字符串# 工具初始化file_read = FileReadTool()dir_read = DirectoryReadTool()code_interpreter = CodeInterpreterTool()# ==================== Agent定义 ====================product_manager = Agent( role="产品经理", goal="分析用户需求，输出清晰的产品需求文档（PRD）", backstory=( "你是一位经验丰富的产品经理，擅长将模糊的用户需求" "转化为清晰、可执行的技术需求。你注重用户体验，" "每个需求都有明确的验收标准。" ), verbose=True, allow_delegation=False, llm=ChatOpenAI(model="gpt-5.5"),)tech_lead = Agent( role="技术架构师", goal="根据需求设计技术方案，输出架构设计文档", backstory=( "你是一位资深技术架构师，精通Python、FastAPI、" "PostgreSQL技术栈。你擅长系统设计，注重代码质量、" "可扩展性和安全性。" ), verbose=True, allow_delegation=True, llm=ChatOpenAI(model="gpt-5.5"),)developer = Agent( role="开发工程师", goal="根据技术方案编写高质量代码", backstory=( "你是一位全栈开发工程师，精通Python、TypeScript。" "你写代码注重规范、可读性和测试覆盖率。" "你习惯先写测试，再写实现。" ), verbose=True, tools=[file_read, dir_read], allow_delegation=False, llm=ChatOpenAI(model="gpt-5.5"),)code_reviewer = Agent( role="代码审查员", goal="审查代码质量，发现潜在问题并给出改进建议", backstory=( "你是一位严格的代码审查专家，关注代码质量、" "安全漏洞、性能问题和最佳实践。" "你的审查报告条理清晰，每个问题都有具体的修改建议。" ), verbose=True, tools=[file_read], allow_delegation=False, llm=ChatOpenAI(model="gpt-5.5"),)qa_engineer = Agent( role="测试工程师", goal="编写测试用例，验证功能正确性", backstory=( "你是一位经验丰富的测试工程师，擅长设计边界测试、" "异常测试和集成测试。你确保每个功能都有充分的测试覆盖。" ), verbose=True, tools=[code_interpreter], allow_delegation=False, llm=ChatOpenAI(model="gpt-5.5"),)### 4.2 定义任务流python# tasks.pyfrom crewai import Taskfrom agents import product_manager, tech_lead, developer, code_reviewer, qa_engineerdef create_tasks(project_description: str) -> list[Task]: """根据项目描述创建任务链""" task_analyze = Task( description=( f"分析以下项目需求，输出PRD文档：\n\n" f"{project_description}\n\n" f"PRD应包含：\n" f"1. 项目背景与目标\n" f"2. 功能需求列表（含优先级）\n" f"3. 非功能需求\n" f"4. 验收标准" ), expected_output="Markdown格式的PRD文档", agent=product_manager, ) task_design = Task( description=( "根据产品经理的PRD文档，设计技术方案：\n" "1. 系统架构图（用文字描述）\n" "2. 数据模型设计\n" "3. API接口设计\n" "4. 技术选型说明\n" "5. 目录结构规划" ), expected_output="Markdown格式的技术设计文档", agent=tech_lead, context=[task_analyze], # 依赖前置任务 ) task_develop = Task( description=( "根据技术方案，实现项目代码：\n" "1. 创建项目目录结构\n" "2. 实现数据模型层\n" "3. 实现API接口层\n" "4. 实现业务逻辑层\n" "5. 添加配置文件和启动入口\n" "确保代码有完整的类型注解和文档字符串。" ), expected_output="完整的项目代码文件列表及内容", agent=developer, context=[task_design], ) task_review = Task( description=( "审查开发工程师提交的代码：\n" "1. 检查代码规范（PEP 8）\n" "2. 检查安全问题（SQL注入、XSS等）\n" "3. 检查性能问题\n" "4. 检查错误处理\n" "5. 给出评分（A/B/C/D）和具体改进建议" ), expected_output="代码审查报告（含问题列表和改进建议）", agent=code_reviewer, context=[task_develop], ) task_test = Task( description=( "为项目编写测试：\n" "1. 单元测试（覆盖核心逻辑）\n" "2. API接口测试\n" "3. 边界和异常测试\n" "4. 运行测试并报告结果" ), expected_output="测试代码文件和测试执行报告", agent=qa_engineer, context=[task_develop, task_review], ) return [task_analyze, task_design, task_develop, task_review, task_test]### 4.3 组建团队并执行python# crew.pyfrom crewai import Crew, Processfrom agents import product_manager, tech_lead, developer, code_reviewer, qa_engineerfrom tasks import create_tasksdef run_dev_crew(project_description: str): """运行开发团队""" # 创建任务 tasks = create_tasks(project_description) # 组建团队 crew = Crew( agents=[product_manager, tech_lead, developer, code_reviewer, qa_engineer], tasks=tasks, process=Process.sequential, # 顺序执行 verbose=True, ) # 启动执行 result = crew.kickoff() return result# 使用示例if __name__ == "__main__": project_desc = """ 开发一个个人记账API服务： - 用户注册/登录（JWT认证） - 记录收支（金额、类别、日期、备注） - 按月/年统计报表 - 预算管理和超支提醒 技术栈：FastAPI + SQLAlchemy + SQLite """ result = run_dev_crew(project_desc) print("=" * 60) print("项目交付物：") print(result)—## 五、进阶：层次化多Agent架构对于更复杂的项目，可以采用层次化架构：python# hierarchical_crew.pyfrom crewai import Crew, Processfrom langchain_openai import ChatOpenAIfrom agents import developer, code_reviewer, qa_engineerfrom tasks import create_tasksdef run_hierarchical_crew(project_description: str): """层次化管理模式""" crew = Crew( agents=[developer, code_reviewer, qa_engineer], tasks=create_tasks(project_description), process=Process.hierarchical, # 层次化：manager自动分配 manager_llm=ChatOpenAI(model="gpt-5.5"), # 管理者模型 verbose=True, ) return crew.kickoff()层次化模式下，一个"经理Agent"会自动：1. 分析任务复杂度2. 决定任务分配给哪个Agent3. 协调Agent之间的依赖关系4. 汇总最终结果—## 六、Agent间通信模式### 6.1 顺序模式（Sequential）Agent_A → Agent_B → Agent_C → 完成适用于流水线式任务，前一个Agent的输出是后一个的输入。### 6.2 层次模式（Hierarchical） Manager Agent / | \ Agent_A Agent_B Agent_C \ | / 汇总结果输出Manager负责分配和汇总。### 6.3 自定义图模式（LangGraph）python# 使用LangGraph实现自定义工作流from langgraph.graph import StateGraph, ENDfrom typing import TypedDict, Annotatedimport operatorclass DevState(TypedDict): requirements: str design: str code: str review: str test_results: str messages: Annotated[list[str], operator.add]def analyze_node(state: DevState) -> DevState: # 需求分析 state["messages"].append("[产品经理] 需求分析完成") return statedef design_node(state: DevState) -> DevState: # 架构设计 state["messages"].append("[架构师] 技术方案设计完成") return statedef develop_node(state: DevState) -> DevState: # 代码实现 state["messages"].append("[开发者] 代码实现完成") return statedef review_node(state: DevState) -> DevState: # 代码审查 - 可以决定是否需要返工 # 实际项目中可根据代码质量动态决定路由方向 state["messages"].append("[审查员] 代码审查通过") state["review"] = "passed" # 或 "needs_rework" return statedef route_after_review(state: DevState) -> str: """根据审查结果决定下一步""" return "test" if state.get("review") != "needs_rework" else "develop"def test_node(state: DevState) -> DevState: state["messages"].append("[测试] 测试通过") return state# 构建图graph = StateGraph(DevState)graph.add_node("analyze", analyze_node)graph.add_node("design", design_node)graph.add_node("develop", develop_node)graph.add_node("review", review_node)graph.add_node("test", test_node)graph.add_edge("analyze", "design")graph.add_edge("design", "develop")graph.add_edge("develop", "review")graph.add_conditional_edges("review", route_after_review, { "test": "test", "develop": "develop", # 返工：回到开发节点})graph.add_edge("test", END)graph.set_entry_point("analyze")app = graph.compile()—## 七、实践建议1. Agent角色要清晰：每个Agent的职责边界明确，避免重叠2. Task描述要具体：包含输入、输出格式、验收标准3. 选择合适的协作模式：简单任务用Sequential，复杂项目用Hierarchical4. 控制Token消耗：多Agent会显著增加API调用，注意成本控制5. 加入人类审核节点：关键决策点加入人工确认（Human-in-the-loop）—## 总结多Agent协作将AI编程从"单兵作战"提升到"团队协作"的层面。本文介绍了：1. 多Agent协作的必要性和优势2. CrewAI框架的核心概念3. 完整的5角色AI开发团队搭建4. 层次化管理和自定义工作流5. 实践建议掌握MCP协议 + 多Agent协作，你已经具备了构建复杂AI应用的基础能力。后续文章将继续探讨AI在代码审查、Prompt工程和DevOps中的实战应用。> 💬 实战交流：你试过多Agent协作吗？遇到什么坑？评论区聊聊，我还准备了一份CrewAI避坑清单，留言"避坑"我私发你。觉得有帮助的话，点赞收藏支持一下！—> 免责声明：本文为AI辅助创作内容，代码示例仅供学习参考。实际项目中多Agent系统的API调用成本可能较高，请根据预算合理规划。文中涉及的工具版本请以官方最新文档为准。专栏：《2026 AI编程效率革命》| 第08篇**发布日期：2026-05-20