推理模式：ReAct与Plan-and-Execute

1

Thought（思考） → Action（行动） → Observation（观察） → Thought → ...

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143

# react_agent.py
# ReAct 推理引擎的纯 Python 实现，不依赖任何框架
import re
from typing import Callable

from openai import OpenAI

client = OpenAI()  # 默认从环境变量 OPENAI_API_KEY 读取密钥
MODEL = "gpt-5.4"
MAX_STEPS = 10  # 安全阀：防止智能体陷入无限循环


# ---------- 工具定义 ----------
def search_wiki(query: str) -> str:
    """模拟维基百科检索工具，实际项目可替换为真实API"""
    fake_db = {
        "北京": "北京是中华人民共和国首都，常住人口约2189万（2020年）。",
        "上海": "上海是中国直辖市，GDP居中国城市首位（2023年约4.72万亿元）。",
        "深圳": "深圳是中国经济特区之一，位于广东省，常住人口约1768万。",
    }
    # 简单的关键词匹配，实际项目用向量检索或API
    for key, val in fake_db.items():
        if key in query:
            return val
    return f"未找到与'{query}'相关的信息。"


def calculate(expression: str) -> str:
    """安全计算器工具，支持基本四则运算"""
    try:
        # 只允许数字和运算符，防止代码注入
        if not re.match(r'^[\d\s+\-*/().]+$', expression):
            return "错误：表达式包含非法字符"
        # ⚠️ eval 有安全风险，这里靠正则白名单防护；生产环境建议用 ast.literal_eval
        result = eval(expression, {"__builtins__": {}}, {})
        return f"{expression} = {result}"
    except Exception as e:
        return f"计算错误：{e}"


# 工具注册表：名字 -> (函数, 描述, 参数说明)
# 智能体通过这张表知道"自己有哪些工具、每个工具怎么用"
TOOLS: dict[str, tuple[Callable, str, str]] = {
    "search_wiki": (search_wiki, "检索维基百科获取城市信息", "query: 检索关键词"),
    "calculate": (calculate, "执行四则运算", "expression: 数学表达式字符串"),
}


# ---------- 提示模板 ----------
# 这是 ReAct 的"灵魂"——用提示词教模型按固定格式输出 Thought 和 Action
REACT_PROMPT = """你是一个使用 ReAct 模式推理的智能体。

可用工具：
{tool_desc}

请严格按以下格式输出，每一步包含两行：
Thought: 你的推理过程，解释为什么这么做
Action: 工具名(参数)
（系统会返回 Observation，然后你继续下一轮 Thought）

当任务完成时，输出：
Thought: 任务已完成的推理
Action: Finish(最终答案)

任务：{task}

{history}
Thought:"""


def build_tool_desc() -> str:
    """构造工具描述文本，注入提示模板"""
    lines = []
    for name, (_, desc, params) in TOOLS.items():
        lines.append(f"- {name}: {desc}，参数：{params}")
    return "\n".join(lines)


def parse_action(text: str) -> tuple[str, str]:
    """从模型输出中解析出 Action 名和参数，例如 search_wiki(北京)"""
    # 用正则匹配 "Action: 工具名(参数)" 的格式
    match = re.search(r'Action:\s*(\w+)\((.*)\)', text)
    if not match:
        # 解析失败：可能是模型没按格式输出，直接当作最终答案返回
        return "Finish", text
    return match.group(1), match.group(2)


def run_tool(action_name: str, args: str) -> str:
    """执行工具调用，返回结果字符串"""
    if action_name == "Finish":
        return args  # Finish 不调用工具，直接返回最终答案
    if action_name not in TOOLS:
        return f"错误：未知工具 {action_name}"
    func, _, _ = TOOLS[action_name]
    try:
        # 简单参数解析：去掉首尾引号后直接传入
        args = args.strip().strip('"').strip("'")
        return func(args)
    except Exception as e:
        return f"工具执行错误：{e}"


def call_llm(prompt: str) -> str:
    """调用大模型，返回文本"""
    resp = client.chat.completions.create(
        model=MODEL,
        messages=[{"role": "user", "content": prompt}],
        temperature=0,  # 推理任务用低温度，保证输出稳定可复现
    )
    return resp.choices[0].message.content


def react_agent(task: str) -> str:
    """ReAct 主循环：Thought → Action → Observation → 重复"""
    history = ""  # 累积的推理轨迹，每轮都塞进提示让模型"看到"完整历史
    for step in range(MAX_STEPS):
        # 构造提示：工具描述 + 任务 + 历史轨迹
        prompt = REACT_PROMPT.format(
            tool_desc=build_tool_desc(),
            task=task,
            history=history,
        )
        # 提示模板末尾预留了 "Thought:"，模型会从这里自然续写推理过程
        output = call_llm(prompt).strip()

        # 把模型输出拼回历史轨迹（包含它写的 Thought 和 Action）
        history += output + "\n"

        # 解析出动作名和参数
        action_name, args = parse_action(output)
        print(f"[Step {step+1}] Action: {action_name}({args})")

        # 如果模型说 Finish，任务完成，返回最终答案
        if action_name == "Finish":
            return args

        # 否则执行工具，把结果作为 Observation 拼回历史
        observation = run_tool(action_name, args)
        history += f"Observation: {observation}\n"
        print(f"[Step {step+1}] Observation: {observation}\n")

    return "达到最大步数，任务未完成。"

1
2
3
4
5

if __name__ == "__main__":
    task = "查询北京和上海的人口（万人），然后计算两地人口之和。"
    result = react_agent(task)
    print("\n=== 最终答案 ===")
    print(result)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

[Step 1] Action: search_wiki(北京)
[Step 1] Observation: 北京是中华人民共和国首都，常住人口约2189万（2020年）。

[Step 2] Action: search_wiki(上海)
[Step 2] Observation: 上海是中国直辖市，GDP居中国城市首位（2023年约4.72万亿元）。

[Step 3] Action: calculate(2189+2487)   # 模型可能记得上海人口约2487万
[Step 3] Observation: 2189+2487 = 4676

[Step 4] Action: Finish(北京约2189万，上海约2487万，两地人口之和约4676万人)

1
2
3
4
5

       Planner（规划器）  →  生成完整步骤列表
            ↓
       Executor（执行器） →  逐步执行每个步骤
            ↓
       Replanner（重规划器）→ 根据执行结果决定下一步

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157

# plan_execute_agent.py
# Plan-and-Execute 推理引擎的纯 Python 实现
import re
import json

# 复用 react_agent.py 中的工具表、工具执行器和 LLM 调用函数
# call_llm 内部已创建好 OpenAI client 并使用 gpt-5.4，这里无需重复创建
from react_agent import TOOLS, run_tool, call_llm, build_tool_desc

MAX_REPLANS = 8  # 重规划次数上限，防止无限重规划


# ---------- Planner：列出"行程单" ----------
PLANNER_PROMPT = """你是一个任务规划器。给定一个任务，把它拆解成有序的执行步骤。

可用工具：
{tool_desc}

要求：
1. 每个步骤应该是一个具体的、可执行的动作
2. 步骤之间可以有依赖，但顺序要合理
3. 最后一步应该是"汇总并给出最终答案"

任务：{task}

请用JSON数组输出步骤列表，每个元素是一个字符串，例如：
["步骤1: ...", "步骤2: ...", "步骤3: 汇总答案"]

只输出JSON，不要其他文字。"""


def planner(task: str) -> list[str]:
    """生成初始计划——相当于出发前列行程单"""
    prompt = PLANNER_PROMPT.format(tool_desc=build_tool_desc(), task=task)
    output = call_llm(prompt).strip()
    # 容错处理：模型可能用 ```json ... ``` 包裹输出，需要剥掉
    output = output.strip("`").replace("json\n", "", 1)
    try:
        steps = json.loads(output)
        if isinstance(steps, list):
            return [str(s) for s in steps]
    except json.JSONDecodeError:
        pass  # JSON 解析失败，走兜底逻辑
    return [f"执行任务：{task}", "汇总答案"]  # 兜底：至少有个最小计划


# ---------- Executor：执行单步 ----------
EXECUTOR_PROMPT = """你是一个步骤执行器。执行给定的单个步骤，可以调用工具。

可用工具：
{tool_desc}

步骤：{step}

已完成的步骤及结果：
{past}

请直接输出该步骤的执行结果（自然语言即可），如果需要调用工具请说明用了哪个工具。"""


def executor(step: str, past: list[tuple[str, str]]) -> str:
    """执行单个步骤——相当于按行程单做一件事"""
    past_text = "\n".join(f"- {s} → {r}" for s, r in past) or "（无）"
    prompt = EXECUTOR_PROMPT.format(
        tool_desc=build_tool_desc(),
        step=step,
        past=past_text,
    )
    output = call_llm(prompt)
    # 简化处理：如果输出里提到调用某工具，尝试解析并执行
    # 实际项目里这里可以嵌套一个完整的 ReAct 循环来执行复杂步骤
    tool_match = re.search(r'(\w+)\(([^)]*)\)', output)
    if tool_match and tool_match.group(1) in TOOLS:
        result = run_tool(tool_match.group(1), tool_match.group(2))
        return f"{output}\n[工具结果] {result}"
    return output


# ---------- Replanner：检查进度，决定是否改"行程单" ----------
REPLANNER_PROMPT = """你是一个重规划器。根据当前进度决定下一步。

原始任务：{task}

已完成的步骤及结果：
{past}

剩余的原始计划：
{remaining}

请判断：
- 如果任务已完成，输出：FINISH: <最终答案>
- 如果需要调整计划，输出新的剩余步骤JSON数组（替换上面的remaining）

只输出 FINISH: ... 或 JSON数组，不要其他文字。"""


def replanner(task: str, past: list[tuple[str, str]], remaining: list[str]) -> tuple[str, list[str]]:
    """重规划：检查进度，返回 (状态, 新剩余步骤)"""
    past_text = "\n".join(f"- {s} → {r}" for s, r in past) or "（无）"
    remaining_text = "\n".join(remaining) or "（无）"
    prompt = REPLANNER_PROMPT.format(
        task=task,
        past=past_text,
        remaining=remaining_text,
    )
    output = call_llm(prompt).strip().strip("`")
    # 优先检查是否输出 FINISH 信号
    if output.startswith("FINISH:"):
        return "finish", [output[len("FINISH:"):].strip()]
    # 否则尝试解析为新的步骤列表
    try:
        steps = json.loads(output.replace("json\n", "", 1))
        if isinstance(steps, list):
            return "continue", [str(s) for s in steps]
    except json.JSONDecodeError:
        pass  # 解析失败则保持原计划不变
    return "continue", remaining


# ---------- 主循环：规划 → 执行 → 重规划 → 重复 ----------
def plan_execute_agent(task: str) -> str:
    """Plan-and-Execute 主循环"""
    # 第一步：Planner 生成初始计划
    plan = planner(task)
    print(f"[初始计划] 共{len(plan)}步：")
    for i, s in enumerate(plan, 1):
        print(f"  {i}. {s}")
    print()

    past: list[tuple[str, str]] = []  # 已完成的步骤及结果
    remaining = plan  # 剩余未执行的步骤

    for _ in range(MAX_REPLANS):
        if not remaining:
            break  # 所有步骤执行完毕
        # 取下一步执行
        current = remaining[0]
        print(f"[执行] {current}")
        result = executor(current, past)
        past.append((current, result))  # 记录已完成步骤
        print(f"[结果] {result}\n")

        # 重规划：检查是否完成、是否需要调整剩余计划
        status, new_remaining = replanner(task, past, remaining[1:])
        if status == "finish":
            return new_remaining[0]  # 任务完成，返回最终答案
        remaining = new_remaining  # 更新剩余计划（可能不变，也可能调整）

    # 兜底：超过重规划上限，让模型基于历史给出答案
    return executor("基于以上所有结果，汇总最终答案", past)


if __name__ == "__main__":
    task = "查询北京和上海的人口（万人），然后计算两地人口之和。"
    result = plan_execute_agent(task)
    print("\n=== 最终答案 ===")
    print(result)

1
2
3
4
5
6

Planner 生成大纲
   ↓
对每个大纲步骤：
   用 ReAct 智能体执行（可调用工具、可循环）
   ↓
Replanner 检查进度，必要时调整剩余大纲

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

# plan_react_agent.py
# 混合模式：Plan-ReAct（外层规划，内层ReAct）
import re

# 复用前两个模块的组件，避免重复造轮子
from react_agent import call_llm, TOOLS, run_tool
from plan_execute_agent import planner, replanner

MAX_OUTER = 6  # 大纲步数上限


def react_on_step(step: str, context: str) -> str:
    """用 ReAct 风格执行单个大纲步骤。
    context 是前序步骤的执行结果，作为这一步的背景信息。"""
    # 把大纲步骤包装成一个"迷你任务"交给智能体
    prompt = f"""你是一个执行单个步骤的智能体，可用工具：{list(TOOLS.keys())}。
步骤：{step}
上下文：{context}
请执行步骤，需要时调用工具，最后给出该步骤的结果。"""
    output = call_llm(prompt)
    # 简化处理：如果输出里提到工具调用，解析并执行
    # 完整实现可嵌套 react_agent 函数，这里为了演示做了简化
    m = re.search(r'(\w+)\(([^)]*)\)', output)
    if m and m.group(1) in TOOLS:
        tool_result = run_tool(m.group(1), m.group(2))
        return f"{output}\n[工具结果] {tool_result}"
    return output


def plan_react_agent(task: str) -> str:
    """混合模式主循环：外层 Plan-and-Execute，内层 ReAct"""
    # 第一步：Planner 生成大纲
    plan = planner(task)
    print(f"[Plan] 大纲共{len(plan)}步：")
    for i, s in enumerate(plan, 1):
        print(f"  {i}. {s}")
    print()

    past: list[tuple[str, str]] = []  # 已完成步骤及结果
    remaining = plan
    context = "（开始）"  # 上下文：把前一步结果传给后一步

    for _ in range(MAX_OUTER):
        if not remaining:
            break
        current = remaining[0]
        # 关键区别：用 ReAct 风格执行这一步（而非简单工具调用）
        print(f"[ReAct执行] {current}")
        result = react_on_step(current, context)
        past.append((current, result))
        context = result  # 把上一步结果作为下一步的上下文
        print(f"[结果] {result}\n")

        # 重规划：检查进度，必要时调整大纲
        status, new_remaining = replanner(task, past, remaining[1:])
        if status == "finish":
            return new_remaining[0]
        remaining = new_remaining

    # 兜底：让智能体基于所有结果汇总答案
    return react_on_step("基于以上结果汇总最终答案", context)


if __name__ == "__main__":
    task = "查询北京和上海的人口（万人），然后计算两地人口之和。"
    print(plan_react_agent(task))