第17课：ReAct模式

📚 ReAct模式概述

本课深入ReAct (Reasoning + Acting)模式——目前最主流的Agent决策范式。ReAct让Agent在每一步都先推理(Reason)下一步该做什么，然后行动(Act)执行工具，再根据观察(Observation)继续推理。

📊 ReAct循环全景

第17课: ReAct模式
├── ReAct循环
│   ├── Thought: 推理当前状态
│   ├── Action: 选择并执行工具
│   └── Observation: 观察工具返回结果
├── 三种模式对比
│   ├── CoT: 只推理不行动
│   ├── Act-only: 只行动不推理
│   └── ReAct: 推理+行动交替
├── ReActAgent
│   ├── 提示词模板
│   ├── 解析器
│   └── 执行器
└── 带反思ReAct
    ├── 自我评估
    ├── 修正行动
    └── 迭代优化

🔍 ReAct循环详解

ReAct核心是推理与行动交替进行。每一步先思考(Thought)，执行行动(Action)，观察结果(Observation)，再进入下一轮思考。

# ReAct循环: Thought → Action → Observation
from openai import OpenAI
import json
import re
from typing import Dict, Callable, List, Optional

client = OpenAI()

class ReActLoop:
    """ReAct循环的最简实现 - 核心流程: Thought → Action → Observation"""
    REACT_PROMPT = """你是一个使用ReAct模式的智能Agent。每一步:1. Thought: 分析当前情况，推理下一步2. Action: 选择工具执行，格式为 Action: 工具名(参数)3. Observation: 你将看到工具的执行结果当你得出最终答案时:Final Answer: 你的最终答案可用工具:{tool_descriptions}"""

    def __init__(self, model="gpt-4o-mini"):
        self.model = model
        self.tools: Dict[str, Callable] = {}
        self.thoughts: List[str] = []
        self.actions: List[str] = []
        self.observations: List[str] = []

    def register_tool(self, name: str, description: str, handler: Callable):
        self.tools[name] = {"description": description, "handler": handler}

    def _get_tool_descriptions(self) -> str:
        return "\n".join(f"- {name}: {info['description']}" for name, info in self.tools.items())

    def _parse_action(self, text: str) -> Optional[tuple]:
        match = re.search(r'Action:\s*(\w+)(?:\((.+?)\))?', text)
        if match: return match.group(1), match.group(2) or ""
        return None

    def _execute_tool(self, tool_name: str, args: str) -> str:
        if tool_name not in self.tools: return f"错误: 未知工具 '{tool_name}'"
        try: return str(self.tools[tool_name]["handler"](args))
        except Exception as e: return f"工具执行错误: {e}"

    def run(self, task: str, max_steps: int = 6) -> str:
        system_prompt = self.REACT_PROMPT.format(tool_descriptions=self._get_tool_descriptions())
        messages = [{"role": "system", "content": system_prompt}, {"role": "user", "content": task}]
        for step in range(max_steps):
            response = client.chat.completions.create(model=self.model, messages=messages, temperature=0.1)
            text = response.choices[0].message.content or ""
            messages.append({"role": "assistant", "content": text})
            final_match = re.search(r'Final Answer:\s*(.+)', text, re.DOTALL)
            if final_match: return final_match.group(1).strip()
            action_result = self._parse_action(text)
            if action_result:
                tool_name, args = action_result
                observation = self._execute_tool(tool_name, args)
                self.thoughts.append(text)
                self.actions.append(f"{tool_name}({args})")
                self.observations.append(observation)
                messages.append({"role": "user", "content": f"Observation: {observation}"})
            else:
                self.thoughts.append(text)
        return "ReAct循环达到最大步数限制"

    def get_trace(self) -> str:
        lines = []
        for i, (t, a, o) in enumerate(zip(self.thoughts, self.actions, self.observations)):
            lines.append(f"Step {i+1}: Thought → Action: {a} → Observation: {o[:100]}")
        return "\n".join(lines)

# 使用
react = ReActLoop()
react.register_tool("search", "搜索信息", lambda q: f"搜索结果: {q}")
react.register_tool("calculate", "计算数学表达式", lambda e: str(eval(e)) if all(c in "0123456789+-*/.() " for c in e) else "错误")
# result = react.run("搜索Python最新版本号，计算它乘以2")

✅ 验证通过：ReAct循环实现了完整的Thought→Action→Observation流程

🧠 ReAct vs CoT vs Act-only

三种推理-行动模式对比：CoT纯思考无工具、Act-only盲目执行无推理、ReAct两者兼顾。

# ReAct vs CoT vs Act-only: 三种模式对比实验
from openai import OpenAI
import json
import time
from typing import Dict, List

client = OpenAI()

class CoTAgent:
    """Chain of Thought: 纯推理，不使用工具"""
    def run(self, task: str) -> dict:
        start = time.time()
        response = client.chat.completions.create(model="gpt-4o-mini",
            messages=[{"role": "system", "content": "请一步步思考并回答问题。"}, {"role": "user", "content": task}])
        return {"mode": "CoT", "answer": response.choices[0].message.content,
                "tokens": response.usage.total_tokens if response.usage else 0,
                "time_ms": (time.time() - start) * 1000, "tool_calls": 0}

class ActOnlyAgent:
    """Act-only: 盲目行动，不推理"""
    def __init__(self, tools: Dict):
        self.tools = tools

    def run(self, task: str, max_actions: int = 5) -> dict:
        start = time.time()
        total_tokens = 0
        actions = []
        messages = [{"role": "system", "content": "你可以使用工具。直接执行操作。"}, {"role": "user", "content": task}]
        tool_schemas = [{"type": "function", "function": {"name": n, "description": i["desc"], "parameters": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}}} for n, i in self.tools.items()]
        for _ in range(max_actions):
            response = client.chat.completions.create(model="gpt-4o-mini", messages=messages, tools=tool_schemas, tool_choice="auto")
            total_tokens += response.usage.total_tokens if response.usage else 0
            msg = response.choices[0].message
            messages.append(msg.to_dict())
            if not msg.tool_calls:
                return {"mode": "Act-only", "answer": msg.content, "tokens": total_tokens, "time_ms": (time.time() - start) * 1000, "tool_calls": len(actions)}
            for tc in msg.tool_calls:
                result = self.tools[tc.function.name]["handler"](json.loads(tc.function.arguments).get("query", ""))
                actions.append(tc.function.name)
                messages.append({"role": "tool", "tool_call_id": tc.id, "content": result})
        return {"mode": "Act-only", "answer": "达到最大操作次数", "tokens": total_tokens, "time_ms": (time.time() - start) * 1000, "tool_calls": len(actions)}

class ReActAgent:
    """ReAct: 推理+行动交替"""
    def __init__(self, tools: Dict):
        self.tools = tools

    def run(self, task: str, max_steps: int = 6) -> dict:
        start = time.time()
        total_tokens = 0
        thoughts, actions = [], []
        tool_desc = "\n".join(f"- {n}: {i['desc']}" for n, i in self.tools.items())
        messages = [{"role": "system", "content": f"你使用ReAct模式工作。每步先思考(Thought)再行动(Action)。
可用工具:
{tool_desc}
最终用 Final Answer: 答案 结束。"},
                    {"role": "user", "content": task}]
        tool_schemas = [{"type": "function", "function": {"name": n, "description": i["desc"], "parameters": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}}} for n, i in self.tools.items()]
        for _ in range(max_steps):
            response = client.chat.completions.create(model="gpt-4o-mini", messages=messages, tools=tool_schemas, tool_choice="auto")
            total_tokens += response.usage.total_tokens if response.usage else 0
            msg = response.choices[0].message
            messages.append(msg.to_dict())
            if not msg.tool_calls:
                return {"mode": "ReAct", "answer": msg.content, "tokens": total_tokens, "time_ms": (time.time() - start) * 1000, "tool_calls": len(actions), "thoughts": thoughts}
            if msg.content: thoughts.append(msg.content[:200])
            for tc in msg.tool_calls:
                result = self.tools[tc.function.name]["handler"](json.loads(tc.function.arguments).get("query", ""))
                actions.append(tc.function.name)
                messages.append({"role": "tool", "tool_call_id": tc.id, "content": result})
        return {"mode": "ReAct", "answer": "达到最大步数", "tokens": total_tokens, "time_ms": (time.time() - start) * 1000, "tool_calls": len(actions), "thoughts": thoughts}

✅ 验证通过：三种模式的Agent有各自独立的决策逻辑

🔧 ReActAgent完整实现

生产级ReActAgent，包含结构化提示词、function calling模式、轨迹追踪。

# ReActAgent: 生产级ReAct模式Agent
from openai import OpenAI
import json
import time
from typing import Callable, Dict, List, Optional
from dataclasses import dataclass, field

client = OpenAI()

@dataclass
class ReActStep:
    """ReAct单步记录"""
    step_number: int
    thought: str = ""
    action_name: str = ""
    action_args: str = ""
    observation: str = ""
    duration_ms: float = 0.0

@dataclass
class ReActTrace:
    """ReAct完整执行轨迹"""
    task: str
    steps: List[ReActStep] = field(default_factory=list)
    final_answer: str = ""
    total_duration_ms: float = 0.0
    total_tokens: int = 0
    success: bool = False

    def format_trace(self) -> str:
        lines = [f"任务: {self.task}"]
        for s in self.steps:
            lines.append(f"Step {s.step_number}: Thought={s.thought[:80]} Action={s.action_name} Obs={s.observation[:80]}")
        lines.append(f"Final Answer: {self.final_answer}")
        return "\n".join(lines)

class ReActAgent:
    """生产级ReAct Agent - 结构化提示词/function calling/执行轨迹"""
    def __init__(self, name: str = "ReActAgent", model: str = "gpt-4o-mini"):
        self.name = name
        self.model = model
        self._tools: Dict[str, dict] = {}
        self._trace: Optional[ReActTrace] = None

    def register(self, name: str, description: str, parameters: dict, handler: Callable) -> "ReActAgent":
        self._tools[name] = {"schema": {"type": "function", "function": {"name": name, "description": description, "parameters": parameters}}, "handler": handler}
        return self

    def _build_system_prompt(self) -> str:
        tool_list = "\n".join(f"  - {name}: {info['schema']['function']['description']}" for name, info in self._tools.items())
        return f"你是{self.name}，使用ReAct模式工作。每步先推理(Thought)再行动(Action)。当你确定最终答案时不再调用工具即可。\n\n可用工具:\n{tool_list}\n\n规则: 每步必须先思考再行动; 优先使用工具获取信息; 最多执行8步。"

    def run(self, task: str, max_steps: int = 8, verbose: bool = False) -> ReActTrace:
        trace = ReActTrace(task=task)
        start_time = time.time()
        messages = [{"role": "system", "content": self._build_system_prompt()}, {"role": "user", "content": task}]
        for step_num in range(1, max_steps + 1):
            step_start = time.time()
            response = client.chat.completions.create(model=self.model, messages=messages, tools=[t["schema"] for t in self._tools.values()], tool_choice="auto")
            step_tokens = response.usage.total_tokens if response.usage else 0
            trace.total_tokens += step_tokens
            msg = response.choices[0].message
            messages.append(msg.to_dict())
            step = ReActStep(step_number=step_num, thought=msg.content or "", duration_ms=(time.time() - step_start) * 1000)
            if not msg.tool_calls:
                trace.steps.append(step)
                trace.final_answer = msg.content or ""
                trace.success = True
                break
            for tc in msg.tool_calls:
                step.action_name = tc.function.name
                step.action_args = tc.function.arguments
                try:
                    args = json.loads(tc.function.arguments)
                    result = self._tools[tc.function.name]["handler"](**args)
                except Exception as e:
                    result = f"执行错误: {e}"
                step.observation = str(result)[:500]
                messages.append({"role": "tool", "tool_call_id": tc.id, "content": step.observation})
            trace.steps.append(step)
            if verbose: print(f"Step {step_num}: {step.action_name} → {step.observation[:100]}")
        trace.total_duration_ms = (time.time() - start_time) * 1000
        if not trace.success: trace.final_answer = "达到最大步数限制"
        self._trace = trace
        return trace

# 使用
agent = ReActAgent(name="ResearchAgent")
agent.register("search", "搜索信息", {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}, lambda query: f"搜索 '{query}' 结果")
# trace = agent.run("搜索Python最新版本", verbose=True)
# print(trace.format_trace())

✅ 验证通过：ReActAgent实现了结构化ReAct提示词、function calling和ReActTrace数据类

🔄 带反思的ReAct

在标准ReAct基础上加入自我反思——Agent在得出结论后评估质量，不满意则调整策略重新执行。

# ReflectiveReAct: 带反思的ReAct Agent
from openai import OpenAI
import json
from typing import Callable, Dict, List
from dataclasses import dataclass, field

client = OpenAI()

@dataclass
class Reflection:
    round_number: int
    initial_answer: str
    self_evaluation: str
    score: float
    improvement_plan: str
    improved_answer: str = ""

class ReflectiveReAct:
    """带反思的ReAct Agent - 执行→评估→反思→改进循环"""
    REFLECTION_PROMPT = "评估任务执行结果:\n任务: {task}\n回答: {answer}\n\n输出JSON: {{"score": 0.8, "evaluation": "评估说明", "improvement_plan": "改进建议"}}"

    def __init__(self, name="ReflectiveReAct", model="gpt-4o-mini"):
        self.name = name
        self.model = model
        self._tools: Dict[str, dict] = {}
        self._reflections: List[Reflection] = []

    def register(self, name: str, description: str, parameters: dict, handler: Callable):
        self._tools[name] = {"schema": {"type": "function", "function": {"name": name, "description": description, "parameters": parameters}}, "handler": handler}

    def _run_react(self, task: str, max_steps: int = 6) -> str:
        tool_list = "\n".join(f"  - {n}: {t['schema']['function']['description']}" for n, t in self._tools.items())
        messages = [{"role": "system", "content": f"你是{self.name}，使用ReAct模式。\n可用工具:\n{tool_list}"}, {"role": "user", "content": task}]
        for _ in range(max_steps):
            resp = client.chat.completions.create(model=self.model, messages=messages, tools=[t["schema"] for t in self._tools.values()], tool_choice="auto")
            msg = resp.choices[0].message
            messages.append(msg.to_dict())
            if not msg.tool_calls: return msg.content or ""
            for tc in msg.tool_calls:
                try:
                    args = json.loads(tc.function.arguments)
                    result = self._tools[tc.function.name]["handler"](**args)
                except Exception as e:
                    result = str(e)
                messages.append({"role": "tool", "tool_call_id": tc.id, "content": str(result)})
        return "ReAct达到最大步数"

    def _reflect(self, task: str, answer: str) -> Reflection:
        prompt = self.REFLECTION_PROMPT.format(task=task, answer=answer)
        resp = client.chat.completions.create(model=self.model, messages=[{"role": "system", "content": "你是严格的质量评估者。"}, {"role": "user", "content": prompt}], response_format={"type": "json_object"})
        try:
            data = json.loads(resp.choices[0].message.content)
            return Reflection(round_number=len(self._reflections) + 1, initial_answer=answer, self_evaluation=data.get("evaluation", ""), score=float(data.get("score", 0.5)), improvement_plan=data.get("improvement_plan", ""))
        except (json.JSONDecodeError, ValueError):
            return Reflection(round_number=len(self._reflections) + 1, initial_answer=answer, self_evaluation="评估失败", score=0.5, improvement_plan="")

    def run(self, task: str, max_reflections: int = 3, score_threshold: float = 0.8) -> str:
        self._reflections = []
        current_task = task
        for round_num in range(1, max_reflections + 1):
            answer = self._run_react(current_task)
            reflection = self._reflect(task, answer)
            self._reflections.append(reflection)
            if reflection.score >= score_threshold:
                reflection.improved_answer = answer
                return answer
            current_task = f"原始任务: {task}\n\n上一轮回答: {answer}\n评估: {reflection.self_evaluation}\n改进: {reflection.improvement_plan}\n\n请根据改进计划重新执行。"
        return answer

    def get_reflections(self) -> List[dict]:
        return [{"round": r.round_number, "score": r.score, "evaluation": r.self_evaluation, "improvement": r.improvement_plan} for r in self._reflections]

✅ 验证通过：ReflectiveReAct实现了执行→评估→反思→改进的完整循环

┌──────────────────────────────────┐
│   ReAct决策流程
├──────────────────────────────────┤
│  Thought → Action → Observation
│      ↓         ↓          ↓
│   推理     执行工具    观察结果
│      ↓___________________↓
│            下一轮 Thought
├──────────────────────────────────┤
│  终止条件: 无工具调用 = 最终答案
└──────────────────────────────────┘

🔗 与其他课程的关系

• 前置知识：第15课「Agent架构」、第16课「工具使用」
• 后续应用：第18课「规划与反思」
• ReAct是后续Multi-Agent、LangGraph的基础决策模式

# 挑战: 构建自适应ReAct Agent
# 简单任务 → 纯CoT
# 需要信息 → 标准ReAct
# 高质量要求 → Reflective ReAct