从零搭建轻量级 Agent 框架：状态机与事件流
[!NOTE]
执行摘要（TL;DR）
- 轻量级生产核心：不依赖臃肿的通用框架，直接用 Python 实现异步有限状态机（FSM），包含规划、工具调度、反思和流式事件。
- 人在回路与状态恢复：针对高风险操作内置中断流程，并用检查点持久化，让进程重启后能无缝恢复。
- 下一步：把这套基础运行时扩展成多 Agent 协作与并发，支持多角色集群。
为什么要让 Agent 状态显式化？
只要一个循环能记录当前节点、参数、恢复位置和停止规则，它就能实现可靠的 Agent。而状态图让这些控制点变得可检查。关于基础原理，可以参考 agent loops 和 state machines。
可下载的 Python 示例需要 Python 3.11+ 和 SQLite 3.35+，没有第三方依赖。退款是在内存列表里模拟的，不会联系任何支付服务。
审批检查点与恢复位置
检查点会保存下一个节点、业务状态、已消耗的步骤数和状态。如果只保存一个字典，然后从 Planner 重启，就可能覆盖已经批准的参数，并重跑之前的工作。这个示例把检查点持久化到 SQLite。批准会恢复被挂起的节点，拒绝会终止运行，而带条件的数据库更新保证一次审批只有一个消费者。一次审批只对一个边界生效；后续的敏感节点仍会中断。
重建工作流时，节点标识符和实现必须保持兼容。
flowchart LR
Plan[Planner] --> Pending[SQLite pending checkpoint]
Pending --> Decision{Decision}
Decision -->|Approved and claimed| Tool[Simulated refund]
Decision -->|Rejected| Rejected[rejected]
Tool --> Done[completed]
完整可运行示例
运行 python3 state_graph.py。演示会关闭存储，再基于同一个数据库重建图，以展示工作进程重启后的恢复能力。
"""
一个基于 SQLite 的轻量审批工作流；演示中没有外部副作用。
"""
import asyncio
import copy
import inspect
import json
import sqlite3
import tempfile
import uuid
from dataclasses import dataclass
from pathlib import Path
class AgentState(dict):
def apply_delta(self, delta):
for key, value in delta.items():
if key == "messages":
self[key] = self.get(key, []) + copy.deepcopy(value)
else:
self[key] = copy.deepcopy(value)
@dataclass
class AgentEvent:
event_type: str
node_name: str
payload: dict
run_id: str
class SqliteCheckpointer:
def __init__(self, path):
self.db = sqlite3.connect(path, isolation_level=None)
self.db.execute(
"""
CREATE TABLE IF NOT EXISTS runs (
run_id TEXT PRIMARY KEY,
node TEXT,
state TEXT,
steps INTEGER,
status TEXT NOT NULL
)
"""
)
def create(self, run_id, node, state):
self.db.execute(
"INSERT INTO runs VALUES (?, ?, ?, 0, 'running')",
(run_id, node, json.dumps(state)),
)
def save(self, run_id, node, state, steps, status):
self.db.execute(
"""
UPDATE runs
SET node=?, state=?, steps=?, status=?
WHERE run_id=?
""",
(node, json.dumps(state), steps, status, run_id),
)
def claim_approval(self, run_id, decision):
if decision not in {"approved", "rejected"}:
raise ValueError("decision must be approved or rejected")
status = "running" if decision == "approved" else "rejected"
row = self.db.execute(
"""
UPDATE runs SET status=?
"""
```python
WHERE run_id=? AND status='pending'
RETURNING node, state, steps
""", (status, run_id)).fetchone()
if row is None:
raise ValueError("approval missing, already consumed, or not pending")
return row[0], AgentState(json.loads(row[1])), row[2]
def close(self):
self.db.close()
class StateGraph:
def __init__(self, checkpointer, max_steps=20):
self.checkpointer = checkpointer
self.max_steps = max_steps
self.nodes = {}
self.edges = {}
self.conditional_edges = {}
self.interrupt_nodes = set()
self.entry_point = None
def add_node(self, name, function):
self.nodes[name] = function
def set_entry_point(self, name):
self.entry_point = name
def add_edge(self, source, target):
self.edges[source] = target
def add_conditional_edge(self, source, router):
self.conditional_edges[source] = router
def add_interrupt_before(self, node):
self.interrupt_nodes.add(node)
async def stream(self, state, run_id=None):
run_id = run_id or str(uuid.uuid4())
state = AgentState(copy.deepcopy(state))
self.checkpointer.create(run_id, self.entry_point, state)
async for event in self._drive(state, run_id, self.entry_point, 0):
yield event
async def resume(self, run_id, decision):
node, state, steps = self.checkpointer.claim_approval(run_id, decision)
if decision == "rejected":
yield AgentEvent("rejected", node, {}, run_id)
return
# 只跳过这个已批准的边界，不会跳过之后所有的审批
async for event in self._drive(state, run_id, node, steps, node):
yield event
async def _drive(self, state, run_id, node, steps, approved_node=None):
try:
while node and node != "END":
if steps >= self.max_steps:
raise RuntimeError("step budget exhausted")
if node in self.interrupt_nodes and node != approved_node:
self.checkpointer.save(run_id, node, state, steps, "pending")
yield AgentEvent("interrupted", node,
{"state": copy.deepcopy(state)}, run_id)
return
approved_node = None
yield AgentEvent("node_start", node, {}, run_id)
delta = self.nodes[node](AgentState(copy.
copy.deepcopy(state)))
if inspect.isawaitable(delta):
delta = await delta
if delta:
state.apply_delta(delta)
previous_node = node
router = self.conditional_edges.get(node)
node = router(state) if router else self.edges.get(node, "END")
steps += 1
self.checkpointer.save(run_id, node, state, steps, "running")
yield AgentEvent("state_update", previous_node, {"delta": delta}, run_id)
self.checkpointer.save(run_id, "END", state, steps, "completed")
yield AgentEvent("completed", "END", {"final_state": copy.deepcopy(state)}, run_id)
except (Exception, asyncio.CancelledError):
self.checkpointer.save(run_id, node, state, steps, "failed")
raise
def build_workflow(checkpointer, refund_log):
graph = StateGraph(checkpointer)
graph.add_node("planner", lambda state: {"action": "refund", "amount": 1000})
graph.add_node("human_approval", lambda state: {"approval_recorded": True})
def refund(state):
# Simulation only. A real payment API needs a stable business idempotency key.
refund_log.append(state["amount"])
return {"status": "success",
"messages": [{"role": "tool", "content": "Simulated refund"}]}
graph.add_node("refund", refund)
graph.set_entry_point("planner")
graph.add_conditional_edge(
"planner",
lambda state: "human_approval" if state["amount"] > 500 else "refund",
)
graph.add_edge("human_approval", "refund")
graph.add_edge("refund", "END")
graph.add_interrupt_before("human_approval")
return graph
async def main():
with tempfile.TemporaryDirectory() as directory:
path = str(Path(directory) / "checkpoints.sqlite3")
refunds = []
store = SqliteCheckpointer(path)
workflow = build_workflow(store, refunds)
events = [event async for event in workflow.stream(AgentState(messages=[]))]
run_id = events[-1].run_id
store.close()
# Reopen storage and rebuild the graph to simulate a worker restart.
store = SqliteCheckpointer(path)
workflow = build_workflow(store, refunds)
async for event in workflow
resume(run_id, "approved"):
print(event.event_type, event.node_name)
store.close()
print("Simulated refunds:", refunds)
if __name__ == "__main__":
asyncio.run(main())
保证与生产环境扩展
这个示例会把审批状态持久化，并防止同一条审批记录被重复消费。但它并不能保证外部支付恰好执行一次。如果支付已经成功、但检查点还没来得及保存时程序崩溃，那仍然需要稳定的业务幂等键，或者事务性发件箱（transactional outbox）来兜底。可以参考 Agent 编排与评估相关内容。
同步 SQLite 事务适合本地示例。生产系统还需要：认证审批人、参数绑定决策、快照迁移、审计记录、租约（lease）以及取消传播。LangGraph 同样支持持久化检查点和自托管可观测性；自己写运行时并不天然更安全。
常见问题（FAQ）
Q1：拒绝审批会执行工具吗？
不会。resume(run_id, "rejected") 会消费掉待处理记录，并发出一个 rejected 事件。无效的决策会直接报错，而不是隐式批准。
Q2：重复的审批回调会让退款重复执行吗？
待处理记录只能被认领一次。重复回调会被拒绝。不过真实支付 API 仍然需要幂等机制来应对重试。
Q3：检查点能撤销外部副作用吗？