大模型本身只会吐文本。给它一段输入，它回一段输出，没有记忆，碰不到外部世界，也不会自己决定下一步干什么。所谓 Agent，就是在模型外面套一层调度器：让它拆任务、挑工具、看结果、再决定继续还是收工。这层调度器不神秘，核心循环几十行代码就能写完。
先明确一件事：Agent 不是万能解法。单轮问答、格式转换、固定流程的批处理，直接调模型或写死代码更快更便宜。Agent 的价值出现在任务步数不确定、需要外部数据、且中间结果会影响后续决策的场合。反过来，对延迟敏感（比如在线客服首响要求 500ms 内）或者工具调用有副作用且不可回滚的场景，上自主循环要格外谨慎——模型多绕两步，成本和风险都成倍放大。
拆开看：Agent 到底由哪几块拼成
抛开框架术语，一个能跑的最小系统只有四件事。
规划：把用户目标翻译成“下一步做什么”。最朴素的实现就是让模型输出结构化决策。
记忆：短期靠对话消息列表，长期靠外部存储检索。前者是上下文窗口里的数组，后者才需要向量库。
工具：一组带名字、描述和参数签名的函数。模型只能看到描述，真正执行靠你的代码。
反思：拿到工具结果后判断是否偏离目标。它可以是循环里的下一次模型调用，也可以是独立的一次评估请求。
这四块里，工具注册和循环控制是骨架，规划和反思都是模型调用，记忆是数据结构。下面按“先能跑、再能扛”的顺序搭。
工具层：注册表加统一执行入口
工具的本质是可被模型“点名”的函数。注册表负责按名字查找，执行入口负责兜住异常——工具抛错不能让整个循环崩掉，要把错误变成字符串喂回模型，让它自己决定换路。
from typing import Callable, Dict, Any
class Tool:
def __init__(self, name: str, description: str, func: Callable):
self.name = name
self.description = description
self.func = func
def run(self, kwargs) -> str:
try:
return str(self.func(kwargs))
except Exception as e:
return f"工具执行出错: {e}"
class ToolRegistry:
def __init__(self):
self._tools: Dict[str, Tool] = {}
def register(self, tool: Tool):
self._tools[tool.name] = tool
def get(self, name: str):
return self._tools.get(name)
def list_tools(self):
return [{"name": t.name, "description": t.description}
for t in self._tools.values()]
工具描述要写清“什么时候用”而不只是“能干什么”。模型选错工具，八成是描述含糊。参数尽量扁平，嵌套结构会显著拉高模型输出格式出错的概率。
循环层：一次决策只走一步
ReAct 的骨架是“思考—行动—观察”反复滚动。每次迭代让模型输出一个 JSON 决策，代码解析后要么执行工具、把结果塞回消息列表，要么判定任务完成直接返回。
import json
from openai import OpenAI
class Agent:
def __init__(self, registry: ToolRegistry):
self.registry = registry
self.client = OpenAI(api_key=OPENAI_API_KEY, base_url=OPENAI_BASE_URL)
self.messages = []
def _system_prompt(self) -> str:
tools_desc = "\n".join(
f"- {t['name']}: {t['description']}"
for t in self.registry.list_tools()
)
return f"""你是一个智能助手，可使用以下工具：
{tools_desc}
严格输出 JSON：
{{"thought": "", "action": "工具名或 none",
"action_input": {{"参数名": "参数值"}}, "final_answer": "完成时的答案"}}
规则：需要工具时 action 填工具名；任务完成时 action 填 none 并填 final_answer；每次只调一个工具。"""
def run(self, user_input: str, max_steps: int = 8) -> str:
self.messages.append({"role": "user", "content": user_input})
for step in range(max_steps):
resp = self.client.chat.completions.create(
model=MODEL_NAME,
messages=[{"role": "system", "content": self._system_prompt()},
*self.messages],
temperature=0,
)
raw = resp.choices[0].message.content
decision = self._parse(raw)
if decision is None:
self.messages.append({"role": "user",
"content": "上一条输出不是合法 JSON，请只输出 JSON，不要加解释。"})
continue
if decision.get("action") == "none":
return decision.get("final_answer", "")
tool = self.registry.get(decision["action"])
result = tool.run(**decision.get("action_input", {})) if tool \
else f"错误: 工具 {decision['action']} 不存在"
self.messages.append({"role": "assistant",
"content": json.dumps(decision, ensure_ascii=False)})
self.messages.append({"role": "user", "content": f"工具执行结果: {result}"})
return "已达到最大步数，任务未完成"
max_steps 是硬刹车，没有它，模型在工具反复报错时会无限重试，账单和延迟一起失控。
模型输出不可信：解析失败才是常态
上面代码里 _parse 不是随手 json.loads。工程上必须假设模型会输出带 Markdown 围栏的 JSON、前后带解释文字、甚至被截断。三个真实坑：
第一，message.content 可能为 None。模型触发拒答（refusal）时内容为空，直接 json.loads(None) 抛 TypeError。判断 if raw is None 后读 resp.choices[0].message.refusal 或直接重试。
第二，finish_reason == "length" 说明输出被 max_tokens 截断，JSON 必然不完整。此时不要解析，直接提高上限或缩短提示词后重发。
第三，围栏污染。安全做法是先剥掉 json 和 包裹，再定位第一个 { 到最后一个 } 的子串解析，失败则把错误信息回灌给模型重试一次。
def _parse(self, raw):
if not raw:
return None
text = raw.strip()
if text.startswith("```"):
text = text.split("\n", 1)[-1].rsplit("```", 1)[0]
start, end = text.find("{"), text.rfind("}")
if start == -1 or end == -1:
return None
try:
return json.loads(text[start:end+1])
except json.JSONDecodeError:
return None
想彻底绕开解析问题，就用模型原生的 Function Calling（tools 参数），让接口层保证结构化输出。代价是绑定具体厂商的接口形态，可移植性下降。原型阶段手写 JSON 更透明，生产环境优先原生工具调用。
工具安全：计算器和数据库都得设防
原文的计算器用 eval 配空 __builtins__，这不够。().__class__.__bases__ 这类属性链仍可能逃逸。更稳的做法是用 ast.parse 白名单校验：只允许 Expression、BinOp、UnaryOp、Constant 和四则运算节点，遇到 Call、Attribute 直接拒绝。
数据库工具的风险更直接。让模型自由拼 SQL，一句 DROP TABLE 就能清库。两层防护：
conn = sqlite3.connect("sales.db")
conn.execute("PRAGMA query_only = ON;")
query_only 让连接只读，写操作直接报错。再叠加 authorizer 回调，只放行 SQLITE_SELECT、SQLITE_READ 等读动作，其余一律拒绝。这样即使模型被提示注入诱导，也改不动数据。
工具输入还要做类型与范围校验。参数是城市名就限制长度和字符集，是数字就检查上下界。校验失败返回结构化错误而不是抛异常，模型看到“参数非法：city 不能为空”比看到堆栈更容易自我纠正。
记忆与反思：别急着上向量库
短期记忆就是 self.messages。上下文一长，两个问题冒出来：token 超限和关键信息被稀释。常见做法是保留最近 N 轮原文，更早的压缩成摘要，摘要本身也占 token，要算进预算。
长期记忆才需要向量检索。余弦相似度那段实现只适合理解原理，真上生产要考虑 embedding 模型版本一致性、维度对齐、以及检索结果的相关性阈值——相似度 0.6 的结果未必有用，无脑 top_k 会把噪声塞进上下文。
反思机制不必单独造一个方法。把“上一步结果是否推进了目标”写进系统提示词的规则里，让模型在每次决策的 thought 字段自评，成本最低。独立反思调用适合长链路任务，但每多一次调用就多一份延迟和费用，权衡后再加。
什么时候该换框架
手写循环的价值在于每一步都看得见，调试时能精确定位是提示词问题、解析问题还是工具问题。等这套跑通、边界摸清，再考虑 LangChain、LlamaIndex 这类框架。它们省的是组件拼装和生态集成，不是替你解决提示词设计和错误处理——那些问题换框架照样存在，只是藏得更深。
多 Agent 协作（AutoGen 那类）适合角色分工明确的场景，比如一个查数据、一个写报告、一个审校。但每多一个 Agent 就多一层消息传递和状态同步，调试复杂度非线性上升。单 Agent 加多工具能解决的事，别急着拆成多角色。
收尾
Agent 的骨架是循环，血肉是工具，神经是提示词。先把 max_steps 设好、把工具异常兜住、把模型输出当不可信输入处理，这三件事做完，系统才算能跑。之后再谈记忆增强、并行工具调用、反思优化。顺序反了，容易在框架的抽象层里迷路，连一次工具调用为什么失败都查不出来。
ID:6651
