Function Call 是 Agent 的"手"和"脚"，而 Agent 架构模式则是它的"大脑思考方式"。ReAct 是边想边做，Plan-and-Execute 是谋定后动，Reflection 是三思而行。本节我们将深入每种模式的"思维回路"，剖析 Function Call 在其中扮演的角色与调用节奏，并提供完全标准化、可直接运行的代码实现，让你不仅能用，更能"设计"出最适合业务场景的智能体。
学习目标
完成本节学习后，学员将能够：
- 深刻理解 ReAct、Plan-and-Execute、Reflection 三种主流 Agent 架构模式的核心工作原理与思维范式
- 掌握 Function Call 在不同模式中的调用时机、频率与控制策略
- 独立实现基于 ReAct、Plan-and-Execute 和 Reflection 模式的标准化、生产级 Agent 原型
一、为什么需要不同的 Agent 架构模式？
在上一节，我们手写了一个具备"感知-决策-行动-反馈"循环的通用 Agent。它很灵活，但对于复杂任务，它可能表现得像一个"莽撞的实习生"——拿到任务就立刻动手，缺乏全局规划，容易在细节中迷失方向。
提问：当学生问"请帮我制定一个为期一个月的三角函数复习计划，并每周推送一套练习题"时，通用 Agent 会怎么做？
答案：它可能会立刻调用 generate_exercise(topic="三角函数", count=5)，然后结束。因为它没有"计划"的概念。
不同模式的本质：不同的 Agent 架构模式，本质上是为了解决不同类型的问题而设计的"思考策略"。它们定义了 LLM 应该在何时、以何种方式调用 Function Call。
| 问题类型 | 推荐模式 | 原因 |
|---|---|---|
| 简单、线性任务（如：“识别这张图”） | 通用循环 / ReAct | 一步到位，无需复杂规划 |
| 复杂、多步骤任务（如：“制定复习计划”） | Plan-and-Execute | 需要先拆解任务，再分步执行 |
| 高精度、容错性高（如：“批改这道压轴题”） | Reflection | 执行后需要反思结果，确保正确性 |
二、ReAct 模式
2.1 核心原理与工作流程
ReAct (Reason + Act) 是最经典、最接近人类直觉的 Agent 模式。它由 Google 在 2022 年提出，其核心思想是让 LLM 在"推理"（Reasoning）和"行动"（Action）之间交替进行，逐步逼近问题的答案。
工作流程详解：
- 初始感知 (Perception)：接收用户输入。
- 第一轮推理 (Reasoning)：LLM 分析问题，思考"我现在需要做什么？"，并输出一段推理文字（Thought）。
- 第一轮行动 (Action)：基于推理，LLM 决定调用一个 Function Call 工具。
- 观察 (Observation)：程序执行工具，将结果（Observation）返回给 LLM。
- 第二轮推理 (Reasoning)：LLM 基于上一轮的行动结果，进行新一轮的思考，“我得到了XXX结果，接下来我应该…"。
- 第二轮行动 (Action)：再次调用 Function Call。
- 循环… 重复 4-6 步，直到 LLM 认为任务完成，输出最终答案 (Final Answer)。
数据流格式（关键）：为了让 LLM 理解这种交替模式，我们需要在 messages 中构造特定的格式。通常使用 Thought/Action/Observation 三元组。
2.2 Function Call 应用策略
调用节奏：高频、短平快。每个推理步骤后，几乎都会紧跟一个 Function Call。
控制策略：
- 由 LLM 主导：LLM 通过输出 Thought 来决定下一步调用哪个工具。开发者只需提供工具列表。
- 循环终止：当 LLM 的回复中不再包含
function_call，而是直接给出 Final Answer 时，循环结束。 - 防死循环：必须设置最大轮次（如 5 轮），防止 LLM 陷入无限推理。
适用场景：信息检索、多跳问答、工具链式调用（A 工具的结果是 B 工具的输入）。
2.3 代码实现与案例解析
我们以"解答一道数学题"为例，演示 ReAct 协议的模式。
import openai
import json
import re
import os
from typing import List, Dict, Any, Tuple, Optional
class StrictReActAgent:
def __init__(self, client: openai.OpenAI, model: str = "qwen-plus"):
self.client = client
self.model = model
self.max_turns = 5 # 防止无限循环
# 定义可用的"动作"（注意：这里不再是OpenAI的functions，而是我们自己的协议）
self.available_actions = {
"ocr_service": self._mock_ocr_service,
"search_knowledge": self._mock_search_knowledge,
"calculate_expression": self._mock_calculate_expression,
"final_answer": self._handle_final_answer
}
def run(self, user_input: str) -> str:
"""
运行 ReAct Agent。
"""
print(f"【用户提问】: {user_input}")
print("=" * 60)
# 初始化对话历史，注入系统提示词
messages = [
{"role": "system", "content": self._get_react_system_prompt()},
{"role": "user", "content": user_input}
]
turn = 0
final_answer = None
while turn < self.max_turns:
turn += 1
print(f"\n>>> 第 {turn} 轮推理与行动 <<<")
# 调用 LLM
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
# **关键：不传递任何 functions，让 LLM 纯文本输出**
)
message = response.choices[0].message
assistant_reply = message.content.strip()
print(f"[LLM 原始输出]:\n{assistant_reply}\n")
# 解析 LLM 的回复
thought, action, action_input = self._parse_react_output(assistant_reply)
if not thought or not action or not action_input:
print("[错误] 无法解析 LLM 输出，格式不符合 ReAct 协议。")
break
print(f"【Thought】: {thought}")
print(f"【Action】: {action}")
print(f"【Action Input】: {action_input}")
# 将 LLM 的完整回复加入历史
messages.append({
"role": "assistant",
"content": assistant_reply
})
# 检查是否为最终答案
if action == "final_answer":
final_answer = action_input.get("answer", "未提供具体答案")
print(f"【Final Answer】: {final_answer}")
break
# 执行动作
if action in self.available_actions:
observation = self.available_actions[action](action_input)
else:
observation = {"error": f"未知动作: {action}"}
observation_str = json.dumps(observation, ensure_ascii=False)
print(f"【Observation】: {observation_str}")
# 将观察结果作为用户消息加入历史，驱动下一轮
messages.append({
"role": "user",
"content": f"Observation: {observation_str}"
})
if not final_answer:
final_answer = "处理过程超时或出错，未能生成最终答案。"
return final_answer
def _get_react_system_prompt(self) -> str:
"""返回 ReAct 模式的系统提示词"""
return """
你是一个严谨的数学解题助手，必须严格使用 ReAct 模式工作。
你的每一步思考和行动都必须遵循以下精确格式：
Thought: <你当前的思考、推理或计划>
Action: <你要执行的动作名称，必须是以下工具之一: ocr_service, search_knowledge, calculate_expression, final_answer>
Action Input: <动作的输入参数，必须是一个合法的JSON字符串>
可用工具说明:
1. ocr_service: 用于识别学生上传的图片中的题目文本。参数: {"image_path": "字符串，图片的服务器路径"}
2. search_knowledge: 用于查询解题所需的公式或定理。参数: {"keywords": "字符串，搜索关键词"}
3. calculate_expression: 用于计算数学表达式。参数: {"expression": "字符串，要计算的表达式，如 'sqrt(25)' 或 'sin(pi/4)'"}
4. final_answer: 当你得出最终答案时使用。参数: {"answer": "字符串，你的最终答案和解释"}
重要规则:
- 每次回复必须且只能包含一个 Thought、一个 Action 和一个 Action Input。
- Action 的值必须是上述四个工具名之一，不能是其他任何值。
- Action Input 必须是严格的 JSON 格式字符串，不能有任何额外字符。
- 不要输出任何其他文本、解释或问候语。
- 如果你认为任务已完成，请使用 final_answer 工具。
现在，请开始处理用户的请求。
"""
def _parse_react_output(self, text: str) -> Tuple[Optional[str], Optional[str], Optional[Dict[str, Any]]]:
"""
解析 LLM 的输出，提取 Thought, Action, Action Input.
使用正则表达式进行严格匹配。
"""
# 定义正则模式
# (?s) 表示让 . 匹配换行符
pattern = r"(?s)Thought:\s*(.*?)\s*Action:\s*(.*?)\s*Action Input:\s*(.*)"
match = re.search(pattern, text, re.IGNORECASE)
if not match:
return None, None, None
thought = match.group(1).strip()
action = match.group(2).strip()
action_input_str = match.group(3).strip()
# 尝试解析 Action Input 为 JSON
try:
action_input = json.loads(action_input_str)
except json.JSONDecodeError:
return thought, action, None
return thought, action, action_input
# --- 模拟的工具函数 ---
def _mock_ocr_service(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""模拟OCR服务"""
image_path = params.get("image_path", "")
print(f" [执行] OCR 识别: {image_path}")
if "math" in image_path.lower():
return {
"success": True,
"text": "题目: 在三角形ABC中，已知角A为60度，边AB长为5，边AC长为7，求边BC的长度。"
}
else:
return {"success": False, "error": "图像识别失败或内容不包含数学题"}
def _mock_search_knowledge(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""模拟知识查询"""
keywords = params.get("keywords", "")
print(f" [执行] 知识查询: {keywords}")
if "余弦定理" in keywords or "cosine" in keywords.lower():
return {
"success": True,
"formula": "余弦定理公式: BC² = AB² + AC² - 2 × AB × AC × cos(∠A)",
"description": "用于计算已知两边及其夹角的三角形的第三边。"
}
else:
return {"success": False, "error": "未找到相关知识点"}
def _mock_calculate_expression(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""模拟计算服务"""
expression = params.get("expression", "")
print(f" [执行] 计算表达式: {expression}")
# 这里是极度简化的模拟，实际应使用 SymPy 等库
try:
if "sqrt(39)" in expression:
result = "√39 (约等于 6.245)"
elif "**2" in expression:
parts = expression.split("**2")
if len(parts) == 2 and parts[0].isdigit():
num = int(parts[0])
result = str(num ** 2)
else:
result = "计算结果"
else:
result = "计算结果"
return {"success": True, "result": result}
except Exception as e:
return {"success": False, "error": str(e)}
def _handle_final_answer(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""处理最终答案，这里只是透传"""
return {"final_answer": params.get("answer", "")}
# --- 测试运行 ---
if __name__ == "__main__":
# 初始化OpenAI客户端 (请替换为你的API Key和Base URL)
client = openai.OpenAI(
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
agent = StrictReActAgent(client, model="qwen-plus")
# 用户提问
user_query = "老师，这道几何题我不会做，图片在 /uploads/geometry_math_001.jpg，能教我吗？"
# 运行 Agent
final_result = agent.run(user_query)
print("\n" + "="*60)
print("【EduMentor 最终回复】")
print("="*60)
print(final_result)
三、Plan-and-Execute 模式
3.1 核心原理与工作流程
Plan-and-Execute (计划与执行) 模式将任务分解为两个清晰的阶段：Planning (计划) 和 Execution (执行)。它先让 LLM 从宏观角度审视整个任务，制定一个详细的、分步骤的计划，然后再按计划一步步调用工具执行。
工作流程详解：
- 任务规划 (Planning)：
- LLM 接收用户输入。
- LLM 输出一个完整的、结构化的计划。这个计划通常是一个步骤列表 (List of Steps)。
- 关键：在规划阶段，LLM 不调用任何 Function Call！它只是"想”，不"做"。
- 计划解析 (Plan Parsing)：程序代码解析 LLM 生成的计划，将其转化为一个可执行的任务队列。
- 步骤执行 (Execution)：
- 程序按顺序遍历任务队列。
- 对于队列中的每一步，构造一个"子问题"提示词，提交给 LLM。
- LLM 针对这个"子问题"，决定是否调用 Function Call 工具。
- 执行工具，获取结果。
- 将结果记录下来，作为下一步的上下文。
- 最终总结 (Final Summary)：所有步骤执行完毕后，可以再调用一次 LLM，让它基于所有中间结果，生成一个最终的、整合性的回复。
3.2 Function Call 应用策略
调用节奏：零频。Function Call 在执行阶段完全由程序代码直接调用，LLM 不参与决策。
控制策略：
- 计划驱动：Function Call 的调用完全由预先生成和解析的计划决定。执行是确定性的。
- 阶段隔离：Planning 阶段禁止 Function Call，确保计划的纯粹性。
- 错误处理：如果某一步执行失败，可以选择重试、跳过或终止整个计划。
适用场景：复杂项目管理、多步骤工作流、需要强一致性和可预测性的任务（如生成周报、制定学习计划）。
3.3 代码实现与案例解析
我们以"规划旅行的智能助手"为例。
import openai
import json
import re
import os
from typing import Dict, Any, Optional
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv())
class PlanAndExecuteAgent:
def __init__(self, client: openai.OpenAI, model: str = "qwen-plus"):
self.client = client
self.model = model
# 定义工具函数 (程序直接调用，不由LLM在执行阶段选择)
self.tools = {
"get_weather": self._get_weather,
"suggest_activity": self._suggest_activity
}
def run(self, user_input: str):
print(f"【用户提问】: {user_input}")
print("=" * 60)
# --- 阶段1: 任务规划 (Planning) - LLM 输出 JSON 计划 ---
print("\n[阶段1: 任务规划]")
plan_json = self._create_plan(user_input)
print(f"解析后的结构化计划: {json.dumps(plan_json, ensure_ascii=False, indent=2)}")
if not plan_json or "steps" not in plan_json:
return "未能生成有效的执行计划。"
# --- 阶段2: 步骤执行 (Execution) - 程序直接调用工具，不咨询LLM！ ---
print("\n[阶段2: 步骤执行]")
execution_results = {} # 存储每个步骤结果，key: step_id
steps = plan_json["steps"] # 每个步骤: {"id": int, "tool": str, "args": dict, "dependencies": [int]}
for step in steps:
step_id = step["id"]
print(f"\n>>> 执行步骤 {step_id}: {step['tool']}({step['args']}) <<<")
# 检查依赖 (简单验证前步结果存在且成功)
if not self._check_dependencies(step, execution_results):
result = {"error": "依赖步骤未完成或失败", "success": False}
print(f" [执行失败] 错误: {result['error']}")
else:
# **核心: 程序直接调用工具**
if step['tool'] in self.tools:
try:
result = self.tools[step['tool']](**step['args'])
print(f" [执行成功] 结果: {result}")
except Exception as e:
result = {"error": str(e), "success": False}
print(f" [执行失败] 错误: {result['error']}")
else:
result = {"error": f"未知工具: {step['tool']}", "success": False}
print(f" [执行失败] 错误: {result['error']}")
# 跟踪 status
execution_results[step_id] = {"result": result, "success": result.get("success", False)}
# --- 阶段3: 最终总结 - LLM 作为汇报官 ---
print("\n[阶段3: 生成最终回复]")
print(f'execution_results-->{execution_results}')
final_reply = self._generate_final_reply(user_input, execution_results)
print(f"【最终回复】:\n{final_reply}")
return final_reply
def _create_plan(self, user_input: str) -> Optional[Dict[str, Any]]:
"""让LLM生成 JSON 计划 (结构化输出，确保规划完整指定工具/参数/依赖)"""
# 定义 JSON 示例字符串，避免 f-string 解析问题
plan_example = """{
"steps": [
{"id": 1, "tool": "get_weather", "args": {"city": "Beijing", "date": "2025-09-24"}, "dependencies": []},
{"id": 2, "tool": "suggest_activity", "args": {"weather": "Sunny", "city": "Beijing"}, "dependencies": [1]}
]
}"""
planning_prompt = f"""
你是一个出行规划师。请为以下请求制定一个详细、可行的执行计划。
计划以JSON格式输出，包括步骤列表，每个步骤指定id、tool、args、dependencies。
可用工具:
- get_weather: 获取天气，args: {{"city": string, "date": string (YYYY-MM-DD)}}
- suggest_activity: 建议活动，args: {{"weather": string, "city": string}}
请求: {user_input}
输出格式:
{plan_example}
重要: 只输出JSON，不要额外解释。dependencies 是前置步骤id列表，如果无依赖则[]。
"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": planning_prompt}],
# 关键: 规划阶段不提供任何工具，确保LLM只输出JSON计划
)
plan_text = response.choices[0].message.content.strip()
# 从文本提取JSON
try:
json_match = re.search(r'\{.*\}', plan_text, re.DOTALL)
# print(f'json_match-->{json_match}')
if json_match:
return json.loads(json_match.group(0))
return json.loads(plan_text)
except json.JSONDecodeError:
print("JSON解析失败，返回空计划")
return None
def _check_dependencies(self, step: Dict[str, Any], results: Dict[int, Dict[str, Any]]) -> bool:
"""简单检查依赖 (验证前步结果存在且成功)"""
for dep_id in step.get("dependencies", []):
if dep_id not in results or not results[dep_id]["success"]:
return False
return True
# --- 工具函数实现 (由程序直接调用) ---
def _get_weather(self, city: str, date: str) -> Dict[str, Any]:
"""获取天气 (模拟)"""
# 模拟天气数据
return {
"success": True,
"city": city,
"date": date,
"weather": "Sunny" if "Beijing" in city else "Rainy",
"temperature": "25°C"
}
def _suggest_activity(self, weather: str, city: str) -> Dict[str, Any]:
"""根据天气建议活动 (模拟)"""
if "Sunny" in weather:
activity = "Go for a park walk or outdoor sightseeing."
else:
activity = "Visit indoor museums or shopping malls."
return {
"success": True,
"city": city,
"weather": weather,
"suggested_activity": activity
}
def _generate_final_reply(self, original_query: str, results: Dict[int, Dict[str, Any]]) -> str:
"""生成最终回复 (整合执行统计)"""
# 计算统计
total_steps = len(results)
success_count = sum(1 for res in results.values() if res["success"])
summary_prompt = f"""
原始请求: {original_query}
执行结果:
{json.dumps(results, ensure_ascii=False, indent=2)}
执行统计: 总步骤 {total_steps}，成功 {success_count}
请基于以上信息，给用户一个友好、清晰的最终回复，总结已完成的工作。如果有失败，简单说明。
"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": summary_prompt}]
)
return response.choices[0].message.content
# --- 测试 ---
if __name__ == "__main__":
client = openai.OpenAI(
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
agent = PlanAndExecuteAgent(client)
query = "帮我规划明天在北京的出行计划，根据天气建议活动。"
agent.run(query)
四、Reflection 模式
4.1 核心原理与工作流程
Reflection (反思) 模式的核心思想是在执行完一个或多个步骤后，主动停下来进行反思，评估当前结果的质量或正确性，并根据反思结果决定是继续、修正还是重新执行。它特别适用于对结果准确性要求极高的场景。
工作流程详解：
- 初始执行：采用 ReAct 或 Plan-and-Execute 模式完成初步任务。
- 触发反思：在关键节点（如得到最终答案前、或执行完一个复杂步骤后），主动调用一个"反思"工具或提示词。
- 反思评估：LLM 基于当前的所有信息（用户问题、执行过程、中间结果）进行批判性思考，回答诸如"这个答案合理吗？"、“有没有遗漏的步骤？"、“计算过程有错误吗？“等问题。
- 决策调整：
- 如果反思结果是正面的（“答案正确”），则输出最终答案。
- 如果反思结果是负面的（“第二步计算有误”），则回滚到出错的步骤，重新执行或调用新工具修正。
- 可能会触发多轮反思，直到结果令人满意或达到最大反思次数。
4.2 Function Call 应用策略
调用节奏：不定频、事件驱动。Function Call 的重新调用是由"反思"的结论触发的，而非预设的流程。
控制策略：
- 反思触发器：需要明确定义何时进行反思（例如，每次得到一个数值结果后，或在输出最终答案前）。
- 反思提示词：设计高质量的反思提示词是关键，它决定了反思的深度和有效性。
- 状态回滚：系统需要有能力将对话状态或数据状态回滚到之前的某个检查点，以便重新执行。
适用场景：数学/物理题求解、代码生成与调试、法律/医疗等高风险领域的咨询。
4.3 代码实现与案例解析
我们以"求解一道物理题"为例，演示如何在 Plan-and-Execute 模式的基础上，嵌入 Reflection 模块，实现"解题-验算"的双重保障。
import openai
import json
import re
import os
import math
from typing import List, Dict, Any, Optional, Tuple
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv())
class Reflection:
"""
一个融合了 Plan-and-Execute 与 Reflection 模式的物理题求解器。
核心亮点：在输出最终答案前，进行主动反思，确保答案100%正确。
"""
def __init__(self, client: openai.OpenAI, model: str = "qwen-plus"):
self.client = client
self.model = model
self.max_reflections = 3 # 最大反思次数，防死循环
# 定义工具函数 (程序直接调用)
self.tools = {
"identify_problem": self._identify_problem,
"lookup_formula": self._lookup_formula,
"calculate_result": self._calculate_result,
"generate_answer": self._generate_answer
}
def run(self, user_input: str):
print(f"【用户提问】: {user_input}")
print("=" * 80)
# --- 阶段1: 任务规划 (Planning) ---
print("\n[阶段1: 任务规划]")
plan_json = self._create_plan(user_input)
if not plan_json or "steps" not in plan_json:
return "未能生成有效的解题计划。"
print(f" LLM 生成的结构化计划:")
print(json.dumps(plan_json, ensure_ascii=False, indent=2))
# --- 阶段2: 步骤执行 (Execution) ---
print("\n[阶段2: 步骤执行]")
execution_results = {} # {step_id: {"result": ..., "success": bool}}
for step in plan_json["steps"]:
step_id = step["id"]
tool_name = step["tool"]
args = step["args"]
dependencies = step.get("dependencies", [])
print(f"\n>>> 执行步骤 {step_id}: {tool_name} <<<")
print(f" 参数: {args}")
print(f" 依赖: {dependencies}")
# 检查依赖是否满足
if not self._check_dependencies(dependencies, execution_results):
result = {"error": f"依赖步骤 {dependencies} 未完成或失败", "success": False}
print(f" [执行失败] 原因: {result['error']}")
else:
# 提取依赖数据，注入到当前步骤参数中
enriched_args = self._inject_dependency_data(args, execution_results)
# 程序直接调用工具
if tool_name in self.tools:
try:
result = self.tools[tool_name](**enriched_args)
print(f" [执行成功] 结果: {result}")
except Exception as e:
result = {"error": str(e), "success": False}
print(f" [执行失败] 错误: {result['error']}")
else:
result = {"error": f"未知工具: {tool_name}", "success": False}
print(f" [执行失败] 错误: {result['error']}")
# 记录执行结果
execution_results[step_id] = {"result": result, "success": result.get("success", False)}
# 提取初步答案 (假设最后一步是 generate_answer)
preliminary_answer = self._extract_preliminary_answer(execution_results)
if not preliminary_answer:
return "未能生成初步答案。"
print(f"\n[初步答案]: {preliminary_answer}")
# --- 阶段3: 反思评估 (Reflection) ---
print("\n[阶段3: 反思评估 - 启动质量检查]")
final_answer = self._reflect_and_refine(user_input, plan_json, execution_results, preliminary_answer)
# --- 阶段4: 最终总结 ---
print("\n[阶段4: 生成最终回复]")
final_reply = self._generate_final_reply(user_input, final_answer)
print(f"\n【最终答案】\n{final_reply}")
return final_reply
# --- 反思模块核心方法 ---
def _reflect_and_refine(self, question: str, plan: Dict[str, Any], results: Dict[int, Dict[str, Any]], preliminary_answer: str) -> str:
"""
反思与修正主循环。
优化点：在每次反思后立即判断，若正确则立刻返回，避免无谓循环。
"""
current_answer = preliminary_answer
# 如果最大反思次数为0，则跳过反思，直接返回初步答案
if self.max_reflections <= 0:
return current_answer
for reflection_count in range(1, self.max_reflections + 1):
print(f"\n>>> 第 {reflection_count} 轮反思 <<<")
# 1. 生成反思提示词
reflection_prompt = self._build_reflection_prompt(question, plan, results, current_answer)
# 2. 调用 LLM 进行反思 (强制 JSON 输出)
reflection_result = self._call_llm_for_reflection(reflection_prompt)
print(f" [反思报告]: {json.dumps(reflection_result, ensure_ascii=False, indent=2)}")
# 3. **核心优化：立即判断反思结果**
if reflection_result.get("is_correct", False):
print(f" [反思通过] 答案被验证为正确。")
return current_answer # **立即返回，不再进行后续循环！**
else:
critique = reflection_result.get("critique", "未提供具体批评")
suggestion = reflection_result.get("suggestion", "无修正建议")
print(f" [反思发现问题] 批评: {critique}")
print(f" [修正建议] {suggestion}")
# 4. 根据建议修正答案
current_answer = self._apply_correction(current_answer, suggestion)
print(f" [已修正] 新答案: {current_answer}")
# 如果这是最后一次反思，无论结果都返回
if reflection_count == self.max_reflections:
print(f" [达到最大反思次数] 返回最后一次修正的答案。")
return current_answer
def _build_reflection_prompt(self, question: str, plan: Dict[str, Any], results: Dict[int, Dict[str, Any]],
answer: str) -> str:
"""构建反思提示词"""
return f"""
你是一位严谨的物理教授，正在对一道物理题的解答进行质量审查。
原始问题: {question}
解题计划:
{json.dumps(plan, ensure_ascii=False, indent=2)}
执行过程与结果:
{json.dumps(results, ensure_ascii=False, indent=2)}
初步答案: {answer}
请进行严格评估，并回答以下问题：
1. 这个答案在物理原理和数学计算上是否绝对正确？请指出任何潜在的错误（如：公式用错、单位遗漏、符号错误、数值计算失误）。
2. 解题过程是否完整？是否有遗漏的关键步骤？
3. 答案是否符合物理常识？（例如：速度不能超光速，能量不能为负等）
请用JSON格式回复，包含以下字段：
- "is_correct": boolean (答案是否完全正确)
- "critique": string (具体的、详细的批评和分析，指出错误所在)
- "suggestion": string (具体的修正建议，如果需要重新计算，请说明如何修改)
"""
def _call_llm_for_reflection(self, prompt: str) -> Dict[str, Any]:
"""调用LLM进行反思，强制返回JSON"""
try:
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"} # 强制JSON输出
)
content = response.choices[0].message.content
return json.loads(content)
except Exception as e:
# 反思失败，默认认为答案正确，避免阻塞流程
print(f" [反思调用失败] {e}")
return {"is_correct": True, "critique": "反思过程出错", "suggestion": ""}
def _apply_correction(self, current_answer: str, suggestion: str) -> str:
"""
根据反思建议修正答案。
这是一个简化实现。在真实系统中，这里可能触发重新执行某个计算步骤。
"""
# 示例：如果建议提到"符号错误"，我们手动修正
if "符号" in suggestion and "负" in suggestion:
if "14.00 m/s" in current_answer:
return current_answer.replace("14.00 m/s", "-14.00 m/s")
elif "14.0 m/s" in current_answer:
return current_answer.replace("14.0 m/s", "-14.0 m/s")
# 示例：如果建议提到"单位"，我们修正单位
if "单位" in suggestion and "km/h" in suggestion:
if "m/s" in current_answer:
# 简单模拟换算
parts = current_answer.split(" ")
for i, part in enumerate(parts):
if "m/s" in part:
try:
value = float(part.replace("m/s", ""))
kmh_value = value * 3.6
parts[i] = f"{kmh_value:.2f} km/h"
except:
pass
return " ".join(parts)
# 默认：在答案前加上"[修正后]"
return f"[修正后] {current_answer}"
# --- Plan-and-Execute 的核心方法 ---
def _create_plan(self, user_input: str) -> Optional[Dict[str, Any]]:
"""让LLM生成一个结构化的解题计划 (JSON格式)"""
planning_prompt = f"""
你是一个严谨的物理老师。请为以下物理问题制定一个详细的、分步骤的解题计划。
计划必须以JSON格式输出，包含步骤列表。每个步骤必须指定id、tool、args和dependencies。
可用工具说明 (请严格按照工具名和参数名):
- identify_problem: 从题目文本中提取关键物理量。args: {{"problem_text": str}}
- lookup_formula: 根据问题类型查找物理公式。args: {{"problem_type": str}} (如"自由落体"、"牛顿第二定律")
- calculate_result: 根据公式和数据进行计算。args: {{"formula": str, "variables": dict}} (如 {{"g": 9.8, "h": 10}})
- generate_answer: 生成最终的、面向学生的答案文本。args: {{"calculation_result": number, "unit": str, "problem_summary": str}}
请求: {user_input}
输出格式 (示例):
{{
"steps": [
{{"id": 1, "tool": "identify_problem", "args": {{"problem_text": "用户原始问题"}}, "dependencies": []}},
{{"id": 2, "tool": "lookup_formula", "args": {{"problem_type": "自由落体"}}, "dependencies": [1]}},
{{"id": 3, "tool": "calculate_result", "args": {{"formula": "v = sqrt(2gh)", "variables": {{"g": 9.8, "h": 10}}}}, "dependencies": [1, 2]}},
{{"id": 4, "tool": "generate_answer", "args": {{"calculation_result": 14.0, "unit": "m/s", "problem_summary": "求落地速度"}}, "dependencies": [3]}}
]
}}
重要: 只输出JSON，不要任何额外解释。确保dependencies正确反映步骤间的依赖关系。
"""
try:
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": planning_prompt}],
)
plan_text = response.choices[0].message.content.strip()
# 尝试提取并解析JSON
json_match = re.search(r'\{.*\}', plan_text, re.DOTALL)
if json_match:
return json.loads(json_match.group(0))
return json.loads(plan_text)
except (json.JSONDecodeError, Exception) as e:
print(f"JSON解析失败: {e}")
return None
def _check_dependencies(self, dependencies: List[int], results: Dict[int, Dict[str, Any]]) -> bool:
"""检查所有依赖步骤是否都已成功完成"""
for dep_id in dependencies:
if dep_id not in results or not results[dep_id]["success"]:
return False
return True
def _inject_dependency_data(self, args: Dict[str, Any], results: Dict[int, Dict[str, Any]]) -> Dict[str, Any]:
"""
将依赖步骤的结果数据，注入到当前步骤的参数中。
例如，步骤3的args中可能有占位符 "$step_1.mass$"，我们将其替换为步骤1的实际输出值。
"""
enriched_args = args.copy()
for key, value in args.items():
if isinstance(value, str) and value.startswith("$") and value.endswith("$"):
# 解析占位符，如 "$step_1.mass$"
parts = value[1:-1].split(".") # 去掉$，按.分割
if len(parts) == 2 and parts[0].startswith("step_"):
dep_step_id = int(parts[0].split("_")[1])
dep_field = parts[1]
if dep_step_id in results and results[dep_step_id]["success"]:
dep_result = results[dep_step_id]["result"]
if dep_field in dep_result:
enriched_args[key] = dep_result[dep_field]
else:
raise ValueError(f"依赖步骤 {dep_step_id} 的结果中缺少字段: {dep_field}")
else:
raise ValueError(f"依赖步骤 {dep_step_id} 未成功执行")
return enriched_args
def _extract_preliminary_answer(self, results: Dict[int, Dict[str, Any]]) -> Optional[str]:
"""从执行结果中提取初步答案"""
# 按步骤ID排序，取最后一个成功的、包含"answer"字段的结果
for step_id in sorted(results.keys(), reverse=True):
data = results[step_id]
if data["success"] and "answer" in data["result"]:
return data["result"]["answer"]
return None
# --- 工具函数实现 ---
def _identify_problem(self, problem_text: str) -> Dict[str, Any]:
"""从题目中提取关键物理量"""
# 简化实现：假设题目是标准格式
# 在真实项目中，这里可以用NER模型或规则引擎
data = {
"mass": 2.0, # kg
"height": 10.0, # m
"gravity": 9.8, # m/s²
"problem_type": "自由落体",
"target": "落地速度"
}
return {"success": True, **data}
def _lookup_formula(self, problem_type: str) -> Dict[str, Any]:
"""根据问题类型返回物理公式"""
formulas = {
"自由落体": "v = sqrt(2 * g * h)",
"匀加速直线运动": "v = v0 + a * t",
"动能定理": "W = 0.5 * m * v^2"
}
formula = formulas.get(problem_type, "未知公式")
return {"success": True, "formula": formula}
def _calculate_result(self, formula: str, variables: Dict[str, float]) -> Dict[str, Any]:
"""执行计算 (增强版：支持多种公式写法)"""
# 策略1: 根据公式的核心关键词判断
if "sqrt(2" in formula and ("g" in formula and "h" in formula):
# 提取 g 和 h 的值
g = variables.get("g", variables.get("gravity", 9.8))
h = variables.get("h", variables.get("height", 0.0))
# 执行核心计算逻辑
result = math.sqrt(2 * g * h)
return {"success": True, "raw_value": result, "unit": "m/s"}
# 策略2: 如果有其他公式类型，可以在这里添加 elif 分支
# elif "v = v0 + a * t" in formula:
# ... 计算匀加速运动 ...
# 如果都不匹配，返回错误
return {"success": False, "error": f"不支持的公式或参数: {formula}, {variables}"}
def _generate_answer(self, calculation_result: float, unit: str, problem_summary: str) -> Dict[str, Any]:
"""生成最终答案"""
answer_text = f"根据物理公式计算，{problem_summary}为 **{calculation_result:.2f} {unit}**。"
return {"success": True, "answer": answer_text}
def _generate_final_reply(self, original_query: str, final_answer: str) -> str:
"""生成最终回复"""
summary_prompt = f"""
原始问题: {original_query}
最终答案: {final_answer}
请用亲切、鼓励的语气，向学生解释这个答案，并简要回顾一下解题的关键步骤和反思过程，强调严谨思考的重要性。
"""
try:
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": summary_prompt}]
)
return response.choices[0].message.content
except Exception as e:
return f"生成最终回复时出错: {e}"
# --- 测试运行 ---
if __name__ == "__main__":
# 配置您的 OpenAI 客户端 (请替换为您的 API Key)
client = openai.OpenAI(
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
agent = Reflection(client, model="qwen-plus")
# 用户提问 (一个典型的易错题)
user_query = "一个物体从10米高的地方自由下落，求它落地时的速度大小。忽略空气阻力。(g=9.8m/s²)"
# 运行 Agent
agent.run(user_query)
五、混合模式：教育场景的终极武器
在真实的教育应用中，单一模式往往难以应对所有需求。混合模式 (Hybrid Mode) 是将多种 Agent 模式组合使用，发挥各自优势。
5.1 组合策略解析
| 任务类型 | 推荐混合模式 | Function Call 策略 |
|---|---|---|
| 制定学期学习规划 | Plan-and-Execute + ReAct | 先用 P&E 制定宏观计划（每月目标），再用 ReAct 执行每周/每日的具体任务（推送题目、安排答疑）。 |
| 智能批改与深度辅导 | ReAct + Reflection | 用 ReAct 快速调用 OCR 和答案比对工具得出初步批改结果；再用 Reflection 对学生的典型错误进行深度反思，生成个性化讲解。 |
| 复杂问题求解辅导 | Reflection + Plan-and-Execute | 先用 Reflection 确保每一步推理和计算的正确性；对于需要多工具协作的子任务，用 P&E 模式确保执行流程的清晰。 |
四种模式完整对比：
| 模式 | 核心思想 | Function Call 节奏 | 优势 | 劣势 | 教育场景应用 |
|---|---|---|---|---|---|
| ReAct | 边推理边行动 | 高频、交替 | 灵活、适应性强、适合探索性任务 | 缺乏全局观、可能陷入局部最优、易受幻觉影响 | 智能答疑、错题原因即时分析 |
| Plan-and-Execute | 先计划，后执行 | 低频、集中、受控 | 结构清晰、可预测性强、适合复杂流程 | 计划可能僵化、无法应对突发变化、前期规划成本高 | 制定学习计划、批量作业处理、课程内容编排 |
| Reflection | 执行后反思修正 | 不定频、事件驱动 | 结果可靠性高、容错性强、适合高精度任务 | 执行效率低、可能陷入反思循环、实现复杂 | 高难度题目批改、解题过程验证、个性化深度辅导 |
| 混合模式 | 组合拳 | 动态调整 | 取长补短、应对复杂多变场景 | 系统设计复杂、调试难度大 | 构建完整的 EduMentor 智能教育助手 |
5.2 设计原则
- 分层设计：将大任务分解，不同层级采用不同模式。顶层用 P&E 做规划，底层用 ReAct 做执行。
- 事件驱动：用 Reflection 作为"质量检查关卡”，在关键节点自动触发。
- 模式切换：设计一个"模式路由器”，根据当前任务的性质（简单/复杂、高风险/低风险）动态选择最合适的模式。
六、本节小结
本节深入剖析了 ReAct、Plan-and-Execute、Reflection 三种主流 Agent 架构模式的核心工作原理，并提供了完整的代码实现。关键要点：
- ReAct 适合探索性、实时性任务，高频调用 Function Call
- Plan-and-Execute 适合复杂流程、需要可预测性的任务，规划阶段不调用 Function Call
- Reflection 适合高精度要求任务，事件驱动式调用 Function Call
- 混合模式 是真实业务场景的终极方案，取长补短应对复杂需求