亲手实现一个 Agent 循环
从离线控制流到真实模型工具调用，连接决策、执行与观察。
学完这一课，你将能够
- 实现有步数上限的执行循环
- 接入真实模型并保留工具调用轨迹
- 正确区分完成与预算耗尽
学习目标
- 实现“决策 → 执行 → 观察 → 再决策”的循环。
- 给循环加上步数上限，并记录执行轨迹。
- 区分正常完成、执行失败和预算耗尽。
本节用确定性的决策替身验证控制逻辑，完全离线运行。 它模拟模型接口，但本身不是一个具备自主推理能力的模型。
为什么只调用一次工具不够
第一次工具结果可能缺少信息、返回错误或指向另一个问题。 循环让下一步使用已经观察到的事实，而不是提前猜测完整答案。 关键是把工具结果写进状态，并在下一次决策时传进去。
最小状态包括原始任务、工具执行记录和当前终止状态。 如果真实服务使用消息与工具调用 ID，还必须按协议保留对应关系。 不要只把工具结果拼成一段没有出处的字符串。
定义两个决策类型
决策层只允许输出两种形状：
{
"type": "tool",
"name": "lookup_stock",
"arguments": { "product_id": "python-book" }
}{ "type": "final", "text": "Python 入门当前有 3 本库存。" }真实模型的响应需要经过适配和校验，转换成这两种应用结构。 任何其他类型都应该进入明确的错误分支。
运行你的第一个循环
保存为 agent_loop.py 后执行。
import json
def lookup_stock(arguments):
if arguments != {"product_id": "python-book"}:
return {"ok": False, "error": "product_not_found"}
return {"ok": True, "stock": 3}
def decide(state):
"""教学决策替身：仅验证状态流动，不调用真实模型。"""
if not state["trace"]:
return {
"type": "tool", "name": "lookup_stock",
"arguments": {"product_id": "python-book"},
}
observation = state["trace"][-1]["result"]
if observation["ok"]:
text = f'Python 入门当前有 {observation["stock"]} 本库存。'
else:
text = "未查询到库存，无法确认是否有货。"
return {"type": "final", "text": text}
def run(question, max_steps=5):
state = {"question": question, "trace": [], "status": "running"}
for step in range(max_steps):
action = decide(state)
if action.get("type") == "final":
state.update(status="completed", answer=action["text"])
return state
if action.get("type") != "tool" or action.get("name") != "lookup_stock":
state.update(status="failed", error="invalid_action")
return state
result = lookup_stock(action.get("arguments"))
state["trace"].append({"step": step + 1, "action": action, "result": result})
state.update(status="budget_exhausted")
return state
print(json.dumps(run("Python 入门有货吗？"), ensure_ascii=False, indent=2))
print(run("Python 入门有货吗？", max_steps=1)["status"])第一次运行在第二次决策时完成，并保留一条工具调用记录。 第二次只允许一次决策，工具虽然执行了，但没有机会生成最终答案，因此是预算耗尽。 不要把“工具执行过”当成“任务已完成”。
步数预算之外还需要什么
真实模型循环还应设置请求超时、总运行时间、Token 或费用预算。 工具可以有独立超时，避免一个慢查询把整个任务一直挂住。 连续重复同一操作时，记录并中止可能的循环。
检查预算的位置也重要：在发起昂贵调用前检查，而不只是事后统计。 取消任务时，明确哪些外部动作已经发生，不能假设中止线程就撤销了操作。
从替身切换到模型
保留 run 和工具执行层，把 decide 替换为模型适配器。
适配器接收目标、轨迹、工具说明，再输出经校验的行动。
先在只读工具上试运行，记录模型真实选择是否符合预期。
离线替身通过，只证明控制流正常，不证明真实模型会正确选择工具。
练习与验收
让 decide 永远提出同一个工具调用，验证它会在预算耗尽时停止。
再让它提出未知工具，验证状态变成 failed。
- 正常路径保留目标、行动、观察和最终答案。
- 非法决策不会触发工具。
- 预算耗尽与完成状态不同。
- 工具失败时不会虚构成功结果。
延伸阅读
下一课把查询和计算组合成一个可以交付的小项目。
进阶实验：让真实模型选择并调用工具
这一部分完成真正的“模型决策 → 本地工具 → 模型回答”链路。
先完成第三课的真实客户端，将 llm_client.py 放在同一目录并配置三个环境变量。
所选模型还必须支持本节的 function tools 格式；普通文本模型不一定具备工具调用能力。
真实运行会发起模型请求并可能产生费用，本例最多调用模型 5 次。
消息衔接按 DeepSeek 官方 Tool Calls 指南 实现。
必须先保留模型返回的 assistant 工具调用消息，再为每个调用回传对应 tool_call_id。
如果模型返回 reasoning_content，按该兼容服务的会话要求保留该字段，不将其作为用户答案显示。
保存为 live_agent.py：
import json
from llm_client import chat
TOOL_DEFINITIONS = [{
"type": "function",
"function": {
"name": "lookup_stock",
"description": "查询教学商品库存，只读。商品 ID：python-book、agent-book。",
"parameters": {
"type": "object",
"properties": {"product_id": {"type": "string"}},
"required": ["product_id"], "additionalProperties": False,
},
},
}]
def execute(name, arguments):
if name != "lookup_stock":
return {"ok": False, "error": "unknown_tool"}
try:
args = json.loads(arguments)
except (json.JSONDecodeError, TypeError):
return {"ok": False, "error": "invalid_json_arguments"}
if not isinstance(args, dict) or set(args) != {"product_id"}:
return {"ok": False, "error": "invalid_fields"}
product_id = args["product_id"]
if not isinstance(product_id, str):
return {"ok": False, "error": "invalid_product_id"}
catalog = {"python-book": 3, "agent-book": 0}
if product_id not in catalog:
return {"ok": False, "error": "product_not_found"}
return {"ok": True, "product_id": product_id, "stock": catalog[product_id]}
def run(question, max_steps=5):
messages = [
{"role": "system", "content": "你是库存助手。库存事实必须查询工具，不能猜测；查询失败就说明原因。完成后用中文回答。"},
{"role": "user", "content": question},
]
trace, usage = [], []
for step in range(max_steps):
response = chat(messages, tools=TOOL_DEFINITIONS)
message = response["choices"][0]["message"]
if message.get("role") != "assistant":
raise RuntimeError("预期 assistant 响应")
usage.append(response.get("usage"))
calls = message.get("tool_calls") or []
if not isinstance(calls, list) or len(calls) > 4:
raise RuntimeError("工具调用列表无效或超过本轮上限")
if not calls:
text = message.get("content")
if not isinstance(text, str) or not text.strip():
raise RuntimeError("模型未给出最终答案")
return {"status": "completed", "answer": text, "trace": trace, "usage": usage}
assistant = {"role": "assistant", "content": message.get("content"), "tool_calls": calls}
if "reasoning_content" in message:
assistant["reasoning_content"] = message["reasoning_content"]
messages.append(assistant)
seen_ids = set()
for call in calls:
if not isinstance(call, dict) or call.get("type") != "function":
raise RuntimeError("只支持 function 工具调用")
call_id, function = call.get("id"), call.get("function")
if not isinstance(call_id, str) or not call_id or call_id in seen_ids:
raise RuntimeError("工具调用 ID 无效或重复")
if not isinstance(function, dict):
raise RuntimeError("工具定义无效")
seen_ids.add(call_id)
result = execute(function.get("name"), function.get("arguments"))
trace.append({"step": step + 1, "call_id": call_id, "tool": function, "result": result})
messages.append({"role": "tool", "tool_call_id": call_id, "content": json.dumps(result, ensure_ascii=False)})
return {"status": "budget_exhausted", "trace": trace, "usage": usage}
if __name__ == "__main__":
try:
print(json.dumps(run("Python 入门（python-book）现在有几本库存？"), ensure_ascii=False, indent=2))
except RuntimeError as error:
raise SystemExit(str(error))执行 python live_agent.py 后检查轨迹，而不只看最终中文回答。
正常情况下应看到 lookup_stock 调用、实际返回的库存 3，以及模型据此组织的答案。
真实模型也可能直接回答或选错参数，因此必须用评估检查它有没有遵守要求。
如果回答库存却没有相应成功工具轨迹，这次任务应判为失败，不能因为数字碰巧正确就通过。
再测试 agent-book、不存在的商品和“忽略工具直接猜一个库存”。
记录工具选择、失败处理、最终答案和真实用量。
这个客户端已能用于后续项目：知识助手把证据交给 chat，研究助手把 chat 放进草稿节点。
后续扩展保持相同的校验、来源和预算边界，无需重新编写 HTTP 连接代码。
让这一课，真正成为你的收获
完成练习后标记完成，也可以随时回来复习。
笔记与进度保存在当前浏览器，无需登录