带有外部影响的任务需要明确的执行边界。收到批准、提交操作和确认操作完成，是不同阶段。以写入一条模拟台账为例，系统先校验参数，再暂停等待人工处理，然后有限重试，并检查台账实际状态。
目标、状态、执行边界和验收标准
| 项目 | 定义 |
|---|---|
| 目标 | 在人工批准具体版本后，为指定task_id写入一条模拟台账记录。 |
| 状态 | task_id标识业务操作；amount保存操作内容；version绑定审批版本；decision记录批准或拒绝；status记录结果；attempts记录尝试次数。 |
| 执行边界 | 金额为1到100的整数；没有批准不得写入；审批必须匹配当前版本；只重试TransientError；最多两次；所有操作仅发生在模拟字典中。 |
| 验收标准 | 批准且版本一致才写入；拒绝与错误版本不产生记录；暂时失败后可恢复；重复task_id不增加记录；越界输入被拒绝。 |
方法与设计思想
目标监控：把完成写成可检查的状态
目标是台账存在指定记录，不能把工具返回“成功”当作唯一依据。commit执行以后再次检查task_id对应的值，确认与目标金额一致才标记completed。
这属于任务级监控。系统级可靠性还需统计失败率、恢复率、等待时长和异常类型。
人工参与：审批必须绑定具体内容
interrupt暂停流程，向调用方展示task_id、金额和version。调用方使用相同thread_id与Command恢复执行。只有approve为true且版本一致才进入commit。
version由task_id和金额计算得到。内容变化就需要新的审批。示例的审批回复由脚本模拟；真实产品还需认证审批人并验证其权限，不能允许模型自行提供批准。
异常恢复：只重试明确的暂时失败
with_retry只捕获TransientError，最多尝试两次。参数错误、版本冲突和业务错误不会被当成暂时故障无限重试。
第一次模拟写入失败，第二次成功。台账以task_id作为幂等键，相同操作再次提交不会产生第二条记录；相同ID绑定不同内容则拒绝。
护栏：在程序层约束操作
输入边界由prepare校验，审批边界由approve判断，内容一致性由commit检查。关键限制由程序执行，而不是依赖模型的自然语言承诺。
中断节点恢复时会重新运行，因此审批之前应避免不可重复的副作用。示例把写入放在独立commit节点中，但内存台账不具备进程崩溃后的耐久性。
为什么加入这些机制
| 机制 | 解决的问题 | 验证方式 |
|---|---|---|
| 输入护栏 | 阻止越界与无效操作 | 金额超范围或task_id为空时立即失败 |
| 人工中断 | 在重要动作之前保留决策权 | 拒绝审批时台账保持不变 |
| 版本绑定 | 防止旧审批批准新内容 | 错误版本无法进入写入节点 |
| 有限重试 | 恢复已分类的暂时故障 | 第一次失败、第二次成功且次数为二 |
| 幂等与结果确认 | 避免重复记录并确认实际完成 | 再次提交同一ID仍只有一条记录 |
LangChain与LangGraph代码实现
主代码在Python 3.12.14、LangChain 1.4.3、langchain-core 1.6.6、LangGraph 1.2.12、Pydantic 2.13.5下运行验证。建议使用独立虚拟环境：
python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install langchain==1.4.3 langchain-core==1.6.6 langgraph==1.2.12 pydantic==2.13.5
将下面完整代码保存为flow05.py，运行python flow05.py。默认模式不需要模型凭据，使用固定数据验证流程和异常分支。
import hashlib
import json
from typing_extensions import TypedDict
from langchain_core.runnables import RunnableLambda
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
class TransientError(Exception):
pass
class State(TypedDict, total=False):
task_id: str
amount: int
version: str
prepared: bool
decision: str
status: str
attempts: int
def digest(task_id: str, amount: int) -> str:
return hashlib.sha256(f"{task_id}:{amount}".encode()).hexdigest()
def build():
# 仅模拟写入台账，真实资金和外部服务均不参与。
ledger = {}
attempts = {}
def prepare(state):
amount = state["amount"]
if type(amount) is not int or not 1 <= amount <= 100:
raise ValueError("金额必须是1到100的整数")
if not state["task_id"].strip():
raise ValueError("task_id不能为空")
return {"prepared": True, "version": digest(state["task_id"], amount)}
def approve(state):
reply = interrupt({"task_id": state["task_id"], "amount": state["amount"],
"version": state["version"], "action": "写入模拟台账"})
approved = (isinstance(reply, dict) and reply.get("approve") is True
and reply.get("version") == state["version"])
return {"decision": "approved" if approved else "rejected"}
def simulated_write(payload):
key = payload["task_id"]
if key in ledger:
if ledger[key] != payload["amount"]:
raise ValueError("同一task_id绑定的金额发生变化")
return "existing"
attempts[key] = attempts.get(key, 0) + 1
if attempts[key] == 1:
raise TransientError("教学用：第一次请求暂时失败")
ledger[key] = payload["amount"]
return "created"
writer = RunnableLambda(simulated_write).with_retry(
retry_if_exception_type=(TransientError,),
stop_after_attempt=2, wait_exponential_jitter=False,
)
def commit(state):
if digest(state["task_id"], state["amount"]) != state["version"]:
raise ValueError("操作内容发生变化")
writer.invoke({"task_id": state["task_id"], "amount": state["amount"]})
if ledger.get(state["task_id"]) != state["amount"]:
raise RuntimeError("写入结果无法确认")
return {"status": "completed", "attempts": attempts[state["task_id"]]}
graph = StateGraph(State)
for name, fn in [("prepare", prepare), ("approve", approve), ("commit", commit)]:
graph.add_node(name, fn)
graph.add_node("reject", lambda _: {"status": "rejected"})
graph.add_edge(START, "prepare")
graph.add_edge("prepare", "approve")
graph.add_conditional_edges("approve", lambda s: s["decision"],
{"approved": "commit", "rejected": "reject"})
graph.add_edge("commit", END)
graph.add_edge("reject", END)
return graph.compile(checkpointer=InMemorySaver()), ledger
def approval(app, task_id, amount, approve=True, wrong_version=False):
cfg = {"configurable": {"thread_id": f"session-{task_id}"}}
paused = app.invoke({"task_id": task_id, "amount": amount}, cfg)
assert "__interrupt__" in paused
version = "wrong" if wrong_version else paused["version"]
return app.invoke(Command(resume={"approve": approve, "version": version}), cfg)
if __name__ == "__main__":
app, ledger = build()
result = approval(app, "t1", 30)
assert result["status"] == "completed" and result["attempts"] == 2
assert approval(app, "t2", 30, approve=False)["status"] == "rejected"
assert approval(app, "t3", 30, wrong_version=True)["status"] == "rejected"
assert set(ledger) == {"t1"}
approval(app, "t1", 30)
assert len(ledger) == 1
try:
app.invoke({"task_id": "invalid", "amount": 200},
{"configurable": {"thread_id": "invalid"}})
except ValueError:
pass
else:
raise AssertionError("越界金额未被拒绝")
print(json.dumps({"status": result["status"], "attempts": result["attempts"],
"ledger": ledger}, ensure_ascii=False, indent=2))
prepare定义可执行参数和版本，approve只收集决策，commit负责可重试写入及结果验证。异常分类集中在writer的重试配置中；模拟台账和尝试次数保存在闭包内。
验收与验证结果
脚本分别执行批准、拒绝、错误版本、重复提交和越界金额。所有结果由台账状态验证。固定脚本的人工决定仅用于复现分支；没有连接任何资金服务，也没有实现真实审批用户界面。
Google ADK实现与LangChain对照
Google ADK对照实现把审批组织成两次调用：第一次写入proposal和awaiting_approval，第二次由可信入口提供decision后执行。它是自建会话状态机，不是LangGraph中断API的替代名称。ADK也提供FunctionTool操作确认能力，真实模型工具审批可按其独立协议接入。
| 比较项 | LangChain／LangGraph | Google ADK |
|---|---|---|
| 等待审批 | interrupt暂停图并保留检查点 | 自定义代理写待审批状态并结束本次调用 |
| 继续执行 | Command(resume=…)与同一thread_id | 同一Session的新调用，通过state_delta接收决定 |
| 版本绑定 | 草稿或操作指纹由程序校验 | 同样校验金额与task_id形成的指纹 |
| 重试与幂等 | with_retry加台账幂等检查 | 自定义循环和台账检查；不是自动框架保证 |
Google ADK完整代码
对照代码已在google-adk==2.11.0与Python 3.12环境运行。安装python -m pip install google-adk==2.11.0，保存为adk05.py后运行。默认不调用真实模型。
import asyncio
import inspect
import json
from collections.abc import Callable
from typing import Any
from google.adk.agents import BaseAgent, SequentialAgent, ParallelAgent, LoopAgent
from google.adk.events import Event, EventActions
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
class Step(BaseAgent):
action: Callable[..., Any]
async def _run_async_impl(self, ctx):
delta = self.action(ctx.session.state)
if inspect.isawaitable(delta):
delta = await delta
yield Event(author=self.name, actions=EventActions(state_delta=delta))
async def run_state(agent, state, session_id="s1", user_id="u1", service=None):
service = service or InMemorySessionService()
await service.create_session(app_name="demo", user_id=user_id,
session_id=session_id, state=state)
runner = Runner(agent=agent, app_name="demo", session_service=service)
async for _ in runner.run_async(user_id=user_id, session_id=session_id,
new_message=types.Content(role="user", parts=[types.Part(text="执行任务")])):
pass
session = await service.get_session(app_name="demo", user_id=user_id,
session_id=session_id)
return dict(session.state)
import hashlib
LEDGER = {}
class ApprovalAgent(BaseAgent):
async def _run_async_impl(self, ctx):
state = ctx.session.state
task, amount = state["task_id"], state["amount"]
if type(amount) is not int or not 1 <= amount <= 100 or not task.strip():
raise ValueError("参数越界")
version = hashlib.sha256(f"{task}:{amount}".encode()).hexdigest()
if state.get("phase") != "awaiting_approval":
yield Event(author=self.name, actions=EventActions(state_delta={
"phase": "awaiting_approval", "version": version,
"proposal": {"task_id": task, "amount": amount, "version": version}}))
return
decision = state.get("decision", {})
if not (decision.get("approve") is True and decision.get("version") == version):
yield Event(author=self.name, actions=EventActions(state_delta={"status": "rejected"}))
return
attempts = 0
if task in LEDGER and LEDGER[task] != amount:
raise ValueError("幂等键冲突")
while task not in LEDGER and attempts < 2:
attempts += 1
if attempts == 1:
continue # 模拟一次暂时失败。
LEDGER[task] = amount
assert LEDGER.get(task) == amount
yield Event(author=self.name, actions=EventActions(state_delta={
"status": "completed", "attempts": attempts}))
async def exercise(task, approve, bad_version=False):
service = InMemorySessionService()
agent = ApprovalAgent(name="approval")
runner = Runner(agent=agent, app_name="demo", session_service=service)
await service.create_session(app_name="demo", user_id="u1", session_id=task,
state={"task_id": task, "amount": 30})
async for _ in runner.run_async(user_id="u1", session_id=task,
new_message=types.Content(role="user", parts=[types.Part(text="准备")])):
pass
session = await service.get_session(app_name="demo", user_id="u1", session_id=task)
assert session.state["phase"] == "awaiting_approval" and task not in LEDGER
decision = {"approve": approve, "version": "wrong" if bad_version else session.state["version"]}
# 真实应用应由经过认证的审批入口提供state_delta。
async for _ in runner.run_async(user_id="u1", session_id=task,
state_delta={"decision": decision},
new_message=types.Content(role="user", parts=[types.Part(text="审批结果")])):
pass
session = await service.get_session(app_name="demo", user_id="u1", session_id=task)
return dict(session.state)
async def main():
good = await exercise("t1", True)
assert good["status"] == "completed" and good["attempts"] == 2
assert (await exercise("t2", False))["status"] == "rejected"
assert (await exercise("t3", True, bad_version=True))["status"] == "rejected"
assert LEDGER == {"t1": 30}
print(json.dumps({"status": good["status"], "ledger": LEDGER}, ensure_ascii=False, indent=2))
if __name__ == "__main__":
asyncio.run(main())
ADK验证第一次调用后没有台账记录，批准才写入，拒绝与错误版本都不写入。审批人身份和跨进程幂等仍属于应用需要实现的能力。
接口与版本边界
对照采用SequentialAgent、ParallelAgent、LoopAgent或BaseAgent模板接口，以便比较相同控制思想。ADK 2.x官方文档已引入图工作流和动态工作流，新项目应评估对应机制；这里的版本运行通过不代表模板API是所有新项目的首选。两种框架的运行结果不能直接等同于真实模型效果。
扩展边界
InMemorySaver、台账和幂等记录都在内存中。生产系统需要持久数据库、唯一约束、事务或外部服务的幂等能力，另外验证重启恢复和“提交成功但响应丢失”的情况。检查点恢复本身不能保证外部副作用只发生一次。
实现依据
ADK原生工具操作确认可参照Action confirmations，需要与模型工具请求和审批入口共同接入。本对照代码采用显式会话状态机，未验证原生工具确认的端到端行为。