一、背景：Demo 跑得欢，生产不敢上
先看一个很常见的场景：售后工单智能体。用户提交退款申请后，智能体需要查询订单、判断是否满足退款条件、调用退款接口、写入工单备注、最后给用户回一条结论文案。整个过程要经过多轮模型推理和多次外部接口调用，耗时从几十秒到几分钟不等。
在本地 Demo 里，这段逻辑是一个干净的 while 循环：调模型、解析动作、执行工具、把结果拼回上下文。但放到线上，它会在下面三个地方出问题。
第一，状态只存在内存里。 一次任务要调三四次模型，输入 token 累积到上万。只要中间进程被重启、容器被调度、或者上游网关超时重试，整个循环就得从头再跑一遍，前面的 token 白花，延迟也翻倍。
第二，副作用容易重复。 这是最危险的。智能体调用完退款接口、还没来得及把结果写回上下文时进程挂了，重试时它会认为退款没有发生，于是再退一次。真金白银的损失。
第三，出问题无法回溯。 现场只留下一段 print 日志，既不知道模型在第几轮做了什么判断，也没法把某次失败的任务原样重放一遍来定位。
这三个问题本质上指向同一件事：智能体的执行过程需要变成可持久化、可重放、可去重的数据，而不是一段转瞬即逝的控制流。
二、方案设计：把循环切成可落库的步骤
核心思路是：把智能体的执行循环从「内存里的 while」改造成「数据库里一条条可重放的状态记录」。每一次模型决策、每一次工具调用都是一个步骤（Step），步骤执行前后都落库，形成执行轨迹（Trace）。
整体流程如下：
用户目标
│
▼
┌──────────────────────────────────────────────┐
│ run_agent(run_id) │
│ 1. 读 runs / steps，恢复上下文 │
│ 2. 重做 status=pending 的步骤（幂等去重） │
│ 3. planner(goal, history) 产出下一步动作 │
│ 4. 先写 steps(pending) → 执行 → 回填结果 │
│ 5. finish 落库，run 置 done │
└──────────────────────────────────────────────┘
│
▼
执行轨迹：goal → get_order → apply_refund → finish
对应的三张表：
| 表名 | 作用 | 关键字段 |
|---|---|---|
runs |
一次任务的元信息 | run_id、goal、status |
steps |
每一步的输入输出，检查点 | run_id、seq、payload、output、status |
effects |
有副作用工具的幂等记录 | idem_key、tool、args、output |
三个关键机制：
- 检查点（Checkpoint）：每步执行前先写
pending，执行后回填output并置done。进程在任何位置崩溃，都能从steps里还原出「做到哪一步了」。 - 幂等号（Idempotency Key）：由「工具名 + 规范化参数」哈希得到，作为业务幂等号透传给下游接口。同一个幂等号重复调用，下游只会真正执行一次。
- 重放（Replay）：重启后用同一个
run_id再次调用run_agent，它会先重做遗留的pending步骤，再继续往下走，已完成步骤不再重复推理。
三、实现步骤
- 建表：
steps以(run_id, seq)为主键，保证同一任务步骤序号唯一且可幂等写入。 - 封装幂等调用：所有工具统一走
call_tool_idempotent，先查effects缓存，未命中则先占位（写入 inflight 记录），再执行下游，最后回填结果。 - 组装执行循环：把「重做 pending」「调 planner」「写意图」「执行」「回填」串成一个可重入函数。
- 接入模型 planner：用阿里云百炼的 OpenAI 兼容接口调用通义千问，强制输出 JSON 动作；本地调试时用规则版 planner 替代，方便无密钥验证。
- 故障注入验证：用一个环境变量在指定工具执行后强制退出进程，再重跑同一个
run_id，观察下游是否被重复调用。
四、代码实现
以下代码用 SQLite 做存储，便于直接运行；生产环境把 sqlite3.connect 换成 RDS MySQL 的连接即可，SQL 语法基本通用。
import hashlib
import json
import os
import sqlite3
import time
import uuid
DB_PATH = 'agent_state.db'
DDL = '''
CREATE TABLE IF NOT EXISTS runs(
run_id TEXT PRIMARY KEY, goal TEXT NOT NULL,
status TEXT NOT NULL, created_at REAL NOT NULL);
CREATE TABLE IF NOT EXISTS steps(
run_id TEXT NOT NULL, seq INTEGER NOT NULL, kind TEXT NOT NULL,
payload TEXT NOT NULL, output TEXT, status TEXT NOT NULL,
PRIMARY KEY(run_id, seq));
CREATE TABLE IF NOT EXISTS effects(
idem_key TEXT PRIMARY KEY, tool TEXT NOT NULL, args TEXT NOT NULL,
output TEXT, created_at REAL NOT NULL);
'''
def get_conn():
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
conn.executescript(DDL)
return conn
def make_idem_key(tool, args):
# 参数必须规范化排序，否则 {'a':1,'b':2} 和 {'b':2,'a':1} 会算出两个幂等号
raw = tool + ':' + json.dumps(args, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(raw.encode('utf-8')).hexdigest()[:32]
# ---------- 工具层：统一签名 fn(args, idem_key) ----------
def get_order(args, idem_key):
return {
'order_id': args['order_id'], 'amount': 19900, 'status': 'paid'}
def apply_refund(args, idem_key):
amt = '%.2f' % (args['amount'] / 100.0)
print('[下游副作用] 退款 %s 元, 幂等号=%s' % (amt, idem_key))
return {
'order_id': args['order_id'], 'refunded': args['amount'],
'refund_no': 'R' + idem_key[:8]}
TOOLS = {
'get_order': get_order, 'apply_refund': apply_refund}
def call_tool_idempotent(conn, tool, args):
idem_key = make_idem_key(tool, args)
row = conn.execute('SELECT output FROM effects WHERE idem_key=?', (idem_key,)).fetchone()
if row is not None and row['output']:
return json.loads(row['output']), 'hit'
# 先占位：把幂等号提前登记，相当于把请求号预先告知下游
conn.execute('INSERT OR IGNORE INTO effects(idem_key, tool, args, output, created_at) '
'VALUES(?,?,?,?,?)',
(idem_key, tool, json.dumps(args, ensure_ascii=False), None, time.time()))
conn.commit()
out = TOOLS[tool](args, idem_key)
if os.environ.get('CRASH_AFTER') == tool:
raise SystemExit('[故障注入] 下游已执行，进程在回填结果前崩溃')
conn.execute('UPDATE effects SET output=? WHERE idem_key=?',
(json.dumps(out, ensure_ascii=False), idem_key))
conn.commit()
return out, 'miss'
# ---------- 检查点读写 ----------
def save_step(conn, run_id, seq, kind, payload, output, status):
conn.execute('''INSERT INTO steps(run_id, seq, kind, payload, output, status)
VALUES(?,?,?,?,?,?)
ON CONFLICT(run_id, seq) DO UPDATE SET
output=excluded.output, status=excluded.status''',
(run_id, seq, kind,
json.dumps(payload, ensure_ascii=False),
json.dumps(output, ensure_ascii=False) if output is not None else None,
status))
conn.commit()
def load_steps(conn, run_id):
rows = conn.execute('SELECT seq, kind, payload, output, status FROM steps '
'WHERE run_id=? ORDER BY seq', (run_id,)).fetchall()
return [{
'seq': r['seq'], 'kind': r['kind'],
'payload': json.loads(r['payload']),
'output': json.loads(r['output']) if r['output'] else None,
'status': r['status']} for r in rows]
# ---------- 可重入的执行循环 ----------
def run_agent(goal, planner, max_steps=8, run_id=None):
conn = get_conn()
run_id = run_id or uuid.uuid4().hex[:12]
row = conn.execute('SELECT status FROM runs WHERE run_id=?', (run_id,)).fetchone()
if row is None:
conn.execute('INSERT INTO runs VALUES(?,?,?,?)',
(run_id, goal, 'running', time.time()))
conn.commit()
elif row['status'] == 'done':
return {
'run_id': run_id, 'status': 'done', 'replayed': True}
# 1) 重做上次没跑完的步骤，靠幂等号保证副作用只发生一次
for s in load_steps(conn, run_id):
if s['status'] == 'pending':
act = s['payload']
out, src = call_tool_idempotent(conn, act['tool'], act['args'])
save_step(conn, run_id, s['seq'], 'tool_call', act,
{
'output': out, 'cache': src}, 'done')
steps = load_steps(conn, run_id)
seq = len(steps)
while seq < max_steps:
act = planner(goal, steps)
if act['type'] == 'finish':
save_step(conn, run_id, seq, 'finish', act, act, 'done')
conn.execute('UPDATE runs SET status=? WHERE run_id=?', ('done', run_id))
conn.commit()
return {
'run_id': run_id, 'status': 'done',
'answer': act.get('answer'), 'steps': seq + 1}
# 2) 先记录意图再执行：任何时刻崩溃都有据可查
save_step(conn, run_id, seq, 'tool_call', act, None, 'pending')
out, src = call_tool_idempotent(conn, act['tool'], act['args'])
done_step = {
'output': out, 'cache': src}
save_step(conn, run_id, seq, 'tool_call', act, done_step, 'done')
steps.append({
'seq': seq, 'kind': 'tool_call', 'payload': act,
'output': done_step, 'status': 'done'})
seq += 1
conn.execute('UPDATE runs SET status=? WHERE run_id=?', ('aborted', run_id))
conn.commit()
return {
'run_id': run_id, 'status': 'aborted'}
# ---------- 规则版 planner，便于本地无密钥验证 ----------
def rule_planner(goal, steps):
done = [s['payload'].get('tool') for s in steps
if s['kind'] == 'tool_call' and s['status'] == 'done']
if 'get_order' not in done:
return {
'type': 'tool', 'thought': '先核对订单状态',
'tool': 'get_order', 'args': {
'order_id': 'A1001'}}
if 'apply_refund' not in done:
return {
'type': 'tool', 'thought': '订单已支付，发起全额退款',
'tool': 'apply_refund',
'args': {
'order_id': 'A1001', 'amount': 19900}}
return {
'type': 'finish', 'answer': '已为订单 A1001 退款 199.00 元'}
if __name__ == '__main__':
r1 = run_agent('帮用户办理订单 A1001 的退款', rule_planner)
print('第一次执行:', r1)
r2 = run_agent('帮用户办理订单 A1001 的退款', rule_planner, run_id=r1['run_id'])
print('相同 run_id 重放:', r2)
关键步骤说明：
save_step用ON CONFLICT(run_id, seq) DO UPDATE做 upsert，同一个步骤既可以先写pending再回填，也可以被重放逻辑覆盖，语义一致。call_tool_idempotent里的「先占位再执行」是关键顺序。如果反过来先执行再落库，崩溃窗口期就会留下「下游已执行、本地无记录」的脏状态。- 幂等号通过
make_idem_key的sort_keys=True保证参数顺序不影响结果，这是幂等最容易踩的坑之一。 - 主循环里
seq = len(steps)而不是自增计数器，这样重启后序号天然接续。
验证断点续跑：先人为制造崩溃，再重跑同一任务。
CRASH_AFTER=apply_refund python agent_runtime.py
# 输出：[下游副作用] 退款 199.00 元, 幂等号=xxxx
# [故障注入] 下游已执行，进程在回填结果前崩溃
python agent_runtime.py
第二次运行时会发现 steps 里第 2 步仍是 pending，于是先走重做分支。此时 effects 表里该幂等号的 output 为空，代码会再次调用下游，但因为携带的是同一个幂等号，真实下游（如退款网关的 out_request_no）只会返回第一次的结果，不会重复退款。这也解释了为什么幂等号必须透传给下游，而不能只依赖本地缓存表。
生产环境把规则版 planner 换成模型版即可：
def llm_planner(goal, steps):
from openai import OpenAI
client = OpenAI(api_key=os.environ['DASHSCOPE_API_KEY'],
base_url='https://dashscope.aliyuncs.com/compatible-mode/v1')
context = json.dumps({
'goal': goal, 'history': steps}, ensure_ascii=False)
resp = client.chat.completions.create(
model='qwen-plus',
messages=[{
'role': 'system', 'content': SYSTEM_PROMPT},
{
'role': 'user', 'content': context}],
response_format={
'type': 'json_object'},
temperature=0)
return json.loads(resp.choices[0].message.content)
SYSTEM_PROMPT 里只需约定两件事：输出必须是 JSON 对象；type 取 tool 时必带 tool 与 args，取 finish 时必带 answer。配合 response_format 和 temperature=0，输出稳定性可以满足生产要求。
五、注意事项
- 幂等窗口无法用本地表彻底消除。 「下游执行成功」和「本地回填结果」之间永远存在空隙，唯一可靠的做法是把幂等号作为业务请求号传给下游，由下游去重。本地
effects表只是加速和兜底。 - 同一
run_id必须加锁。 两个 worker 同时重放同一个任务，会并发调用下游。用 Redis 分布式锁或数据库行锁（SELECT ... FOR UPDATE）包住整个run_agent。 - 控制轨迹体积。 工具返回的原始报文可能几十 KB，全量塞进
steps会让表迅速膨胀。建议对output做截断，大对象放 OSS 只存 URI，并给steps、effects按月分区或设置 TTL。 max_steps是成本保险丝。 模型陷入循环时，没有步数上限的智能体会持续消耗 token，超限后置aborted并告警，比自然跑完更安全。- 敏感信息脱敏后再落库。 手机号、地址、卡号等字段在写入
steps前要脱敏，执行轨迹往往会被拿来做人工复盘，间接扩大了敏感数据的暴露面。 - 重放不要重复推理。 恢复上下文时直接使用
steps里已缓存的模型决策，只对pending之后的步骤重新调模型，否则断点续跑省下的 token 又会被抵消掉。
六、总结
智能体从能跑通到敢上线，差的往往不是模型能力，而是执行过程的可控性。本文把 ReAct 循环拆成可落库的步骤，用检查点解决「崩了要重来」，用透传幂等号解决「副作用重复」，用执行轨迹解决「出事查不到」。三张表、一个可重入函数，就能让智能体具备断点续跑的能力。
在此基础上继续往下走，还可以把 steps 扩展成完整的可观测体系：记录每步的模型耗时与 token 消耗做成本分析，把失败轨迹沉淀成评测集做回归测试，或者让多个智能体共享同一套检查点机制来编排长任务。执行轨迹本身，就是智能体工程化最有价值的资产之一。
