ReAct Prompt 模式：推理与行动的交织

问题 → Thought → Action → Observation → Thought → Action → ... → Answer
        ↑                                    ↑
     内部推理                            环境反馈

Thought 1: 分析问题，决定第一步
Action 1: 调用工具
Observation 1: 工具返回结果

Thought 2: 基于观察决定下一步
Action 2: 调用工具
Observation 2: 工具返回结果

Thought N: 已有足够信息
Action: FINISH
Answer: 最终答案

问题: 2024年奥斯卡最佳影片导演出生城市人口？
Thought 1: 先找最佳影片和导演
Action 1: search("2024 Oscar best picture")
Observation 1: 《奥本海默》获最佳影片，导演诺兰。
Thought 2: 诺兰出生城市
Action 2: search("Nolan birthplace")
Observation 2: 出生于伦敦。
Thought 3: 伦敦人口
Action 3: search("London population")
Observation 3: 约898万。
Thought 4: 信息齐全
Action 4: FINISH
Answer: 诺兰出生于伦敦，人口约898万。

from dataclasses import dataclass
from typing import Callable

@dataclass
class Tool:
    name: str
    description: str
    func: Callable
    parameters: dict
    
    def execute(self, **kwargs) -> str:
        try:
            return str(self.func(**kwargs))
        except Exception as e:
            return f"Error: {e}"

tools = [
    Tool("search", "搜索网络", search_web, 
         {"query": {"type": "string"}}),
    Tool("calculate", "数学计算", calculate,
         {"expression": {"type": "string"}}),
]

REACT_PROMPT = """你是能使用工具的智能助手。

可用工具:
{tools_description}

格式:
Thought: 推理过程
Action: 工具名称
Action Input: JSON参数
Observation: (系统自动填入)

完成后:
Thought: 最终推理
Action: FINISH
Final Answer: 最终答案

最多 {max_iterations} 轮。

问题: {question}
Thought 1:"""

class ReActEngine:
    def __init__(self, llm, tools, max_iter=10):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
        self.max_iter = max_iter
    
    def run(self, question: str) -> dict:
        history = ""
        for i in range(self.max_iter):
            response = self.llm.generate(self._build_prompt(question, history), temperature=0)
            parsed = self._parse(response)
            if parsed.get('action') == 'FINISH':
                return {'answer': parsed.get('final_answer', ''), 'iterations': i+1}
            name = parsed.get('action', '')
            obs = self.tools[name].execute(**parsed.get('input', {})) if name in self.tools else f"未知工具: {name}"
            history += f"\nThought: {parsed['thought']}\nAction: {name}\nObservation: {obs}\n"
        return {'answer': '达到最大迭代', 'timeout': True}

class RobustReActEngine(ReActEngine):
    def _check_loop(self, trace, window=3) -> bool:
        if len(trace) < window * 2:
            return False
        recent = trace[-window:]
        for i in range(len(trace) - window):
            if all(r['action'] == s['action'] for r, s in zip(recent, trace[i:i+window])):
                return True
        return False