ReAct 与思维链：AI 的推理模式

用户: 25 * 4 + 10 = ?
AI: 110

用户: 25 * 4 + 10 = ?
AI: 先算 25 * 4 = 100，再算 100 + 10 = 110。所以答案是 110。

prompt = """
问题：小明有 5 个苹果，小红给了他 3 个，小明吃了 2 个，现在有多少个？
思考过程：
1. 小明原来有 5 个苹果
2. 小红给了他 3 个，5 + 3 = 8
3. 小明吃了 2 个，8 - 2 = 6
答案：6 个

问题：如果每袋子里有 12 个苹果，小明买了 3 袋子，又买了 2 个散装的，现在有多少个？
思考过程：
"""

问题：如果列文虎克最早发现的细菌是 1676 年，而青霉素是 1928 年发现的，两者相差多少年？
让我们一步步思考。

请按以下格式回答：
1. 理解问题：...
2. 分析条件：...
3. 推导过程：...
4. 得出结论：...

用户：帮我查一下今天北京的天气，如果下雨就提醒我带伞

ReAct 循环：
1. Thought：我需要先查询北京今天的天气
2. Action：调用天气 API
3. Observation：结果显示北京今天有雨，气温 15-20 度
4. Thought：根据天气结果，今天有雨，我需要提醒用户带伞
5. Action：生成提醒文本
6. Output：带伞提醒

prompt = """
你是一个智能助手，可以调用工具来完成任务。

可用工具：
- search(query): 搜索信息
- calculate(expression): 数学计算

请按照以下格式回答：
Thought: 你对当前情况的分析
Action: 你要执行的行动（如果是最终答案，输出 Finish）
Action Input: 行动的输入
Observation: 行动的结果

开始！

问题：{question}
"""

def react_agent(question, max_iterations=10):
    context = []
    
    for i in range(max_iterations):
        # 1. 生成 Thought + Action
        response = llm.generate(context + [f"问题：{question}"])
        
        # 2. 解析动作
        thought = extract_thought(response)
        action = extract_action(response)
        
        # 3. 执行动作
        if action == "Finish":
            return extract_answer(response)
        
        result = execute_tool(action)
        
        # 4. 添加观察结果
        context.append(f"Thought: {thought}")
        context.append(f"Action: {action}")
        context.append(f"Observation: {result}")
    
    return "达到最大迭代次数"

available_tools = {
    "search": search_function,
    "calculate": calculate_function,
    "lookup": lookup_function,
    "send_email": send_email_function,
}

def execute_tool(action):
    tool_name, tool_input = parse_action(action)
    return available_tools[tool_name](tool_input)

Thought：首先让我分析这个问题...（CoT 模式）
Action：基于分析，我需要...（ReAct 模式）

用户：爱因斯坦什么时候出生？
Self-Ask：我应该先查一下爱因斯坦的出生日期...
Action：search("爱因斯坦 出生 日期")
Observation：1879年3月14日
回答：爱因斯坦出生于 1879 年 3 月 14 日

Thought：这个任务可以分成三步：1) 查天气 2) 判断是否下雨 3) 生成提醒
Action：先执行第1步
...

def hybrid_agent(task):
    # 第一阶段：用 CoT 分析任务
    analysis = cot_analyze(task)
    
    # 第二阶段：根据分析结果选择模式
    if needs_external_data(analysis):
        return react_execute(analysis)
    elif has_clear_steps(analysis):
        return plan_execute(analysis)
    else:
        return cot_generate(analysis)

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain.tools import tool

@tool
def search_wiki(query: str):
    """搜索维基百科"""
    # 实现搜索逻辑
    return wikipedia.search(query)

@tool  
def calculate(expr: str):
    """计算数学表达式"""
    return eval(expr)

# 定义状态
class AgentState(dict):
    question: str
    thought: str
    action: str
    action_input: str
    observation: str
    answer: str
    iterations: int

# 推理节点
def reason_node(state: AgentState):
    llm = ChatOpenAI(model="gpt-4")
    
    prompt = f"""
    问题：{state['question']}
    之前的观察：{state.get('observation', '无')}
    
    你需要决定下一步做什么。可用工具：search_wiki, calculate
    请按以下格式回答：
    Thought: 你的分析
    Action: 工具名 或 Finish
    Action Input: 工具输入 或 最终答案
    """
    
    response = llm.invoke(prompt)
    # 解析 response...
    return {"thought": thought, "action": action, "action_input": input_}

# 执行节点
def act_node(state: AgentState):
    tool = get_tool(state["action"])
    result = tool.invoke(state["action_input"])
    return {"observation": result, "iterations": state["iterations"] + 1}

# 构建图
graph = StateGraph(AgentState)
graph.add_node("reason", reason_node)
graph.add_node("act", act_node)
graph.set_entry_point("reason")

# 添加条件边
def should_continue(state):
    if state["action"] == "Finish":
        return "end"
    return "act"

graph.add_conditional_edges("reason", should_continue)
graph.add_edge("act", "reason")

app = graph.compile()