AI Agent 架构设计模式：从 ReAct 到多智能体协作

┌─────────────────────────────────────────┐
│              Agent 核心                  │
│                                         │
│  感知（Perception）                      │
│    └── 接收用户输入、工具返回、环境状态     │
│                                         │
│  规划（Planning）                        │
│    └── LLM 推理：下一步做什么？           │
│                                         │
│  行动（Action）                          │
│    └── 调用工具、执行代码、生成内容        │
│                                         │
│  记忆（Memory）                          │
│    ├── 短期：当前对话上下文               │
│    └── 长期：向量数据库、结构化存储        │
└─────────────────────────────────────────┘

[Thought] → [Action] → [Observation] → [Thought] → ...

用户问题：帮我找出最近一个月Python最受欢迎的10个开源项目

Thought: 我需要搜索GitHub上近期Star增长最快的Python项目
Action: search_web("GitHub trending Python repositories last month")
Observation: 返回搜索结果，包含多个项目列表

Thought: 结果不够精确，应该直接查询GitHub Trending API
Action: github_api("trending/repositories?language=python&since=monthly")
Observation: 返回结构化的项目列表，包含名称、Star数、描述

Thought: 数据已获取，需要按Star增量排序并格式化
Action: sort_and_format(data, key="stars_gained", limit=10)
Observation: 格式化完成的TOP10列表

回答: 以下是本月GitHub上最受欢迎的10个Python开源项目...

Planner（规划者）→ 分解任务为子步骤列表
Executor（执行者）→ 逐步执行每个子任务
Re-Planner（重规划者）→ 根据执行结果动态调整计划

# 第一步：Planner 制定计划
plan = planner_llm("""
任务：为我们的SaaS产品撰写竞品分析报告

请制定详细的执行计划：
1. 搜索市场上主要竞争对手
2. 收集每个竞品的功能特性、定价、用户评价
3. 分析竞品的优劣势
4. 总结我们的差异化机会
5. 撰写结构化报告
""")

# 第二步：Executor 逐步执行
for step in plan.steps:
    result = executor_llm(step, previous_results)
    results.append(result)
    
    # 第三步：Re-Planner 动态调整
    if needs_replanning(result):
        plan = replanner_llm(original_task, completed_steps, result)

Generator → 生成初步输出
Critic    → 评估质量，指出问题
Generator → 根据反馈修改
（循环直到满足标准或达到最大轮次）

max_rounds = 3
output = generator_llm(task)

for round in range(max_rounds):
    critique = critic_llm(f"""
    原始任务：{task}
    当前输出：{output}
    
    请从以下维度评估：
    1. 准确性：内容是否正确？
    2. 完整性：是否覆盖所有要求？
    3. 质量：表达是否清晰专业？
    
    如果发现问题，列出具体改进点。
    如果质量已达标，回复 "PASS"
    """)
    
    if "PASS" in critique:
        break
    
    output = generator_llm(f"根据以下反馈修改输出：{critique}\n原输出：{output}")

失败案例 → 生成反思摘要 → 存入记忆数据库
新任务开始 → 检索相关历史反思 → 注入当前上下文 → 避免重蹈覆辙

Orchestrator（编排者）
    ├── Research Agent（负责搜索和信息收集）
    ├── Analysis Agent（负责数据分析）
    ├── Writing Agent（负责内容生成）
    └── Review Agent（负责质量检查）

Agent A 提出方案 → Agent B 评估批判 → Agent C 整合优化
（适用于需要多角度思考的决策场景）

from langgraph.graph import StateGraph, END

# 定义状态
class ResearchState(TypedDict):
    task: str
    search_results: list
    analysis: str
    final_report: str

# 定义各节点
def research_node(state):
    results = search_agent(state["task"])
    return {"search_results": results}

def analysis_node(state):
    analysis = analysis_agent(state["search_results"])
    return {"analysis": analysis}

def writing_node(state):
    report = writing_agent(state["task"], state["analysis"])
    return {"final_report": report}

# 构建图
graph = StateGraph(ResearchState)
graph.add_node("research", research_node)
graph.add_node("analysis", analysis_node)
graph.add_node("writing", writing_node)

graph.add_edge("research", "analysis")
graph.add_edge("analysis", "writing")
graph.add_edge("writing", END)

app = graph.compile()

任务复杂度如何？
├── 简单（单工具、1-3步）
│   └── → ReAct 即可
│
├── 中等（多工具、4-10步）
│   ├── 需要自我改进？ → Reflection + ReAct
│   └── 步骤顺序明确？ → Plan-and-Execute
│
└── 复杂（需专业分工、长流程）
    ├── 任务可并行？ → Multi-Agent（并行模式）
    └── 需要多角度决策？ → Multi-Agent（协作/辩论模式）

# 必须实现的容错机制
@retry(max_attempts=3, backoff=exponential)
def tool_call_with_retry(tool, args):
    ...

# 人工干预节点
if confidence < threshold:
    result = await human_approval(action_plan)

每一步记录：时间戳 | 节点名称 | 输入 | 输出 | Token用量 | 延迟
推荐工具：LangSmith、Langfuse、Arize Phoenix