第6章-推理控制-CoT ReAct与自主推理策略-《Agentic AI 智能体应用开发》
·
第6章：推理控制 —— CoT、ReAct 与自主推理策略
本章导引
如果说 Context Manager 决定 Agent “看到什么”，Tool Executor 决定 Agent “能做什么”，那么 Reasoning Controller 决定 Agent “怎么想”。它是 Agent 的元认知层——控制推理策略、管理迭代循环、保障推理质量。
本章的核心问题是：如何让 Agent 高效地、可靠地、可控制地完成多步推理任务？ 我们将从基础策略（CoT、ReAct、Plan-Execute）起步，深入高级技术（ToT、Self-Consistency、Reflection），最终构建一个生产级的推理控制系统。
6.1 推理策略基础
6.1.1 Chain-of-Thought (CoT)：让模型"想清楚再说"
Chain-of-Thought 是最基础也最广泛使用的推理增强技术。核心思想极其简单：让模型在给出最终答案之前，先生成中间推理步骤。
CoT 的本质
不经推理的回答：
Q: 一个长方形长8米宽6米，面积是多少？
A: 48平方米。
经过 CoT 推理的回答：
Q: 一个长方形长8米宽6米，面积是多少？
A: 让我一步步思考：
1. 长方形面积公式 = 长 × 宽
2. 长 = 8米, 宽 = 6米
3. 面积 = 8 × 6 = 48
所以，答案是48平方米。
CoT 之所以有效，是因为它将复杂问题分解为子问题，每个子问题都比原问题更简单，且中间结果可用于后续推理。这本质上是在模拟人类的"慢慢想"过程。
CoT 的变体
Zero-shot CoT：只加一句 “Let’s think step by step”（让我们逐步思考），不需要任何示例。
Few-shot CoT：在 Prompt 中给出几个带推理过程的示例。
Auto-CoT：自动从问题库中检索相似问题及其推理过程作为 Few-shot 示例。
# Python: Auto-CoT 实现
class AutoCoT:
def __init__(self, vector_store):
self.vector_store = vector_store
self.example_store = vector_store # 存储历史成功推理
async def build_prompt(self, question: str, k: int = 3) -> str:
# 1. 检索最相似的 k 个成功推理案例
similar = await self.example_store.search(question, top_k=k)
# 2. 构建 Few-shot CoT Prompt
prompt = "以下是几个带推理过程的示例：\n\n"
for i, example in enumerate(similar):
prompt += f"示例 {i+1}:\n"
prompt += f"问题: {example['question']}\n"
prompt += f"推理过程: {example['reasoning']}\n"
prompt += f"答案: {example['answer']}\n\n"
prompt += f"现在请回答问题，并展示你的推理过程：\n{question}\n"
return prompt
6.1.2 ReAct 模式：Reasoning + Acting 的协同循环
ReAct (Reasoning + Acting) 是 Agent 推理策略的真正核心。它将"思考"和"行动"交织在一起：思考产生行动，行动结果反馈给思考。
ReAct 循环模式：
Thought: 我需要了解问题的更多信息...
Action: search_code("authentication middleware")
Observation: [找到 3 个相关文件: auth.ts, middleware.ts, jwt.ts]
Thought: auth.ts 看起来是入口。我需要读取它。
Action: read_file("src/auth.ts")
Observation: [文件内容...使用了 JWT + bcrypt]
Thought: 认证使用了 JWT，没有明显问题。但需要检查 token 刷新逻辑...
Action: search_code("refreshToken")
Observation: [在 jwt.ts 第 45 行发现 refreshToken 未设置过期时间]
Thought: 这就是安全问题！refreshToken 永不过期是一个安全漏洞。
Answer: 发现安全问题：jwt.ts 第 45 行的 refreshToken 没有设置过期时间...
ReAct 策略的完整实现
// TypeScript: ReAct 推理策略
class ReActStrategy implements ReasoningStrategy {
name = 'react';
systemInstruction = `
使用 ReAct (Reasoning + Acting) 模式解决问题：
循环格式：
Thought: [你对当前状态的分析和下一步计划]
Action: [需要调用的工具名称]
Action Input: [工具参数的 JSON]
Observation: [工具执行的结果]
... (重复 Thought → Action → Observation)
Thought: [基于所有观察的最终分析]
Final Answer: [最终答案]
规则：
1. 每次只执行一个 Action
2. Action 之后必须等待 Observation 再继续
3. 如果连续 3 次 Action 结果相同，说明陷入循环，改变策略
4. 当你确信有了足够的信息时，给出 Final Answer
`;
/** 解析 Model 的 ReAct 输出 */
parseOutput(output: string): ParsedReAct {
const result: ParsedReAct = {
thoughts: [],
actions: [],
finalAnswer: null,
};
const lines = output.split('\n');
let currentSection: 'thought' | 'action' | 'action_input' | null = null;
let currentThought = '';
let currentAction: { name: string; input: string } | null = null;
for (const line of lines) {
if (line.startsWith('Thought:')) {
if (currentThought) result.thoughts.push(currentThought.trim());
currentThought = line.replace('Thought:', '').trim();
currentSection = 'thought';
} else if (line.startsWith('Action:')) {
if (currentThought) result.thoughts.push(currentThought.trim());
currentThought = '';
currentAction = { name: line.replace('Action:', '').trim(), input: '' };
currentSection = 'action';
} else if (line.startsWith('Action Input:')) {
if (currentAction) {
currentAction.input = line.replace('Action Input:', '').trim();
}
} else if (line.startsWith('Final Answer:')) {
result.finalAnswer = line.replace('Final Answer:', '').trim();
break;
} else if (currentSection === 'thought') {
currentThought += ' ' + line.trim();
}
}
return result;
}
/** 格式化 Observation 注入上下文 */
formatObservation(action: string, result: ToolResult): string {
return `Observation: ${result.content}`;
}
}
# Python: ReAct 推理引擎
class ReActEngine:
def __init__(self, model: ModelProvider, tools: ToolExecutor, max_steps: int = 15):
self.model = model
self.tools = tools
self.max_steps = max_steps
self.strategy = ReActStrategy()
async def run(self, task: str, context: dict = None) -> ReActResult:
scratchpad = []
history = [{"role": "user", "content": task}]
for step in range(self.max_steps):
# 1. 构建包含历史推理的上下文
prompt = self._build_prompt(task, scratchpad, step)
# 2. Model 推理
response = await self.model.reason(ModelInput(
system_prompt=self.strategy.system_instruction,
messages=[{"role": "user", "content": prompt}],
tools=self.tools.get_definitions(),
))
# 3. 解析输出
parsed = self.strategy.parse_output(response.text_content)
scratchpad.append({
"step": step,
"thought": parsed.thoughts[-1] if parsed.thoughts else "",
"action": parsed.actions[-1] if parsed.actions else None,
})
# 4. 如果有 Final Answer，返回
if parsed.final_answer:
return ReActResult(
answer=parsed.final_answer,
steps=step + 1,
trace=scratchpad,
success=True,
)
# 5. 如果有 Action，执行
if parsed.actions:
action = parsed.actions[-1]
result = await self.tools.execute(ToolCall(
id=f"step_{step}",
name=action["name"],
arguments=action["input"],
))
scratchpad[-1]["observation"] = result.content
# 检测循环：连续3步相同 Action
if self._detect_loop(scratchpad):
break
# 6. 没有 Action 也没有 Final Answer → 强制结束
else:
break
return ReActResult(
answer="无法在最大步数内完成任务",
steps=step + 1,
trace=scratchpad,
success=False,
)
def _detect_loop(self, scratchpad: list) -> bool:
if len(scratchpad) < 3:
return False
recent = scratchpad[-3:]
actions = [s.get("action", {}).get("name") for s in recent]
return len(set(actions)) == 1 # 连续3步同一工具
def _build_prompt(self, task: str, scratchpad: list, step: int) -> str:
prompt = f"任务: {task}\n\n"
if scratchpad:
prompt += "之前的推理:\n"
for s in scratchpad[-5:]: # 最近 5 步
prompt += f"Step {s['step']+1}:\n"
prompt += f"Thought: {s['thought']}\n"
if s.get('action'):
prompt += f"Action: {s['action']['name']}\n"
if s.get('observation'):
prompt += f"Observation: {s['observation'][:500]}\n"
prompt += "\n"
prompt += f"Step {step+1}: 请继续推理，使用 Thought/Action 或给出 Final Answer"
return prompt
6.1.3 Plan-and-Execute：先规划再执行
与 ReAct 的"边想边做"不同，Plan-and-Execute 采用"先完整规划，再逐步执行"的策略。它更接近人类的项目管理方式。
Plan-and-Execute 流程：
Phase 1: PLAN（规划阶段）
Planner Agent 分解任务为子任务序列：
Task: "为新项目搭建 CI/CD 流水线"
Plan:
1. 分析项目结构和技术栈
2. 确定 CI/CD 平台选型
3. 编写构建配置（Dockerfile）
4. 编写流水线配置（.github/workflows/*.yml）
5. 添加代码质量检查（lint + test）
6. 配置自动部署
7. 测试流水线是否正常工作
Phase 2: EXECUTE（执行阶段）
Executor Agent 逐步执行每个子任务：
Step 1: ✓ 分析完成，技术栈: Node.js + TypeScript + Docker
Step 2: ✓ 选择 GitHub Actions
Step 3: ✓ Dockerfile 已创建
Step 4: ✓ workflow 配置已创建
...
Phase 3: VERIFY（验证阶段）
验证最终结果是否符合预期
# Python: Plan-and-Execute 引擎
class PlanAndExecuteEngine:
def __init__(self, planner_model, executor_model):
self.planner = planner_model # 规划用强模型
self.executor = executor_model # 执行可用较弱模型（降成本）
async def run(self, task: str) -> PEResult:
# Phase 1: PLAN
plan = await self._plan(task)
print(f"计划: {len(plan.steps)} 步")
# Phase 2: EXECUTE
results = []
context = {"task": task, "previous_results": []}
for i, step in enumerate(plan.steps):
print(f"执行 {i+1}/{len(plan.steps)}: {step.description[:80]}")
result = await self._execute_step(step, context)
results.append(result)
context["previous_results"].append({
"step": i + 1,
"description": step.description,
"status": result.status,
"output": result.output,
})
# 某步失败时评估是否需要重新规划
if result.status == "failed" and step.critical:
print(f"关键步骤失败，重新规划剩余步骤...")
remaining = plan.steps[i+1:]
new_plan = await self._replan(task, context, remaining)
plan.steps = plan.steps[:i+1] + new_plan.steps
# Phase 3: VERIFY
verification = await self._verify(task, results, context)
return PEResult(plan=plan, results=results, verification=verification)
async def _plan(self, task: str) -> Plan:
prompt = f"""请将以下任务分解为可执行的子任务序列。每个子任务应该具体、可验证、有明确产出。
任务: {task}
输出格式 (JSON):
{{
"steps": [
{{
"id": 1,
"description": "子任务描述",
"expected_output": "预期产出",
"dependencies": [],
"critical": true,
"estimated_tools": ["需要使用的工具"]
}}
]
}}
要求:
- 每个子任务应该尽可能独立
- 标注子任务间的依赖关系
- 标注哪些是关键步骤（失败则任务失败）
- 总步数控制在 3-10 步"""
response = await self.planner.reason(ModelInput(
system_prompt="你是一个专业的任务规划专家。",
messages=[{"role": "user", "content": prompt}],
))
return Plan.parse_raw(response.text_content)
6.1.4 各策略对比：准确性、延迟、成本综合评估
| 维度 | CoT | ReAct | Plan-Execute |
|---|---|---|---|
| 推理模式 | 先想后答 | 边想边做 | 先规划后执行 |
| 准确性 | 中（适合推理题） | 高（有工具辅助） | 最高（结构化管理） |
| 延迟 | 低（单次调用） | 中（多轮迭代） | 高（规划+执行+验证） |
| 成本 | 低 | 中 | 高 |
| Token 消耗 | 低 | 中（多轮对话） | 高（规划 Token + 执行 Token） |
| 工具使用 | 不支持 | 完整支持 | 完整支持 |
| 错误恢复 | 弱（一次失败全完） | 强（可重试） | 最强（可重新规划） |
| 适合场景 | 简单推理、数学题 | 中等复杂度的多工具任务 | 复杂多步任务、大型项目 |
| 不适合场景 | 需要外部信息 | 简单的一步问答 | 简单任务（过度设计） |
选型建议：
任务复杂度
▲
│ Plan-Execute ← 复杂多步任务（>5步，多工具）
│ "搭建CI/CD流水线"
│
│ ReAct ← 中等复杂度（2-5步，2-4个工具）
│ "分析并修复这个Bug"
│
│ CoT ← 简单推理（一步推理）
│ "分析这段代码的时间复杂度"
└────────────────────────────►
6.2 高级推理技术
6.2.1 Tree-of-Thought (ToT)：多路径探索与回溯
CoT 是线性的——一条路走到黑。Tree-of-Thought 则维护一棵推理树，在每一步探索多个可能的推理方向，评估每个方向的前景，选择最优的继续深入。
传统 CoT (线性): Tree-of-Thought (树状):
┌── 方案A-1 ── 评估: 0.7 ✓
思路 → 步骤1 ──► 思路 → 步骤2 方案A ──┼── 方案A-2 ── 评估: 0.3 ✗
──► 思路 → 步骤3 │
问题 ──┼── 方案B ── 评估: 0.9 ✓✓
│ └── 方案B-1 ── 最终答案
│
└── 方案C ── 评估: 0.4 ✗
# Python: Tree-of-Thought 实现
import heapq
from dataclasses import dataclass
@dataclass
class ThoughtNode:
id: str
content: str
parent: Optional['ThoughtNode']
children: list['ThoughtNode']
score: float # 评估分数（0-1）
depth: int
class TreeOfThought:
def __init__(self, model, beam_width: int = 3, max_depth: int = 5):
self.model = model
self.beam_width = beam_width # 每层保留最优的 N 个节点
self.max_depth = max_depth
async def solve(self, problem: str) -> ToTResult:
# 创建根节点
root = ThoughtNode(
id="root",
content=f"问题: {problem}",
parent=None,
children=[],
score=1.0,
depth=0,
)
# BFS + Beam Search
frontier = [root]
for depth in range(self.max_depth):
# 1. 为每个 frontier 节点生成下一步思路
candidates = []
for node in frontier:
next_thoughts = await self._generate_next_thoughts(node, problem)
for thought in next_thoughts:
child = ThoughtNode(
id=f"{node.id}.{len(node.children)}",
content=thought,
parent=node,
children=[],
score=0, # 稍后评估
depth=depth + 1,
)
node.children.append(child)
candidates.append(child)
if not candidates:
break
# 2. 评估所有候选节点
for candidate in candidates:
candidate.score = await self._evaluate_thought(candidate, problem)
# 3. Beam Search: 保留分数最高的 beam_width 个节点
frontier = sorted(candidates, key=lambda n: n.score, reverse=True)[:self.beam_width]
# 4. 检查是否有节点达到终止条件
for node in frontier:
if await self._is_solution(node, problem):
return ToTResult(solution=node, path=self._trace_path(node))
# 没有找到满意解，返回最优路径
best = max(frontier, key=lambda n: n.score)
return ToTResult(solution=best, path=self._trace_path(best))
async def _generate_next_thoughts(self, node: ThoughtNode, problem: str) -> list[str]:
"""生成当前节点的下一步推理方向（2-3个）"""
prompt = f"""基于以下推理路径，提出 {self.beam_width} 个不同的下一步推理方向：
问题: {problem}
当前推理路径:
{self._format_path(self._trace_path(node))}
请提出 {self.beam_width} 个不同的下一步推理方向。每个方向应该是有实质差异的，而非简单改写。
输出格式: 每行一个方向，以 "- " 开头。"""
response = await self.model.reason(ModelInput(
system_prompt="你是一个创意推理专家，擅长从多个角度思考问题。",
messages=[{"role": "user", "content": prompt}],
))
# 解析输出
directions = [
line.strip()[2:] for line in response.text_content.split('\n')
if line.strip().startswith('- ')
]
return directions[:self.beam_width]
async def _evaluate_thought(self, node: ThoughtNode, problem: str) -> float:
"""评估推理方向的可行性（0-1）"""
prompt = f"""评估以下推理方向对解决问题的帮助程度（0-1之间的分数）：
问题: {problem}
当前推理: {node.content}
评分标准:
- 1.0: 这很可能就是正确答案
- 0.7-0.9: 方向正确，有很大希望
- 0.4-0.6: 可能有用，但不确定
- 0.1-0.3: 可能偏离正确方向
- 0.0: 明显错误的方向
只输出一个 0-1 之间的数字。"""
response = await self.model.reason(ModelInput(
system_prompt="你是一个严格的推理评估专家。",
messages=[{"role": "user", "content": prompt}],
temperature=0.0,
))
try:
return float(response.text_content.strip())
except ValueError:
return 0.5
def _trace_path(self, node: ThoughtNode) -> list[ThoughtNode]:
path = []
current = node
while current:
path.append(current)
current = current.parent
return list(reversed(path))
6.2.2 Self-Consistency：多次采样 + 投票提升准确性
对于有明确答案的推理任务（数学题、逻辑题、代码 bug 定位），Self-Consistency 是一种简单但极其有效的技术：
- 对同一问题，用较高 temperature 生成 N 个推理链
- 每个推理链给出一个最终答案
- 投票选择出现最多的答案
# Python: Self-Consistency 实现
class SelfConsistency:
def __init__(self, model, samples: int = 5, temperature: float = 0.7):
self.model = model
self.samples = samples
self.temperature = temperature
async def solve(self, problem: str) -> SCCResult:
# 1. 生成 N 个推理链
chains = []
for i in range(self.samples):
response = await self.model.reason(ModelInput(
system_prompt="请逐步推理并给出最终答案。最终答案用 <<<答案>>> 包裹。",
messages=[{"role": "user", "content": problem}],
temperature=self.temperature, # 非零温度产生多样性
))
chains.append({
"index": i,
"reasoning": response.text_content,
"answer": self._extract_answer(response.text_content),
})
# 2. 统计答案分布
from collections import Counter
answer_counts = Counter(c["answer"] for c in chains if c["answer"])
# 3. 选择最常见的答案
most_common = answer_counts.most_common(1)[0]
# 4. 计算一致性分数
consistency = most_common[1] / self.samples
return SCCResult(
answer=most_common[0],
confidence=consistency,
vote_distribution=dict(answer_counts),
reasoning_chains=chains,
)
def _extract_answer(self, text: str) -> Optional[str]:
import re
match = re.search(r'<<<(.+?)>>>', text)
return match.group(1).strip() if match else None
6.2.3 Reflection 与 Self-Correction：Agent 的自我反思
Reflection 是让 Agent 审查自己的输出，发现错误并自我纠正的能力。这是 Agent 从 “好用” 到 “可靠” 的关键跃迁。
# Python: Reflection 模式
class ReflectionEngine:
def __init__(self, model, reflection_prompt: str = None):
self.model = model
self.reflection_prompt = reflection_prompt or """
请严格审查以下 Agent 的输出，检查：
1. **事实准确性**: 陈述的事实是否准确？是否有幻觉？
2. **逻辑一致性**: 推理是否自洽？是否有逻辑跳跃？
3. **完整性**: 是否遗漏了重要方面？
4. **安全性**: 是否包含危险建议或敏感信息？
5. **可执行性**: 如果给出了操作步骤，是否真的可执行？
请指出发现的问题，并给出修正后的版本。如果没有问题，直接回复 "NO_ISSUES"。
"""
async def run_with_reflection(self, task: str, max_rounds: int = 3) -> ReflectedResult:
# Round 1: 初始执行
response = await self.model.reason(ModelInput(
system_prompt="完成以下任务。",
messages=[{"role": "user", "content": task}],
))
best_output = response.text_content
reflection_history = []
for round_num in range(max_rounds):
# 自我反思
reflection = await self.model.reason(ModelInput(
system_prompt=self.reflection_prompt,
messages=[
{"role": "user", "content": f"原始任务: {task}\n\nAgent 的输出:\n{best_output}"},
],
))
reflection_text = reflection.text_content
reflection_history.append({
"round": round_num + 1,
"output": best_output,
"reflection": reflection_text,
})
# 如果没有问题，结束
if "NO_ISSUES" in reflection_text:
return ReflectedResult(
final_output=best_output,
rounds=round_num + 1,
history=reflection_history,
improved=(round_num > 0),
)
# 有问题：基于反思重新生成
improved = await self.model.reason(ModelInput(
system_prompt="根据以下反馈改进你的输出。",
messages=[
{"role": "user", "content": f"{task}\n\n上一次输出:\n{best_output}\n\n反馈:\n{reflection_text}\n\n请给出改进后的完整输出。"},
],
))
best_output = improved.text_content
return ReflectedResult(
final_output=best_output,
rounds=max_rounds,
history=reflection_history,
improved=True,
)
6.2.4 实践案例：可插拔的推理策略框架
// TypeScript: 可插拔推理策略框架
interface ReasoningStrategy {
name: string;
systemInstruction: string;
parseOutput(output: string): ParsedReasoning;
shouldContinue(state: AgentState): boolean;
}
class ReasoningStrategyManager {
private strategies: Map<string, ReasoningStrategy> = new Map();
private currentStrategy: string;
private fallbackStrategies: string[] = [];
register(strategy: ReasoningStrategy): void {
this.strategies.set(strategy.name, strategy);
}
/** 动态切换策略 */
switchTo(strategy: string): void {
if (!this.strategies.has(strategy)) {
throw new Error(`未知策略: ${strategy}`);
}
this.currentStrategy = strategy;
console.log(`[Reasoning] 切换策略: ${strategy}`);
}
/** 设置降级链 */
setFallbackChain(chain: string[]): void {
this.fallbackStrategies = chain;
}
/** 自动降级：当前策略连续失败时切换 */
async autoFallback(state: AgentState): Promise<ReasoningStrategy> {
const current = this.strategies.get(this.currentStrategy)!;
// 检测是否连续失败
if (state.consecutiveErrors >= 3) {
const currentIndex = this.fallbackStrategies.indexOf(this.currentStrategy);
if (currentIndex < this.fallbackStrategies.length - 1) {
this.switchTo(this.fallbackStrategies[currentIndex + 1]);
}
}
return this.strategies.get(this.currentStrategy)!;
}
getAllStrategies(): string[] {
return Array.from(this.strategies.keys());
}
}
// 注册策略
const manager = new ReasoningStrategyManager();
manager.register(new CoTStrategy());
manager.register(new ReActStrategy());
manager.register(new PlanExecuteStrategy());
// 配置降级链：PlanExecute → ReAct → CoT
// （越复杂的策略越容易在部分场景失败，降级到更简单的策略）
manager.setFallbackChain(['plan-execute', 'react', 'cot']);
6.3 推理质量保障
6.3.1 幻觉检测：事实校验、来源追溯、置信度评估
幻觉（Hallucination）是 LLM 最棘手的问题之一。在 Agent 场景中，幻觉可能导致错误的工具调用、错误的文件修改、错误的安全判断。
# Python: 幻觉检测器
class HallucinationDetector:
def __init__(self, model, fact_check_model=None):
self.model = model
self.fact_checker = fact_check_model or model
async def check(self, text: str, context: Optional[str] = None) -> HallucinationReport:
"""检测文本中的潜在幻觉"""
checks = []
# 1. 提取事实性陈述
claims = await self._extract_claims(text)
# 2. 逐条验证
for claim in claims:
result = await self._verify_claim(claim, context)
checks.append(result)
# 3. 整体评估
hallu_count = sum(1 for c in checks if not c.verified)
confidence = 1.0 - (hallu_count / max(len(checks), 1))
return HallucinationReport(
total_claims=len(checks),
hallucinated=hallu_count,
confidence=confidence,
claims=checks,
recommendation=self._recommend(confidence),
)
async def _extract_claims(self, text: str) -> list[str]:
"""提取文本中的事实性声明"""
prompt = f"""从以下文本中提取所有事实性声明（而非观点或建议）。
每行一个声明，只提取可以被客观验证的事实。
文本:
{text}
事实性声明:"""
response = await self.model.reason(ModelInput(
system_prompt="你是一个严格的事实核查员。",
messages=[{"role": "user", "content": prompt}],
))
return [line.strip() for line in response.text_content.split('\n') if line.strip()]
async def _verify_claim(self, claim: str, context: Optional[str]) -> ClaimCheck:
"""验证单条声明"""
prompt = f"""验证以下声明是否可以从提供的上下文中得到支持：
声明: "{claim}"
上下文:
{context or '无额外上下文。基于你的常识判断。'}
请以 JSON 格式回答:
{{
"verified": true/false,
"confidence": 0-1,
"reason": "验证或否定的理由"
}}"""
response = await self.fact_checker.reason(ModelInput(
system_prompt="你是一个事实核查专家。只基于给定信息做出判断。",
messages=[{"role": "user", "content": prompt}],
temperature=0.0,
))
try:
data = json.loads(response.text_content)
return ClaimCheck(
claim=claim,
verified=data["verified"],
confidence=data["confidence"],
reason=data["reason"],
)
except:
return ClaimCheck(claim=claim, verified=False, confidence=0.0, reason="无法解析验证结果")
def _recommend(self, confidence: float) -> str:
if confidence > 0.95: return "高度可信，可以直接使用"
if confidence > 0.8: return "基本可信，建议人工复核关键事实"
if confidence > 0.6: return "存在较多不确定内容，建议主要事实人工复核"
return "可信度低，不建议直接使用"
6.3.2 推理过程可视化：决策树、推理链追踪
# Python: 推理链可视化
class ReasoningVisualizer:
def generate_mermaid(self, trace: list[dict]) -> str:
"""生成 Mermaid 格式的推理流程图"""
mermaid = "```mermaid\ngraph TD\n"
for step in trace:
step_id = f"step{step['step']}"
# 思考节点
thought = step.get('thought', '')[:50]
mermaid += f' {step_id}_think["Thought: {thought}"]\n'
# 行动节点
if step.get('action'):
action_name = step['action'].get('name', 'unknown')
mermaid += f' {step_id}_act["Action: {action_name}"]\n'
mermaid += f' {step_id}_think --> {step_id}_act\n'
# 观察节点
if step.get('observation'):
obs = step['observation'][:50]
mermaid += f' {step_id}_obs["Obs: {obs}"]\n'
if step.get('action'):
mermaid += f' {step_id}_act --> {step_id}_obs\n'
# 连接到下一步
if step['step'] > 0:
prev_id = f"step{step['step']-1}"
last_prev_node = f"{prev_id}_obs" if trace[step['step']-1].get('observation') else f"{prev_id}_think"
mermaid += f' {last_prev_node} --> {step_id}_think\n'
mermaid += "```\n"
return mermaid
def format_trace_table(self, trace: list[dict]) -> str:
"""生成推理追踪表格"""
rows = []
for step in trace:
rows.append({
"步数": step['step'] + 1,
"思考": step.get('thought', '-')[:80],
"行动": step.get('action', {}).get('name', '-'),
"结果": step.get('observation', '-')[:80],
"耗时(ms)": step.get('duration_ms', '-'),
})
# 转为 Markdown 表格
if not rows:
return "无推理记录"
headers = list(rows[0].keys())
table = "| " + " | ".join(headers) + " |\n"
table += "| " + " | ".join(["---"] * len(headers)) + " |\n"
for row in rows:
table += "| " + " | ".join(str(row[h]) for h in headers) + " |\n"
return table
6.3.3 推理超时与熔断：防止无限推理循环
// TypeScript: 推理超时与熔断控制器
class ReasoningGuard {
private config: GuardConfig;
private state: GuardState;
constructor(config: GuardConfig) {
this.config = {
maxSteps: config.maxSteps ?? 50,
maxTimeMs: config.maxTimeMs ?? 300000, // 5 分钟
maxTokens: config.maxTokens ?? 200000, // 20万 tokens
loopDetectionWindow: config.loopDetectionWindow ?? 5,
loopThreshold: config.loopThreshold ?? 0.8, // 80% 相似即判定循环
};
this.state = {
startTime: Date.now(),
totalTokens: 0,
steps: [],
};
}
/** 每步推理前检查 */
check(step: number, context: string): GuardResult {
const now = Date.now();
// 检查1: 步数限制
if (step >= this.config.maxSteps) {
return {
allow: false,
reason: `达到最大步数限制 (${this.config.maxSteps})`,
action: 'force_stop',
};
}
// 检查2: 时间限制
const elapsed = now - this.state.startTime;
if (elapsed > this.config.maxTimeMs) {
return {
allow: false,
reason: `达到时间限制 (${this.config.maxTimeMs}ms)`,
action: 'force_stop',
};
}
if (elapsed > this.config.maxTimeMs * 0.8) {
return {
allow: true,
reason: `接近时间限制，还剩 ${this.config.maxTimeMs - elapsed}ms`,
action: 'warn',
};
}
// 检查3: Token 限制
if (this.state.totalTokens > this.config.maxTokens * 0.9) {
return {
allow: true,
reason: '接近 Token 限制，请尽快给出最终答案',
action: 'warn',
};
}
// 检查4: 推理循环检测
if (this.detectLoop()) {
return {
allow: false,
reason: '检测到推理循环，Agent 陷入了重复模式',
action: 'force_switch_strategy',
};
}
// 检查5: 进展检测
if (this.detectStagnation()) {
return {
allow: true,
reason: '推理似乎停滞，建议改变策略或给出当前最优答案',
action: 'suggest_switch',
};
}
return { allow: true, reason: '', action: 'continue' };
}
/** 记录当前步骤 */
recordStep(step: AgentStep): void {
this.state.steps.push(step);
this.state.totalTokens += step.tokensUsed;
}
private detectLoop(): boolean {
if (this.state.steps.length < this.config.loopDetectionWindow) {
return false;
}
const recent = this.state.steps.slice(-this.config.loopDetectionWindow);
// 检查最近几步的工具调用模式是否高度相似
const toolPatterns = recent.map(s =>
(s.toolCalls || []).map(t => t.name).join(',')
);
const uniquePatterns = new Set(toolPatterns);
// 如果所有步骤的模式都一样（或只有1种），判定为循环
return uniquePatterns.size <= 1;
}
private detectStagnation(): boolean {
if (this.state.steps.length < 8) return false;
const recent = this.state.steps.slice(-8);
// 检查最后 8 步是否都没有实质性进展（都是读取/搜索，没有写入/执行）
const productiveActions = recent.filter(s =>
(s.toolCalls || []).some(t =>
['write_file', 'run_command', 'execute'].includes(t.name)
)
);
return productiveActions.length === 0;
}
}
6.3.4 实践案例：带质量保障的生产级推理控制器
// TypeScript: 生产级推理控制器
class ProductionReasoningController {
private strategyManager: ReasoningStrategyManager;
private guard: ReasoningGuard;
private hallucinationDetector: HallucinationDetector;
private visualizer: ReasoningVisualizer;
async executeReasoning(
task: string,
context: AgentContext,
): Promise<ReasoningResult> {
const trace: AgentStep[] = [];
let strategy = this.strategyManager.getCurrent();
for (let step = 0; step < this.config.maxSteps; step++) {
// 1. 安全检查
const guardResult = this.guard.check(step, context);
if (guardResult.action === 'force_stop') {
return this.wrapUp(guardResult.reason, trace, step);
}
if (guardResult.action === 'force_switch_strategy') {
strategy = this.strategyManager.autoFallback(context);
console.warn(`[推理] 强制切换策略: ${strategy.name}`);
}
// 2. Model 推理
const startTime = Date.now();
const modelOutput = await this.model.reason({
...context,
systemPrompt: context.systemPrompt + '\n' + strategy.systemInstruction,
});
const stepTime = Date.now() - startTime;
// 3. 记录步骤
const agentStep: AgentStep = {
stepIndex: step,
strategy: strategy.name,
input: context.messages[context.messages.length - 1]?.content,
output: modelOutput.textContent || '',
toolCalls: modelOutput.toolCalls || [],
tokensUsed: modelOutput.usage.totalTokens,
durationMs: stepTime,
};
this.guard.recordStep(agentStep);
trace.push(agentStep);
// 4. 处理工具调用
if (modelOutput.toolCalls?.length) {
for (const call of modelOutput.toolCalls) {
const result = await context.toolExecutor.execute(call);
// 注入结果到上下文
context.appendToolResult(result);
}
continue; // 继续循环
}
// 5. 幻觉检测
if (modelOutput.textContent) {
const hallucinationCheck = await this.hallucinationDetector.check(
modelOutput.textContent,
JSON.stringify(context.messages)
);
if (hallucinationCheck.confidence < 0.6) {
console.warn(`[推理] 检测到低置信度输出 (${hallucinationCheck.confidence})，触发反思`);
// 注入反思提示
context.injectReflection(hallucinationCheck);
continue;
}
}
// 6. 判断是否是最终答案
const parsed = strategy.parseOutput(modelOutput.textContent || '');
if (parsed.finalAnswer) {
return {
answer: parsed.finalAnswer,
steps: step + 1,
trace,
...this.summarizeTrace(trace),
};
}
}
return this.wrapUp('达到最大步数限制', trace, this.config.maxSteps);
}
}
6.4 本章小结
本章深入了 Agent 的"大脑皮层"——Reasoning Controller，覆盖了从基础到高级的推理策略和完整的质量保障体系：
-
基础策略：CoT（想清楚再说）、ReAct（边想边做）、Plan-Execute（先规划后执行）。三种策略分别适合不同复杂度级别的任务，可以通过策略管理器动态切换。
-
高级技术：Tree-of-Thought（多路径探索+剪枝）、Self-Consistency（多次采样+投票）、Reflection（自我反思+改进）。这些技术显著提升推理的准确性和可靠性。
-
质量保障：幻觉检测（事实提取+逐条验证）、推理可视化（Mermaid 图+表格）、推理超时与循环检测。推理质量保障不是加分项，而是生产级 Agent 的必选项。
推理控制的终极目标是：让 Agent 的思考过程高效、可靠、可解释、可干预。
关键术语
| 术语 | 英文 | 定义 |
|---|---|---|
| CoT | Chain-of-Thought | 让模型生成中间推理步骤的技术 |
| ReAct | Reasoning + Acting | 将推理和行动交织的 Agent 推理模式 |
| Plan-and-Execute | Plan-and-Execute | 先完整规划再逐步执行的策略 |
| ToT | Tree-of-Thought | 同时探索多个推理方向并剪枝的技术 |
| Self-Consistency | Self-Consistency | 多次采样+多数投票的准确性提升技术 |
| Reflection | Reflection | Agent 审查自身输出并改进的能力 |
| 幻觉检测 | Hallucination Detection | 识别和标记模型生成的不实陈述 |
思考与练习
-
策略对比实验：选择一个问题（如"分析这个项目的安全性"），分别用 CoT、ReAct、Plan-Execute 三种策略执行，比较结果的质量、耗时和 Token 消耗。
-
ReAct 解析器：实现一个更健壮的 ReAct 输出解析器，能处理各种格式变体（中英文混合、不同的标签格式、嵌套的思考-行动）。
-
Tree-of-Thought 实验：为一个逻辑推理问题（如"谁是凶手"类谜题）运行 Tree-of-Thought，观察不同分支的评分和最终剪枝结果。
-
幻觉检测实践：让 Agent 分析一个你不熟悉的领域的问题，运行幻觉检测器，逐条手动验证检测器的判断是否准确。
-
推理循环检测：设计一个会导致 Agent 陷入推理循环的场景（如互相冲突的指令），测试推理控制器的循环检测是否能正确触发和中断。
这里是“一人公司”的成长家园。我们提供从产品曝光、技术变现到法律财税的全栈内容，并连接云服务、办公空间等稀缺资源，助你专注创造，无忧运营。
更多推荐
1
0- 0
已为社区贡献73条内容
所有评论(0)