AI Agent 工具调用模式：ReAct、Plan-and-Execute 与反思循环

1
import json
2
import re
3
from typing import Callable, Any
4
from dataclasses import dataclass, field
5

6

7
# ============ 1. 工具定义 ============
8
@dataclass
9
class Tool:
10
    """工具的标准化描述，让 LLM 知道怎么调用"""
11
    name: str
12
    description: str
13
    func: Callable[..., Any]
14

15
    def to_openai_schema(self) -> dict:
16
        """转成 OpenAI tools 格式"""
17
        return {
18
            "type": "function",
19
            "function": {
20
                "name": self.name,
21
                "description": self.description,
22
                "parameters": {
23
                    "type": "object",
24
                    "properties": {
25
                        "input": {
26
                            "type": "string",
27
                            "description": f"输入参数: {self.description}"
28
                        }
29
                    },
30
                    "required": ["input"]
31
                }
32
            }
33
        }
34

35

36
# ============ 2. 真实工具实现 ============
37
def calculator(input: str) -> str:
38
    """计算数学表达式"""
39
    try:
40
        # 安全的 eval 限制
41
        allowed = set("0123456789+-*/()., ")
42
        if not all(c in allowed for c in input):
43
            return "错误：只支持数字和四则运算"
44
        return str(eval(input))
45
    except Exception as e:
46
        return f"计算错误: {e}"
47

48

49
def web_search(input: str) -> str:
50
    """模拟网络搜索（生产环境用 Tavily/SerpAPI）"""
51
    knowledge = {
52
        "Python": "Python 是 1991 年 Guido van Rossum 发布的解释型编程语言",
53
        "Rust": "Rust 是 Mozilla 2010 年发布的系统级语言，主打内存安全",
54
        "北京": "北京今日 25°C，晴，西北风 3 级",
55
        "上海": "上海今日 28°C，多云，东南风 2 级",
56
    }
57
    for key, val in knowledge.items():
58
        if key in input:
59
            return val
60
    return f"未找到关于 '{input}' 的信息"
61

62

63
def write_file(input: str) -> str:
64
    """把内容写入文件，input 格式 'filename|content'"""
65
    try:
66
        filename, content = input.split("|", 1)
67
        with open(f"/tmp/{filename.strip()}", "w") as f:
68
            f.write(content.strip())
69
        return f"已写入 /tmp/{filename.strip()}"
70
    except Exception as e:
71
        return f"写入失败: {e}"
72

73

74
# ============ 3. ReAct Agent 核心 ============
75
class ReActAgent:
76
    def __init__(self, llm_client, tools: list[Tool], max_steps: int = 8):
77
        self.llm = llm_client
78
        self.tools = {t.name: t for t in tools}
79
        self.max_steps = max_steps
80
        self.history = []  # ReAct 的关键：维护思考-行动的历史
81

82
    def _format_tools_desc(self) -> str:
83
        """把工具列表拼成 prompt 描述"""
84
        lines = []
85
        for name, tool in self.tools.items():
86
            lines.append(f"- {name}: {tool.description}")
87
        return "\n".join(lines)
88

89
    def _build_prompt(self, question: str) -> str:
90
        tools_desc = self._format_tools_desc()
91
        history = "\n".join(self.history)
92
        return f"""你是一个可以使用工具的 AI Agent。可用工具：
93
{tools_desc}
94

95
严格按以下格式回答（不要有任何额外内容）：
96
Thought: 你在想什么，为什么需要这个工具
97
Action: 工具名称({{"input": "工具参数"}})
98
Observation: （由系统填充，不要自己写）
99
... (Thought/Action/Observation 可以重复 N 次)
100
Thought: 我已经收集到所有信息
101
Final Answer: 给出最终答案
102

103
历史记录：
104
{history}
105

106
问题：{question}
107
"""
108

109
    def _parse_action(self, text: str) -> tuple[str, dict] | None:
110
        """解析 LLM 输出的 Action 行"""
111
        match = re.search(r"Action:\s*(\w+)\((\{.*?\})\)", text, re.DOTALL)
112
        if not match:
113
            return None
114
        try:
115
            return match.group(1), json.loads(match.group(2))
116
        except json.JSONDecodeError:
117
            return None
118

119
    def step(self, question: str) -> str:
120
        """执行 ReAct 循环"""
121
        for i in range(self.max_steps):
122
            prompt = self._build_prompt(question)
123
            response = self.llm.complete(prompt)
124

125
            # 情况 1：Agent 给出最终答案
126
            if "Final Answer:" in response:
127
                final = response.split("Final Answer:")[-1].strip()
128
                self.history.append(f"Thought: 已完成\nFinal Answer: {final}")
129
                return final
130

131
            # 情况 2：解析 Action 并执行
132
            parsed = self._parse_action(response)
133
            if not parsed:
134
                # 解析失败：把它当作 Thought 让 LLM 重新思考
135
                self.history.append(f"Thought: 上一轮输出格式错误，重试")
136
                continue
137

138
            tool_name, args = parsed
139
            if tool_name not in self.tools:
140
                observation = f"错误：工具 {tool_name} 不存在"
141
            else:
142
                try:
143
                    observation = self.tools[tool_name].func(**args)
144
                except Exception as e:
145
                    observation = f"工具执行异常: {e}"
146

147
            # 关键：把 Observation 追加到历史，形成 ReAct 的"链"
148
            self.history.append(f"Thought: {response.split('Action:')[0].strip()}")
149
            self.history.append(f"Action: {tool_name}({json.dumps(args)})")
150
            self.history.append(f"Observation: {observation}")
151

152
        return "达到最大步数限制，未能完成"
153

154

155
# ============ 4. 用法示例 ============
156
# 模拟 LLM（生产环境替换成 OpenAI/Claude API）
157
class MockLLM:
158
    def complete(self, prompt: str) -> str:
159
        # 这里简化：实际中调用 GPT-4 / Claude
160
        # 根据 prompt 中的关键词模拟响应
161
        if "北京" in prompt and "weather" not in str(prompt).lower():
162
            return """Thought: 用户问北京天气，我需要调用 web_search
163
Action: web_search({"input": "北京"})
164
"""
165
        if "北京" in prompt and "weather" in prompt:
166
            return """Thought: 我已经获取了北京的天气
167
Final Answer: 北京今日 25°C，晴，西北风 3 级。"""
168
        if "计算" in prompt or "多少" in prompt:
169
            return """Thought: 需要计算数学表达式
170
Action: calculator({"input": "25 * 3 + 100"})
171
"""
172
        return "Final Answer: 我不知道"
173

174

175
# 启动 Agent
176
tools = [
177
    Tool("calculator", "执行数学计算，input 为表达式字符串", calculator),
178
    Tool("web_search", "搜索互联网信息", web_search),
179
    Tool("write_file", "写入文件，input 格式'文件名|内容'", write_file),
180
]
181

182
agent = ReActAgent(MockLLM(), tools)
183
result = agent.step("北京今天天气怎么样？")
184
print(result)  # → 北京今日 25°C，晴，西北风 3 级

1
from typing import TypedDict, Annotated
2
from langgraph.graph import StateGraph, END
3
import operator
4

5

6
# ============ 状态定义 ============
7
class PlanExecuteState(TypedDict):
8
    """Agent 状态：贯穿整个工作流"""
9
    input: str                              # 用户问题
10
    plan: list[str]                         # 当前计划步骤
11
    past_steps: Annotated[list, operator.add]  # 已完成步骤 (step, result)
12
    response: str                           # 最终响应
13

14

15
# ============ 1. Planner 节点 ============
16
def plan_step(state: PlanExecuteState):
17
    """Planner LLM：把问题拆解成可执行步骤"""
18
    prompt = f"""把以下任务拆解成 3-7 个具体步骤，每个步骤要明确、可执行：
19
任务：{state['input']}
20

21
可用工具：search, calculator, file_writer
22
请输出 JSON 数组：["步骤1", "步骤2", ...]
23
"""
24
    plan_text = llm.complete(prompt)
25
    plan = json.loads(plan_text)
26
    return {"plan": plan}
27

28

29
# ============ 2. Executor 节点 ============
30
def execute_step(state: PlanExecuteState):
31
    """Executor LLM：执行当前步骤"""
32
    plan = state["plan"]
33
    past_steps = state["past_steps"]
34
    task = plan[0]  # 总是取第一个未完成步骤
35

36
    # 给 Executor 历史上下文
37
    history = "\n".join([f"已完成：{s} → {r}" for s, r in past_steps])
38

39
    # 让 LLM 决定调用哪个工具
40
    tool_call = llm.complete(f"""根据当前任务决定调用哪个工具：
41
历史：
42
{history}
43
当前任务：{task}
44
可用工具：calculator(数字表达式), web_search(关键词), write_file(文件名|内容)
45
请输出 JSON：{{"tool": "工具名", "args": "参数"}}
46
""")
47
    tool_call = json.loads(tool_call)
48

49
    # 执行工具
50
    if tool_call["tool"] == "calculator":
51
        result = calculator(tool_call["args"])
52
    elif tool_call["tool"] == "web_search":
53
        result = web_search(tool_call["args"])
54
    else:
55
        result = "未知工具"
56

57
    return {
58
        "past_steps": [(task, result)],
59
        "plan": plan[1:]  # 弹出已完成的步骤
60
    }
61

62

63
# ============ 3. Replanner 节点 ============
64
def replan_step(state: PlanExecuteState):
65
    """Replanner LLM：根据已执行结果判断是否需要重新规划"""
66
    past_steps = state["past_steps"]
67
    original_plan = state["plan"]
68

69
    history = "\n".join([f"完成：{s} → {r}" for s, r in past_steps])
70

71
    # 两种情况：要么重新规划，要么任务完成
72
    response = llm.complete(f"""基于以下执行结果，决定下一步：
73
原计划：{original_plan}
74
执行历史：
75
{history}
76

77
情况 1：任务完成 → 输出 {{"response": "最终答案"}}
78
情况 2：需要继续 → 输出 {{"plan": ["下一步1", "下一步2", ...]}}
79
情况 3：原计划失败需要重规划 → 输出 {{"plan": ["新步骤1", ...]}}
80
""")
81
    output = json.loads(response)
82

83
    if "response" in output:
84
        return {"response": output["response"]}
85
    return {"plan": output["plan"]}
86

87

88
# ============ 4. 条件边 ============
89
def should_end(state: PlanExecuteState) -> str:
90
    """决定是继续执行还是结束"""
91
    if "response" in state and state["response"]:
92
        return "end"
93
    if not state["plan"]:
94
        return "replan"  # 没计划了也要反思一下
95
    return "execute"
96

97

98
# ============ 5. 编排工作流（用 LangGraph）============
99
workflow = StateGraph(PlanExecuteState)
100

101
workflow.add_node("planner", plan_step)
102
workflow.add_node("executor", execute_step)
103
workflow.add_node("replanner", replan_step)
104

105
workflow.set_entry_point("planner")
106
workflow.add_edge("planner", "executor")
107
workflow.add_edge("executor", "replanner")
108
workflow.add_conditional_edges(
109
    "replanner",
110
    should_end,
111
    {"execute": "executor", "replan": "replanner", "end": END}
112
)
113

114
app = workflow.compile()
115

116
# 运行
117
result = app.invoke({"input": "北京和上海今天的气温相差多少度？"})
118
print(result["response"])

1
class ReflexionAgent:
2
    def __init__(self, llm, tools, max_trials=3, memory_path="reflexion_memory.jsonl"):
3
        self.llm = llm
4
        self.tools = tools
5
        self.max_trials = max_trials
6
        self.memory_path = memory_path
7
        self.short_term = []  # 短期：当前任务的 ReAct 历史
8
        self.long_term = self._load_memory()  # 长期：反思记录
9

10
    def _load_memory(self) -> list[str]:
11
        try:
12
            with open(self.memory_path) as f:
13
                return [json.loads(line) for line in f]
14
        except FileNotFoundError:
15
            return []
16

17
    def _save_memory(self, reflection: str):
18
        with open(self.memory_path, "a") as f:
19
            f.write(json.dumps({
20
                "task": self.current_task,
21
                "reflection": reflection,
22
                "timestamp": time.time()
23
            }) + "\n")
24

25
    def _reflect(self, task: str, trajectory: list, success: bool) -> str:
26
        """反思阶段：让 LLM 分析自己哪里做错了"""
27
        traj_text = "\n".join([f"{t['thought']}\n→ {t['action']} → {t['observation']}"
28
                                for t in trajectory])
29

30
        prompt = f"""任务：{task}
31
执行轨迹：
32
{traj_text}
33
结果：{"成功" if success else "失败"}
34

35
请反思：
36
1. 如果失败，哪里出错了？为什么？
37
2. 下次应该怎么改进？
38
3. 这个教训是否适用于其他任务？
39

40
用一段话总结（不超过 100 字）：
41
"""
42
        return self.llm.complete(prompt)
43

44
    def run(self, task: str) -> str:
45
        self.current_task = task
46
        for trial in range(self.max_trials):
47
            self.short_term = []
48

49
            # 第 1 阶段：执行（ReAct 循环）
50
            for step in range(8):
51
                # 在 prompt 里同时注入短期和长期记忆
52
                memory_text = "\n".join([
53
                    f"过去教训 {i+1}: {m['reflection']}"
54
                    for i, m in enumerate(self.long_term[-3:])  # 取最近 3 条
55
                ])
56

57
                prompt = f"""任务：{task}
58

59
【重要：以下是你过去的反思教训，请避免重蹈覆辙】
60
{memory_text}
61

62
当前轨迹：
63
{self._format_traj()}
64
请输出下一步 Thought 和 Action：
65
"""
66
                response = self.llm.complete(prompt)
67
                action = self._parse_action(response)
68

69
                if "Final Answer:" in response:
70
                    # 任务完成
71
                    if self._verify(response.split("Final Answer:")[-1]):
72
                        return response.split("Final Answer:")[-1]
73
                    else:
74
                        # 自评失败：触发反思
75
                        break
76

77
                # 执行工具
78
                obs = self._execute_tool(action)
79
                self.short_term.append({
80
                    "thought": response,
81
                    "action": action,
82
                    "observation": obs
83
                })
84

85
            # 第 2 阶段：反思
86
            reflection = self._reflect(task, self.short_term, success=False)
87
            self._save_memory(reflection)
88
            self.long_term.append({"reflection": reflection})
89

90
        return f"任务失败，已尝试 {self.max_trials} 次并记录反思"