A6 的循环能完成「查价格 → 算账」这样的小任务。任务一长、工具里出现下单这类会扣钱的操作，新问题就来了：模型走一步看一步，容易绕远路；算完不检查就回答；同一个调用反复做；没问过人就下了单。这一课在 A6 的 run_agent 上把这几处逐个补上。
先计划，再执行
让模型先写一个 2–5 步的计划，再开始调用工具。这样做有三个好处：方向先定下来，少走弯路；计划留在对话里，后面每一步都能对照；你和日志都能看到它打算做什么，计划明显不对时可以早点拦下。
python
SYSTEM = """你是小蜂咖啡的点单助手，可以使用工具。
1. 按计划逐步执行；价格和规则必须用 search_docs 查，金额必须用 calculator 算，不要心算。
2. 某一步失败时，先说明原因并调整剩下的计划，不要用同样的参数再试一次。
3. place_order 的 total 必须等于 calculator 算出的结果。"""
def make_plan(question):
resp = client.chat.completions.create(model=MODEL, max_completion_tokens=300, messages=[
{"role": "system", "content": SYSTEM},
{"role": "user", "content": f"任务：{question}\n先不要执行，只列出 2–5 步计划，每步一行。"},
])
return resp.choices[0].message.content计划不是合同。查不到资料、参数错了、用户拒绝了，都应该让模型说明原因、改写剩下的步骤，而不是硬按原计划走，也不是把失败的调用原样再来一遍。这条规则写在 system 里就够了，循环本身不用改。
- 计划要短：十几步的计划往往到第三步就过时了。
- 只给多步任务用：一问一答的请求多一次计划调用，只是多花钱、多等一会儿。
回答前先自查
模型说「完成了」不代表真的完成了。常见情况：只做了一半（用户说「直接下单」，它只报了价）；数字是心算的而不是工具算的；工具报错了却当作成功。
办法是在模型给出最终回答时，加一次检查：把目标和过程交给模型（可以是便宜一点的模型），用结构化输出返回是否通过、问题在哪。没通过就把问题作为新消息交回去，让它接着做。
python
CHECK_SCHEMA = {"name": "check", "strict": True, "schema": {
"type": "object",
"properties": {"ok": {"type": "boolean"}, "problem": {"type": "string", "description": "没通过时写明问题，通过时为空字符串"}},
"required": ["ok", "problem"], "additionalProperties": False}}
def reflect(question, messages):
history = "\n".join(f"{m['role']}: {m.get('content') or m.get('tool_calls')}" for m in messages[1:])
resp = client.chat.completions.create(model=MODEL, messages=[
{"role": "system", "content": "你是检查员。对照用户的目标检查最后的回答：要求是否全部完成？金额是否来自工具结果？有没有把失败当成功？"},
{"role": "user", "content": f"目标：{question}\n\n过程：\n{history}"},
], response_format={"type": "json_schema", "json_schema": CHECK_SCHEMA})
data = json.loads(resp.choices[0].message.content)
return "" if data["ok"] else data["problem"]- 只查一两次：否则可能「改了又查、查了又改」停不下来。
- 能用代码查的就别用模型：「下单金额是否等于计算器的结果」这种，直接比对数字更可靠。
预算和停止条件
A6 只有步数上限。任务一长，还要再加几道闸：
| 条件 | 为什么 | 做法 |
|---|---|---|
| 步数上限 | 防止死循环 | max_steps |
| token / 费用上限 | 每一步都会把前面的对话再发一遍，越往后单步越贵 | 累加 usage.total_tokens，超过预算就停 |
| 重复的相同调用 | 模型卡住时，常用同样的参数反复调同一个工具 | 记录「工具名 + 参数」出现的次数，第二次提醒，第三次停 |
| 总耗时 | 用户不会一直等 | 记下开始时间，超时就停 |
停下来不等于白做：返回已经拿到的部分结果，说明停在哪一步、为什么停。
有副作用的操作：模型提议，人来批准
工具分两类。只读的（查资料、计算）可以直接执行；有副作用的（下单、发消息、改数据、删文件），由你的代码在执行前暂停，把模型要做的事展示给用户，等用户批准。
python
ORDER_TOOL = {"type": "function", "function": {"name": "place_order",
"description": "为顾客下单并扣款。items 写饮品、杯型和数量，例如「大杯拿铁 x2」",
"parameters": {"type": "object", "properties": {"items": {"type": "string"}, "total": {"type": "number", "description": "应付金额（元）"}},
"required": ["items", "total"]}}}
SIDE_EFFECT_TOOLS = {"place_order"} # 这些工具执行前必须经过 approve
def approve_in_terminal(name, args): # 网页里换成弹窗或按钮，结果再回到这里
print(f"Agent 想执行 {name}：{json.dumps(args, ensure_ascii=False)}")
return input("批准吗？(y/n) ").strip().lower() == "y"- 批准写在代码里，不写在提示词里。提示词里写「下单前先问用户」，模型大多时候会照做，但不能保证；代码拦住才是保证。
- 给用户看具体参数：买什么、多少钱，而不是模型的一段话。批准后执行的就是用户看到的那份参数，不再让模型改。
- 拒绝也是工具结果：把「用户拒绝了，没有执行」交回模型，它才会去问用户想怎么改，而不会以为已经下单成功。
把它们放进循环
在 A6 的基础上加了计划、自查、三道停止条件、人工确认、存档和日志。search_docs、calculator、TOOLS 沿用 A6，前面几节的 SYSTEM、make_plan、reflect、ORDER_TOOL 也放进同一个文件：
python
import json, os, time
from openai import OpenAI
client = OpenAI(base_url="https://hivegpt.cn/v1", api_key=os.environ["HIVEGPT_API_KEY"])
MODEL = "gpt-5.5"
def save(path, state):
with open(path, "w", encoding="utf-8") as f:
json.dump(state, f, ensure_ascii=False)
def log_step(step, name, args, result, tokens, seconds):
with open("agent_log.jsonl", "a", encoding="utf-8") as f:
f.write(json.dumps({"step": step, "tool": name, "args": args, "result": result[:200],
"tokens": tokens, "seconds": round(seconds, 2)}, ensure_ascii=False) + "\n")
def execute(name, raw_args, functions, approve):
try:
args = json.loads(raw_args)
if name in SIDE_EFFECT_TOOLS and not approve(name, args):
return "用户拒绝了这次操作，没有执行。请询问用户想怎么调整。"
return str(functions[name](**args))
except Exception as e:
return f"调用失败：{e}"
def run_agent(question, tools, functions, approve, state_file="agent_state.json",
max_steps=8, max_tokens=30000, max_repeats=2, max_seconds=180):
if os.path.exists(state_file): # 有存档：从上次停下的地方继续
with open(state_file, encoding="utf-8") as f:
state = json.load(f)
else:
state = {"step": 0, "tokens": 0, "seen": {}, "checked": False, "messages": [
{"role": "system", "content": SYSTEM},
{"role": "user", "content": question},
{"role": "assistant", "content": "计划：\n" + make_plan(question)},
{"role": "user", "content": "按计划开始执行。"},
]}
messages, start = state["messages"], time.time()
while state["step"] < max_steps:
if state["tokens"] > max_tokens or time.time() - start > max_seconds:
return f"停止：超出预算（已用 {state['tokens']} token）。进度已保存，可以稍后继续。"
state["step"] += 1
t0 = time.time()
resp = client.chat.completions.create(model=MODEL, messages=messages, tools=tools)
state["tokens"] += resp.usage.total_tokens # 计划和自查的 token 也可以累加进来
msg = resp.choices[0].message
entry = {"role": "assistant", "content": msg.content or ""}
if msg.tool_calls:
entry["tool_calls"] = [c.model_dump() for c in msg.tool_calls]
messages.append(entry)
if not msg.tool_calls: # 模型想结束了：先自查一次
if not state["checked"]:
state["checked"] = True
problem = reflect(question, messages)
if problem:
messages.append({"role": "user", "content": f"检查发现问题：{problem}\n请修正后再回答。"})
save(state_file, state)
continue
if os.path.exists(state_file):
os.remove(state_file) # 任务完成，存档不再需要
return msg.content
for call in msg.tool_calls:
name, raw = call.function.name, call.function.arguments
key = name + raw
state["seen"][key] = state["seen"].get(key, 0) + 1
if state["seen"][key] > max_repeats:
return f"停止：{name} 用相同参数调用了 {state['seen'][key]} 次，可能卡住了。"
result = execute(name, raw, functions, approve)
if state["seen"][key] == max_repeats:
result += "\n（提醒：这个调用和之前完全相同，请换个做法。）"
log_step(state["step"], name, raw, result, resp.usage.total_tokens, time.time() - t0)
messages.append({"role": "tool", "tool_call_id": call.id, "content": result})
save(state_file, state) # 一步完整结束后再存档
return "停止：步数用完了。进度已保存，可以稍后继续。"
functions = {"search_docs": search_docs, "calculator": calculator,
"place_order": lambda items, total: f"下单成功：{items}，共 {total} 元，订单号 A1024"}
print(run_agent("帮我点 2 杯大杯燕麦拿铁，用 100 积分抵扣，直接下单",
TOOLS + [ORDER_TOOL], functions, approve_in_terminal))动手：批准还是拒绝
下面的运行框带上了下单工具。Agent 想下单时会停下来，给你看要买什么、多少钱，由你点 批准 或 拒绝。下单是演示，不会真的扣款。每一步都是一次模型调用，从你选的 Key 按用量计费。
▶ 运行 Agent 循环
第 1 步 · search_docs({"query":"燕麦拿铁 大杯"}) → 菜单：拿铁 28 元，美式 22 元，燕麦拿铁 32 元；大杯加 4 元。
第 2 步 · search_docs({"query":"积分 抵扣"}) → 会员规则：会员每消费 1 元积 1 分，每 100 分可抵扣 5 元。
第 3 步 · calculator({"expression":"(32+4)*2-100/100*5"}) → 67
第 4 步 · place_order({"items":"大杯燕麦拿铁 x2","total":67}) → （已批准）下单成功（演示，不会真的扣款）：大杯燕麦拿铁 x2，共 67 元，订单号 DEMO-4821
第 5 步 · 最终回答：计划：查价格 → 查积分规则 → 计算 → 下单。2 杯大杯燕麦拿铁共 72 元，100 积分抵 5 元，实付 67 元，已下单，订单号 DEMO-4821。两种都试一次：
- 点 批准：看最终回答里的金额是否和计算器的结果一致。
- 点 拒绝：工具会返回「顾客拒绝了这次下单，没有扣款」，看 Agent 是不是改为询问你想怎么调整，而没有再下一次单。
再试试「帮我点 3 杯美式，积分不用了」「来一杯超大杯拿铁」（菜单里没有超大杯），看它在查不到时怎么调整计划。运行框里的循环只有步数上限和人工确认；计划由 system 提示词要求，自查和重复检测在上面的 Python 代码里。
存档和日志
长任务可能跑几十步，也可能要等审批，中途难免中断：进程重启、网络出错、审批人下午才回复。上面的代码每一步结束后把 messages 和各项计数存进文件，下次用同一个 state_file 调用就从断点继续。
- 一步完整结束后再存。如果存下的是「有 tool_calls 却没有工具结果」的半截对话，恢复时接口会报错。
- 等审批时可以存档退出。审批结果到了，把它作为工具结果补进
messages，再接着跑，不必让进程一直挂着。 - 有副作用的工具要防重复执行。恢复时如果不确定上次下单成没成功，先查订单状态；给每次操作一个唯一编号，下游按编号去重，不要直接再下一次。
日志每一步写一行 JSON：步数、工具、参数、结果摘要、token、耗时。出了问题对着日志看，就能分清是计划错了、工具错了，还是模型理解错了。结果里可能有用户的个人信息，存之前要截断或脱敏。
小结
- 多步任务先让模型写简短计划，某步失败时调整计划；回答前自查一次，能用代码查的就用代码查。
- 三道停止条件：步数、token、重复调用（再加总耗时）；停下时返回部分结果和原因。
- 有副作用的工具由代码暂停等人批准，拒绝作为工具结果交回；每步存档、记日志，长任务可以恢复。
下一课是 H 路线的最后一课：生产化。把这一课的日志扩展成完整的链路追踪，再用缓存、模型路由和成本控制，让应用在线上跑得稳、花得少。