用纯 Python 手写一个完整的 ReAct Agent,理解 Agent = LLM + Tools + Loop 的本质。
这是整个教程最重要的模块。理解这一章后,你会发现 LangChain、LangGraph 等框架只是"简化了这些步骤"。
simple_llm_call.py → tool_schema.py → react_loop.py → memory.py → error_handling.py → full_agent.py → agent.py
(基础调用) (定义工具) (核心循环) (记忆) (容错) (完整组装) (最终版本)
simple_llm_call.py)LLM 的消息格式:
from langchain_core.messages import SystemMessage, HumanMessage
messages = [
SystemMessage(content="你是一个自动驾驶专家。"),
HumanMessage(content="什么是 BEV 感知?"),
]
response = llm.invoke(messages)
三种消息类型:
- SystemMessage: 设定 AI 的角色和行为规则
- HumanMessage: 用户输入
- AIMessage: AI 的回复(用于多轮对话)
| 参数 | 作用 | Agent 开发建议 |
|---|---|---|
temperature |
控制随机性 (0=确定, 1=创意) | Agent 中设为 0,确保行为可预测 |
max_tokens |
输出长度上限 | 根据任务复杂度设置 |
python module_01_from_scratch/simple_llm_call.py
tool_schema.py)
LLM (大脑) ──决定调用什么工具──> Tool (手脚) ──执行具体操作──> 返回结果
@dataclass
class Tool:
name: str # 工具名称 (LLM 用这个名字来调用)
description: str # 功能描述 (LLM 根据这个判断是否使用)
parameters: dict # 参数 JSON Schema (LLM 按这个格式传参)
function: Callable # 实际的 Python 函数
LLM 完全依赖 description 来决定何时调用哪个工具。一个好的 description 应该:
1. 清楚说明工具的功能
2. 暗示什么时候应该用这个工具
3. 描述参数的含义
TOOL_REGISTRY = [
Tool(name="calculator", description="执行数学计算...", function=calculator),
Tool(name="get_current_time", description="获取当前时间...", function=get_current_time),
Tool(name="search_papers", description="搜索论文...", function=search_papers),
]
python module_01_from_scratch/tool_schema.py
react_loop.py) ⭐ 核心这是 Ian AI Agent 的"心脏":
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Thought │ --> │ Action │ --> │ Observe │
│ (推理) │ │ (调用工具)│ │ (观察结果)│
└──────────┘ └──────────┘ └──────────┘
^ │
└───────────────────────────────────┘
循环直到 LLM 给出最终答案
Step 1: System Prompt 告诉 LLM 怎么做
你是一个智能助手。当需要获取信息时:
1. 思考需要什么
2. 调用工具 (以 JSON 格式输出)
3. 观察结果
4. 重复直到能回答
Step 2: 解析 LLM 的响应
LLM 返回 JSON:
{"action": "calculator", "action_input": {"expression": "sqrt(144)"}}
代码解析后调用 calculator(expression="sqrt(144)")。
Step 3: 执行工具并反馈
工具返回: "计算结果: 12.0"
→ 追加到对话历史
→ LLM 继续思考或给出最终答案
Step 4: 循环直到 final_answer
{"action": "final_answer", "action_input": "答案是 12"}
class ReActAgent:
def run(self, user_query: str) -> str:
messages = [SystemMessage(content=system_prompt),
HumanMessage(content=user_query)]
for iteration in range(max_iterations):
response = self.llm.invoke(messages)
action = parse_llm_action(response.content)
if action["action"] == "final_answer":
return action["action_input"]
# 执行工具
result = tool.run(**action["action_input"])
# 反馈结果
messages.append(AIMessage(content=response.content))
messages.append(HumanMessage(content=f"工具结果: {result}"))
python module_01_from_scratch/react_loop.py
memory.py)- LLM 上下文窗口有限(DeepSeek: 64K tokens)
- 长对话超出限制导致错误或高昂成本
- 聪明的记忆策略让 Agent 在长任务中保持高效
策略 1: 滑动窗口(Sliding Window)
只保留最近 N 轮对话,旧的直接丢弃。
- 优点:简单,Token 可控
- 缺点:旧信息丢失
策略 2: 摘要记忆(Summary Memory)
用 LLM 将旧对话压缩为摘要。
- 优点:保留关键信息,Token 高效
- 缺点:需要额外 LLM 调用
python module_01_from_scratch/memory.py
error_handling.py)1. LLM 输出格式错误:返回的不是合法 JSON
2. API 调用失败:网络波动、限流
3. 工具执行异常:参数错误、超时
# 1. 多重 JSON 解析策略
strategies = [
extract_json_from_markdown,
extract_first_brace_pair,
fix_single_quotes,
]
# 2. API 重试 (指数退避)
for attempt in range(max_retries):
try:
return llm.invoke(messages)
except RateLimitError:
time.sleep(2 ** attempt) # 2s, 4s, 8s
# 3. 最大迭代保护
if iteration >= max_iterations:
return "Agent 停止,请简化问题。"
python module_01_from_scratch/error_handling.py
agent.py)
Agent
├── LLM (DeepSeek via ChatOpenAI)
├── Tools (calculator, search, time, file)
├── Memory (summary/sliding window)
├── ReAct Loop (thought → action → observe)
└── Error Handling (retry, parse fallback, guard)
from module_01_from_scratch.agent import Agent
agent = Agent(verbose=True)
answer = agent.chat("帮我算一下 256 的平方根")
print(answer)
# 多轮对话
answer = agent.chat("搜索 BEV 感知的论文")
# Agent 记住了之前的上下文
python module_01_from_scratch/agent.py
| 概念 | 说明 |
|---|---|
| Agent = LLM + Tools + Loop | Agent 的本质是让 LLM 在"思考-行动-观察"循环中使用工具 |
| System Prompt | 定义 Agent 是谁、有什么工具、如何输出 |
| Tool Schema | 工具的 name + description + parameters,LLM 据此决策 |
| ReAct Loop | 核心循环,LLM 反复推理和调用工具直到得到答案 |
| Memory | 管理对话历史,防止超出 token 限制 |
| Error Handling | JSON 解析容错、API 重试、迭代上限保护 |