主题
原文：https://www.mstx.cn/harness/437.html · 创建：2026-09-08 · 更新：2026-09-08 · 阅读：1063
一个值得深思的现象
在6.1和6.2节中，你构建了ReAct模式和规划能力。Agent现在能够：
- 在每一步显式推理
- 在开始前制定计划
- 按计划推进执行
但有一个问题仍然存在：Agent会坚持错误的方向。
考虑这个场景：
用户：帮我查一下我们公司上季度的销售数据，并生成一份总结报告
Agent制定计划：
1. 查询销售数据库
2. 分析数据趋势
3. 生成总结报告
执行：
步骤1：调用query_sales_data(quarter="Q2") → 返回空
步骤2：调用query_sales_data(quarter="第二季度") → 仍然返回空
步骤3：调用query_sales_data(quarter="2季度") → 还是空
步骤4：调用query_sales_data(quarter="last quarter") → 继续空
...
Agent陷入了一个循环：不断尝试不同的参数格式，但没有停下来思考"也许问题不在参数格式，而在于数据库本身没有数据"或者"也许我应该换一个工具"。这个场景暴露了ReAct和规划的共同局限：它们关注"下一步做什么"，但不关注"我是不是走错了路"。
反思机制解决的核心问题：让Agent定期停下来，评估自己的执行轨迹，判断是否需要调整策略。
反思是什么
反思（Reflection）是Agent在执行过程中或执行后对自己行为的元认知评估。它回答三个问题：
- 我做了什么？——回顾已执行的操作
- 效果如何？——评估这些操作是否有效
- 需要改变什么？——判断是否需要调整策略
人类在解决复杂问题时经常使用反思。当你尝试了几种方法都失败后，你会停下来想："也许我应该换一个思路。"Agent也需要这种能力。
反思的两个时机
反思可以在两个时机发生：
时机一：执行中反思（在线反思）
在Agent循环的特定节点触发反思。比如：
- 连续3次工具调用失败后
- 同一个工具被重复调用超过3次时
- 计划执行到一半时
- Agent的循环达到预设轮次时
时机二：执行后反思（离线反思）
在Agent完成任务后，对整个执行过程进行回顾。这种反思的结果不用于当前任务，而是用于改进未来的执行。比如：
- 这次执行中哪些步骤是多余的？
- 哪些工具调用是无效的？
- 计划中有哪些步骤是可以合并的？
实现在线反思
反思的触发条件
python
from enum import Enum
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
import json
import time
class ReflectionTrigger(Enum):
"""反思的触发条件"""
CONSECUTIVE_FAILURES = "consecutive_failures" # 连续失败
REPEATED_TOOL_CALL = "repeated_tool_call" # 重复调用同一工具
MID_PLAN = "mid_plan" # 计划执行到一半
MAX_ITERATIONS_APPROACHING = "max_iterations_approaching" # 接近最大迭代
USER_FEEDBACK_NEGATIVE = "user_feedback_negative" # 用户负面反馈
class ReflectionManager:
"""反思管理器：决定何时触发反思"""
def __init__(self):
self.triggers = {
ReflectionTrigger.CONSECUTIVE_FAILURES: 3, # 连续3次失败触发
ReflectionTrigger.REPEATED_TOOL_CALL: 3, # 同一工具重复3次触发
ReflectionTrigger.MID_PLAN: True, # 计划过半时触发
ReflectionTrigger.MAX_ITERATIONS_APPROACHING: 2, # 剩2轮时触发
}
self.consecutive_failures = 0
self.tool_call_history: List[str] = []
self.reflection_count = 0
self.max_reflections = 2 # 最多反思2次，避免过度反思
def should_reflect(self, context: Dict[str, Any]) -> bool:
"""判断当前是否应该触发反思"""
if self.reflection_count >= self.max_reflections:
return False
# 检查连续失败
if context.get("last_tool_success") is False:
self.consecutive_failures += 1
else:
self.consecutive_failures = 0
if self.consecutive_failures >= self.triggers[ReflectionTrigger.CONSECUTIVE_FAILURES]:
self.reflection_count += 1
self.consecutive_failures = 0
return True
# 检查重复调用
tool_name = context.get("tool_name")
if tool_name:
self.tool_call_history.append(tool_name)
recent = self.tool_call_history[-self.triggers[ReflectionTrigger.REPEATED_TOOL_CALL]:]
if len(recent) >= self.triggers[ReflectionTrigger.REPEATED_TOOL_CALL]:
if len(set(recent)) == 1: # 全是同一个工具
self.reflection_count += 1
self.tool_call_history.clear()
return True
# 检查是否接近最大迭代
if context.get("iterations_remaining", 999) <= self.triggers[ReflectionTrigger.MAX_ITERATIONS_APPROACHING]:
self.reflection_count += 1
return True
return False
def reset(self):
"""重置反思状态"""
self.consecutive_failures = 0
self.tool_call_history.clear()
self.reflection_count = 0反思提示词
触发反思后，需要向LLM注入一个反思指令，要求它评估当前状态并决定是否需要调整策略：
python
REFLECTION_PROMPT = """
[反思时刻]
在继续执行前，请停下来回顾你的执行轨迹，回答以下问题：
1. 到目前为止你做了什么？
2. 你的方法有效吗？
3. 是否出现了反复尝试但无效的情况？
4. 你是否陷入了某种循环或错误的方向？
如果发现当前策略无效，请：
- 说明问题出在哪里
- 提出一个不同的策略
- 然后继续执行
如果当前策略有效，直接继续执行。保持简洁。
"""集成到Runtime中
python
class ReflectiveRuntime:
"""带反思能力的Agent运行时"""
def __init__(self, llm_client, tool_registry, memory,
max_iterations: int = 15,
verbose: bool = True):
self.client = llm_client
self.tools = tool_registry
self.memory = memory
self.max_iterations = max_iterations
self.verbose = verbose
self.reflection_manager = ReflectionManager()
self.execution_history = []
def run(self, user_input: str) -> str:
self.memory.initialize(user_input)
self.reflection_manager.reset()
for iteration in range(self.max_iterations):
if self.verbose:
print(f"\n--- 第{iteration + 1}轮 ---")
# 检查是否需要反思
remaining = self.max_iterations - iteration - 1
context = {
"iterations_remaining": remaining,
"last_tool_success": self._get_last_tool_success(),
"tool_name": self._get_last_tool_name(),
}
if self.reflection_manager.should_reflect(context):
if self.verbose:
print(" [触发反思]")
# 注入反思提示
self.memory.add_system_message(REFLECTION_PROMPT)
# 调用LLM
response = self._call_llm()
message = response.choices[0].message
if message.tool_calls:
self.memory.add_llm_message(message)
for tool_call in message.tool_calls:
name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
result = self.tools.execute(name, args)
# 记录执行历史
self.execution_history.append({
"iteration": iteration,
"tool": name,
"arguments": args,
"success": result.success,
"result": result.content[:100],
})
# 更新反思管理器的状态
if result.success:
self.reflection_manager.consecutive_failures = 0
else:
self.reflection_manager.consecutive_failures += 1
self.memory.add_tool_result(result)
else:
final_answer = message.content or ""
return final_answer
return "达到最大迭代次数。"
def _get_last_tool_success(self) -> bool:
if not self.execution_history:
return True
return self.execution_history[-1].get("success", True)
def _get_last_tool_name(self) -> Optional[str]:
if not self.execution_history:
return None
return self.execution_history[-1].get("tool")
def _call_llm(self):
return self.client.chat.completions.create(
model="gpt-4o-mini",
messages=self.memory.get_context(),
tools=self.tools.get_schemas(),
temperature=0.0,
)运行示例：反思如何打破循环
用户：帮我查一下上季度的销售数据
--- 第1轮 ---
执行：query_sales_data({"quarter": "Q2"})
观察：返回空
--- 第2轮 ---
执行：query_sales_data({"quarter": "第二季度"})
观察：返回空
--- 第3轮 ---
执行：query_sales_data({"quarter": "2季度"})
观察：返回空
--- 第4轮 ---
[触发反思]
[反思时刻]
思考：我已经连续3次尝试用不同的参数格式查询销售数据，但每次都返回空。
这可能不是参数格式的问题，而是数据本身不存在或存储在不同的位置。
我应该换一个策略：先检查数据库中有什么数据，或者询问用户确认数据源。
执行：list_available_datasets({})
观察：数据库中有以下数据集：产品数据、用户数据、营销数据...
思考：没有"销售数据"这个数据集。但有"营销数据"，可能包含销售相关信息。
我应该询问用户确认是否查"营销数据"。
回答：数据库中未找到"销售数据"数据集。可用的数据集包括：产品数据、用户数据、营销数据。
请问您指的是"营销数据"吗？实现离线反思
离线反思发生在任务完成后，目标是从经验中学习。
python
class OfflineReflector:
"""离线反思器：在任务完成后分析执行质量"""
def reflect(self, user_input: str, execution_history: List[Dict[str, Any]],
final_result: str) -> Dict[str, Any]:
"""生成反思报告"""
report = {
"task": user_input,
"total_tool_calls": len(execution_history),
"successful_calls": sum(1 for h in execution_history if h.get("success")),
"failed_calls": sum(1 for h in execution_history if not h.get("success")),
"unique_tools_used": len(set(h.get("tool") for h in execution_history)),
"repeated_calls": self._count_repeated_calls(execution_history),
"suggestions": [],
}
# 分析失败
failures = [h for h in execution_history if not h.get("success")]
if failures:
report["suggestions"].append(
f"有{len(failures)}次工具调用失败。考虑为这些工具添加参数预处理或更详细的错误提示。"
)
# 分析重复调用
repeated = report["repeated_calls"]
if repeated:
report["suggestions"].append(
f"检测到重复调用模式。考虑添加缓存机制避免重复查询相同数据。"
)
# 分析效率
if len(execution_history) > 5:
report["suggestions"].append(
f"任务需要{len(execution_history)}次工具调用。考虑是否有更高效的执行路径。"
)
return report
def _count_repeated_calls(self, history: List[Dict[str, Any]]) -> int:
"""统计重复调用"""
from collections import Counter
tool_counts = Counter(h.get("tool") for h in history)
return sum(count - 1 for count in tool_counts.values() if count > 1)离线反思的产出可以用于：
- 改进系统提示词（发现LLM常见的误解模式）
- 优化工具描述（发现LLM总是用错某个参数）
- 调整计划策略（发现某些步骤总是需要重做）
- 更新评估集（将"坏案例"加入回归测试）
反思的局限性
局限一：反思本身也可能出错。
LLM的反思是基于它对自己执行轨迹的"理解"。如果它的理解有偏差（比如错误地认为"数据库没有数据"而实际上只是查询语句有问题），反思会引导它走向错误的方向。
应对：反思不是"最终判断"，而是"调整建议"。反思后仍然需要工具结果来验证新策略是否有效。
局限二：过度反思会消耗资源。
每次反思都需要额外的LLM调用和token消耗。如果触发太频繁，成本会显著增加。
应对：设置反思次数上限（如最多2次），只在明确的触发条件下反思。
局限三：反思可能导致过度保守。
过度反思的Agent可能变得"犹豫不决"——频繁改变方向，无法坚持执行一个计划。
应对：反思应该关注"策略层面"的调整，而非"步骤层面"的微调。只有在明显的失败模式下才触发反思。
反思与ReAct、规划的关系
三者不是互斥的，而是递进的：
ReAct（单步推理）： 思考→行动→观察
↓ 增加全局视野
规划（多步计划）： 计划→执行→跟踪
↓ 增加元认知
反思（自我评估）： ...→停下来→评估→调整- ReAct回答"这一步怎么做"
- 规划回答"整体怎么做"
- 反思回答"我做得对吗？需要改吗"
一个完整的Agent应该在ReAct的每一轮中推理，在任务开始时规划，在执行中定期反思。三者结合，Agent才真正具备"聪明的执行力"。
本节小结
反思机制让Agent具备了元认知能力——不仅能执行任务，还能评估自己的执行质量并调整策略。本节实现了：
- 反思触发条件：连续失败、重复调用、接近迭代上限
- 在线反思：执行中停下来评估，必要时改变策略
- 离线反思：任务完成后分析执行质量，提取改进建议
- 与ReAct和规划的结合：三层递进的智能执行架构
核心原则：反思的价值不在于"找到完美的答案"，而在于识别失败模式并跳出思维定式。当Agent陷入"反复尝试同样的事情却期待不同结果"时，反思是它唯一的出路。
在下一节中，我们将讨论多Agent协作——当单个Agent的能力不足以完成任务时，如何让多个Agent分工合作。