多智能体协作（Multi-Agent）：从团队协作到自主组织的技术实现

pip install crewai crewai-tools langchain-openai

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool

# 工具
search = SerperDevTool()

# 定义 Agent
researcher = Agent(
    role="产品研究员",
    goal="深入研究目标用户、竞品和市场机会",
    backstory="你是一位有 10 年经验的 SaaS 产品研究员，擅长提炼用户痛点。",
    tools=[search],
    verbose=True,
)

writer = Agent(
    role="产品文档撰写者",
    goal="基于研究成果撰写高质量的产品需求文档",
    backstory="你是一位资深的 Agile 产品负责人，擅长将模糊的想法转化为可执行的 PRD。",
    verbose=True,
)

reviewer = Agent(
    role="产品文档审查者",
    goal="审查 PRD 的完整性和可执行性",
    backstory="你是一位严格的技术 VP，确保每个 PRD 都有清晰的验收标准。",
    verbose=True,
)

# 定义任务
task1 = Task(
    description="研究 '个人开发者 AI 工具 SaaS' 市场。包括：1) 目标用户画像；2) TOP5 竞品及其定价；3) 核心痛点未满足”。输出结构化的调研报告。",
    expected_output="500 字以上的结构化调研报告，包含用户画像、竞品分析和 3 个核心痛点。",
    agent=researcher,
)

task2 = Task(
    description="基于调研报告，撰写产品需求文档（PRD），包括：产品概述、核心功能列表、用户故事、验收标准。",
    expected_output="完整的 Markdown PRD 文档，不少于 800 字。",
    agent=writer,
)

task3 = Task(
    description="审查 PRD：检查是否包含用户故事、是否有明确验收标准、技术可行性如何。给出修改建议。",
    expected_output="审查报告，列出 5-10 条详细的修改建议。",
    agent=reviewer,
)

# 组装 Crew
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[task1, task2, task3],
    process=Process.sequential,  # 顺序执行（也可以 Process.parallel）
    verbose=True,
)

# 运行
result = crew.kickoff(inputs={"topic": "个人开发者 AI 工具 SaaS"})
print(result)

from crewai import Process

# Process.parallel — 所有 Agent 同时工作
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[task1, task2, task3],
    process=Process.parallel,
)

# Process.hierarchical — Manager Agent 动态调度
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[task1, task2, task3],
    process=Process.hierarchical,
    manager_llm="gpt-4o",  # 指定 Manager 的模型
)

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

# 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    next_step: str
    research_done: bool
    code_done: bool
    test_passed: bool

# 定义节点（Agent 函数）
def research_agent(state):
    # 调用 LLM 做研究
    return {"messages": ["Research completed"], "research_done": True, "next_step": "code"}

def code_agent(state):
    if not state["research_done"]:
        return {"messages": ["Need research first"], "next_step": "research"}
    # 生成代码
    return {"messages": ["Code generated"], "code_done": True, "next_step": "test"}

def test_agent(state):
    if not state["code_done"]:
        return {"messages": ["Need code first"], "next_step": "code"}
    # 运行测试
    passed = True  # 模拟
    return {
        "messages": ["Tests passed" if passed else "Tests failed"],
        "test_passed": passed,
        "next_step": "end" if passed else "code",
    }

# 定义条件路由
def router(state):
    return state["next_step"]

# 构建图
builder = StateGraph(AgentState)
builder.add_node("research", research_agent)
builder.add_node("code", code_agent)
builder.add_node("test", test_agent)
builder.add_edge("research", "code")
builder.add_edge("code", "test")
builder.add_conditional_edges("test", router, {"end": END, "code": "code"})
builder.set_entry_point("research")

graph = builder.compile()

# 运行
result = graph.invoke({
    "messages": ["Implement REST API for user management"],
    "research_done": False,
    "code_done": False,
    "test_passed": False,
})

from tenacity import retry, stop_after_attempt, wait_exponential

class FaultTolerantAgent:
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=10)
    )
    def invoke_with_retry(self, agent, task):
        try:
            return agent.run(task)
        except Exception as e:
            # 降级策略：跳过当前任务，流式报告错误
            return {"status": "error", "fallback": str(e)}