ReAct：推理与行动的循环

# 纯推理（CoT）的局限：无法获取外部信息
prompt = """
问：埃菲尔铁塔有多高？
答：让我一步步思考。
埃菲尔铁塔是巴黎的地标建筑...
（模型只能基于训练数据回答，可能不准确或过时）
"""

# 纯行动（Toolformer）的局限：缺少策略思考
prompt = """
问：比较北京和上海今天的天气
[调用 weather_api("北京")]
[调用 weather_api("上海")]
（如果API返回异常，模型没有推理能力来处理）

# ReAct 范式：推理与行动交替
react_trace = """
Thought: 用户想知道北京和上海今天的天气差异，
        我需要分别查询两个城市的天气信息。
Action: weather_api
Action Input: {"city": "北京"}
Observation: {"temp": 22, "condition": "晴"}
Thought: 已获取北京天气，现在查询上海。
Action: weather_api
Action Input: {"city": "上海"}
Observation: {"temp": 25, "condition": "多云"}
Thought: 北京22度晴天，上海25度多云，
        上海比北京高3度。可以给出答案了。
Answer: 上海今天25度多云，北京22度晴天，上海比北京高3度。
"""

# ReAct 论文中的提示模板结构
REACT_PROMPT = """
Answer the following questions as best you can.
You have access to the following tools:

{tools}

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Begin!

Question: {input}
Thought: {agent_scratchpad}
"""

# 论文中的 Few-shot 示例（简化版）
FEW_SHOT_EXAMPLE = """
Question: What profession does Nicholas Ray and Elia Kazan have in common?
Thought: I need to search Nicholas Ray and Elia Kazan, find their
        professions, then find the common one.
Action: Search
Action Input: "Nicholas Ray"
Observation: Nicholas Ray (born Raymond Nicholas Kienzle Jr., 1911-1979)
             was an American film director, screenwriter, and actor.
Thought: Nicholas Ray is a film director. Now I need to search
        Elia Kazan.
Action: Search
Action Input: "Elia Kazan"
Observation: Elia Kazan was a Greek-American director, producer,
             screenwriter, and actor.
Thought: Both are film directors. I can answer now.
Final Answer: Film director.
"""

# Thought-Action-Observation 循环的伪代码实现
class ReActLoop:
    def __init__(self, llm, tools, max_iterations=10):
        self.llm = llm
        self.tools = tools
        self.max_iterations = max_iterations
    
    def run(self, question: str) -> str:
        scratchpad = ""
        for i in range(self.max_iterations):
            # Thought: 模型决定下一步
            response = self.llm.generate(
                prompt=question, scratchpad=scratchpad
            )
            thought, action, action_input = response.parse()
            
            if response.is_final_answer:
                return response.final_answer
            
            # Action: 执行工具调用
            observation = self.tools.execute(action, action_input)
            
            # 更新 scratchpad，进入下一轮
            scratchpad += f"\nThought: {thought}\n"
            scratchpad += f"Action: {action}\n"
            scratchpad += f"Action Input: {action_input}\n"
            scratchpad += f"Observation: {observation}\n"
        
        return "达到最大步数，无法完成"

# 一个完整的 TAO 循环示例（追踪天气比较）
# Iteration 1
thought_1 = "我需要查询北京和上海的天气来比较"
action_1 = "weather_query"
input_1 = '{"city": "北京", "field": "temperature"}'
obs_1 = '{"city": "北京", "temp": 22, "unit": "celsius"}'

# Iteration 2
thought_2 = "北京是22度，还需要查上海的温度"
action_2 = "weather_query"
input_2 = '{"city": "上海", "field": "temperature"}'
obs_2 = '{"city": "上海", "temp": 25, "unit": "celsius"}'

# Iteration 3
thought_3 = "北京22度，上海25度，上海更高"
final_answer = "上海比北京高3度（25°C vs 22°C）"

print(f"共执行 {3} 轮 TAO 循环得到答案")

# CoT vs ReAct 的 prompt 对比
# CoT prompt - 只思考不行动
COT_PROMPT = """
Question: 如果一家公司年收入增长20%，成本增长15%，
        利润率会如何变化？
Let's think step by step.
（模型在内部推理，不调用任何工具）
"""

# ReAct prompt - 边思考边行动
REACT_PROMPT = """
Question: Apple 公司 2024 年 Q3 的营收是多少？
Thought: 我需要查询 Apple 最新的财报数据。
Action: search
Action Input: "Apple Q3 2024 revenue"
（模型先思考，然后调用搜索工具获取真实数据）
"""

# ToT vs ReAct 的搜索策略对比
# ToT：广度优先探索多个推理路径
def tot_search(problem, depth=3, breadth=5):
    """Tree of Thoughts - 维护一棵推理树"""
    tree = {problem: []}  # 根节点
    for level in range(depth):
        for node in get_current_leaves(tree):
            # 每个节点生成多个候选
            candidates = generate_thoughts(node, breadth)
            # 评估并选择最优
            scored = [(c, evaluate(c)) for c in candidates]
            tree[node] = sorted(scored, key=lambda x: -x[1])[:2]
    return extract_best_path(tree)

# ReAct：线性但带真实反馈
def react_search(problem, max_steps=6):
    """ReAct - 线性推理但每步有真实工具反馈"""
    scratchpad = ""
    for step in range(max_steps):
        thought, action = generate_next(problem, scratchpad)
        if is_answer(thought):
            return extract_answer(thought)
        observation = execute_tool(action)  # 真实反馈
        scratchpad += f"\n{thought}\n{action}\n{observation}"

# 搜索工具集成（使用 DuckDuckGo）
from duckduckgo_search import DDGS
from typing import Dict

class SearchTool:
    name = "search"
    description = "搜索互联网获取实时信息"
    
    def execute(self, query: str) -> str:
        """执行搜索并返回格式化的结果"""
        with DDGS() as ddgs:
            results = list(ddgs.text(query, max_results=3))
        
        formatted = []
        for i, r in enumerate(results, 1):
            formatted.append(
                f"[{i}] {r['title']}\n"
                f"    {r['body']}\n"
                f"    URL: {r['href']}"
            )
        return "\n\n".join(formatted)

tool = SearchTool()
print(tool.execute("2024年诺贝尔物理学奖"))

# 代码执行工具（带沙箱安全控制）
import subprocess
import re

class CodeExecutionTool:
    name = "python_repl"
    description = "执行 Python 代码并返回结果"
    
    ALLOWED_MODULES = {"math", "json", "re", "datetime", "collections"}
    FORBIDDEN_PATTERNS = [
        r"import\s+os", r"import\s+sys", r"subprocess",
        r"open\(", r"__import__", r"eval\(", r"exec\("
    ]
    
    def _is_safe(self, code: str) -> bool:
        for pattern in self.FORBIDDEN_PATTERNS:
            if re.search(pattern, code):
                return False
        return True
    
    def execute(self, code: str) -> str:
        if not self._is_safe(code):
            return "Error: 代码包含不安全的操作"
        result = subprocess.run(
            ["python3", "-c", code],
            capture_output=True, text=True, timeout=10
        )
        return result.stdout or result.stderr

# 用 LangChain 构建 ReAct Agent 的完整示例
from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain.tools import Tool
from langchain import hub

# 1. 定义工具
def search_wikipedia(query: str) -> str:
    """搜索 Wikipedia 获取百科知识"""
    import wikipedia
    try:
        return wikipedia.summary(query, sentences=3)
    except:
        return f"未找到关于 {query} 的信息"

def calculate(expression: str) -> str:
    """计算数学表达式"""
    return str(eval(expression, {"__builtins__": {}}, {}))

tools = [
    Tool(
        name="Wikipedia",
        func=search_wikipedia,
        description="搜索 Wikipedia 获取人物、事件、概念的百科信息"
    ),
    Tool(
        name="Calculator",
        func=calculate,
        description="计算数学表达式，如 2**10 或 3.14*5*5"
    ),
]

# 2. 创建 ReAct Agent
llm = ChatOpenAI(model="gpt-4", temperature=0)
prompt = hub.pull("hwchase17/react")
agent = create_react_agent(llm, tools, prompt)
executor = AgentExecutor(
    agent=agent, tools=tools, verbose=True,
    max_iterations=8, handle_parsing_errors=True
)

# 3. 运行
result = executor.invoke({
    "input": "如果一个圆的半径是5cm，它的面积是多少？"
})
print(result["output"])

# 自定义 ReAct 提示词模板
from langchain.prompts import PromptTemplate

CUSTOM_REACT_PROMPT = PromptTemplate.from_template("""
你是一个专业的研究助手，擅长使用工具来回答问题。
你可以使用以下工具：

{tools}

请按以下格式回答：

Question: 你需要回答的问题
Thought: 思考下一步应该做什么
Action: 工具名称，必须是 [{tool_names}] 之一
Action Input: 工具的输入
Observation: 工具执行的结果
... （可以重复 Thought/Action/Observation 多轮）
Thought: 我现在知道最终答案了
Final Answer: 对原始问题的最终回答

开始！

Question: {input}
Thought: {agent_scratchpad}
""")

# 使用自定义提示词创建 Agent
agent = create_react_agent(llm, tools, CUSTOM_REACT_PROMPT)
executor = AgentExecutor(
    agent=agent, tools=tools, verbose=True,
    max_iterations=6
)

# ReAct Agent 的多维度评估
from dataclasses import dataclass
from typing import List

@dataclass
class ReactMetrics:
    accuracy: float           # 答案正确率
    avg_steps: float          # 平均步骤数
    avg_tokens: float         # 平均 token 消耗
    tool_error_rate: float    # 工具调用失败率
    parse_error_rate: float   # 解析失败率
    self_correction_rate: float  # 自我纠正率

def evaluate_agent(agent, test_cases: List[dict]) -> ReactMetrics:
    results = []
    for case in test_cases:
        trace = agent.run_with_trace(case["question"])
        results.append({
            "correct": trace.final_answer == case["answer"],
            "steps": len(trace.ta_iterations),
            "tokens": trace.total_tokens,
            "tool_errors": trace.tool_errors,
            "parse_errors": trace.parse_errors,
            "corrections": trace.self_corrections,
        })
    
    n = len(results)
    return ReactMetrics(
        accuracy=sum(r["correct"] for r in results) / n,
        avg_steps=sum(r["steps"] for r in results) / n,
        avg_tokens=sum(r["tokens"] for r in results) / n,
        tool_error_rate=sum(r["tool_errors"] for r in results) / n,
        parse_error_rate=sum(r["parse_errors"] for r in results) / n,
        self_correction_rate=sum(r["corrections"] for r in results) / n,
    )

# Reflexion 改进：自我反思式 ReAct
class ReflexionAgent:
    """在 ReAct 基础上增加自我反思机制"""
    def __init__(self, base_agent, reflection_model=None):
        self.base_agent = base_agent
        self.reflection_model = reflection_model or base_agent.llm
        self.reflections = []  # 历史反思记录
    
    def run(self, question: str) -> str:
        # 第一次尝试
        result = self.base_agent.run(question)
        
        if not result.success:
            # 反思：为什么失败了？
            reflection = self.reflection_model.generate(
                f"任务：{question}\n"
                f"执行轨迹：{result.trace}\n"
                f"结果：失败 - {result.error}\n"
                f"请分析失败原因并给出改进建议。"
            )
            self.reflections.append(reflection)
            
            # 第二次尝试（带上反思）
            result = self.base_agent.run(
                question, context=f"之前的教训：{reflection}"
            )
        return result