Agent 的本质 = LLM + while 循环 + 终止守卫 — 模型只负责"出意图", 执行、记录、刹车全部发生在你的代码里
Agent 像一个只会动嘴的实习生: 模型负责说"帮我查一下 X"(tool_call), 但跑腿(工具执行)、记账(messages[])、设闹钟(终止守卫)全是你写的代码。所谓 Agent 框架, 剥开包装后就是 while + stop_reason + messages.append 三件套——框架省的是脚手架, 代替不了你对循环收敛与成本的负责。理解了这一页, LangGraph / OpenAI Agents SDK / Claude Agent SDK 在你眼里都是同一副骨架的不同皮。
while steps < 25: resp = llm(messages) # 关键: 模型只出意图, 不干活 if resp.stop_reason == "end_turn": break # → 循环收敛, 返回回答
assistant = {"role": "assistant", "content": "先查天气再定穿搭",
"tool_calls": [get_weather("上海")]} # Reason+Act 同一条消息
observation = {"role": "tool", "content": '{"temp": 31}'} # Observation
messages += [assistant, observation] # → 循环推进一格messages = [system, user] # 起点只有 2 条 for _ in range(3): messages.append(assistant_msg) # 只追加, 不改写历史 messages.append(tool_msg) print(len(messages)) # → 8 (2 + 3×2)
arguments 是字符串不是 dict, 必须 json.loads 后才能用。 call = resp.message.tool_calls[0] call.function.name # → "search_orders" call.function.arguments # → '{"status": "pending"}' 是 JSON 字符串! args = json.loads(call.function.arguments) # 关键: 先 parse 再用
role:"tool" 消息作回执, 靠 tool_call_id 关联。漏一条, 下一轮 API 直接 400。(Anthropic 的变体: user 消息内嵌 tool_result 块) messages.append({"role": "tool", "tool_call_id": call.id,
"content": '{"total": 3}'})
# 漏回填 → 400: assistant message with 'tool_calls' must be
# followed by tool messages responding to eachtool_use 继续、end_turn 收尾、max_tokens 是截断不是完成、length 同理。OpenAI 侧对应 finish_reason。 match resp.stop_reason: case "tool_use": execute(resp.message.tool_calls) # 继续循环 case "end_turn": return resp.text # 正常收尾 case "max_tokens": raise Truncated(resp) # 截断≠完成
assert step < MAX_STEPS, "步数超限" assert cost < MAX_COST, "预算超限" assert sig not in seen, "重复调用" # 关键: 三道守卫都在你的代码里, 模型一个都指望不上
usage = [2000, 4100, 6300, 8600] # 第 1~4 轮的 input tokens print(sum(usage)) # → 21000 (计费按总和) # 关键: 20 轮任务的输入费 ≈ 单轮的 40 倍
def truncate(text: str, limit=4000) -> str: if len(text) <= limit: return text return text[:limit//2] + "\n...[截断]...\n" + text[-limit//4:] # 关键: 头尾各留一段 — 计数与结论常在首尾
system 消息出现一次、位置固定在最前。用户输入永远走 user, 绝不拼进 system——这是权限分层。 messages = [{"role": "system", "content": RULES}] # 只出现一次
n = len([m for m in messages if m["role"] == "system"])
print(n) # → 1 (每轮重复 append 是事故)sort_keys 归一化参数, 否则 dict 序列化顺序不稳会造成漏判。 sig = (call.name, hash(json.dumps(call.input, sort_keys=True))) if sig == last_sig: # 与上一次完全相同 repeat += 1 # 关键: sort_keys 保证参数顺序无关, 否则误判"不同调用"
state = {"messages": messages, "step": step, "spent": cost}
db.save(task_id, json.dumps(state)) # 每轮后持久化
# 关键: 不落盘 = 发版即丢, 从头重跑还要付双份 token 钱内部工单系统要一个"能查库、能建单"的机器人。不引框架, 先把骨架写对, 后面所有高级能力都往这副骨架上挂。
def run_agent(goal: str, tools: dict, max_steps: int = 25) -> str: messages = [{"role": "system", "content": SYS_PROMPT}, {"role": "user", "content": goal}] for step in range(max_steps): # 守卫①: 步数写进 for, 天然兜底 resp = llm.create(messages=messages, tools=list(tools)) messages.append(resp.message) # 关键: assistant 消息先入列 if resp.stop_reason != "tool_use": # end_turn / max_tokens 都视为收尾 return resp.message.content for call in resp.message.tool_calls: result = dispatch(tools, call) # 真正干活的是你的代码 messages.append({"role": "tool", "tool_call_id": call.id, "content": truncate(json.dumps(result), 4000)}) return "已达步数上限, 附中间结论收尾" # 不裸奔报错, 带状态退出
用户反馈"机器人转圈不出结果"。把每步的 Thought/Action/Observation 打成结构化日志, 卡在哪一轮一目了然。
# 排查口诀: 先看 step 几开始重复, 再看那一步的观察是什么 for step, msg in enumerate(messages, 1): if msg["role"] == "assistant" and msg.get("tool_calls"): for tc in msg["tool_calls"]: log.info("step=%d think=%r act=%s(%s)", step, (msg["content"] or "")[:80], # 思考截断, 防日志爆炸 tc.function.name, tc.function.arguments) elif msg["role"] == "tool": log.info("step=%d obs=%s", step, msg["content"][:120]) # → step=7 think='又查了上海' act=search(city="上海") ×3 → 定位到死循环
模型调用量放开后, 偶发任务会"越跑越欢"。步数守卫防死循环, 预算守卫防"每步合法但总量爆炸"。
class Budget: max_steps = 25 # 单任务步数上限, 按业务复杂度定 max_cost = 0.50 # 单任务费用上限(美元), 用高价模型要按比缩 spent = 0.0 def guarded_loop(messages, budget: Budget): for step in range(budget.max_steps): resp = llm.create(messages=messages) budget.spent += cost_of(resp.usage) # usage 里带 input/output tokens if budget.spent > budget.max_cost: return bail("超预算熔断", step, budget.spent) # → $0.52 > $0.50 # ...正常 tool_use 分支...
新手把工具异常当 500 抛出去, 任务直接死。老手把异常序列化成"观察"喂回模型——它下一轮会自己改参数重试。
def safe_dispatch(tools, call) -> str: try: result = dispatch(tools, call) return json.dumps(result, ensure_ascii=False) except Exception as e: # 关键: 异常是给模型看的观察, 不是给用户看的 500 return json.dumps({"error": type(e).__name__, "message": str(e)[:200], # 截断防上下文爆炸 "hint": "检查参数后重试, 或改用其他工具"}) # → 模型读到 KeyError: 'order_id' 后自动补参数, 自愈而非崩死
模型偶尔会"忘了自己刚查过"。第 2 次重复就掐断, 不要等第 400 次。
import hashlib seen = {} # (tool, args_hash) → 次数 key = (call.function.name, hashlib.md5(call.function.arguments.encode()).hexdigest()) seen[key] = seen.get(key, 0) + 1 if seen[key] >= 2: # 同工具同参数出现 2 次 return json.dumps({"error": "重复调用已熔断, 请换思路或直接收尾"}) # → 卡死循环在第 2 次重复时被掐断, 而不是烧到守卫②
20 轮的深度任务, 46k tokens 的历史让每轮又慢又贵。超过阈值就把老历史压成摘要, 最近几轮原样保留。
def compact(messages, keep_last=6, token_budget=24000): if count_tokens(messages) <= token_budget: return messages head, tail = messages[1:-keep_last], messages[-keep_last:] summary = llm_create([{"role": "user", "content": "把以下过程压缩成要点, 必须保留所有数字结论:\n" + text_of(head)}]) return [messages[0], # system 原样保留 {"role": "user", "content": "[历史摘要] " + summary}, *tail] # → 46k 压到 9k, 费用降 5 倍, 注意力重新集中在最近几轮
不开流式, 用户盯着白屏等 8 秒。流式下文字 token 直接上屏, 工具调用渲染成状态条, 体感天壤之别。
with client.messages.stream(messages=messages, tools=TOOLS) as stream: for event in stream: if event.type == "content_block_delta": yield {"kind": "text", "data": event.delta.text} # 文字直接上屏 elif event.type == "content_block_start" \ and event.content_block.type == "tool_use": yield {"kind": "status", "data": f"调用 {event.content_block.name} ..."} # 状态条 # → 首字延迟 8s → 0.6s; 循环内每轮都要重开流
长任务跑到第 13 步赶上发版重启。没有 checkpoint 就得从 0 重跑——双倍 token 钱外加用户重等。
key = f"agent:{task_id}:messages" redis.set(key, json.dumps(messages), ex=86400) # 每轮结束后落盘, TTL 24h # 进程被 OOM kill / 滚动发布后, 新 worker 接管: saved = redis.get(key) if saved: messages = json.loads(saved) # → 从第 13 步继续, 而不是从 0 重跑 step = int(redis.get(f"agent:{task_id}:step") or 0) # 注意: 有副作用的工具要幂等, 否则续跑会重复下单/重复扣款
复杂任务里模型经常"做完了还继续折腾"。注册一个显式的收尾工具, 终止从"赌 stop_reason"变成"明确动作"。
TOOLS.append({
"name": "task_done",
"description": "目标已完成时调用: 提交最终报告并结束任务, 之后不得再调其他工具",
"input_schema": {"type": "object",
"properties": {"report": {"type": "string"}},
"required": ["report"]}})
# 循环里最先判它:
if call.name == "task_done":
return call.input["report"] # → 模型自己举手"我做完了"
任务级 trace 一条线看不出问题, 步骤级 span 才能回答"慢在哪一步、贵在哪一步"。
with tracer.start_as_current_span("agent.step") as span: span.set_attribute("agent.step", step) span.set_attribute("llm.model", "claude-sonnet-4-5") span.set_attribute("llm.input_tokens", resp.usage.input_tokens) span.set_attribute("llm.output_tokens", resp.usage.output_tokens) for call in resp.message.tool_calls: with tracer.start_as_current_span("tool." + call.name) as ts: ts.set_attribute("tool.args_sha1", args_hash(call)) result = dispatch(tools, call) ts.set_attribute("tool.bytes", len(str(result))) # → Jaeger 里一次任务展开 20 个 step span, 慢/贵步骤一眼定位
while True 没有出口, 模型可以永远"再查一次". 正解: max_steps + 预算双守卫。 # 错: while True: run(llm, tools) # → step=4217 · cost $84 # 对: for step in range(25): ... # → 25 步强制收尾
SELECT * 的 2MB 结果直接 json.dumps 进 messages. 正解: 截断 + 只返回模型需要的字段。 # 错: content=json.dumps(rows) # → 2,048,000 字符入列 # 对: content=truncate(json.dumps(rows[:20]), 4000)
tool_calls 却没有对应 tool 消息. 正解: 每个 tool_call_id 必须有回执。 # 错: messages.append(resp.message); continue # → 400: 'tool_calls' must be followed by tool messages # 对: for tc in resp.message.tool_calls: messages.append(tool_msg(tc.id, run(tc)))
max_tokens 截断被当正常结束展示. 正解: 分支处理后再输出。 # 错: return resp.message.content # → '{"orders": [{"id": 1' 半截 # 对: if resp.stop_reason != "end_turn": handle_not_done(resp)
append(system_msg). 正解: system 固定在 messages[0], 循环只追加 assistant/tool。 # 错: messages += [system_msg] # → 10 轮后 10 份规则 # 对: # 初始化放一次, 循环体内只 append assistant / tool
temperature=1.0 采样发散. 正解: 工具型任务压到 0~0.3。 # 错: llm.create(..., temperature=1.0) # → 参数随机漂移 # 对: llm.create(..., temperature=0.2) # → 决策稳定可复现
json.dumps(..., sort_keys=True) 归一化。 # 错: hash(str(call.input)) # → {"a":1,"b":2} 与 {"b":2,"a":1} 视为不同 # 对: hash(json.dumps(call.input, sort_keys=True))
# 错: result = http.get(url, timeout=10) # → ConnectionError 冒泡, 任务死亡 # 对: except Exception as e: return {"error": str(e)[:200]}
arguments 解析失败. 原因: 输出上限给太小, tool_call 的 JSON 被腰斩. 正解: 上限给足 + parse 失败降级回喂。 # 错: llm.create(..., max_tokens=64) # → '{"city": "上海", "date"' 解析炸 # 对: max_tokens=1024; except JSONDecodeError: 回喂"参数不完整, 请重发"
tool_call_id 关联。 # 错: for call, r in zip(calls, await gather(*tasks)): ... # → 错配 # 对: # 回填时按 call.id 找各自的 r, 与完成顺序无关
# 错: gather(*[loop(shared_messages) for t in tasks]) # → 交错 append # 对: gather(*[loop(deepcopy(messages)) for t in tasks])
"Tool"/"ASSISTANT". 正解: role 收敛成常量。 # 错: {"role": "Tool", "content": ...} # → 400: invalid value for 'role' # 对: ROLE_TOOL = "tool"; {"role": ROLE_TOOL, ...}
# 错: append 60 条不处理 # → 每轮 input 46k tokens # 对: if count_tokens(messages) > 24000: messages = compact(messages)
print 到控制台, 从未进 messages. 正解: 结果必须以 tool 消息入列。 # 错: print(run_tool(call)) # → 模型上下文里没有, 只能编 # 对: messages.append(tool_msg(call.id, run_tool(call)))
trace_id 贯穿所有 step 日志。 # 错: log.info("step=3 act=search") # → 与其他任务混在一起 # 对: log.info("trace=%s step=3 act=%s", trace_id, act)
Retry-After。 # 错: for _ in range(3): llm.create(...) # → 429 风暴, 封号警告 # 对: wait = min(2 ** attempt + random.random(), 30)
len(text)//4 对中文严重失真(1 汉字 ≈ 1~2 token). 正解: 官方 tokenizer 计数。 # 错: est = len(text) // 4 # → 中文误差 2~3 倍, 预算形同虚设 # 对: est = count_tokens(text) # → 官方 tokenizer, 计费一致
end_turn/task_done 才对用户输出。 # 错: return resp.message.content # → "好的, 我来帮您查询"(就完了) # 对: if resp.stop_reason == "end_turn": return resp.message.content
user 消息。 # 错: system = RULES + user_input # → 注入文本获得系统级权限 # 对: messages = [{"role": "system", "content": RULES}, {"role": "user", ...}]
# 错: TASK_STATE = {} # → 重启即清零, 用户重等+重计费 # 对: redis.set(f"agent:{task_id}", json.dumps(state), ex=86400)