最小Agent实现学习笔记
一、最小可运行 Agent 代码
代码基于文章中的 Agent Loop 设计，使用 OpenAI SDK 的 Function Calling 机制，约 120 行，可直接运行（只需填入你的 API Key 和 Base URL）。
python
import json, time, signal, os, ast, operator
from openai import OpenAI
# ============================================================
# 配置
# ============================================================
API_KEY = os.getenv("OPENAI_API_KEY") # 替换为你的 Key
BASE_URL = "https://maas.qianwenaiapi.com/compatible-mode/v1" # 可替换为兼容 OpenAI 的端点
MODEL = "qwen3.8-flash" # 或任何支持 function calling 的模型
ALLOWED_ROOT = "/tmp/agent_sandbox" # read_file 的允许根目录
MAX_STEPS = 8
TIMEOUT = 60 # 总超时（秒）
TOOL_TIMEOUT = 5 # 单次工具超时（秒）
client = OpenAI(api_key=API_KEY, base_url=BASE_URL)
# ============================================================
# 1. 工具执行函数
# ============================================================
def search(query: str) -> str:
"""模拟搜索。实际可接入真实搜索 API。"""
return f"[模拟搜索结果] 关于「{query}」的摘要信息……"
# 安全计算器：白名单运算符递归求值，不使用 eval
_SAFE_OPS = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
ast.Pow: operator.pow,
ast.USub: operator.neg,
}
def _eval_node(node):
if isinstance(node, ast.Expression):
return _eval_node(node.body)
if isinstance(node, ast.Constant):
return node.value
if isinstance(node, ast.BinOp):
op_type = type(node.op)
if op_type not in _SAFE_OPS:
raise ValueError(f"不允许的运算符：{op_type.__name__}")
return _SAFE_OPS[op_type](_eval_node(node.left), _eval_node(node.right))
if isinstance(node, ast.UnaryOp):
op_type = type(node.op)
if op_type not in _SAFE_OPS:
raise ValueError(f"不允许的一元运算符：{op_type.__name__}")
return _SAFE_OPS[op_type](_eval_node(node.operand))
raise ValueError(f"不允许的表达式节点：{type(node).__name__}")
def calculator(expression: str) -> str:
"""安全计算四则运算表达式，如 \"(3 + 5) * 2\"。"""
try:
tree = ast.parse(expression, mode="eval")
result = _eval_node(tree)
return str(result)
except Exception as e:
return f"计算失败：{e}"
def read_file(file_path: str) -> str:
"""读取沙箱目录内的文本文件。"""
abs_path = os.path.abspath(file_path)
if not abs_path.startswith(os.path.abspath(ALLOWED_ROOT)):
return "错误：路径不在允许范围内"
try:
with open(abs_path, "r", encoding="utf-8") as f:
return f.read()
except Exception as e:
return f"读取失败：{e}"
TOOL_FUNCTIONS = {
"search": search,
"calculator": calculator,
"read_file": read_file,
}
# ============================================================
# 2. 工具 Schema（LLM 看到的部分）
# ============================================================
TOOLS = [
{
"type": "function",
"function": {
"name": "search",
"description": "搜索互联网信息。输入搜索关键词，返回结果摘要。",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"}
},
"required": ["query"],
},
},
},
{
"type": "function",
"function": {
"name": "calculator",
"description": "计算数学表达式。输入一个合法的数学表达式字符串，返回计算结果。",
"parameters": {
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式，如 (3+5)*2"}
},
"required": ["expression"],
},
},
},
{
"type": "function",
"function": {
"name": "read_file",
"description": f"读取沙箱目录（{ALLOWED_ROOT}）内的文件内容。",
"parameters": {
"type": "object",
"properties": {
"file_path": {"type": "string", "description": "文件的绝对路径或相对于沙箱根目录的路径"}
},
"required": ["file_path"],
},
},
},
]
# ============================================================
# 3. 工具超时保护（Unix 主线程有效）
# ============================================================
class ToolTimeoutError(Exception):
pass
def _alarm_handler(signum, frame):
raise ToolTimeoutError("工具执行超时")
def run_with_timeout(func, args, timeout):
old = signal.signal(signal.SIGALRM, _alarm_handler)
signal.alarm(timeout)
try:
return func(**args)
finally:
signal.alarm(0)
signal.signal(signal.SIGALRM, old)
# ============================================================
# 4. Agent Loop 核心
# ============================================================
def run_agent(user_input: str) -> str:
messages = [{"role": "user", "content": user_input}]
start = time.time()
for step in range(MAX_STEPS):
# 总超时检查
if time.time() - start > TIMEOUT:
return f"[兜底] 总超时（{TIMEOUT}s），已执行 {step} 步。"
# ---- Think: 调 LLM ----
response = client.chat.completions.create(
model=MODEL,
messages=messages,
tools=TOOLS,
tool_choice="auto",
)
msg = response.choices[0].message
# 无工具调用 → 直接返回文本，结束
if not msg.tool_calls:
return msg.content or "[空回复]"
# ---- 有工具调用：先把 assistant 消息加入历史（必须 model_dump） ----
messages.append(msg.model_dump())
# ---- Act: 逐个执行工具 ----
for tool_call in msg.tool_calls:
tool_name = tool_call.function.name
try:
tool_args = json.loads(tool_call.function.arguments)
except json.JSONDecodeError as e:
observation = json.dumps({"success": False, "error": f"参数 JSON 解析失败：{e}"})
else:
# 未知工具处理
if tool_name not in TOOL_FUNCTIONS:
observation = json.dumps({"success": False, "error": f"未知工具：{tool_name}"})
else:
# 执行 + 双层超时保护
try:
result = run_with_timeout(
TOOL_FUNCTIONS[tool_name], tool_args, TOOL_TIMEOUT
)
observation = json.dumps({"success": True, "result": result}, ensure_ascii=False)
except ToolTimeoutError:
observation = json.dumps({"success": False, "error": f"工具 {tool_name} 执行超时（{TOOL_TIMEOUT}s）"})
except Exception as e:
observation = json.dumps({"success": False, "error": f"工具执行失败：{e}"}, ensure_ascii=False)
# ---- Observe: 喂回工具结果（必须带 tool_call_id） ----
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": observation,
})
# 循环回到 Think，LLM 看到工具结果后决定下一步
return f"[兜底] 达到最大步数（{MAX_STEPS}），未能完成。"
# ============================================================
# 5. 运行
# ============================================================
if __name__ == "__main__":
os.makedirs(ALLOWED_ROOT, exist_ok=True)
with open(os.path.join(ALLOWED_ROOT, "note.txt"), "w") as f:
f.write("这是沙箱内的示例文件内容。")
# 测试用例：需要多步工具调用
print(run_agent("先帮我计算 (15 + 27) * 3 是多少，然后用 search 查一下 'Python Agent Loop' 是什么。"))运行前准备：pip install openai，填入有效的 API_KEY 和 BASE_URL。如果不在 Unix 主线程运行，signal.alarm 会报错，可去掉工具级超时或改用 concurrent.futures 实现。
二、学习笔记
1. Agent 循环的本质：一个带状态转移的 for 循环
文章把 Agent Loop 概括为 Think → Act → Observe 的循环。用最朴素的话说，Agent 并不是什么魔法——它就是一个 for step in range(max_steps) 循环，每一轮做三件事：
- Think：把当前对话历史（含工具定义）发给 LLM，让模型决定“下一步说什么或调什么工具”；
- Act：如果模型返回了
tool_calls，就按名字找到本地函数并执行； - Observe：把执行结果以
role: "tool"的身份追加回messages，下一轮 LLM 就能看到。
终止条件只有两个：LLM 不再请求工具（返回纯文本），或者触发了兜底（超时/最大步数）。理解这一点，Agent 的神秘感就消失了大半。
2. 工具定义：Schema 是给 LLM 看的“接口文档”
每个工具需要两样东西：一个 Python 函数（真正干活）和一个 JSON Schema（告诉 LLM 这个工具叫什么、干什么、参数是什么）。文章特别强调：description 写得好不好，直接决定 LLM 会不会调错工具，“像给初级开发者写 docstring 一样认真”。
在代码中，TOOL_FUNCTIONS 字典是“执行侧”的映射，TOOLS 列表是“描述侧”的载荷，两者必须对齐——名字、参数名、必填项都要一致，否则会出现“LLM 按 Schema 传了参数，但 Python 函数不认”的错位。
3. Function Calling 的“配对键”：tool_call_id
这是文章点出的一个关键细节：role: "tool" 的消息必须带 tool_call_id，且该 ID 必须与上一轮 assistant 消息中对应 tool_call 的 id 完全一致。OpenAI API 靠这个 ID 把“工具请求”和“工具结果”配对起来。漏掉它，下一轮调用会直接报错。
另一个配套细节是：msg 是 SDK 返回的 ChatCompletionMessage 对象，必须调用 msg.model_dump()转成 dict 再 append 进 messages，否则后续序列化可能出问题。
4. 两层超时：总超时不够，工具超时必须有
文章指出一个容易被忽略的陷阱：循环开头的 time.time() - start > timeout 检查只在每轮 LLM 调用前生效。如果某个工具执行本身卡死（读超大文件、网络请求挂起），总超时检查根本不会被触发，整个 Agent 就永久卡住了。
因此需要两层超时：
- 总超时：整个循环的墙上时间上限；
- 工具超时：单次工具执行的墙上时间上限。
示例代码用 signal.alarm 实现了工具级超时。注意它仅限 Unix 主线程；跨平台场景可以改用 concurrent.futures.ThreadPoolExecutor 配合 future.result(timeout=...)。
5. 安全细节：计算器不用 eval，文件路径防遍历
文章用两个具体例子展示了“工具不是随便写的”：
计算器：即使你用 ast.parse 拦截了 ast.Call，eval 仍然不安全——__import__('os').system('rm -rf /') 并不走 ast.Call 节点。正确做法是白名单运算符递归求值：只注册 Add/Sub/Mult/Div/Pow 等允许的 AST 节点类型，遇到白名单外的节点直接抛异常。示例代码中的 _SAFE_OPS 和 _eval_node 就是这一思路的实现。
read_file：必须检查 os.path.abspath(file_path) 是否以允许的根目录开头，否则 ../../etc/passwd这样的路径可以轻松跳出限制。示例代码在 read_file 开头做了这个检查，返回错误而不是抛出异常，让 LLM 能看到并自行调整。
6. 容错哲学：把错误变成 Observation，让 LLM 自己兜底
文章反复强调一个 Agent 设计原则：工具执行失败不要崩，把错误信息喂回去。具体来说：
- 工具内部抛出的异常 → 包成
{"success": false, "error": "..."}作为 observation； - LLM 幻觉出不存在的工具名 → 返回
{"error": "未知工具：xxx"}，LLM 下一轮通常会换正确的工具； - 参数 JSON 解析失败 → 同样喂回错误信息。
这背后的逻辑是：LLM 本身就是最好的“错误恢复器”。你不需要在代码里预判所有失败模式，只需要保证任何失败都不会让循环崩溃，并把失败原因如实告诉模型。
7. 完整流程一句话总结
用户输入 → 进入循环 → LLM 看到历史 + 工具描述 → 要么直接回答（结束），要么请求工具 → 代码执行工具（带超时、带沙箱、带异常兜底）→ 结果以 tool_call_id 配对的方式喂回 → 重复，直到 LLM 给出最终回答或触发兜底。整个过程的核心代码不到 50 行，剩下的都是工具本身和安全/超时的工程细节。
评论
还没有评论，来说两句？
评论已关闭