Skip to content
Gains Summary
Main Navigation 首页 / Home
C++ 编程 / C++ Programming
系统与高性能 / Systems & Performance
Web 开发 / Web Development
人工智能 / Artificial Intelligence
工业软件 / Industrial Software
其他内容 / Other Topics
C++ 编程 / C++系统与性能 / SystemsWeb 开发 / Web人工智能 / AI工业软件 / Industrial

外观

Sidebar Navigation

← 人工智能 / Artificial Intelligence

智能体工程 / Agent Engineering

1. Agent 工程体系全景 / Agent Engineering System Overview

2. Function Calling - 让 LLM 具备行动能力 / Function Calling for Giving LLMs the Ability to Act

3. Agent 框架演进 - 从裸 SDK 到 LangGraph / The Evolution of Agent Frameworks from Raw SDKs to LangGraph

4. RAG 基础 - 让 Agent 拥有"知识" / Retrieval-Augmented Generation Fundamentals for Agent Knowledge

5. 记忆管理 - Agent 的大脑 / Memory Management as the Brain of an Agent

6. Agent 工作流 - 从单步到复杂的执行编排 / Agent Workflows from Single Steps to Complex Orchestration

7. 多 Agent 系统 - 多个 Agent 协作 / Multi-Agent Systems and Agent Collaboration

8. RAG 进阶 - 企业级知识库实战 / Advanced RAG for Enterprise Knowledge Bases

9. 真实 Agent 应用场景 / Real-World AI Agent Applications

10. Structured Output - 让 LLM 输出可控的结构化数据 / Structured Output for Controllable, Machine-Readable LLM Responses

11. Tools Design Best Practices - AI Agent 工具设计最佳实践 / Tools Design Best Practices for AI Agents

12. Agent 架构模式 - 从单 Agent 到多 Agent 的工程范式 / Agent Architecture Patterns

13. Agent Modes — 编程 Agent 的交互模式设计 / Designing Interaction Modes for Coding Agents

14. Agent Workflow 编排:从循环到持久化执行的演进

15. Context Engineering - 从 Prompt 设计到上下文编排 / Context Engineering: From Prompt Design to Context Orchestration

16. Agent 缓存工程:从 KV Cache、Prompt Cache 到语义缓存 / Agent Caching Engineering

17. Harness Engineering, Skills, and Loop Engineering — 从信任模型到验证系统 / From Trusting Models to Verifying Systems

18. MCP 协议 - AI 工具的"USB 接口" / Model Context Protocol for AI Tool Integration

19. Agent 评估与测试 — 如何衡量一个"不可预测"的系统 / Agent Evaluation and Testing — How to Measure an "Unpredictable" System

20. 安全沙箱 - Agent 的安全边界 / Secure Sandboxes as Agent Safety Boundaries

21. 权限与门卫 - Agent 的安全控制中枢 / Permissions and Policy Gates for Agent Control

22. API Key 管理与安全 - Agent 的密钥生命周期的管理 / API Key Lifecycle Management and Security for Agents

23. 提示词注入防护 - Agent 的防御前沿 / Prompt Injection Defense for AI Agents

24. 可观测性与调试 - Agent 运行的透明度保障 / Observability and Debugging for Transparent Agent Operations

25. 模型路由 - 让正确的模型做正确的事 / Model Routing for Matching Models to Tasks

26. OpenClaw 设计深度分析 - 为什么它让人觉得"活"了 / OpenClaw Design Analysis and the Illusion of Liveliness

27. Claude Code 泄露源码深度分析 - 512,000 行代码揭示的生产级 Agent 架构 / Claude Code Source Analysis and Production Agent Architecture

28. LobeChat 设计深度分析 - 全栈 Agent Chat 应用工程实践 / LobeChat Design Analysis and Full-Stack Agent Chat Engineering

29. 编程 Agent 全面对比:从 Claude Code 到 Pi 的设计哲学 / Coding Agents Comparison: Design Philosophies from Claude Code to Pi

30. 领域 Agent 的确定性工具编译与延迟执行——从自然语言规格到单次 CAE 提交

31. Agent 工程学习指南 / An AI Agent Engineering Learning Guide

本页目录

Agent 工作流 - 从单步到复杂的执行编排 / Agent Workflows from Single Steps to Complex Orchestration ​

📅 创建时间:2026-05-08 🏷️ 标签:#AgentWorkflow #ReAct #PlanExecute #工作流模式 📚 前置知识:[[04-memory-management]]


📋 本章目标 ​

  • 理解 Agent 工作流的核心概念
  • 掌握三种主流工作流范式:单步 / ReAct / Plan-and-Execute
  • 理解子任务拆分与聚合的方法
  • 掌握错误处理与重试机制
  • 理解 Agent 的"终止条件"设计
  • 能够为具体场景选择合适的工作流

第0部分:ReAct 不是在做一个神秘的事情——它就是 Function Calling 外面套了一层命名 ​

你在 Function Calling 那篇已经看到了 Agent 循环:

while LLM返回了tool_calls:
    执行工具
    把结果返回 LLM
return LLM的最终回答
1
2
3
4

ReAct 只是给这个循环里的每一步起了名字。 不是新技术,是一个观察框架:

┌─────────────────────────────────────────────────────────────┐
│               ReAct 就是把 Function Calling 循环拆成三步       │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  你已知的 Function Calling 循环      ReAct 给它起的名字       │
│  ──────────────────────────────      ─────────────────       │
│                                                             │
│  LLM 说:"我需要调用 get_weather     → Thought(思考)       │
│  来查北京温度"                                              │
│                                                             │
│  LLM 输出 tool_calls:               → Action(行动)        │
│  [{name: "get_weather", args: ...}]                         │
│                                                             │
│  你的程序执行函数,得到结果           → Observation(观察)   │
│  {temp: 25, condition: "晴"}                                │
│                                                             │
│  把结果传回 LLM,LLM 决定下一步      → 下一轮 Thought        │
│                                                             │
└─────────────────────────────────────────────────────────────┘
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

用一次真实的 API 调用序列来看 ReAct 的每一步到底发生了什么。

0.1 ReAct 的完整 API 调用序列 ​

假设用户问:"北京今天多少度?用这个温度来判断应该穿什么衣服。"

第1轮:

→ POST /v1/chat/completions
  messages: [
    {role: "system", content: "你是助手。每次行动前先解释你的思考。"},
    {role: "user", content: "北京今天多少度?用这个温度来判断穿什么。"}
  ],
  tools: [{name: "get_weather", ...}]

← LLM 响应:
  finish_reason: "tool_calls"    ← 不是 "stop",说明它想调工具
  message: {
    content: "用户想知道北京的温度,然后用温度来判断穿什么。我需要先查询北京当前的天气。"  ← 这就是 "Thought"
    tool_calls: [{function: {name: "get_weather", arguments: '{"city":"北京"}'}}]  ← 这就是 "Action"
  }
1
2
3
4
5
6
7
8
9
10
11
12
13

注意:content 和 tool_calls 同时存在。LLM 先输出了一段自然语言思考("我需要先查询..."),然后又输出了工具调用。这就是 ReAct 的关键特征——模型被训练成在调工具之前先"说"出它在想什么。这个 content 字段就是 ReAct 论文里说的 Thought。

第2轮:

→ POST /v1/chat/completions
  messages: [
    ...前两轮的历史...,
    {role: "assistant", content: "我需要先查询北京当前天气。", tool_calls: [...]},
    {role: "tool", tool_call_id: "xxx", content: '{"temp": 25, "condition": "晴"}'}  ← 这就是 "Observation"
  ],
  tools: [{name: "get_weather", ...}]

← LLM 响应:
  finish_reason: "stop"           ← 这次是 "stop",结束了
  message: {
    content: "北京今天 25°C,晴天。这个温度很舒适,建议穿薄外套或长袖 T 恤。"
  }
1
2
3
4
5
6
7
8
9
10
11
12
13

整个循环就两轮 API 调用,结束了。 ReAct 没有改变任何底层机制——LLM 还是那个 LLM,Function Calling 还是那个 Function Calling。它只是多了一个"让 LLM 在调工具前先说出自己的思考"的约定。

0.2 Plan-and-Execute 又是什么? ​

Plan-and-Execute 的思路更简单:不要边想边做,先完整列出计划,再一步步执行。

┌─────────────────────────────────────────────────────────────┐
│            Plan-and-Execute 就是两阶段的 Function Calling      │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  阶段一:Plan(规划)—— 一次 LLM 调用                        │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 用户:"分析本月的销售数据并写一份报告"               │   │
│  │                                                     │   │
│  │ → LLM 被 prompt 引导,不执行工具,只输出计划:       │   │
│  │   1. 查询本月销售数据(调 get_sales_data)           │   │
│  │   2. 分析数据趋势(用 Python 执行分析脚本)          │   │
│  │   3. 生成图表(调 generate_chart)                  │   │
│  │   4. 撰写报告摘要                                   │   │
│  │   5. 把报告发邮件给老板(调 send_email)             │   │
│  └─────────────────────────────────────────────────────┘   │
│                            ↓                                │
│  阶段二:Execute(执行)—— 按计划逐步 Function Calling       │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 对计划中的每一步:                                   │   │
│  │   步骤1 → 调 get_sales_data → 拿到数据              │   │
│  │   步骤2 → 调 execute_python → 拿到分析结果          │   │
│  │   步骤3 → 调 generate_chart → 拿到图表              │   │
│  │   步骤4 → LLM 写报告(不需要工具,直接生成文本)     │   │
│  │   步骤5 → 调 send_email → 发送完成                  │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

和 ReAct 的区别:

  • ReAct:做一步 → 想一想 → 做下一步(每步都调一次 LLM 来"想")
  • Plan-Execute:一次性想好所有步骤 → 闷头执行(执行阶段尽量不调 LLM 来"想")

Plan-Execute 的好处是省 token(不用每步都让 LLM 思考一次),代价是僵化(中间发现计划不行时不好调整)。

0.3 单步 Agent——就是 Function Calling 只循环一次 ​

单步 Agent 其实不配叫一个独立的"范式"——它就是 max_iterations=1 的 Function Calling:

python
# 单步 Agent = 不让它循环
executor = AgentExecutor(agent=agent, tools=tools, max_iterations=1)

# 完整 Agent = 允许循环
executor = AgentExecutor(agent=agent, tools=tools, max_iterations=10)
1
2
3
4
5

所有 Agent 在底层都是同一个东西:一个 while 循环,不停地 POST messages 到 API,直到 finish_reason == "stop"。 所谓"工作流范式",只是在这个循环上加了不同的约束——允许多少次迭代、是否先做计划、是否要求 LLM 在行动前先说出思考。

带着这个认知,下面的细节就不会觉得神秘了。

1.1 从单次调用到工作流 ​

单次 LLM 调用:

┌─────────────────────────────────────────────────────────────┐
│                    单次调用示意                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  输入 → LLM → 输出                                           │
│                                                             │
│  用户:"翻译'hello world'成中文"                             │
│  → LLM → "你好世界"                                         │
│                                                             │
│  特点:一步到位,没有中间过程                                │
│                                                             │
└─────────────────────────────────────────────────────────────┘
1
2
3
4
5
6
7
8
9
10
11
12

Agent 工作流:

┌─────────────────────────────────────────────────────────────┐
│                    Agent 工作流示意                          │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  用户复杂请求:"帮我写一篇关于 AI 的报告,要有数据支撑"       │
│         ↓                                                   │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ Step 1: 搜索 AI 相关数据                              │   │
│  └─────────────────────────────────────────────────────┘   │
│         ↓                                                   │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ Step 2: 分析整理数据                                   │   │
│  └─────────────────────────────────────────────────────┘   │
│         ↓                                                   │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ Step 3: 撰写报告                                      │   │
│  └─────────────────────────────────────────────────────┘   │
│         ↓                                                   │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ Step 4: 检查质量,不满意则修改                         │   │
│  └─────────────────────────────────────────────────────┘   │
│         ↓                                                   │
│      最终报告                                               │
│                                                             │
└─────────────────────────────────────────────────────────────┘
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

1.2 为什么需要工作流? ​

┌─────────────────────────────────────────────────────────────┐
│                    工作流解决的核心问题                      │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  问题1:复杂任务无法一步完成                                │
│  "帮我分析这个月的销售数据并写报告"                         │
│  → 涉及:数据获取 → 数据分析 → 报告撰写 → 检查修改           │
│                                                             │
│  问题2:Agent 可能走入死胡同                                │
│  Agent 尝试了一种方法,失败了,需要换方法                    │
│  → 需要:重试机制 + 备选方案                                 │
│                                                             │
│  问题3:Agent 不知道该停在哪里                               │
│  "帮我优化这个代码" → 可以无限优化下去                       │
│  → 需要:明确的终止条件                                     │
│                                                             │
│  问题4:中间结果需要被利用                                   │
│  搜索的结果要传给写作模块,分析的结果要传给总结模块           │
│  → 需要:状态传递机制                                        │
│                                                             │
└─────────────────────────────────────────────────────────────┘
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

第2部分:三种工作流范式 ​

2.1 范式1:单步 Agent(Simple Agent) ​

最简单:一个 Prompt + 一次调用

┌─────────────────────────────────────────────────────────────┐
│                    单步 Agent 结构                           │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  用户输入 → System Prompt → LLM + Tools → 输出              │
│                                                             │
│  特点:                                                     │
│  • 没有循环                                                │
│  • 没有中间状态                                            │
│  • 工具调用最多一次                                         │
│                                                             │
└─────────────────────────────────────────────────────────────┘
1
2
3
4
5
6
7
8
9
10
11
12

适用场景:

  • 简单明确的任务(翻译、格式化、简单问答)
  • 只需要一个工具就能完成的任务
  • 不需要复杂推理的场景

代码示例:

python
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")
tools = [get_weather, send_email]

agent = create_openai_functions_agent(llm, tools, system_prompt)
executor = AgentExecutor(agent=agent, tools=tools, max_iterations=1)

result = executor.invoke({"input": "北京天气怎么样?"})
1
2
3
4
5
6
7
8
9
10

2.2 范式2:ReAct Agent ​

ReAct = Reasoning + Acting(推理驱动的行动)

┌─────────────────────────────────────────────────────────────┐
│                    ReAct 循环                                │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│    ┌─────────────────────────────────────────────────┐      │
│    │                                                 │      │
│    │     ┌────────────────────────────────┐         │      │
│    │     │         LLM 思考                │         │      │
│    │     │  "用户想知道天气,我需要先调用   │         │      │
│    │     │   天气工具获取北京的气温..."    │         │      │
│    │     └────────────────┬───────────────┘         │      │
│    │                      ↓ Thought                  │      │
│    │              ┌───────────────┐                  │      │
│    │              │   调用工具     │                  │      │
│    │              └───────┬───────┘                  │      │
│    │                      ↓ Action                   │      │
│    │              ┌───────────────┐                  │      │
│    │              │  执行 get_    │                  │      │
│    │              │  weather     │                  │      │
│    │              └───────┬───────┘                  │      │
│    │                      ↓ Observation              │      │
│    │              ┌───────────────┐                  │      │
│    │              │  返回:25°C   │                  │      │
│    │              │  晴天         │                  │      │
│    │              └───────┬───────┘                  │      │
│    │                      ↓                          │      │
│    │     ┌────────────────────────────────┐         │      │
│    │     │ LLM 看到结果,决定下一步        │         │      │
│    │     │ "已获取天气,现在可以回答用户"  │         │      │
│    │     └────────────────────────────────┘         │      │
│    │                      ↓                          │      │
│    │              是否继续?                          │      │
│    │              ↓     ↓                           │      │
│    │            是      否                          │      │
│    │             ↓       ↓                         │      │
│    │      返回思考      输出最终回答                  │      │
│    │                                                 │      │
│    └─────────────────────────────────────────────────┘      │
│                                                             │
└─────────────────────────────────────────────────────────────┘
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

核心思想:

┌─────────────────────────────────────────────────────────────┐
│                    ReAct 的核心                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  每次循环,LLM 思考三个问题:                                │
│                                                             │
│  1. Thought(思考):现在的情况是什么?我该做什么?           │
│  2. Action(行动):调用哪个工具?参数是什么?                │
│  3. Observation(观察):工具返回了什么结果?                │
│                                                             │
│  然后进入下一轮循环,直到任务完成                            │
│                                                             │
└─────────────────────────────────────────────────────────────┘
1
2
3
4
5
6
7
8
9
10
11
12
13

ReAct 代码示例:

python
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_core.agents import AgentFinish

def run_react_agent(user_input, tools, max_iterations=10):
    llm = ChatOpenAI(model="gpt-4o")
    agent = create_openai_functions_agent(llm, tools, system_prompt)
    executor = AgentExecutor(agent=agent, tools=tools, max_iterations=max_iterations)

    result = executor.invoke({"input": user_input})
    return result["output"]

# 调用
result = run_react_agent("帮我查北京天气,然后发邮件告诉老板", tools)
1
2
3
4
5
6
7
8
9
10
11
12
13

2.3 范式3:Plan-and-Execute Agent ​

Plan-and-Execute = 先规划再执行(Think then Act)

┌─────────────────────────────────────────────────────────────┐
│                    Plan-and-Execute 结构                     │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │  Plan 阶段(规划)                                   │   │
│  │  ┌─────────────────────────────────────────────┐   │   │
│  │  │ LLM 分析任务,生成执行计划                     │   │   │
│  │  │ "任务:写报告                                  │   │   │
│  │  │  1. 搜索 AI 数据                               │   │   │
│  │  │  2. 分析数据                                    │   │   │
│  │  │  3. 撰写报告草稿                               │   │   │
│  │  │  4. 检查质量"                                  │   │   │
│  │  └─────────────────────────────────────────────┘   │   │
│  └─────────────────────────────────────────────────────┘   │
│         ↓                                                   │
│  ┌─────────────────────────────────────────────────────┐   │
│  │  Execute 阶段(执行)                               │   │
│  │  按计划顺序执行每个步骤                              │   │
│  │  步骤1 → 步骤2 → 步骤3 → 步骤4                     │   │
│  └─────────────────────────────────────────────────────┘   │
│         ↓                                                   │
│      汇总结果                                               │
│                                                             │
└─────────────────────────────────────────────────────────────┘
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

Plan-and-Execute vs ReAct 对比:

┌─────────────────────────────────────────────────────────────┐
│                    两种范式对比                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  │ 特性           │ ReAct          │ Plan-and-Execute       │
│  ├────────────────┼────────────────┼──────────────────────┤
│  │ 决策时机        │ 边做边想        │ 先想好再做            │
│  │ 灵活性          │ 高(可中途改)  │ 低(按计划执行)      │
│  │ 可解释性        │ 低(过程混乱)  │ 高(计划清晰可见)    │
│  │ 适合场景        │ 探索性任务      │ 流程明确的任务        │
│  │ Token 消耗      │ 较低           │ 较高(要规划)        │
│  │ 错误恢复        │ 自然(下一轮)  │ 需要显式处理          │
│                                                             │
│  选择建议:                                                  │
│  • 任务流程明确(如数据分析报告)→ Plan-and-Execute        │
│  • 任务路径不确定(如研究探索)→ ReAct                      │
│                                                             │
└─────────────────────────────────────────────────────────────┘
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

Plan-and-Execute 代码示例:

python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o")

def plan_and_execute(task, tools):
    # Step 1: Plan - 生成执行计划
    plan_prompt = f"""分析以下任务,生成具体的执行步骤:
任务:{task}

输出格式:
1. 第一步:...
2. 第二步:...
3. 第三步:...

请生成详细的执行计划:"""

    plan_response = llm.invoke(plan_prompt)
    steps = parse_steps(plan_response.content)  # 解析步骤

    # Step 2: Execute - 按计划执行
    results = []
    for step in steps:
        # 构建执行 Prompt
        execute_prompt = f"""执行以下步骤:
{step}

之前的执行结果:
{results}

{'请根据之前结果继续执行' if results else '这是第一步,请执行'}"""

        result = llm_with_tools.invoke(execute_prompt)  # 调用带工具的 LLM
        results.append({"step": step, "result": result})

    # Step 3: 汇总
    return aggregate_results(results)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

2.4 三种范式适用场景总结 ​

┌─────────────────────────────────────────────────────────────┐
│                    范式选择指南                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  单步 Agent:                                               │
│  • 翻译一段文字                                             │
│  • 回答一个简单问题                                         │
│  • 格式化一段 JSON                                          │
│                                                             │
│  ReAct Agent:                                             │
│  • 用户问题不确定需要几步解决                                │
│  • 需要多次工具调用                                          │
│  • 探索性任务(如研究分析)                                   │
│                                                             │
│  Plan-and-Execute:                                        │
│  • 任务流程相对固定                                          │
│  • 需要高可解释性                                            │
│  • 分步结果需要汇总                                          │
│                                                             │
│  实际项目中,可以组合使用:                                   │
│  • 外层用 Plan-and-Execute(大方向)                        │
│  • 内层用 ReAct(每个步骤内部探索)                          │
│                                                             │
└─────────────────────────────────────────────────────────────┘
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

第3部分:子任务拆分与聚合 ​

3.1 为什么需要拆分? ​

┌─────────────────────────────────────────────────────────────┐
│                    子任务拆分的价值                          │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  复杂任务 → 拆成简单子任务 → 降低难度                        │
│                                                             │
│  示例:"帮我分析竞品并写报告"                                │
│                                                             │
│  拆分为:                                                   │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ 子任务1: 搜索竞品 A 的信息                            │   │
│  │ 子任务2: 搜索竞品 B 的信息                            │   │
│  │ 子任务3: 搜索竞品 C 的信息                            │   │
│  │ 子任务4: 对比分析(依赖 1,2,3)                        │   │
│  │ 子任务5: 撰写报告(依赖 4)                           │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  1,2,3 可以并行执行                                         │
│  4 依赖 1,2,3 完成                                          │
│  5 依赖 4 完成                                              │
│                                                             │
└─────────────────────────────────────────────────────────────┘
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

3.2 依赖关系处理 ​

python
from collections import defaultdict

class TaskGraph:
    """简单的任务依赖图"""
    def __init__(self):
        self.tasks = {}  # task_id -> task_def
        self.dependencies = defaultdict(list)  # task_id -> [depends_on]

    def add_task(self, task_id, task_def, depends_on=None):
        self.tasks[task_id] = task_def
        if depends_on:
            self.dependencies[task_id] = depends_on

    def get_execution_order(self):
        """获取拓扑排序后的执行顺序"""
        visited = set()
        order = []

        def dfs(task_id):
            if task_id in visited:
                return
            visited.add(task_id)
            for dep in self.dependencies[task_id]:
                dfs(dep)
            order.append(task_id)

        for task_id in self.tasks:
            dfs(task_id)

        return order

    def execute(self, task_id):
        """执行任务及其依赖"""
        order = self.get_execution_order()
        results = {}

        for tid in order:
            # 执行任务,使用依赖结果
            deps_results = [results[d] for d in self.dependencies[tid] if d in results]
            results[tid] = self.tasks[tid](deps_results)

        return results[task_id]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

第4部分:错误处理与重试机制 ​

4.1 常见错误类型 ​

┌─────────────────────────────────────────────────────────────┐
│                    Agent 常见错误                            │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  错误1:工具调用失败                                         │
│  • 网络超时、API 不可用、参数格式错误                         │
│  → 重试 + 降级策略                                          │
│                                                             │
│  错误2:工具返回无效结果                                     │
│  • 返回空数据、格式不符合预期                                │
│  → 解析错误处理 + 重新尝试                                   │
│                                                             │
│  错误3:LLM 陷入死循环                                       │
│  • 反复调用同一工具                                          │
│  • 无限循环                                                  │
│  → 最大迭代次数限制                                          │
│                                                             │
│  错误4:LLM 放弃任务                                         │
│  • 任务太难,LLM 直接说"做不到"                              │
│  → 尝试简化任务或改写 Prompt                                │
│                                                             │
└─────────────────────────────────────────────────────────────┘
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

4.2 重试策略实现 ​

python
import time

class ResilientExecutor:
    """带重试机制的 Agent 执行器"""
    def __init__(self, max_retries=3, backoff_factor=1.5):
        self.max_retries = max_retries
        self.backoff_factor = backoff_factor

    def execute_with_retry(self, agent, tools, user_input):
        last_error = None

        for attempt in range(self.max_retries + 1):
            try:
                result = agent.invoke({"input": user_input})

                # 检查结果是否有效
                if self._is_valid_result(result):
                    return result

                # 结果无效,尝试修复 Prompt 后重试
                if attempt < self.max_retries:
                    user_input = self._improve_prompt(user_input, result)

            except Exception as e:
                last_error = e
                if attempt < self.max_retries:
                    wait_time = self.backoff_factor ** attempt
                    time.sleep(wait_time)

        raise Exception(f"Agent 执行失败:{last_error}")

    def _is_valid_result(self, result):
        """验证结果是否有效"""
        # 根据具体业务定义有效性的判断标准
        return result and result.get("output")

    def _improve_prompt(self, original, failed_result):
        """基于失败结果改进 Prompt"""
        return f"""{original}

注意:之前的尝试没有成功完成任务。
请重新思考并尝试不同的方法。"""
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

4.3 最大迭代次数保护 ​

python
from langchain.agents import AgentExecutor

# LangChain 的内置保护
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=10,  # 最多 10 次循环
    max_execution_time=60,  # 或最多 60 秒
    early_stopping_method="force",  # "force" 或 "generate"
    handle_parsing_errors=True  # 自动处理解析错误
)
1
2
3
4
5
6
7
8
9
10
11

第5部分:Agent 的终止条件 ​

5.1 何时应该停止? ​

┌─────────────────────────────────────────────────────────────┐
│                    终止条件设计                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  条件1:任务完成                                             │
│  • LLM 明确输出了预期的结果                                  │
│  • 工具调用返回了所需信息                                    │
│                                                             │
│  条件2:达到资源上限                                         │
│  • 最大迭代次数                                              │
│  • 最大 Token 消耗                                           │
│  • 最大执行时间                                              │
│                                                             │
│  条件3:检测到无法完成                                       │
│  • 工具反复失败                                              │
│  • LLM 明确表示无法完成                                      │
│                                                             │
│  条件4:用户主动取消                                         │
│  • 用户发送了停止指令                                        │
│                                                             │
└─────────────────────────────────────────────────────────────┘
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

5.2 终止条件的实现 ​

python
from langchain.agents import AgentFinish, AgentFinish

class StoppingAgentExecutor:
    def run(self, user_input, max_turns=10):
        messages = [{"role": "user", "content": user_input}]
        turn_count = 0

        while turn_count < max_turns:
            turn_count += 1

            response = llm_with_tools.invoke(messages)

            # 检查是否完成任务(AgentFinish)
            if isinstance(response, AgentFinish):
                return response.return_values["output"]

            # 检查是否有工具调用
            if not response.tool_calls:
                # 没有工具调用但也没完成任务
                return f"无法完成任务:{response.content}"

            # 执行工具
            for tool_call in response.tool_calls:
                tool_result = execute_tool(tool_call)
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": tool_result
                })

        return f"达到最大轮次限制({max_turns}),任务未完成"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

核心总结 ​

总结1:三种工作流范式 ​

范式特点适用场景
单步 Agent无循环,一次调用简单任务
ReAct边做边想,灵活性高探索性任务
Plan-and-Execute先想后做,可解释性强流程固定的任务

总结2:子任务拆分 ​

复杂任务 → 按依赖关系拆分 → 可并行部分并行执行
→ 聚合结果 → 最终输出
1
2

总结3:错误处理 ​

重试机制:指数退避 + 最大次数限制
终止条件:任务完成 / 资源上限 / 无法完成 / 用户取消
1
2

章节测试 ​

测试1:范式选择 ​

什么情况下应该选择 ReAct 而不是 Plan-and-Execute?

测试2:ReAct 循环 ​

ReAct 的三个核心步骤是什么?

测试3:依赖处理 ​

子任务 A 依赖 B 和 C,B 和 C 互不依赖,以下哪个执行顺序是正确的? A. A → B → C B. B → C → A C. A → B 和 C 并行

测试4:错误处理 ​

AgentExecutor 中 max_iterations 参数的作用是什么?

测试5:终止条件 ​

以下哪个不是 Agent 的合理终止条件? A. 达到最大迭代次数 B. LLM 返回了预期的结果 C. Token 消耗超过 100 万 D. 用户发送停止指令


参考答案 ​

测试1答案 ​

答案:当任务路径不确定、需要灵活探索时选择 ReAct。例如:"帮我研究一下 AI 最新进展",没有固定流程,需要边探索边调整方向。


测试2答案 ​

答案:Thought(思考当前情况)→ Action(执行工具)→ Observation(观察结果),然后进入下一轮循环。


测试3答案 ​

答案:B(B → C → A)

解析:A 依赖 B 和 C,所以必须 B 和 C 都完成后才能执行 A。B 和 C 互不依赖,谁先执行都可以。


测试4答案 ​

答案:限制 Agent 最大循环次数,防止无限循环。例如 max_iterations=10 表示最多执行 10 次工具调用循环。


测试5答案 ​

答案:C(Token 消耗超过 100 万)

解析:上下文窗口通常在 10 万左右,100 万 Token 限制没有实际意义。其他选项都是合理的终止条件。


相关笔记 ​

  • [[01-function-calling]] - 工作流中的工具调用
  • [[02-framework-evolution]] - LangGraph 中的工作流实现
  • [[06-multi-agent]] - 多 Agent 协作中的工作流

下一步学习 ​

  • [ ] 阅读 06 - 多 Agent 系统

学习状态:🟡 开始学习

最后更新于:

Pager
上一篇5. 记忆管理 - Agent 的大脑 / Memory Management as the Brain of an Agent
下一篇7. 多 Agent 系统 - 多个 Agent 协作 / Multi-Agent Systems and Agent Collaboration

持续记录,持续成长

Copyright © Tidenflow