13523 字
约 45 分钟
1
第5章 ReAct-让Agent学会思考
2026-09-18
第5章 ReAct-让Agent学会思考
来源：https://ai-agent-guide.xiaofuge.cn/chapters/ch04-react-pattern.html 所属：第二篇-Agent的大脑
让 Agent 学会"边想边做"的经典范式
🔗 从上一章延续的问题
第3章的天气 Agent 能查天气、能规划行程，但它不会思考为什么要查——用户问"明天适合跑步吗"，它直接调 get_weather，却没有先推理"适合跑步需要看天气+空气质量"。这导致了两个问题：① 可能漏掉该调的工具，② 可能调了不必要的工具浪费 token。ReAct 就是为了解决这两个问题而诞生的——让 Agent 在行动前先思考，观察后再反思。 🎯 本章先掌握 3 件事
这一章内容很多，但如果你是第一次接触 ReAct，只要先搞懂三件事就够了：为什么要先想再做、Thought / Action / Observation 如何形成闭环、它和上一章工具调用有什么关系。
你可以先读 4.1 到 4.4，把 ReAct 的核心循环吃透；4.5 之后的 ReWOO、LLM Compiler、Reflexion 和生产级主循环更适合作为进阶阅读，不需要一次全吞下去。
5.1 从一个问题开始
假设用户问 Agent："2024年诺贝尔物理学奖得主是谁？他们的研究领域是什么？"
一个没有规划能力的 Agent 可能直接让 LLM 回答——但 LLM 的知识截止到训练日期，可能不知道 2024 年的诺奖。它需要搜索。但什么时候搜？搜什么？搜完之后做什么？
这就是 ReAct 要解决的问题。
ReAct = Reasoning + Acting
让 LLM 交替进行"推理（Thought）"和"行动（Action）"，通过"观察（Observation）"获取反馈，循环直到得出最终答案。
5.2 ReAct 的诞生背景
在 ReAct 之前，Agent 面临两难：
只推理（Reasoning-only）
LLM 用 CoT 一步步推理
✓ 推理能力强
✗ 无法获取外部信息
✗ 可能产生幻觉
→ 想得到，做不到
只行动（Acting-only）
LLM 直接调用工具
✓ 能获取外部信息
✗ 没有推理规划
✗ 不知道下一步做什么
→ 做得到，想不到
2022 年，Yao 等人在论文 "ReAct: Synergizing Reasoning and Acting in Language Models" 中提出了 ReAct 范式，将两者结合： 💡 ReAct 的核心洞察
**推理指导行动，行动反馈推理。**LLM 先"想"该做什么（Thought），再"做"（Action），然后"看"结果（Observation），基于结果再"想"下一步——形成闭环。
5.3 ReAct 循环：Thought → Action → Observation
ReAct 的核心是三个角色的循环：
逐步演示：一个完整的 ReAct 执行
1
Thought #1
我需要先搜索2024年诺贝尔物理学奖得主。
← 1 / 7 → 重播
5.4 ReAct 的 Prompt 模板
ReAct 的实现核心是 System Prompt。来看一个标准的 ReAct prompt：
ReAct System Prompt
尽可能回答以下问题。你可以使用以下工具：
可用工具：
- search(query): 搜索互联网获取信息
- lookup(keyword): 在上一次搜索结果中查找关键词
- finish(answer): 提交最终答案
格式要求：
每一步必须使用以下格式：
Thought: 你对下一步该做什么的推理
Action: 工具名称(参数)
Observation: 工具返回的结果
... (Thought/Action/Observation 可以重复多次)
Thought: 我现在知道答案了
Action: finish(最终答案)
开始！
问题: {user_question}
关键设计点
1. 强制结构化输出 用 "Thought:" "Action:" "Observation:" 标记，让 LLM 的输出可被程序解析。
2. 工具结果由框架注入 "Observation:" 不是 LLM 生成的，是框架执行工具后拼接上去的。
3. 循环终止条件 当 LLM 输出 finish(answer) 时结束循环。也可设置最大步数防止死循环。
4. 推理可见性 Thought 部分是 LLM 的"内心独白"，让推理过程可审计、可调试。
5.5 ReWOO 模式（Reasoning Without Observation）
ReAct 的核心循环是"边想边做"——每一步 Thought 之后立刻 Action，获得 Observation 后再想下一步。这很直观，但有一个严重的问题：每一步的 Observation 都会被塞入上下文，随着步数增加，token 消耗线性膨胀。
ReWOO 提出了一个反直觉的思路：先想好所有步骤，再统一执行。
ReWOO = Reasoning Without Observation
2023 年由 Xu 等人提出。核心思想：让 LLM 在没有 Observation 的情况下，先规划出完整的推理链（Planner），然后由 Worker 统一执行所有工具调用，最后由 Solver 综合所有结果给出答案。
Planner → Worker → Solver 三阶段
ReWOO 把 ReAct 的循环拆成了三个独立阶段： 🔍 三阶段详解
Planner（规划器）
LLM 一次性生成完整的推理链，每一步用 #E 标记期望的工具调用。因为还没有 Observation，Planner 必须靠自身推理决定步骤——这就是 "Without Observation" 的含义。
Worker（执行器）
逐个执行 Planner 规划的工具调用，将结果填入 #E 占位符。Worker 是程序化的，不需要 LLM 参与——纯工具执行。
Solver（求解器） LLM 接收已填入结果的完整推理链，综合所有信息生成最终答案。此时 LLM 看到的是"规划+真实结果"的完整视图。
ReAct vs ReWOO 流程对比
ReAct：边想边做
- 每一步 Thought 都需要携带完整历史上下文
- Observation 逐步累积，token 随步数线性增长
- LLM 调用次数 = 步数 × 2（Thought + 解析）
- 优势：灵活调整，每步可基于 Observation 改变策略
ReWOO：先想后做
- Planner 只调用 LLM 1次，生成完整推理链
- Worker 执行工具，不消耗 LLM token
- Solver 只调用 LLM 1次，综合答案
- 总 LLM 调用 = 2次，远少于 ReAct 的 N 次
ReWOO 代码示例
from openai import OpenAI
client = OpenAI()
TOOLS = {"search": search_web, "weather": get_weather, "calc": calculate}
# ===== Phase 1: Planner =====
PLANNER_PROMPT = """你是一个规划器。根据用户问题，生成一系列步骤来解决问题。
每一步用 #E 标记需要执行的工具调用。
格式：
Plan: 第一步的推理
#E1 = tool_name("参数")
Plan: 第二步的推理（可引用 #E1 的结果）
#E2 = tool_name("参数")
...
问题: {question}"""
def plan_steps(question):
"""Phase 1: LLM 一次性规划所有步骤"""
resp = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "system", "content": PLANNER_PROMPT},
{"role": "user", "content": question}]
)
return resp.choices[0].message.content
# ===== Phase 2: Worker =====
def execute_steps(plan_text):
"""Phase 2: 解析并执行所有 #E 工具调用"""
import re
results = {}
# 查找所有 #E 标记
steps = re.findall(r'#E(\d+)\s*=\s*(\w+)\((".*?"|\'.*?\'|.*?)\)', plan_text)
for step_id, tool_name, tool_input in steps:
tool_input = tool_input.strip('"\'')
if tool_name in TOOLS:
results[f"#E{step_id}"] = TOOLStool_name
else:
results[f"#E{step_id}"] = f"Error: {tool_name} not found"
return results
# ===== Phase 3: Solver =====
SOLVER_PROMPT = """根据以下规划和执行结果，给出最终答案。
规划:
{plan}
执行结果:
{results}
请综合以上信息，给出完整答案。"""
def solve_answer(plan_text, results):
"""Phase 3: LLM 综合所有结果生成答案"""
# 将结果填入规划的 #E 占位符
filled_plan = plan_text
for key, value in results.items():
filled_plan = filled_plan.replace(key, f"[结果: {value}]")
resp = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "system", "content": SOLVER_PROMPT},
{"role": "user", "content":
f"规划:\n{plan_text}\n\n执行结果:\n{filled_plan}"}]
)
return resp.choices[0].message.content
# ===== 完整流程 =====
def rewoo_agent(question):
"""ReWOO: Planner → Worker → Solver"""
# 1. 规划
plan = plan_steps(question)
# 2. 执行
results = execute_steps(plan)
# 3. 求解
answer = solve_answer(plan, results)
return answer
# 使用
answer = rewoo_agent("北京今天天气如何？适合户外活动吗？")
import OpenAI from 'openai';
const client = new OpenAI();
const TOOLS: Record string> = {
search: search_web,
weather: get_weather,
calc: calculate
};
// ===== Phase 1: Planner =====
const PLANNER_PROMPT = `你是一个规划器。根据用户问题，生成一系列步骤来解决问题。
每一步用 #E 标记需要执行的工具调用。
格式：
Plan: 第一步的推理
#E1 = tool_name("参数")
Plan: 第二步的推理（可引用 #E1 的结果）
#E2 = tool_name("参数")
...
问题: {question}`;
async function plan_steps(question: string): Promise {
// Phase 1: LLM 一次性规划所有步骤
const resp = await client.chat.completions.create({
model: 'gpt-4o',
messages: [
{ role: 'system', content: PLANNER_PROMPT },
{ role: 'user', content: question }
]
});
return resp.choices[0].message.content!;
}
// ===== Phase 2: Worker =====
function execute_steps(planText: string): Record {
// Phase 2: 解析并执行所有 #E 工具调用
const results: Record = {};
const regex = /#E(\d+)\s*=\s*(\w+)\(["']?(.*?)["']?\)/g;
let match: RegExpExecArray | null;
while ((match = regex.exec(planText)) !== null) {
const stepId = match[1];
const toolName = match[2];
const toolInput = match[3];
if (toolName in TOOLS) {
results[`#E${stepId}`] = TOOLStoolName;
} else {
results[`#E${stepId}`] = `Error: ${toolName} not found`;
}
}
return results;
}
// ===== Phase 3: Solver =====
const SOLVER_PROMPT = `根据以下规划和执行结果，给出最终答案。
规划:
{plan}
执行结果:
{results}
请综合以上信息，给出完整答案。`;
async function solve_answer(
planText: string,
results: Record
): Promise {
// Phase 3: LLM 综合所有结果生成答案
let filledPlan = planText;
for (const [key, value] of Object.entries(results)) {
filledPlan = filledPlan.replaceAll(key, `[结果: ${value}]`);
}
const resp = await client.chat.completions.create({
model: 'gpt-4o',
messages: [
{ role: 'system', content: SOLVER_PROMPT },
{ role: 'user', content: `规划:\n${planText}\n\n执行结果:\n${filledPlan}` }
]
});
return resp.choices[0].message.content!;
}
// ===== 完整流程 =====
async function rewoo_agent(question: string): Promise {
// ReWOO: Planner → Worker → Solver
const plan = await plan_steps(question); // 1. 规划
const results = execute_steps(plan); // 2. 执行
const answer = await solve_answer(plan, results); // 3. 求解
return answer;
}
// 使用
const answer = await rewoo_agent('北京今天天气如何？适合户外活动吗？');
package main
import (
"context"
"fmt"
"regexp"
"strings"
openai "github.com/sashabaranov/go-openai"
)
var client = openai.NewClient()
func searchWeb(input string) string { return "search: " + input }
func getWeather(input string) string { return "weather: " + input }
func calculate(input string) string { return "calc: " + input }
var tools = map[string]func(string) string{
"search": searchWeb,
"weather": getWeather,
"calc": calculate,
}
// ===== Phase 1: Planner =====
const plannerPrompt = `你是一个规划器。根据用户问题，生成一系列步骤来解决问题。
每一步用 #E 标记需要执行的工具调用。
格式：
Plan: 第一步的推理
#E1 = tool_name("参数")
Plan: 第二步的推理（可引用 #E1 的结果）
#E2 = tool_name("参数")
...
问题: %s`
func planSteps(question string) (string, error) {
// Phase 1: LLM 一次性规划所有步骤
resp, err := client.CreateChatCompletion(context.Background(), openai.ChatCompletionRequest{
Model: openai.GPT4o,
Messages: []openai.ChatCompletionMessage{
{Role: openai.ChatMessageRoleSystem, Content: fmt.Sprintf(plannerPrompt, "{question}")},
{Role: openai.ChatMessageRoleUser, Content: question},
},
})
if err != nil {
return "", err
}
return resp.Choices[0].Message.Content, nil
}
// ===== Phase 2: Worker =====
func executeSteps(planText string) map[string]string {
// Phase 2: 解析并执行所有 #E 工具调用
results := make(map[string]string)
re := regexp.MustCompile(`#E(\d+)\s*=\s*(\w+)\(["']?(.*?)["']?\)`)
matches := re.FindAllStringSubmatch(planText, -1)
for _, m := range matches {
stepID, toolName, toolInput := m[1], m[2], m[3]
if fn, ok := tools[toolName]; ok {
results["#E"+stepID] = fn(toolInput)
} else {
results["#E"+stepID] = fmt.Sprintf("Error: %s not found", toolName)
}
}
return results
}
// ===== Phase 3: Solver =====
const solverPrompt = `根据以下规划和执行结果，给出最终答案。
规划:
%s
执行结果:
%s
请综合以上信息，给出完整答案。`
func solveAnswer(planText string, results map[string]string) (string, error) {
// Phase 3: LLM 综合所有结果生成答案
filledPlan := planText
for key, value := range results {
filledPlan = strings.ReplaceAll(filledPlan, key, fmt.Sprintf("[结果: %s]", value))
}
resultStr := ""
for k, v := range results {
resultStr += fmt.Sprintf("%s: %s\n", k, v)
}
resp, err := client.CreateChatCompletion(context.Background(), openai.ChatCompletionRequest{
Model: openai.GPT4o,
Messages: []openai.ChatCompletionMessage{
{Role: openai.ChatMessageRoleSystem, Content: solverPrompt},
{Role: openai.ChatMessageRoleUser, Content: fmt.Sprintf("规划:\n%s\n\n执行结果:\n%s", planText, filledPlan)},
},
})
if err != nil {
return "", err
}
return resp.Choices[0].Message.Content, nil
}
// ===== 完整流程 =====
func rewooAgent(question string) (string, error) {
// ReWOO: Planner → Worker → Solver
plan, err := planSteps(question) // 1. 规划
if err != nil {
return "", err
}
results := executeSteps(plan) // 2. 执行
answer, err := solveAnswer(plan, results) // 3. 求解
if err != nil {
return "", err
}
return answer, nil
}
// 使用
func main() {
answer, _ := rewooAgent("北京今天天气如何？适合户外活动吗？")
fmt.Println(answer)
}
import com.openai.client.OpenAIClient;
import com.openai.models.*;
import java.util.*;
import java.util.regex.*;
public class AgentCode {
static OpenAIClient client = new OpenAIClient();
static Map> tools = new HashMap<>();
static {
tools.put("search", AgentCode::searchWeb);
tools.put("weather", AgentCode::getWeather);
tools.put("calc", AgentCode::calculate);
}
static String searchWeb(String input) { return "search: " + input; }
static String getWeather(String input) { return "weather: " + input; }
static String calculate(String input) { return "calc: " + input; }
// ===== Phase 1: Planner =====
static final String PLANNER_PROMPT = """
你是一个规划器。根据用户问题，生成一系列步骤来解决问题。
每一步用 #E 标记需要执行的工具调用。
格式：
Plan: 第一步的推理
#E1 = tool_name("参数")
Plan: 第二步的推理（可引用 #E1 的结果）
#E2 = tool_name("参数")
...
问题: %s""";
static String planSteps(String question) {
// Phase 1: LLM 一次性规划所有步骤
ChatCompletionCreateParams params = ChatCompletionCreateParams.builder()
.model("gpt-4o")
.addMessage(ChatCompletionMessage.builder()
.role(ChatCompletionMessage.Role.SYSTEM)
.content(String.format(PLANNER_PROMPT, "{question}"))
.build())
.addMessage(ChatCompletionMessage.builder()
.role(ChatCompletionMessage.Role.USER)
.content(question)
.build())
.build();
ChatCompletion resp = client.chat().completions().create(params);
return resp.choices().get(0).message().content().orElse("");
}
// ===== Phase 2: Worker =====
static Map executeSteps(String planText) {
// Phase 2: 解析并执行所有 #E 工具调用
Map results = new LinkedHashMap<>();
Pattern pattern = Pattern.compile("#E(\\\\d+)\\\\s*=\\\\s*(\\\\w+)\\\\([\"']?(.*?)[\"']?\\\\)");
Matcher matcher = pattern.matcher(planText);
while (matcher.find()) {
String stepId = matcher.group(1);
String toolName = matcher.group(2);
String toolInput = matcher.group(3);
if (tools.containsKey(toolName)) {
results.put("#E" + stepId, tools.get(toolName).apply(toolInput));
} else {
results.put("#E" + stepId, "Error: " + toolName + " not found");
}
}
return results;
}
// ===== Phase 3: Solver =====
static final String SOLVER_PROMPT = """
根据以下规划和执行结果，给出最终答案。
规划:
%s
执行结果:
%s
请综合以上信息，给出完整答案。""";
static String solveAnswer(String planText, Map results) {
// Phase 3: LLM 综合所有结果生成答案
String filledPlan = planText;
for (Map.Entry entry : results.entrySet()) {
filledPlan = filledPlan.replace(entry.getKey(),
"[结果: " + entry.getValue() + "]");
}
ChatCompletionCreateParams params = ChatCompletionCreateParams.builder()
.model("gpt-4o")
.addMessage(ChatCompletionMessage.builder()
.role(ChatCompletionMessage.Role.SYSTEM)
.content(SOLVER_PROMPT)
.build())
.addMessage(ChatCompletionMessage.builder()
.role(ChatCompletionMessage.Role.USER)
.content("规划:\\n" + planText + "\\n\\n执行结果:\\n" + filledPlan)
.build())
.build();
ChatCompletion resp = client.chat().completions().create(params);
return resp.choices().get(0).message().content().orElse("");
}
// ===== 完整流程 =====
static String rewooAgent(String question) {
// ReWOO: Planner → Worker → Solver
String plan = planSteps(question); // 1. 规划
Map results = executeSteps(plan); // 2. 执行
String answer = solveAnswer(plan, results); // 3. 求解
return answer;
}
// 使用
public static void main(String[] args) {
String answer = rewooAgent("北京今天天气如何？适合户外活动吗？");
System.out.println(answer);
}
}
⚠️ ReWOO 的局限
ReWOO 的 Planner 必须在没有 Observation 的条件下规划所有步骤，这意味着它无法根据中间结果调整策略。如果第一步的搜索结果出乎意料，Planner 无法"转向"——它只能沿着预设路径走到底。这是 ReWOO 相比 ReAct 的核心劣势。
5.6 LLM Compiler 模式
ReAct 是串行的：Thought₁ → Action₁ → Observation₁ → Thought₂ → Action₂ → …。如果 Agent 需要调用 5 个独立的工具，即使它们之间没有依赖关系，也必须一个一个等。
LLM Compiler 提出了并行执行的思路：让 LLM 一次性输出多个独立的工具调用，然后并行执行它们。
LLM Compiler
2023 年由 Kim 等人提出。核心思想：将 Agent 的执行计划编译成依赖图（DAG），识别可并行的步骤，同时执行没有依赖关系的工具调用，显著减少总执行时间。
依赖图分析：识别可并行的步骤
关键洞察：很多工具调用之间没有数据依赖。比如：
串行（ReAct）
Thought → search("北京天气") → Obs₁
Thought → search("上海天气") → Obs₂
Thought → search("广州天气") → Obs₃
Thought → 比较三城市 → Answer
总计：4轮LLM + 3次搜索 = ~12s
并行（LLM Compiler）
Planner → 并行:
search("北京天气") ─┐
search("上海天气") ─┤→ Solver → Answer
search("广州天气") ─┘
总计：1轮LLM + 1次并行搜索 = ~4s
并行调用多个工具：代码示例
import asyncio
from openai import OpenAI
client = OpenAI()
# ===== Step 1: 编译器生成并行计划 =====
COMPILER_PROMPT = """根据用户问题，生成工具调用计划。
标记依赖关系：如果一个调用需要另一个的结果，标注 depends_on。
输出JSON格式:
[
{"id": 1, "tool": "search", "args": {"query": "北京天气"},
"depends_on": []},
{"id": 2, "tool": "search", "args": {"query": "上海天气"},
"depends_on": []},
{"id": 3, "tool": "compare", "args": {"data_from": [1, 2]},
"depends_on": [1, 2]}
]
问题: {question}"""
async def compile_plan(question):
"""LLM 编译：生成带依赖关系的工具调用计划"""
resp = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "system", "content": COMPILER_PROMPT},
{"role": "user", "content": question}],
response_format={"type": "json_object"}
)
import json
return json.loads(resp.choices[0].message.content)
# ===== Step 2: 并行执行独立工具 =====
async def execute_parallel(plan):
"""按依赖图并行执行：无依赖的步骤同时运行"""
results = {}
completed = set()
while len(completed) = 0.8: # 质量达标
return result
# 3. 反思推理过程
reflection = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": REFLECT_PROMPT},
{"role": "user", "content":
f"之前的尝试:\n{result['trajectory']}\n\n"
f"评估反馈:\n{evaluation['feedback']}"}
]
).choices[0].message.content
reflections.append(reflection)
print(f"Trial {trial+1}: score={evaluation['score']:.2f}")
print(f"Reflection: {reflection[:100]}...")
return result # 达到最大尝试次数，返回最后结果
def evaluate_result(question, result):
"""LLM 评估答案质量"""
resp = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content":
"评估以下答案的质量。输出JSON: {\"score\": 0-1, \"feedback\": \"...\"}"},
{"role": "user", "content":
f"问题: {question}\n答案: {result['answer']}"}
],
response_format={"type": "json_object"}
)
import json
return json.loads(resp.choices[0].message.content)
# 使用
answer = reflexion_agent(
"解释量子纠缠的原理，并说明它与经典关联的区别"
)
import OpenAI from 'openai';
const client = OpenAI();
const REFLECT_PROMPT = """你刚刚完成了一个任务，但结果不够好。;
// 回顾你的推理过程和执行结果，找出问题所在。
// 之前的尝试:
// {trajectory}
// 评估反馈:
// {evaluation}
// 请反思:
// 1. 哪一步推理出了问题？
// 2. 应该怎么做才更好？
// 3. 下次尝试时需要注意什么？
// 输出格式:
// Reflection: 你的反思内容"""
function react_agent_with_context(question, context="") {
/** docstring */
const messages = [{"role": "system", "content": REACT_PROMPT + context}];
// messages.append({"role": "user", "content": question})
const trajectory = [];
for (const step of range(10)) {
const resp = client.chat.completions.create(model="gpt-4o", messages=messages);
const thought = resp.choices[0].message.content;
// trajectory.append(thought)
if ("finish" in thought.lower()) {
return {"answer": thought, "trajectory": "\n".join(trajectory)};
// messages.append({"role": "assistant", "content": thought})
return {"answer": thought, "trajectory": "\n".join(trajectory)};
function reflexion_agent(question, max_trials=3) {
/** docstring */
const reflections = [] # 累积的反思经验;
for (const trial of range(max_trials)) {
// 1. 执行任务（使用 ReAct + 历史反思作为额外上下文）
const context = "";
if (reflections) {
const context = `\n之前的反思经验:\n" + "\n`;
const result = react_agent_with_context(question, context);
// 2. 评估结果质量
const evaluation = evaluate_result(question, result);
if (evaluation["score"] >= 0.8) {
return result;
// 3. 反思推理过程
const reflection = client.chat.completions.create(;
const model = "gpt-4o",;
const messages = [;
// {"role": "system", "content": REFLECT_PROMPT},
// {"role": "user", "content":
// f"之前的尝试:\n{result['trajectory']}\n\n"
// f"评估反馈:\n{evaluation['feedback']}"}
// ]
// ).choices[0].message.content
// reflections.append(reflection)
console.log(`Trial {trial+1}: score={evaluation['score']:.2f}`);
console.log(`Reflection: {reflection[:100]}...`);
return result # 达到最大尝试次数，返回最后结果;
function evaluate_result(question, result) {
/** docstring */
const resp = client.chat.completions.create(;
const model = "gpt-4o",;
const messages = [;
// {"role": "system", "content":
// "评估以下答案的质量。输出JSON: {\"score\": 0-1, \"feedback\": \"...\"}"},
// {"role": "user", "content":
// f"问题: {question}\n答案: {result['answer']}"}
// ],
const response_format = {"type": "json_object"};
// )
// Node.js built-in or npm package for: json
return json.loads(resp.choices[0].message.content);
// 使用
const answer = reflexion_agent(;
// "解释量子纠缠的原理，并说明它与经典关联的区别"
// )
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// from openai import OpenAI
// Python: client = OpenAI()
// Python: REFLECT_PROMPT = """你刚刚完成了一个任务，但结果不够好。
// Python: 回顾你的推理过程和执行结果，找出问题所在。
// Python: 之前的尝试:
// Python: {trajectory}
// Python: 评估反馈:
// Python: {evaluation}
// Python: 请反思:
// Python: 1. 哪一步推理出了问题？
// Python: 2. 应该怎么做才更好？
// Python: 3. 下次尝试时需要注意什么？
// Python: 输出格式:
// Python: Reflection: 你的反思内容"""
func react_agent_with_context() {
// Python: messages = [{"role": "system", "content": REACT_PROMPT + context}]
// Python: messages.append({"role": "user", "content": question})
// Python: trajectory = []
for _, step := range range(10) {
// Python: resp = client.chat.completions.create(model="gpt-4o", messages=messages)
// Python: thought = resp.choices[0].message.content
// Python: trajectory.append(thought)
if "finish" in thought.lower() {
return {"answer": thought, "trajectory": "\n".join(trajectory)}
// Python: messages.append({"role": "assistant", "content": thought})
return {"answer": thought, "trajectory": "\n".join(trajectory)}
func reflexion_agent() {
// Python: reflections = [] # 累积的反思经验
for _, trial := range range(max_trials) {
// 1. 执行任务（使用 ReAct + 历史反思作为额外上下文）
// Python: context = ""
if reflections {
// Python: context = f"\n之前的反思经验:\n" + "\n".join(reflections)
// Python: result = react_agent_with_context(question, context)
// 2. 评估结果质量
// Python: evaluation = evaluate_result(question, result)
if evaluation["score"] >= 0.8 {
return result
// 3. 反思推理过程
// Python: reflection = client.chat.completions.create(
// Python: model="gpt-4o",
// Python: messages=[
// Python: {"role": "system", "content": REFLECT_PROMPT},
// Python: {"role": "user", "content":
// Python: f"之前的尝试:\n{result['trajectory']}\n\n"
// Python: f"评估反馈:\n{evaluation['feedback']}"}
// Python: ]
// Python: ).choices[0].message.content
// Python: reflections.append(reflection)
fmt.Println(f"Trial {trial+1}: score={evaluation['score']:.2f}")
fmt.Println(f"Reflection: {reflection[:100]}...")
return result # 达到最大尝试次数，返回最后结果
func evaluate_result() {
// Python: resp = client.chat.completions.create(
// Python: model="gpt-4o",
// Python: messages=[
// Python: {"role": "system", "content":
// Python: "评估以下答案的质量。输出JSON: {\"score\": 0-1, \"feedback\": \"...\"}"},
// Python: {"role": "user", "content":
// Python: f"问题: {question}\n答案: {result['answer']}"}
// Python: ],
// Python: response_format={"type": "json_object"}
// Python: )
// import json
return json.loads(resp.choices[0].message.content)
// 使用
// Python: answer = reflexion_agent(
// Python: "解释量子纠缠的原理，并说明它与经典关联的区别"
// Python: )
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// from openai import OpenAI
// Python: client = OpenAI()
// Python: REFLECT_PROMPT = """你刚刚完成了一个任务，但结果不够好。
// Python: 回顾你的推理过程和执行结果，找出问题所在。
// Python: 之前的尝试:
// Python: {trajectory}
// Python: 评估反馈:
// Python: {evaluation}
// Python: 请反思:
// Python: 1. 哪一步推理出了问题？
// Python: 2. 应该怎么做才更好？
// Python: 3. 下次尝试时需要注意什么？
// Python: 输出格式:
// Python: Reflection: 你的反思内容"""
public static void react_agent_with_context() {
// Python: messages = [{"role": "system", "content": REACT_PROMPT + context}]
// Python: messages.append({"role": "user", "content": question})
// Python: trajectory = []
for (var step : range(10)) {
// Python: resp = client.chat.completions.create(model="gpt-4o", messages=messages)
// Python: thought = resp.choices[0].message.content
// Python: trajectory.append(thought)
if ("finish" in thought.lower()) {
return {"answer": thought, "trajectory": "\n".join(trajectory)};
// Python: messages.append({"role": "assistant", "content": thought})
return {"answer": thought, "trajectory": "\n".join(trajectory)};
public static void reflexion_agent() {
// Python: reflections = [] # 累积的反思经验
for (var trial : range(max_trials)) {
// 1. 执行任务（使用 ReAct + 历史反思作为额外上下文）
// Python: context = ""
if (reflections) {
// Python: context = f"\n之前的反思经验:\n" + "\n".join(reflections)
// Python: result = react_agent_with_context(question, context)
// 2. 评估结果质量
// Python: evaluation = evaluate_result(question, result)
if (evaluation["score"] >= 0.8) {
return result;
// 3. 反思推理过程
// Python: reflection = client.chat.completions.create(
// Python: model="gpt-4o",
// Python: messages=[
// Python: {"role": "system", "content": REFLECT_PROMPT},
// Python: {"role": "user", "content":
// Python: f"之前的尝试:\n{result['trajectory']}\n\n"
// Python: f"评估反馈:\n{evaluation['feedback']}"}
// Python: ]
// Python: ).choices[0].message.content
// Python: reflections.append(reflection)
System.out.println(String.format("$1"));
System.out.println(String.format("$1"));
return result # 达到最大尝试次数，返回最后结果;
public static void evaluate_result() {
// Python: resp = client.chat.completions.create(
// Python: model="gpt-4o",
// Python: messages=[
// Python: {"role": "system", "content":
// Python: "评估以下��案的质量。输出JSON: {\"score\": 0-1, \"feedback\": \"...\"}"},
// Python: {"role": "user", "content":
// Python: f"问题: {question}\n答案: {result['answer']}"}
// Python: ],
// Python: response_format={"type": "json_object"}
// Python: )
// import json
return json.loads(resp.choices[0].message.content);
// 使用
// Python: answer = reflexion_agent(
// Python: "解释量子纠缠的原理，并说明它与经典关联的区别"
// Python: )
}
💡 Reflexion 的核心价值
Reflexion 不只是"多做几遍"。它的关键是反思内容会累积——每轮尝试的反思都被保存下来，作为下一轮的额外上下文。这意味着 Agent 在"从失败中学习"，而不是盲目重试。这就像人类做题：第一次做错，反思原因，第二次带着反思去做，大概率比第一次好。
5.8 Structured Output（结构化输出）
ReAct 用 "Thought:" "Action:" 标记强制 LLM 输出特定格式，但文本标记解析极不稳定——LLM 可能输出 "Thought: ..." 或 "思考: ..." 或干脆不写标记。这是 Agent 开发中最头疼的问题之一。
结构化输出（Structured Output）就是为了彻底解决 LLM 输出不可靠的问题。
Structured Output
通过 JSON Schema 约束，让 LLM 的输出严格符合预定义的结构——字段名、类型、必填项、枚举值都被锁定。LLM 不再"自由发挥"，而是像填表一样输出。
三种方式对比
📊 JSON Mode vs Function Calling vs Structured Output | 方式 | 约束强度 | 字段类型保证 | 必填项保证 | 适用场景 | | --- | --- | --- | --- | --- | | JSON Mode | 弱（只保证是合法JSON） | ✗ 不保证 | ✗ 不保证 | 简单JSON输出 | | Function Calling | 中（参数有Schema） | △ 部分保证 | △ 部分保证 | 工具调用 | | Structured Output | 强（完整Schema约束） | ✓ 严格保证 | ✓ 严格保证 | Agent结构化推理 | ### OpenAI 结构化输出 + Pydantic 验证
from pydantic import BaseModel, Field
from openai import OpenAI
from typing import List, Optional
client = OpenAI()
# ===== 1. 用 Pydantic 定义输出 Schema =====
class AgentStep(BaseModel):
"""Agent 单步推理的结构化输出"""
thought: str = Field(
description="对当前状态的推理分析"
)
action: str = Field(
description="要调用的工具名称",
enum=["search", "lookup", "calculate", "finish"]
)
action_input: str = Field(
description="工具的输入参数"
)
confidence: float = Field(
description="对当前推理的置信度 0-1",
ge=0, le=1
)
class AgentResponse(BaseModel):
"""Agent 完整响应的结构化输出"""
steps: List[AgentStep] = Field(
description="推理步骤列表"
)
final_answer: Optional[str] = Field(
description="最终答案（仅当 action=finish 时提供）",
default=None
)
needs_more_info: bool = Field(
description="是否需要更多信息才能回答"
)
# ===== 2. 用 OpenAI Structured Output 调用 =====
def structured_agent(question):
"""使用 Structured Output 的 Agent"""
response = client.beta.chat.completions.parse(
model="gpt-4o-2024-08-06",
messages=[
{"role": "system", "content":
"你是一个研究助手。分析问题并给出推理步骤。"},
{"role": "user", "content": question}
],
response_format=AgentResponse # Pydantic 模型直接作为 Schema
)
# 3. 返回的是已验证的 Pydantic 对象
result = response.choices[0].message.parsed
# 类型安全！result.thought 是 str，result.confidence 是 float
for step in result.steps:
print(f"Thought: {step.thought}")
print(f"Action: {step.action}({step.action_input})")
print(f"Confidence: {step.confidence:.2f}")
if result.final_answer:
print(f"Answer: {result.final_answer}")
return result
# ===== 3. Pydantic 自动验证 =====
# 如果 LLM 输出的 confidence 超出 0-1 范围，Pydantic 会自动拒绝
# 如果 action 不是枚举值之一，Pydantic 会自动拒绝
# 如果 steps 是空列表，可以根据业务规则添加额外验证
def validate_agent_response(response: AgentResponse):
"""业务层面的额外验证"""
if not response.steps:
raise ValueError("Agent 必须至少输出一个推理步骤")
if response.needs_more_info and response.final_answer:
raise ValueError("标注需要更多信息时不应有最终答案")
if response.steps[-1].action == "finish" and not response.final_answer:
raise ValueError("finish 动作必须提供最终答案")
return True
# 使用
result = structured_agent("量子计算和经典计算的本质区别是什么？")
validate_agent_response(result)
import {BaseModel, Field} from 'pydantic';
import OpenAI from 'openai';
// TypeScript has built-in types, no import needed for List, Optional
const client = OpenAI();
// ===== 1. 用 Pydantic 定义输出 Schema =====
class AgentStep {
/** docstring */
// thought: str = Field(
// description = "对当前状态的推理分析"
// )
// action: str = Field(
// description = "要调用的工具名称",
// enum = ["search", "lookup", "calculate", "finish"]
// )
// action_input: str = Field(
// description = "工具的输入参数"
// )
// confidence: float = Field(
// description = "对当前推理的置信度 0-1",
// ge = 0, le=1
// )
class AgentResponse {
/** docstring */
// steps: List[AgentStep] = Field(
// description = "推理步骤列表"
// )
// final_answer: Optional[str] = Field(
// description = "最终答案（仅当 action=finish 时提供）",
// default = None
// )
// needs_more_info: bool = Field(
// description = "是否需要更多信息才能回答"
// )
// ===== 2. 用 OpenAI Structured Output 调用 =====
function structured_agent(question) {
/** docstring */
// response = client.beta.chat.completions.parse(
// model = "gpt-4o-2024-08-06",
// messages = [
// {"role": "system", "content":
// "你是一个研究助手。分析问题并给出推理步骤。"},
// {"role": "user", "content": question}
// ],
// response_format = AgentResponse # Pydantic 模型直接作为 Schema
// )
// 3. 返回的是已验证的 Pydantic 对象
// result = response.choices[0].message.parsed
// 类型安全！result.thought 是 str，result.confidence 是 float
for (const step of result.steps) {
console.log(`Thought: {step.thought}`);
console.log(`Action: {step.action}({step.action_input})`);
console.log(`Confidence: {step.confidence:.2f}`);
if (result.final_answer) {
console.log(`Answer: {result.final_answer}`);
return result;
// ===== 3. Pydantic 自动验证 =====
// 如果 LLM 输出的 confidence 超出 0-1 范围，Pydantic 会自动拒绝
// 如果 action 不是枚举值之一，Pydantic 会自动拒绝
// 如果 steps 是空列表，可以根据业务规则添加额外验证
function validate_agent_response(response: AgentResponse) {
/** docstring */
if (!response.steps) {
// raise ValueError("Agent 必须至少输出一个推理步骤")
if (response.needs_more_info && response.final_answer) {
// raise ValueError("标注需要更多信息时不应有最终答案")
if (response.steps[-1].action == "finish" && !response.final_answer) {
// raise ValueError("finish 动作必须提供最终答案")
return true;
// 使用
// result = structured_agent("量子计算和经典计算的本质区别是什么？")
// validate_agent_response(result)
}
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// from pydantic import BaseModel, Field
// from openai import OpenAI
// from typing import List, Optional
// Python: client = OpenAI()
// ===== 1. 用 Pydantic 定义输出 Schema =====
// AgentStep - CLI Agent class
type AgentStep struct {
// Python: thought: str = Field(
// Python: description="对当前状态的推理分析"
// Python: )
// Python: action: str = Field(
// Python: description="要调用的工具名称",
// Python: enum=["search", "lookup", "calculate", "finish"]
// Python: )
// Python: action_input: str = Field(
// Python: description="工具的输入参数"
// Python: )
// Python: confidence: float = Field(
// Python: description="对当前推理的置信度 0-1",
// Python: ge=0, le=1
// Python: )
// AgentResponse - CLI Agent class
type AgentResponse struct {
// Python: steps: List[AgentStep] = Field(
// Python: description="推理步骤列表"
// Python: )
// Python: final_answer: Optional[str] = Field(
// Python: description="最终答案（仅当 action=finish 时提供）",
// Python: default=None
// Python: )
// Python: needs_more_info: bool = Field(
// Python: description="是否需要更多信息才能回答"
// Python: )
// ===== 2. 用 OpenAI Structured Output 调用 =====
func structured_agent() {
// Python: response = client.beta.chat.completions.parse(
// Python: model="gpt-4o-2024-08-06",
// Python: messages=[
// Python: {"role": "system", "content":
// Python: "你是一个研究助手。分析问题并给出推理步骤。"},
// Python: {"role": "user", "content": question}
// Python: ],
// Python: response_format=AgentResponse # Pydantic 模型直接作为 Schema
// Python: )
// 3. 返回的是已验证的 Pydantic 对象
// Python: result = response.choices[0].message.parsed
// 类型安全！result.thought 是 str，result.confidence 是 float
for _, step := range result.steps {
fmt.Println(f"Thought: {step.thought}")
fmt.Println(f"Action: {step.action}({step.action_input})")
fmt.Println(f"Confidence: {step.confidence:.2f}")
if result.final_answer {
fmt.Println(f"Answer: {result.final_answer}")
return result
// ===== 3. Pydantic 自动验证 =====
// 如果 LLM 输出的 confidence 超出 0-1 范围，Pydantic 会自动拒绝
// 如果 action 不是枚举值之一，Pydantic 会自动拒绝
// 如果 steps 是空列表，可以根据业务规则添加额外验证
func validate_agent_response() {
if not response.steps {
// Python: raise ValueError("Agent 必须至少输出一个推理步骤")
if response.needs_more_info and response.final_answer {
// Python: raise ValueError("标注需要更多信息时不应有最终答案")
if response.steps[-1].action == "finish" and not response.final_answer {
// Python: raise ValueError("finish 动作必须提供最终答案")
return true
// 使用
// Python: result = structured_agent("量子计算和经典计算的本质区别是什么？")
// Python: validate_agent_response(result)
}
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// from pydantic import BaseModel, Field
// from openai import OpenAI
// from typing import List, Optional
// Python: client = OpenAI()
// ===== 1. 用 Pydantic 定义输出 Schema =====
public class AgentStep {
// Python: thought: str = Field(
// Python: description="对当前状态的推理分析"
// Python: )
// Python: action: str = Field(
// Python: description="要调用的工具名称",
// Python: enum=["search", "lookup", "calculate", "finish"]
// Python: )
// Python: action_input: str = Field(
// Python: description="工具的输入参数"
// Python: )
// Python: confidence: float = Field(
// Python: description="对当前推理的置信度 0-1",
// Python: ge=0, le=1
// Python: )
public class AgentResponse {
// Python: steps: List[AgentStep] = Field(
// Python: description="推理步骤列表"
// Python: )
// Python: final_answer: Optional[str] = Field(
// Python: description="最终答案（仅当 action=finish 时提供）",
// Python: default=None
// Python: )
// Python: needs_more_info: bool = Field(
// Python: description="是否需要更多信息才能回答"
// Python: )
// ===== 2. 用 OpenAI Structured Output 调用 =====
public static void structured_agent() {
// Python: response = client.beta.chat.completions.parse(
// Python: model="gpt-4o-2024-08-06",
// Python: messages=[
// Python: {"role": "system", "content":
// Python: "你是一个研究助手。分析问题并给出推理步骤。"},
// Python: {"role": "user", "content": question}
// Python: ],
// Python: response_format=AgentResponse # Pydantic 模型直接作为 Schema
// Python: )
// 3. 返回的是已验证的 Pydantic 对象
// Python: result = response.choices[0].message.parsed
// 类型安全！result.thought 是 str，result.confidence 是 float
for (var step : result.steps) {
System.out.println(String.format("$1"));
System.out.println(String.format("$1"));
System.out.println(String.format("$1"));
if (result.final_answer) {
System.out.println(String.format("$1"));
return result;
// ===== 3. Pydantic 自动验证 =====
// 如果 LLM 输出的 confidence 超出 0-1 范围，Pydantic 会自动拒绝
// 如果 action 不是枚举值之一，Pydantic 会自动拒绝
// 如果 steps 是空列表，可以根据业务规则添加额外验证
public static void validate_agent_response() {
if (!response.steps) {
// Python: raise ValueError("Agent 必须至少输出一个推理步骤")
if (response.needs_more_info && response.final_answer) {
// Python: raise ValueError("标注需要更多信息时不应有最终答案")
if (response.steps[-1].action == "finish" && !response.final_answer) {
// Python: raise ValueError("finish 动作必须提供最终答案")
return true;
// 使用
// Python: result = structured_agent("量子计算和经典计算的本质区别是什么？")
// Python: validate_agent_response(result)
}
}
✓ Structured Output 的核心优势
- 100% 格式保证：LLM 输出严格符合 Schema，无需正则解析
- 类型安全：字段类型、枚举值、范围约束全部锁定
- 可验证：Pydantic 自动验证 + 业务规则二次校验
- 开发效率：不用写解析代码，直接用 Pydantic 模型
5.9 上下文压缩实战
ReAct 的致命问题是上下文膨胀。每一步 Thought + Action + Observation 都被追加到消息历史中，10步任务可能积累上千 token 的上下文。到第20步，LLM 可能因为上下文太长而"遗忘"最初的问题。 ⚠️ 上下文膨胀的真实数据
Step 1: ~200 tokens (system + question + thought₁ + action₁ + obs₁)
Step 5: ~1,000 tokens (累积5轮)
Step 10: ~2,000 tokens
Step 20: ~4,000 tokens → LLM 开始遗忘早期信息
Step 50: ~10,000 tokens → 接近模型上下文窗口上限
三种压缩策略
1. 截断（Truncation） 最粗暴：直接丢弃最早的 N 条消息。优点是简单，缺点是可能丢失关键信息。
messages = messages[-K:]
只保留最近K条
2. 摘要（Summarization） 用 LLM 将早期对话压缩成摘要。保留语义，但消耗额外 LLM 调用。
summary = llm.summarize(old_msgs)
messages = [summary] + recent_msgs
3. 语义保护型压缩 最精细：锁定核心指令，只压缩中间步骤。确保关键信息不丢失。
locked = [system, question]
compressed = compress(middle)
messages = locked + compressed + recent
语义保护型压缩：锁定核心指令
压缩前后 Token 对比
from openai import OpenAI
client = OpenAI()
def semantic_compress(messages, keep_recent=3):
"""语义保护型压缩：锁定核心指令，压缩中间步骤"""
# 1. 锁定：System Prompt 和原始问题
locked = messages[:2] # system + user question
# 2. 保留：最近K步（完整不压缩）
recent = messages[-(keep_recent * 3):] # 每步3条(thought+action+obs)
# 3. 压缩：中间所有步骤 → 一段摘要
middle = messages[2:-(keep_recent * 3)]
if len(middle) > 6: # 只有中间步骤足够多才压缩
summary = client.chat.completions.create(
model="gpt-4o-mini", # 用小模型压缩，节省成本
messages=[
{"role": "system", "content":
"将以下对话历史压缩为简短摘要，保留关键决策和发现。"
"忽略冗余的推理过程，只保留结论性信息。"},
{"role": "user", "content":
f"对话历史:\n{format_messages(middle)}"}
]
).choices[0].message.content
compressed_middle = [
{"role": "system", "content": f"[历史摘要] {summary}"}
]
else:
compressed_middle = middle
# 4. 组合：锁定 + 压缩 + 保留
return locked + compressed_middle + recent
def format_messages(messages):
"""格式化消息列表为可读文本"""
text = ""
for msg in messages:
role = msg["role"]
content = msg["content"][:200] # 每条截断到200字
text += f"[{role}] {content}\n"
return text
# ===== Token 对比示例 =====
def compare_compression(original_msgs, compressed_msgs):
"""压缩前后 token 对比"""
def count_tokens(msgs):
# 粗略估算：1个中文字≈1.5token，1个英文词≈1token
total = 0
for msg in msgs:
content = msg["content"]
chinese_chars = sum(1 for c in content if '\u4e00' 6) {
const summary = client.chat.completions.create(;
const model = "gpt-4o-mini", # 用小模型压缩，节省成本;
const messages = [;
// {"role": "system", "content":
// "将以下对话历史压缩为简短摘要，保留关键决策和发现。"
// "忽略冗余的推理过程，只保留结论性信息。"},
// {"role": "user", "content":
// f"对话历史:\n{format_messages(middle)}"}
// ]
// ).choices[0].message.content
const compressed_middle = [;
// {"role": "system", "content": f"[历史摘要] {summary}"}
// ]
} else {
const compressed_middle = middle;
// 4. 组合：锁定 + 压缩 + 保留
return locked + compressed_middle + recent;
function format_messages(messages) {
/** docstring */
const text = "";
for (const msg of messages) {
const role = msg["role"];
const content = msg["content"][:200] # 每条截断到200字;
// text += f"[{role}] {content}\n"
return text;
// ===== Token 对比示例 =====
function compare_compression(original_msgs, compressed_msgs) {
/** docstring */
function count_tokens(msgs) {
// 粗略估算：1个中文字≈1.5token，1个英文词≈1token
const total = 0;
for (const msg of msgs) {
const content = msg["content"];
const chinese_chars = sum(1 for c in content if '\u4e00' 6 {
// Python: summary = client.chat.completions.create(
// Python: model="gpt-4o-mini", # 用小模型压缩，节省成本
// Python: messages=[
// Python: {"role": "system", "content":
// Python: "将以下对话历史压缩为简短摘要，保留关键决策和发现。"
// Python: "忽略冗余的推理过程，只保留结论性信息。"},
// Python: {"role": "user", "content":
// Python: f"对话历史:\n{format_messages(middle)}"}
// Python: ]
// Python: ).choices[0].message.content
// Python: compressed_middle = [
// Python: {"role": "system", "content": f"[历史摘要] {summary}"}
// Python: ]
} else {
// Python: compressed_middle = middle
// 4. 组合：锁定 + 压缩 + 保留
return locked + compressed_middle + recent
func format_messages() {
// Python: text = ""
for _, msg := range messages {
// Python: role = msg["role"]
// Python: content = msg["content"][:200] # 每条截断到200字
// Python: text += f"[{role}] {content}\n"
return text
// ===== Token 对比示例 =====
func compare_compression() {
func count_tokens() {
// 粗略估算：1个中文字≈1.5token，1个英文词≈1token
// Python: total = 0
for _, msg := range msgs {
// Python: content = msg["content"]
// Python: chinese_chars = sum(1 for c in content if '\u4e00' 6) {
// Python: summary = client.chat.completions.create(
// Python: model="gpt-4o-mini", # 用小模型压缩，节省成本
// Python: messages=[
// Python: {"role": "system", "content":
// Python: "将以下对话历史压缩为简短摘要，保留关键决策和发现。"
// Python: "忽略冗余的推理过程，只保留结论性信息。"},
// Python: {"role": "user", "content":
// Python: f"对话历史:\n{format_messages(middle)}"}
// Python: ]
// Python: ).choices[0].message.content
// Python: compressed_middle = [
// Python: {"role": "system", "content": f"[历史摘要] {summary}"}
// Python: ]
} else {
// Python: compressed_middle = middle
// 4. 组合：锁定 + 压缩 + 保留
return locked + compressed_middle + recent;
public static void format_messages() {
// Python: text = ""
for (var msg : messages) {
// Python: role = msg["role"]
// Python: content = msg["content"][:200] # 每条截断到200字
// Python: text += f"[{role}] {content}\n"
return text;
// ===== Token 对比示例 =====
public static void compare_compression() {
public static void count_tokens() {
// 粗略估算：1个中文字≈1.5token，1个英文词≈1token
// Python: total = 0
for (var msg : msgs) {
// Python: content = msg["content"]
// Python: chinese_chars = sum(1 for c in content if '\u4e00' string> = {
search: searchWeb,
lookup: lookupKeyword,
};
const REACT_PROMPT = `...`; // ReAct 系统提示词
async function reactAgent(question: string, maxSteps = 10): Promise {
const messages: any[] = [
{ role: "system", content: REACT_PROMPT },
{ role: "user", content: question },
];
for (let step = 0; step > tools = new HashMap<>();
static {
tools.put("search", ReActAgent::searchWeb);
tools.put("lookup", ReActAgent::lookupKeyword);
}
static final String REACT_PROMPT = "..."; // ReAct 系统提示词
public static String reactAgent(String question, int maxSteps) {
List messages = new ArrayList<>();
messages.add(ChatCompletionMessageParam.ofSystem(
ChatCompletionSystemMessageParam.builder()
.systemMessage(REACT_PROMPT).build()));
messages.add(ChatCompletionMessageParam.ofUser(
ChatCompletionUserMessageParam.builder()
.userMessage(question).build()));
Pattern actionRe = Pattern.compile("Action: (\\w+)\\((.*?)\\)");
for (int step = 0; step tool = tools.get(toolName);
String result = tool != null ? tool.apply(toolInput)
: "Error: tool " + toolName + " not found";
// 5. 注入 Observation
messages.add(ChatCompletionMessageParam.ofUser(
ChatCompletionUserMessageParam.builder()
.userMessage("Observation: " + result).build()));
}
return "达到最大步数限制";
}
}
核心就这几步：LLM 生成 → 解析 Action → 执行工具 → 注入 Observation → 循环。
5.11 ReAct 的优势与局限
✓ 优势
- 简单直观：Thought-Action-Observation 三段式，易于理解和实现
- 推理可审计：Thought 让每步推理可见，方便调试
- 错误可恢复：Observation 反馈让 Agent 能发现错误并调整
- 工具可组合：多步推理中可调用不同工具
- 通用性强：不依赖特定模型，任何 LLM 都能用
✗ 局限
- Token 消耗大：每步都带完整历史，长任务 token 消耗线性增长
- 串行执行：Thought→Action→Observation 严格串行，无法并行
- 依赖 LLM 质量：推理质量直接取决于 LLM 能力
- 格式不稳定：LLM 可能输出错误格式导致解析失败
- 容易循环：某些情况下 Agent 会陷入循环不收敛
5.12 ReAct 的演进与变体
本章介绍了 ReAct 的多个演进变体，下面是完整的 ReAct 家族演进图谱： 🔄 ReAct 家族演进 | 变体 | 改进点 | 核心机制 | 年份 | | --- | --- | --- | --- | | ReAct | 原始版：Thought-Action-Observation | 推理+行动交替循环 | 2022 | | ReWOO | 先规划再执行，减少token消耗 | Planner→Worker→Solver | 2023 | | LLM Compiler | 并行执行独立工具调用 | 依赖图(DAG)编译+并行 | 2023 | | Reflexion | 加入自我反思，从失败中学习 | 执行→评估→反思→重试循环 | 2023 | | LATS | 结合蒙特卡洛树搜索+ReAct | 树搜索+推理行动 | 2023 | | Function Calling ReAct | 用原生 Function Calling 替代文本解析 | 结构化工具调用 | 2023+ | ## 5.13 工程深度：生产级 Agent 主循环设计
前面我们学习了 ReAct、ReWOO、Reflexion 等推理模式，但它们都是概念模型。生产环境中，Agent 主循环需要解决一系列工程问题：烧钱怎么控制？死循环怎么检测？AI 输出被截断怎么办？接口超时怎么重试？本节基于 WaLiCode 项目的 streamingAgent.ts 真实实现，讲解生产级 Agent 主循环的六大工程机制。
5.13.1 双模式设计：Chat vs Agent
WaLiCode 的 Agent 主循环支持两种模式，用对比表格展示差异： | 维度 | Chat 模式 | Agent 模式 | | --- | --- | --- | | 交互方式 | 一问一答，不自动续写 | 自动多轮，直到任务完成 | | 最大轮次 | 1 轮 | 20 轮（可配置） | | 工具调用 | 每轮最多 5 次 | 每轮最多 5 次，总计最多 200 次 | | 重试次数 | 3 次 | 5 次（更宽容） | | 适用场景 | 简单问答、闲聊 | 编码任务、文件操作、复杂工作流 | | Token 预算 | 无全局限制 | 200k token 全局预算 | 为什么需要两种模式？如果用 Agent 模式处理简单问答，AI 可能会"过度思考"——明明一句话能回答的问题，它非要调几个工具验证一下，浪费 token 和时间。反过来，用 Chat 模式处理编码任务，AI 无法自动连续执行多步操作，用户体验很差。
// 伪代码：双模式入口
interface AgentLoopConfig {
mode: 'chat' | 'agent';
maxRounds: number; // chat: 1, agent: 20
maxToolCallsPerRound: number; // 默认 5
maxTotalToolCalls: number; // agent: 200
maxRetries: number; // chat: 3, agent: 5
tokenBudget: number; // agent: 200000
}
async function runAgentLoop(
messages: Message[],
config: AgentLoopConfig,
callbacks: StreamingCallbacks
) {
let round = 0;
let totalToolCalls = 0;
let totalTokens = 0;
while (round = GLOBAL_TOKEN_BUDGET) {
callbacks.onText('\n⚠️ 已达到全局 Token 预算限制，自动停止。');
break;
}
// 流式接收 AI 输出
for await (const event of ai.stream(currentMessages)) {
if (event.type === 'text') {
const deltaTokens = countTokens(event.content);
totalTokensAcrossRounds += deltaTokens;
// ... 实时累加
}
}
round++;
}
// // 伪代码：Token Budget 管理
// const GLOBAL_TOKEN_BUDGET = 200_000; // 20万 token 全局预算
// let totalTokensAcrossRounds = 0;
// while (round = GLOBAL_TOKEN_BUDGET) {
// callbacks.onText('\n⚠️ 已达到全局 Token 预算限制，自动停止。');
// break;
// }
// // 流式接收 AI 输出
// for await (const event of ai.stream(currentMessages)) {
// if (event.type === 'text') {
// const deltaTokens = countTokens(event.content);
// totalTokensAcrossRounds += deltaTokens;
// // ... 实时累加
// }
// }
// round++;
// }
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// Python: // 伪代码：Token Budget 管理
// Python: const GLOBAL_TOKEN_BUDGET = 200_000; // 20万 token 全局预算
// Python: let totalTokensAcrossRounds = 0;
// Python: while (round = GLOBAL_TOKEN_BUDGET) {
// Python: callbacks.onText('\n⚠️ 已达到全局 Token 预算限制，自动停止。');
// Python: break;
// Python: }
// Python: // 流式接收 AI 输出
// Python: for await (const event of ai.stream(currentMessages)) {
// Python: if (event.type === 'text') {
// Python: const deltaTokens = countTokens(event.content);
// Python: totalTokensAcrossRounds += deltaTokens;
// Python: // ... 实时累加
// Python: }
// Python: }
// Python: round++;
// Python: }
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// Python: // 伪代码：Token Budget 管理
// Python: const GLOBAL_TOKEN_BUDGET = 200_000; // 20万 token 全局预算
// Python: let totalTokensAcrossRounds = 0;
// Python: while (round = GLOBAL_TOKEN_BUDGET) {
// Python: callbacks.onText('\n⚠️ 已达到全局 Token 预算限制，自动停止。');
// Python: break;
// Python: }
// Python: // 流式接收 AI 输出
// Python: for await (const event of ai.stream(currentMessages)) {
// Python: if (event.type === 'text') {
// Python: const deltaTokens = countTokens(event.content);
// Python: totalTokensAcrossRounds += deltaTokens;
// Python: // ... 实时累加
// Python: }
// Python: }
// Python: round++;
// Python: }
}
关键设计点：
- 全局预算：不是单轮限制，而是整个会话的总量限制。20 万 token 约等于 15 万字中文，足够完成大部分编码任务
- 实时累加：每收到一段流式输出就立即计数，不等一轮结束才统计。这样即使单轮输出就超预算也能及时停止
- 优雅停止：超预算时不直接中断，而是注入提示让 AI 自然收尾，避免输出半截代码
5.13.3 死循环保护（Diminishing Returns Detection）
Agent 模式下 AI 可能陷入死循环——反复调用同一个工具、传完全相同的参数、得到完全相同的错误结果。这不罕见：AI 看到 FileNotFoundError，于是去 read_file，文件确实不存在，于是又去 write_file，写错了路径，又看到 FileNotFoundError……
WaLiCode 的检测策略：连续 2 轮执行完全相同的失败工具调用时自动终止。
// 伪代码：死循环检测
let lastRoundToolCalls: string[] = [];
let lastRoundErrors: string[] = [];
while (round 0
&& currentRoundErrors.every(e => lastRoundErrors.includes(e))
&& currentRoundErrors.length === lastRoundErrors.length;
if (sameErrors) {
callbacks.onText('\n⚠️ 检测到重复失败，自动停止以避免死循环。');
break;
}
lastRoundToolCalls = currentRoundToolCalls;
lastRoundErrors = currentRoundErrors;
round++;
}
为什么阈值设为 2 而不是 1？因为有些问题需要重试一次才能确认——比如网络抖动导致的临时失败，第二次重试可能就成功了。设为 2 是在"尽早发现死循环"和"允许合理重试"之间的平衡。
5.13.4 max_tokens 截断恢复
AI 模型有输出长度限制（如 4096 token）。当输出被截断时，finish_reason 会变成 'length' 而不是 'stop'。如果直接忽略，用户会看到半截代码或不完整的回答。WaLiCode 的方案是注入 "please continue" 自动恢复：
// 伪代码：截断恢复
for await (const event of ai.stream(currentMessages)) {
if (event.type === 'text') {
assistantContent += event.content;
}
if (event.type === 'finish') {
if (event.finish_reason === 'length') {
// 输出被 max_tokens 截断
wasTruncated = true;
}
}
}
if (wasTruncated && round IDLE_TIMEOUT_MS) {
callbacks.onText('\n⚠️ 空闲超时，自动停止。');
break;
}
// 每次收到事件时更新
for await (const event of ai.stream(...)) {
lastActivityTime = Date.now();
// ...
}
// const IDLE_TIMEOUT_MS = 10 * 60 * 1000; // 10 分钟
// let lastActivityTime = Date.now();
// // 在主循环中检查
// if (Date.now() - lastActivityTime > IDLE_TIMEOUT_MS) {
// callbacks.onText('\n⚠️ 空闲超时，自动停止。');
// break;
// }
// // 每次收到事件时更新
// for await (const event of ai.stream(...)) {
const lastActivityTime = Date.now();;
// // ...
// }
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// Python: const IDLE_TIMEOUT_MS = 10 * 60 * 1000; // 10 分钟
// Python: let lastActivityTime = Date.now();
// Python: // 在主循环中检查
// Python: if (Date.now() - lastActivityTime > IDLE_TIMEOUT_MS) {
// Python: callbacks.onText('\n⚠️ 空闲超时，自动停止。');
// Python: break;
// Python: }
// Python: // 每次收到事件时更新
// Python: for await (const event of ai.stream(...)) {
// Python: lastActivityTime = Date.now();
// Python: // ...
// Python: }
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// Python: const IDLE_TIMEOUT_MS = 10 * 60 * 1000; // 10 分钟
// Python: let lastActivityTime = Date.now();
// Python: // 在主循环中检查
// Python: if (Date.now() - lastActivityTime > IDLE_TIMEOUT_MS) {
// Python: callbacks.onText('\n⚠️ 空闲超时，自动停止。');
// Python: break;
// Python: }
// Python: // 每次收到事件时更新
// Python: for await (const event of ai.stream(...)) {
// Python: lastActivityTime = Date.now();
// Python: // ...
// Python: }
}
413 反应式压缩：当 API 返回 413（请求体过大）时，紧急压缩上下文后重试。
try {
for await (const event of ai.stream(currentMessages)) {
// ...
}
} catch (err) {
if (err instanceof HTTPError && err.status === 413) {
// 请求体过大，紧急压缩上下文
const summary = await generateAiSummary(currentMessages);
currentMessages = [
{ role: 'system', content: '之前的对话已压缩：' + summary },
...recentMessages // 保留最近的几轮
];
// 用压缩后的上下文重试
continue;
}
throw err;
}
// try {
// for await (const event of ai.stream(currentMessages)) {
// // ...
// }
// } catch (err) {
// if (err instanceof HTTPError && err.status === 413) {
// // 请求体过大，紧急压缩上下文
// const summary = await generateAiSummary(currentMessages);
const currentMessages = [;
// { role: 'system', content: '之前的对话已压缩：' + summary },
// ...recentMessages // 保留最近的几轮
// ];
// // 用压缩后的上下文重试
// continue;
// }
// throw err;
// }
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// Python: try {
// Python: for await (const event of ai.stream(currentMessages)) {
// Python: // ...
// Python: }
// Python: } catch (err) {
// Python: if (err instanceof HTTPError && err.status === 413) {
// Python: // 请求体过大，紧急压缩上下文
// Python: const summary = await generateAiSummary(currentMessages);
// Python: currentMessages = [
// Python: { role: 'system', content: '之前的对话已压缩：' + summary },
// Python: ...recentMessages // 保留最近的几轮
// Python: ];
// Python: // 用压缩后的上下文重试
// Python: continue;
// Python: }
// Python: throw err;
// Python: }
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// Python: try {
// Python: for await (const event of ai.stream(currentMessages)) {
// Python: // ...
// Python: }
// Python: } catch (err) {
// Python: if (err instanceof HTTPError && err.status === 413) {
// Python: // 请求体过大，紧急压缩上下文
// Python: const summary = await generateAiSummary(currentMessages);
// Python: currentMessages = [
// Python: { role: 'system', content: '之前的对话已压缩：' + summary },
// Python: ...recentMessages // 保留最近的几轮
// Python: ];
// Python: // 用压缩后的上下文重试
// Python: continue;
// Python: }
// Python: throw err;
// Python: }
}
413 压缩与正常压缩的区别：
- 触发时机：正常压缩在 token 达到 80% 时主动触发；413 压缩在请求被拒绝后被动触发
- 压缩力度：正常压缩保留近期 3-5 轮；413 压缩可能只保留最近 1-2 轮，更激进
- 用户感知：正常压缩用户无感；413 压缩可能丢失重要上下文，影响回答质量 🔗 4.13 核心要点
双模式：Chat 简单高效，Agent 自动多轮，按场景选择避免浪费。
Token Budget：全局预算防止烧钱，实时累加及时停止。
死循环保护：连续 2 轮相同失败自动终止，阈值 2 平衡了灵敏度和容错。
截断恢复：max_tokens 截断时注入 continue 自动续写，有限次恢复避免无限循环。
指数退避：临时错误自动重试，2^n 秒退避避免雪崩，区分可重试和不可重试错误。
413 压缩：请求过大时紧急压缩重试，是正常压缩的兜底方案。 📋 八股总结 — 面试高频考点
Q1: 什么是 ReAct 模式？它解决了什么问题？
ReAct = Reasoning + Acting，让 LLM 交替进行**推理（Thought）、行动（Action）、观察（Observation）**的循环，直到得出最终答案。
解决的问题：纯推理模型（CoT）无法获取外部信息，容易幻觉；纯行动模型直接调用工具但缺乏规划。ReAct 将推理和行动结合，推理指导行动，行动反馈推理。
Q2: ReAct 循环中 Thought、Action、Observation 各自的作用？
Thought（推理）：LLM 的"内心独白"，分析当前状态，决定下一步该做什么。是推理过程的核心，也是可审计性的基础。
Action（行动）：基于 Thought 的决策，调用具体工具。格式为 Action: tool_name(input)。由 Agent 框架解析和执行。
Observation（观察）：工具执行后返回的结果。注意：这不是 LLM 生成的，而是框架执行工具后注入到对话中的。LLM 基于 Observation 进行下一轮 Thought。
Q3: ReAct 的 Prompt 是怎么设计的？为什么需要强制格式？
Prompt 通过 "Thought:" "Action:" "Observation:" 标记强制结构化输出。
需要强制格式的原因：Agent 框架需要从 LLM 输出中解析出"要调用什么工具"和"传什么参数"。如果 LLM 输出自然语言，解析极其困难且不可靠。强制格式让输出可程序化解析，同时保持推理过程的可读性。
Q4: ReAct 模式有哪些局限性？后续如何改进？
局限：① Token 消耗线性增长（每步带完整历史）；② 严格串行无法并行；③ 格式解析不稳定；④ 可能陷入死循环。
改进方向：① ReWOO 先规划后执行减少token；② LLM Compiler 并行执行独立步骤；③ Reflexion 加入自我反思从失败学习；④ Structured Output 解决格式不稳定；⑤ 上下文压缩控制 token 消耗。
Q5: ReAct 中的 Observation 是 LLM 生成的吗？为什么？
不是。 Observation 是 Agent 框架执行工具后，将真实结果注入到对话历史中的。
这是最容易踩坑的点。如果让 LLM 自己"想象" Observation，就失去了与真实世界交互的能力，变成纯粹的推理游戏。ReAct 的核心价值就在于 Action 真实执行、Observation 真实反馈。这也是 ReAct 相比纯 CoT 的根本优势。
Q6: ReWOO 与 ReAct 的核心区别是什么？
ReAct 是边思考边执行（Thought→Action→Observation 循环），每步都需要 LLM 参与，token 消耗随步数线性增长。
ReWOO 是先规划后执行（Planner→Worker→Solver），LLM 只调用2次（规划+求解），Worker 执行工具不消耗 LLM token。
ReWOO 的优势是 token 消耗大幅减少；劣势是 Planner 无法根据中间结果调整策略，如果第一步规划错误，后续只能沿着错误路径走到底。
Q7: LLM Compiler 如何实现并行执行？它的适用条件是什么？
LLM Compiler 让 LLM 一次性输出多个工具调用，构建依赖图（DAG），识别没有数据依赖的步骤同时执行。
适用条件：工具调用之间确实没有依赖关系（如并行搜索多个城市天气）。如果有依赖（如先搜结果再基于结果做计算），则依赖步骤必须等前置步骤完成。并行加速效果取决于独立步骤的占比。
Q8: Structured Output 相比文本解析（ReAct的 "Action:" 标记）有什么本质优势？
文本解析：LLM 输出 "Action: search("query")"，用正则解析。不稳定——LLM 可能输出格式偏差（少冒号、多空格、换行位置不同），导致解析失败。
Structured Output：通过 JSON Schema + Pydantic 约束，LLM 输出严格符合预定义结构。字段名、类型、枚举值、范围全部锁定，100% 可靠解析，无需正则。
本质区别：文本解析是"尽力而为"，Structured Output 是"强制保证"。
Q9: 上下文压缩的三种策略分别是什么？语义保护型压缩的核心思想？
截断：直接丢弃最早的消息，简单但可能丢失关键信息。
摘要：用 LLM 将早期对话压缩成摘要，保留语义但消耗额外调用。
语义保护型压缩：将消息分为三层——锁定层（System Prompt + 原始问题，不压缩）、压缩层（中间步骤，摘要压缩）、保留层（最近K步，原样保留）。核心思想是确保关键指令永远不丢失，只压缩可牺牲的中间推理过程。
第5章 ReAct-让Agent学会思考
http://www.clxhxhhr.top/posts/708/ 评论
0 条
还没有评论，先写一条吧。