Agentic Design Patterns-模式17：推理技术（Reasoning Techniques）的代码实现

目录

1. 概述

2. 价值

3. 关键技术

4. 代码实现

扩展模块：添加外部工具支持

使用说明

核心特点

1. 概述

推理技术让智能体的内部推理过程变得显式化，使其能够拆解问题、考虑中间步骤，并得出更稳健、准确的结论。一个核心原则是，在推理阶段分配更多的计算资源，即允许智能体或底层大模型拥有更多的处理时间或步骤来分析问题并生成响应。智能体可以进行迭代优化、探索多种解决路径，或调用外部工具。推理时增加计算资源，尤其在需要深入分析和思考的复杂问题上，能显著提升准确性、连贯性和健壮性。

图1：推理设计模式

2. 价值

复杂问题解决不仅仅是直接给出答案，AI面临的核心挑战是如何分解、推理和规划多步任务。显式化智能体的"思考"过程，使其能系统性地解决难题。

推理扩展定律为智能体系统的高效、经济部署提供理论依据。它挑战"模型越大越好"的直觉，强调合理分配推理资源可优化性能、响应延迟和运维成本。开发者可据此做出更精细的资源分配和性能优化决策，实现更经济高效的AI部署。推理扩展定律表明智能体性能不仅取决于模型大小，还取决于分配的"思考时间"，实现更高质量的自主行动。

3. 关键技术

提升AI模型问题解决能力的核心推理技术包括链式思维（Chain‑of‑Thought,CoT）、树式思维（Tree‑of‑Thought,ToT）、自我纠错（Self‑correction）、可验证奖励强化学习（RLVR）、ReAct（推理与行动）、CoD（辩论链）、GoD（辩论图）、MASS（多智能体系统搜索）等。

链式思维（CoT）是智能体的内部独白，通过分步规划将复杂目标拆解为可执行动作序列；树式思维和自我纠错赋予智能体深度思考能力，可评估多种策略、纠错并优化方案；ReAct框架赋予智能体核心操作循环，使其能动态行动并与环境交互；协作框架如辩论链（CoD）推动从单体到多智能体系统，团队协作能解决更复杂问题并减少偏见；DeepResearch等应用展示了这些技术如何让智能体自主执行复杂、长期任务，如深入调查；MASS框架自动优化智能体提示和交互结构，确保多智能体系统整体性能最优；集成这些推理技术，打造真正自主、可托付的智能体，能独立规划、行动和解决复杂问题。

4. 代码实现

以下是一个关于推理技术的Python程序示例，实现了基于不同推理技术（CoT, ReAct, ToT等）的智能体框架，支持动态调用大模型进行多步推理：

from typing import List, Dict, Any, Optional
from datetime import datetime
import json
import re

# 模拟外部模型客户端（根据你的实际调用方式调整）
try:
from model_factory import get_model_client
client = get_model_client()
except ImportError:
# 模拟客户端（用于测试）
class MockClient:
class chat:
class completions:
@staticmethod
def create(model, messages, stream=False):
class Response:
class Choice:
class Message:
content = "这是模拟的模型响应，包含推理步骤和最终答案。"
message = Message()
choices = [Choice()]
return Response()
client = MockClient()

class ReasoningAgent:
"""推理智能体基类"""

def __init__(self, model: str = "gpt-4o"):
self.model = model
self.conversation_history = []
self.step_count = 0

def call_model(self, system_prompt: str, user_prompt: str) -> str:
"""调用大模型获取响应"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
]

try:
response = client.chat.completions.create(
model=self.model,
messages=messages,
stream=False
)
result = response.choices[0].message.content

# 记录交互历史
self.conversation_history.append({
"timestamp": datetime.now().isoformat(),
"step": self.step_count,
"system": system_prompt[:100] + "…" if len(system_prompt) > 100 else system_prompt,
"user": user_prompt[:100] + "…" if len(user_prompt) > 100 else user_prompt,
"response": result[:200] + "…" if len(result) > 200 else result
})
self.step_count += 1

return result
except Exception as e:
return f"模型调用错误: {str(e)}"

def solve(self, problem: str) -> str:
"""解决问题（子类需重写此方法）"""
raise NotImplementedError

class ChainOfThoughtAgent(ReasoningAgent):
"""链式思维（CoT）智能体"""

def solve(self, problem: str) -> Dict[str, Any]:
"""使用CoT分步推理解决问题"""
system_prompt = """你是一个推理专家。请使用链式思维（Chain-of-Thought）分步推理解决问题。
按以下格式输出：
1. 问题重述
2. 步骤分解
3. 逐步推理
4. 最终答案

确保每一步都清晰、逻辑严密。"""

user_prompt = f"问题：{problem}\\n\\n请使用链式思维进行推理："

reasoning_process = self.call_model(system_prompt, user_prompt)

return {
"technique": "Chain-of-Thought",
"problem": problem,
"reasoning_process": reasoning_process,
"steps": self.extract_steps(reasoning_process),
"final_answer": self.extract_final_answer(reasoning_process)
}

def extract_steps(self, reasoning: str) -> List[str]:
"""从推理过程中提取步骤"""
steps = []
lines = reasoning.split('\\n')
for line in lines:
if re.match(r'^\\d+[\\.\\)]', line.strip()) or re.match(r'^步骤\\d+', line) or re.match(r'^Step', line):
steps.append(line.strip())
return steps if steps else ["步骤提取失败"]

def extract_final_answer(self, reasoning: str) -> str:
"""提取最终答案"""
markers = ["最终答案：", "答案：", "结论：", "Final Answer:", "Answer:"]
for marker in markers:
if marker in reasoning:
parts = reasoning.split(marker, 1)
if len(parts) > 1:
return parts[1].strip().split('\\n')[0]
return reasoning[-200:] # 返回最后200字符作为备选

class ReActAgent(ReasoningAgent):
"""ReAct（推理+行动）智能体"""

def __init__(self, model: str = "gpt-4o", max_steps: int = 5):
super().__init__(model)
self.max_steps = max_steps
self.actions_log = []

def solve(self, problem: str) -> Dict[str, Any]:
"""使用ReAct框架解决问题"""
system_prompt = """你是一个ReAct智能体，可以交替进行推理（Thought）和行动（Action）。
行动格式：Action: [工具名] [参数]
可用工具：calculate(执行计算), search(搜索信息), analyze(分析数据)
推理格式：Thought: [你的推理]
最终答案格式：Final Answer: [答案]"""

user_prompt = f"问题：{problem}\\n\\n请开始ReAct流程："

full_process = ""
for step in range(self.max_steps):
# 获取思考或行动
response = self.call_model(system_prompt, user_prompt + full_process)
full_process += f"\\n{response}"

# 记录行动
if "Action:" in response:
action_part = response.split("Action:")[1].split("\\n")[0].strip()
self.actions_log.append({
"step": step,
"action": action_part
})

# 检查是否得出最终答案
if "Final Answer:" in response:
break

return {
"technique": "ReAct",
"problem": problem,
"full_process": full_process,
"actions_taken": self.actions_log,
"steps_used": step + 1
}

class TreeOfThoughtAgent(ReasoningAgent):
"""树式思维（ToT）智能体"""

def solve(self, problem: str, branches: int = 3) -> Dict[str, Any]:
"""使用树式思维探索多个推理路径"""
system_prompt = """你是一个树式思维推理系统。请为问题生成多个不同的推理思路。
输出格式：
问题分析：
可能思路1：[思路描述]
可能思路2：[思路描述]
…
评估与综合：[对各思路的评估和综合]
最终答案：[答案]"""

user_prompt = f"问题：{problem}\\n\\n请生成{branches}个不同的推理思路："

reasoning = self.call_model(system_prompt, user_prompt)

# 提取不同思路
thoughts = []
lines = reasoning.split('\\n')
current_thought = ""

for line in lines:
if re.match(r'^可能思路\\d+', line) or re.match(r'^思路\\d+', line) or re.match(r'^Approach\\d+', line):
if current_thought:
thoughts.append(current_thought)
current_thought = line
elif current_thought and not re.match(r'^(评估|最终|结论)', line):
current_thought += "\\n" + line

if current_thought:
thoughts.append(current_thought)

return {
"technique": "Tree-of-Thought",
"problem": problem,
"branches_explored": len(thoughts),
"thoughts": thoughts,
"full_reasoning": reasoning,
"final_answer": self.extract_final_answer(reasoning)
}

class SelfCorrectionAgent(ReasoningAgent):
"""自我纠错智能体"""

def solve(self, problem: str) -> Dict[str, Any]:
"""使用自我纠错技术迭代优化答案"""
system_prompt = """你是一个具有自我纠错能力的AI。请按以下步骤：
1. 首次尝试解决问题
2. 检查答案中的潜在错误
3. 修正错误并重新推理
4. 给出最终验证过的答案"""

# 第一轮：初始推理
user_prompt1 = f"问题：{problem}\\n\\n请给出你的初始解答："
initial_answer = self.call_model(system_prompt, user_prompt1)

# 第二轮：自我检查
user_prompt2 = f"""初始解答：{initial_answer}

请仔细检查以上解答，找出可能的逻辑错误、计算错误或假设问题。
列出所有发现的问题："""

error_analysis = self.call_model(system_prompt, user_prompt2)

# 第三轮：修正答案
user_prompt3 = f"""问题：{problem}
初始解答：{initial_answer}
发现的问题：{error_analysis}

请基于以上分析，给出修正后的最终答案："""

final_answer = self.call_model(system_prompt, user_prompt3)

return {
"technique": "Self-Correction",
"problem": problem,
"initial_answer": initial_answer,
"error_analysis": error_analysis,
"corrected_answer": final_answer,
"iteration_steps": 3
}

class MultiAgentDebate:
"""多智能体辩论系统（CoD/GoD简化实现）"""

def __init__(self, model: str = "gpt-4o", num_agents: int = 3):
self.model = model
self.num_agents = num_agents
self.agents = [ReasoningAgent(model) for _ in range(num_agents)]
self.debate_history = []

def solve(self, problem: str, rounds: int = 2) -> Dict[str, Any]:
"""通过多智能体辩论解决问题"""

positions = []

# 第一轮：各智能体提出初始观点
for i in range(self.num_agents):
system_prompt = f"""你是专家{i+1}，请从你的专业角度分析问题。
请提出有说服力的观点和推理。"""

user_prompt = f"问题：{problem}\\n\\n请给出你的分析和解决方案："

response = self.agents[i].call_model(system_prompt, user_prompt)
positions.append({
"agent": f"专家{i+1}",
"position": response
})

# 辩论轮次
debate_log = []
for round_num in range(rounds):
round_log = {"round": round_num + 1, "exchanges": []}

for i in range(self.num_agents):
# 收集其他智能体的观点
other_positions = "\\n\\n".join([
f"{pos['agent']}的观点：{pos['position'][:500]}…"
for j, pos in enumerate(positions) if j != i
])

system_prompt = f"""你是专家{i+1}，正在进行辩论。
请回应其他专家的观点，并强化或修正你的立场。"""

user_prompt = f"""问题：{problem}
你的当前观点：{positions[i]['position'][:500]}…

其他专家的观点：
{other_positions}

请回应并进行深入讨论："""

response = self.agents[i].call_model(system_prompt, user_prompt)
positions[i]['position'] = response

round_log["exchanges"].append({
"agent": f"专家{i+1}",
"response": response[:300] + "…" if len(response) > 300 else response
})

debate_log.append(round_log)

# 最终综合
system_prompt = """作为最终裁判，请综合所有专家的辩论观点，给出最终的最佳答案。"""

all_positions = "\\n\\n".join([
f"{pos['agent']}：{pos['position']}"
for pos in positions
])

user_prompt = f"""问题：{problem}

专家辩论记录：
{all_positions}

请给出综合所有观点后的最终答案："""

final_response = self.agents[0].call_model(system_prompt, user_prompt)

return {
"technique": "Multi-Agent Debate (CoD/GoD)",
"problem": problem,
"num_agents": self.num_agents,
"initial_positions": positions,
"debate_rounds": debate_log,
"final_synthesis": final_response
}

class ReasoningOrchestrator:
"""推理技术协调器"""

def __init__(self):
self.techniques = {
"cot": ChainOfThoughtAgent(),
"react": ReActAgent(),
"tot": TreeOfThoughtAgent(),
"self_correct": SelfCorrectionAgent(),
"debate": MultiAgentDebate()
}

def solve_problem(self, problem: str, technique: str = "cot", **kwargs) -> Dict[str, Any]:
"""使用指定推理技术解决问题"""
if technique not in self.techniques:
available = ", ".join(self.techniques.keys())
return {
"error": f"未知推理技术。可用技术: {available}",
"available_techniques": list(self.techniques.keys())
}

agent = self.techniques[technique]

if technique == "debate":
if "rounds" in kwargs:
return agent.solve(problem, rounds=kwargs["rounds"])
return agent.solve(problem)
elif technique == "tot":
if "branches" in kwargs:
return agent.solve(problem, branches=kwargs["branches"])
return agent.solve(problem)
else:
return agent.solve(problem)

def compare_techniques(self, problem: str, techniques: List[str] = None) -> Dict[str, Any]:
"""比较不同推理技术的效果"""
if techniques is None:
techniques = ["cot", "react", "tot", "self_correct"]

results = {}
for tech in techniques:
if tech in self.techniques:
print(f"正在使用 {tech} 技术解决问题…")
results[tech] = self.solve_problem(problem, tech)

return {
"problem": problem,
"comparison": results,
"summary": self._generate_comparison_summary(results)
}

def _generate_comparison_summary(self, results: Dict[str, Dict]) -> str:
"""生成比较摘要"""
summary_lines = ["推理技术比较摘要："]

for tech, result in results.items():
if "error" not in result:
technique_name = result.get("technique", tech)
summary_lines.append(f"\\n{technique_name}：")

if "final_answer" in result:
answer_preview = result["final_answer"][:100] + "…" if len(result["final_answer"]) > 100 else result["final_answer"]
summary_lines.append(f" 答案预览：{answer_preview}")

if "steps" in result:
summary_lines.append(f" 推理步骤数：{len(result['steps'])}")

if "iteration_steps" in result:
summary_lines.append(f" 迭代次数：{result['iteration_steps']}")

return "\\n".join(summary_lines)

def main():
"""主函数：演示推理技术的使用"""

# 示例问题
sample_problems = [
"如果一辆火车以每小时80公里的速度从A站开往B站，距离为240公里，同时另一辆火车以每小时100公里的速度从B站开往A站，它们多久会相遇？",
"请分析气候变化对全球经济的主要影响，并提出三项应对策略。",
"一个数加上它的一半等于30，这个数是多少？",
]

problem = sample_problems[0]
print(f"问题：{problem}\\n")

orchestrator = ReasoningOrchestrator()

# 1. 使用链式思维
print("=" * 60)
print("1. 链式思维（CoT）示例：")
cot_result = orchestrator.solve_problem(problem, "cot")
print(f"技术：{cot_result.get('technique', 'N/A')}")
print(f"最终答案：{cot_result.get('final_answer', 'N/A')}")
print(f"推理步骤：{len(cot_result.get('steps', []))}步")

# 2. 使用ReAct
print("\\n" + "=" * 60)
print("2. ReAct示例：")
react_result = orchestrator.solve_problem(problem, "react")
print(f"技术：{react_result.get('technique', 'N/A')}")
print(f"执行步骤：{react_result.get('steps_used', 'N/A')}")
print(f"行动记录：{react_result.get('actions_taken', [])}")

# 3. 比较不同技术
print("\\n" + "=" * 60)
print("3. 推理技术比较：")
comparison = orchestrator.compare_techniques(problem, ["cot", "self_correct"])
print(comparison["summary"])

# 4. 多智能体辩论
print("\\n" + "=" * 60)
print("4. 多智能体辩论示例（简化）：")
print("（注意：完整辩论会消耗较多token，这里仅演示结构）")

# 使用简单问题演示
simple_problem = "人工智能对人类就业的影响主要是正面还是负面？"
debate_result = MultiAgentDebate(num_agents=2).solve(simple_problem, rounds=1)
print(f"辩论主题：{simple_problem}")
print(f"参与专家：{debate_result.get('num_agents', 'N/A')}位")
print(f"辩论轮次：{len(debate_result.get('debate_rounds', []))}")

# 5. 保存推理历史
print("\\n" + "=" * 60)
print("5. 推理历史记录示例：")
cot_agent = ChainOfThoughtAgent()
cot_agent.solve(problem)

# 保存历史到文件
with open("reasoning_history.json", "w", encoding="utf-8") as f:
json.dump(cot_agent.conversation_history, f, ensure_ascii=False, indent=2)
print("推理历史已保存到 reasoning_history.json")

if __name__ == "__main__":
main()

扩展模块：添加外部工具支持

# tools.py – 外部工具模块
class ExternalTools:
"""外部工具集，供ReAct等智能体调用"""

@staticmethod
def calculate(expression: str) -> str:
"""执行数学计算（简化版）"""
try:
# 注意：实际使用中应使用安全的方式评估表达式
# 这里仅为演示
if "+" in expression:
parts = expression.split("+")
return str(float(parts[0]) + float(parts[1]))
elif "-" in expression:
parts = expression.split("-")
return str(float(parts[0]) – float(parts[1]))
elif "*" in expression:
parts = expression.split("*")
return str(float(parts[0]) * float(parts[1]))
elif "/" in expression:
parts = expression.split("/")
return str(float(parts[0]) / float(parts[1]))
else:
return f"无法解析表达式: {expression}"
except Exception as e:
return f"计算错误: {str(e)}"

@staticmethod
def search(query: str) -> str:
"""模拟搜索功能"""
# 实际应用中应连接搜索引擎API
return f"模拟搜索结果：关于'{query}'，找到相关信息。"

@staticmethod
def analyze(data_description: str) -> str:
"""模拟数据分析"""
return f"对'{data_description}'的分析完成。发现关键模式：…"

使用说明

核心特点

这个框架可以作为智能体推理系统的基础，根据实际需求进行扩展和优化。

参照书籍《Agentic Design Patterns》的基本概念和观点。