路线 02：Agent 执行、暂停与恢复

学习场景

客户要求查询订单，随后申请换货。沿着同一个 run_id 阅读请求受理、模型调用、工具执行、客户补充信息、客户确认和主管审批，观察每一步怎样保存进度、检查权限和继续执行。

图中后端路径相对 src/zhifu/。连续箭头表示直接函数调用，箭头文字注明调用条件；间断箭头明确标注 HTTP、框架调度、对象返回或持久数据关系。框架节点和存储节点单独标明类型。load、model、followup 是 build_graph 内部函数，execute 是 build_tool_graph 内部函数。返回路线目录。

图 1：请求受理、发布配置与工作进程调度

flowchart TD
    HTTP["HTTP 请求<br/>POST /openapi/v1/agent/runs"]
    A["modules/agent_runtime/api.py<br/>enqueue<br/>身份、店铺权限和幂等检查"]
    B["modules/agent_runtime/service.py<br/>queue_run<br/>配置快照、任务数量和运行期限"]
    C["modules/merchant/service.py<br/>current_release_version<br/>读取店铺当前发布版本"]
    R[("MySQL<br/>modules/merchant/models.py<br/>AgentRelease：发布配置")]
    D[("MySQL<br/>modules/agent_runtime/models.py<br/>AgentRun：QUEUED 与 release_snapshot")]
    E["modules/agent_runtime/worker.py<br/>main → execute_one<br/>持续领取运行"]
    F["modules/agent_runtime/business_wait.py<br/>reconcile<br/>核对业务等待，见图 3"]
    G["modules/agent_runtime/worker.py<br/>claim<br/>租户调度、期限和执行租约"]
    H["modules/agent_runtime/capacity.py<br/>acquire<br/>申请并发额度"]
    I["modules/agent_runtime/runtime.py<br/>build_graph → build_tool_graph<br/>构建主图及工具子图"]
    J["LangGraph 框架<br/>graph.ainvoke<br/>按 next_node 调度，见图 2"]
    HTTP -. "HTTP / FastAPI 路由分发" .-> A
    A -->|新请求| B
    A -. "相同幂等键及内容：读取原运行并复核权限" .-> D
    B --> C
    B -. "按版本读取 manifest" .-> R
    B -. "事务提交：保存 graph_version 和模型配置" .-> D
    E -->|每轮首先调用| F
    E -->|核对之后领取| G
    D -. "跨进程：读取排队任务或过期租约" .-> G
    G -->|选定租户后| H
    G -. "事务提交：RUNNING，lease_token 增加" .-> D
    E -->|领取成功后| I
    E -->|调用已构建的图| J
    I -. "返回已编译图对象" .-> J

enqueue 对重复请求执行 validate_result_access，当前权限撤销后不能继续读取旧工具结果。新运行保存店铺发布配置；没有店铺发布记录时使用 AgentSettings 与当前 GRAPH_VERSION。queue_run 验证 graph_version 已登记在 GRAPH_RELEASES 中，当前默认版本为 after-sales-v10。

claim 优先考虑商家预留额度，再按租户最近获得调度机会的时间选择租户，租户内部按创建时间选择任务。Redis 额度不足时保持排队。租约更新后，旧工作进程无法通过 locked_run 的 lease_token 检查。

会话消息的另一个入口是 conversations/api.py 的 accept_message：它准备最近消息、调用同一个 queue_run，再保存 conversation_id、control_version 与 active_run_id。该调用与独立运行入口共用后续执行流程。

图 1 源码导航

api.py：enqueue、validate_result_access：找到相同幂等键的查询和权限复核。
service.py：queue_run、merchant/service.py：current_release_version：追踪 release_snapshot、graph_version 与 deadline_at 的来源。
config.py：GRAPH_RELEASES：核对不同版本的 clarification、business_wait、followup 开关。
worker.py：execute_one、claim、capacity.py：acquire：分别追踪工作进程、数据库领取和 Redis 并发控制。
runtime.py：build_graph、conversations/api.py：accept_message：查看图构建与会话入口的连接位置。

图 2：LangGraph 调度模型、工具与最终答复

flowchart TD
    G["LangGraph 框架<br/>graph.ainvoke / add_conditional_edges<br/>依据节点返回的 next_node 调度"]
    L["modules/agent_runtime/runtime.py<br/>load<br/>读取已保存的 next_node"]
    M["modules/agent_runtime/runtime.py<br/>model<br/>校验引用、读取记忆、调用模型并保存响应"]
    T["modules/agent_runtime/runtime.py<br/>execute<br/>工具子图中的执行节点"]
    F["modules/agent_runtime/runtime.py<br/>followup<br/>处理单独的回访登记工具"]
    K["modules/agent_runtime/runtime.py<br/>locked_run<br/>身份、控制权、租约和店铺权限检查"]
    V["modules/agent_runtime/tools.py<br/>model_tools → tool_definitions<br/>按 graph_version 声明工具"]
    P["LangChain / 模型服务<br/>ChatOpenAI.bind_tools(...).ainvoke<br/>返回答复或 tool_calls"]
    I["modules/agent_runtime/tools.py<br/>invoke_tool<br/>参数校验与业务服务调用"]
    W["modules/agent_runtime/inputs.py：pause_for_input<br/>modules/agent_runtime/business_wait.py：pause_for_business<br/>保存等待状态，见图 3"]
    C["modules/conversations/service.py<br/>finish_run<br/>保存最终会话事件并释放 active_run_id"]
    G -. "框架调度：START → load" .-> L
    G -. "框架调度：next_node=model" .-> M
    G -. "框架调度：next_node=tools，进入工具子图" .-> T
    G -. "框架调度：next_node=followup" .-> F
    L --> K
    M -->|调用模型前及提交结果前| K
    T -->|读取工具请求及提交工具结果前| K
    F --> K
    M --> V
    M -. "SDK 调用并通过 HTTP 访问模型服务" .-> P
    T -->|订单、工单、创建方案等工具| I
    F -->|create_followup_request| I
    T -->|补充信息，或方案创建后的业务等待| W
    M -->|无 tool_calls：设置 COMPLETED 后调用| C
    L -. "节点返回 next_node" .-> G
    M -. "节点返回 tools 或 done" .-> G
    T -. "节点返回 model、followup 或 wait" .-> G
    F -. "节点返回 model" .-> G

build_graph 通过 add_node 注册内部函数，通过 add_conditional_edges 定义运行方向。工具节点返回后，由 LangGraph 根据 next_node 调度下一节点。done 与 wait 都连接 END；前者对应完成，后者保留等待状态，后续通过图 3 的入口继续原运行。

工具执行与权限检查使用同一套版本清单。execute 对每个工具参数调用 Pydantic 校验，并用 agent:{run_id}:{step}:{index} 生成业务幂等标识。工具结果、步骤编号和下一节点在数据库事务中共同提交。create_followup_request 必须单独调用；工具节点先保存 next_node=followup，由框架继续执行回访节点。知识检索使用专用分支：事务外检索，返回后调用 verified_results，具体流程见路线 04。

model 在网络调用前保存 ModelAttempt，通过 BudgetCallback 检查预算与记录计量。模型网络请求处于两个短事务之间；返回后保存已经产生的用量，再检查运行权限、知识引用和记忆版本。工作进程遇到异常时保存 FAILED 和 error_code，并在 finally 中释放并发额度。

图 2 源码导航

runtime.py：build_graph、load、model：结合函数末尾的图注册代码阅读调度关系。
runtime.py：locked_run：查看 token_hash、会话控制版本、租约与店铺状态检查。
runtime.py：build_tool_graph / execute、followup：追踪工具结果、幂等标识和下一节点的共同提交。
tools.py：model_tools、tool_definitions、invoke_tool：比较工具声明和执行时采用的同一版本规则。
budget_callback.py：BudgetCallback、conversations/service.py：finish_run：阅读 LangChain 回调和最终会话事件的保存位置。

图 3：补充信息、客户确认与主管审批

flowchart TD
    T["modules/agent_runtime/runtime.py<br/>execute<br/>执行需要等待的工具"]
    P["modules/agent_runtime/inputs.py<br/>pause_for_input<br/>等待客户补充信息"]
    B["modules/agent_runtime/business_wait.py<br/>pause_for_business<br/>等待方案确认或审批"]
    R[("MySQL<br/>modules/agent_runtime/models.py<br/>AgentRun + RunInputEvent：状态和输入收据")]
    H["HTTP / 客户操作<br/>POST /openapi/v1/agent/runs/{run_id}/inputs"]
    A["modules/agent_runtime/api.py<br/>inputs → owned_run<br/>校验本人运行与当前权限"]
    I["modules/agent_runtime/inputs.py<br/>accept_input<br/>校验等待类型、版本、期限与幂等键"]
    X["HTTP / 客户或主管操作<br/>确认方案、修订方案、提交审批决定"]
    Y["modules/after_sales/api.py<br/>confirm / revise / decide → command<br/>提交相应业务操作"]
    S["modules/after_sales/service.py<br/>AfterSalesService.execute<br/>调用 confirm、revise 或 decide"]
    E[("MySQL<br/>modules/after_sales/models.py<br/>Ticket + Confirmation + Approval + BusinessEvent")]
    W["modules/agent_runtime/worker.py<br/>execute_one<br/>每轮核对业务等待"]
    Q["modules/agent_runtime/business_wait.py<br/>reconcile → apply_event<br/>核验当前方案、业务事件与授权"]
    C["modules/agent_runtime/worker.py<br/>claim<br/>重新领取同一个 run_id"]
    T -->|request_clarification| P
    T -->|创建方案且版本启用 business_wait| B
    P -. "同事务：WAITING_USER，expected_input_type=clarification" .-> R
    B -. "同事务：WAITING_USER 或 WAITING_APPROVAL" .-> R
    H -. "HTTP / FastAPI 路由分发" .-> A
    A -->|owned_run 返回后| I
    I -. "同事务：保存输入收据并设置 QUEUED/model" .-> R
    X -. "HTTP / FastAPI 路由分发" .-> Y
    Y --> S
    S -. "事务提交：当前业务状态、确认和审批证据" .-> E
    W --> Q
    R -. "读取业务等待运行" .-> Q
    E -. "跨进程：读取已提交业务事件和当前版本" .-> Q
    Q -->|DRAFT、NEEDS_REVISION 或 WAITING_REVIEW| B
    Q -. "其他有效业务状态：保存收据并设置 QUEUED/model" .-> R
    Q -. "权限或业务证据检查失败：设置 FAILED" .-> R
    R -. "跨进程：领取 QUEUED，等待到期则记录 FAILED" .-> C

图中的 A、Y、Q 分别合并显示同一文件内的直接调用；这些调用对应各自入口函数的实际代码。inputs 调用 owned_run 后才调用 accept_input。售后 command 调用 AfterSalesService.execute；后者根据 action 调用对应业务方法。reconcile 调用 apply_event，统一处理业务检查异常。

补充信息只允许 expected_input_type=clarification，必须匹配当前 interrupt_id 与 run_version。方案确认和审批等待通过售后接口产生业务记录。apply_event 检查当前工单版本、客户确认、主管审批及主管当前权限；相同业务事件由 business_cursor 和输入收据约束重复处理。方案修订后，等待信息改用新的方案版本及等待标识。

规则失效或订单版本变化时，apply_event 可以将工单置为 NEEDS_REVISION 并继续等待；订单快照仅仅超过时效时返回 False，等待订单同步。审批人权限撤销时，reconcile 保存 FAILED 和 APPROVER_REVOKED。会话接管与取消通过状态、控制版本和租约检查阻止后续提交。

api.py 的 resume 供 SERVICE 身份核对已保存输入收据；输入受理或业务核对事务已经完成重新排队。该接口返回原收据与当前运行状态。

图 3 源码导航

inputs.py：pause_for_input、accept_input：追踪等待标识、输入收据和重新排队。
api.py：inputs、owned_run、resume：核对客户输入与内部收据查询各自的身份要求。
business_wait.py：pause_for_business、reconcile、apply_event：按调用方向阅读等待保存、事件核对和业务分支。
after_sales/api.py：command、confirm、decide、AfterSalesService.execute：找到产生业务证据的接口事务。

关键数据与状态

AgentRun：messages、step、next_node 保存执行进度；graph_version 与 release_snapshot 保存本次运行采用的配置；恢复时继续使用这些配置。
status 包括 QUEUED、RUNNING、WAITING_USER、WAITING_APPROVAL、COMPLETED、FAILED、CANCELLED。version 处理客户端并发；lease_token 处理工作进程执行权；control_version 处理会话接管。
RunInputEvent：通过租户、运行、幂等键和等待标识约束输入记录。业务恢复收据还保存 source_event_id、方案版本和工单版本。
ModelAttempt：区分模型请求已经开始、响应已收到和结果未知，保留调用次数及实际用量。
新运行及恢复后的执行期限为 10 分钟；补充信息和业务等待期限为 24 小时。工作进程单次执行限制 120 秒，领取租约为 125 秒。每次运行最多 6 次模型调用、20 个执行步骤，每轮最多 4 个工具调用；单个账户未结束运行数量限制为 3。

阅读检查问题

工作进程在工具事务提交后终止，新的工作进程依据哪些字段继续执行？同一方案为何不会重复创建？
店铺发布了新的 Agent 配置，正在等待客户补充信息的运行恢复后采用哪个版本？工具声明与执行怎样使用这个版本？
客户补充信息、客户确认方案、主管审批分别保存什么记录？审批后撤销主管权限，运行恢复过程中哪一步会拒绝继续？

测试文件阅读入口

本路线仅阅读以下用例及其断言，没有执行测试或模型调用。测试文件中的运行条件和断言用于理解设计边界。

test_agent_runtime.py：test_tool_graph_creates_one_proposal_on_replay 与 test_expired_lease_resumes_committed_tool_without_duplicate：阅读工具重复执行、租约过期及原工具结果复用。
test_agent_inputs.py：test_resume_same_run_atomic_idempotency_and_event 与 test_registered_old_graph_remains_executable：阅读相同输入并发重试和旧图版本继续执行。
test_business_wait.py：test_confirmation_approval_restore_same_run_once、test_revision_replaces_wait_and_requires_new_confirmation、test_takeover_and_approver_revocation_prevent_resume：阅读业务恢复、方案修订、会话接管与授权撤销。

02-agent-runtime.md 18 KB Trvalý odkaz Historie Surový