AI辅助创作 | 专栏《2026 AI编程效率革命》第08篇

前言

上一篇我们学习了MCP协议，掌握了AI与工具交互的标准化方法。本文将更进一步，探讨如何让多个AI Agent协同工作——就像组建一个AI开发团队，每个Agent负责不同的角色，通过协作完成复杂任务。

---

一、为什么需要多Agent协作？

单Agent在面对复杂任务时有明显局限：

问题	单Agent	多Agent协作
任务复杂度	长任务容易跑偏	分而治之，各司其职
上下文窗口	单个窗口有限	各Agent独立上下文
专业性	通用但不精深	各Agent专精一个领域
并发性	串行执行	可并行处理

典型场景：一个软件开发项目需要产品经理、架构师、开发工程师、测试工程师、代码审查员等角色协作完成。

---

二、主流多Agent框架对比

2.1 框架对比表

框架	特点	适用场景
CrewAI	角色驱动，上手简单	快速搭建团队协作
AutoGen	微软出品，对话式协作	研究、对话式任务
LangGraph	图状态机，灵活可控	复杂工作流编排
Camel	角色扮演，多Agent沟通	创意生成、辩论

2.2 本文选择CrewAI

CrewAI以其直观的角色定义和简洁的API成为入门多Agent的最佳选择。

---

三、CrewAI核心概念

"""
CrewAI的三个核心概念：
1. Agent（智能体）：扮演特定角色，拥有目标、工具和 backstory
2. Task（任务）：具体的工作内容，有明确的预期输出
3. Crew（团队）：由多个Agent组成，协调完成一系列Task
"""

# 安装
# uv add crewai crewai-tools

┌──────────────────────────────────────────────┐
│                   Crew                        │
│                                               │
│  ┌─────────┐  ┌─────────┐  ┌──────────┐      │
│  │ 产品经理 │  │ 开发工程师│  │ 测试工程师│      │
│  │  Agent   │  │  Agent   │  │  Agent   │      │
│  └────┬────┘  └────┬────┘  └────┬─────┘      │
│       │            │            │              │
│  ┌────▼────┐  ┌────▼────┐  ┌───▼─────┐       │
│  │ 需求分析 │  │ 代码实现  │  │ 测试验证  │      │
│  │  Task   │──│  Task   │──│  Task   │       │
│  └─────────┘  └─────────┘  └─────────┘       │
└──────────────────────────────────────────────┘

---

四、实战：搭建AI软件开发团队

4.1 定义Agent角色

# agents.py
from crewai import Agent, Task, Crew, Process
from crewai_tools import (
    FileReadTool,
    DirectoryReadTool,
    CodeInterpreterTool,
)

# 工具初始化
file_read = FileReadTool()
dir_read = DirectoryReadTool()
code_interpreter = CodeInterpreterTool()

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

product_manager = Agent(
    role="产品经理",
    goal="分析用户需求，输出清晰的产品需求文档（PRD）",
    backstory=(
        "你是一位经验丰富的产品经理，擅长将模糊的用户需求"
        "转化为清晰、可执行的技术需求。你注重用户体验，"
        "每个需求都有明确的验收标准。"
    ),
    verbose=True,
    allow_delegation=False,
    llm="gpt-5.5",
)

tech_lead = Agent(
    role="技术架构师",
    goal="根据需求设计技术方案，输出架构设计文档",
    backstory=(
        "你是一位资深技术架构师，精通Python、FastAPI、"
        "PostgreSQL技术栈。你擅长系统设计，注重代码质量、"
        "可扩展性和安全性。"
    ),
    verbose=True,
    allow_delegation=True,
    llm="gpt-5.5",
)

developer = Agent(
    role="开发工程师",
    goal="根据技术方案编写高质量代码",
    backstory=(
        "你是一位全栈开发工程师，精通Python、TypeScript。"
        "你写代码注重规范、可读性和测试覆盖率。"
        "你习惯先写测试，再写实现。"
    ),
    verbose=True,
    tools=[file_read, dir_read],
    allow_delegation=False,
    llm="gpt-5.5",
)

code_reviewer = Agent(
    role="代码审查员",
    goal="审查代码质量，发现潜在问题并给出改进建议",
    backstory=(
        "你是一位严格的代码审查专家，关注代码质量、"
        "安全漏洞、性能问题和最佳实践。"
        "你的审查报告条理清晰，每个问题都有具体的修改建议。"
    ),
    verbose=True,
    tools=[file_read],
    allow_delegation=False,
    llm="claude-4-sonnet",
)

qa_engineer = Agent(
    role="测试工程师",
    goal="编写测试用例，验证功能正确性",
    backstory=(
        "你是一位经验丰富的测试工程师，擅长设计边界测试、"
        "异常测试和集成测试。你确保每个功能都有充分的测试覆盖。"
    ),
    verbose=True,
    tools=[code_interpreter],
    allow_delegation=False,
    llm="gpt-5.5",
)

4.2 定义任务流

# tasks.py
from crewai import Task

def 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 组建团队并执行

# crew.py
from crewai import Crew, Process
from agents import product_manager, tech_lead, developer, code_reviewer, qa_engineer
from tasks import create_tasks

def 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架构

对于更复杂的项目，可以采用层次化架构：

# hierarchical_crew.py
from crewai import Crew, Process

def 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="gpt-5.5",        # 管理者模型
        verbose=True,
    )
    
    return crew.kickoff()

层次化模式下，一个"经理Agent"会自动：

1. 分析任务复杂度
2. 决定任务分配给哪个Agent
3. 协调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）

# 使用LangGraph实现自定义工作流
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class 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 state

def design_node(state: DevState) -> DevState:
    # 架构设计
    state["messages"].append("[架构师] 技术方案设计完成")
    return state

def develop_node(state: DevState) -> DevState:
    # 代码实现
    state["messages"].append("[开发者] 代码实现完成")
    return state

def review_node(state: DevState) -> dict:
    # 代码审查 - 可以决定是否需要返工
    state["messages"].append("[审查员] 代码审查通过")
    return {"next": "test"}  # 或 "rework" 返回开发

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", lambda s: s.get("next", "test"), {
    "test": "test",
    "rework": "develop",
})
graph.add_edge("test", END)

graph.set_entry_point("analyze")
app = graph.compile()

---

七、实践建议

1. Agent角色要清晰：每个Agent的职责边界明确，避免重叠
2. Task描述要具体：包含输入、输出格式、验收标准
3. 选择合适的协作模式：简单任务用Sequential，复杂项目用Hierarchical
4. 控制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中的实战应用。

---

免责声明：本文为AI辅助创作内容，代码示例仅供学习参考。实际项目中多Agent系统的API调用成本可能较高，请根据预算合理规划。文中涉及的工具版本请以官方最新文档为准。

专栏：《2026 AI编程效率革命》| 第08篇
发布日期：2026-05-03