客户要求查询订单,随后申请换货。沿着同一个 run_id 阅读请求受理、模型调用、工具执行、客户补充信息、客户确认和主管审批,观察每一步怎样保存进度、检查权限和继续执行。
图中后端路径相对 src/zhifu/。连续箭头表示直接函数调用,箭头文字注明调用条件;间断箭头明确标注 HTTP、框架调度、对象返回或持久数据关系。框架节点和存储节点单独标明类型。load、model、followup 是 build_graph 内部函数,execute 是 build_tool_graph 内部函数。返回路线目录。
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。该调用与独立运行入口共用后续执行流程。
release_snapshot、graph_version 与 deadline_at 的来源。clarification、business_wait、followup 开关。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 中释放并发额度。
token_hash、会话控制版本、租约与店铺状态检查。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 身份核对已保存输入收据;输入受理或业务核对事务已经完成重新排队。该接口返回原收据与当前运行状态。
messages、step、next_node 保存执行进度;graph_version 与 release_snapshot 保存本次运行采用的配置;恢复时继续使用这些配置。status 包括 QUEUED、RUNNING、WAITING_USER、WAITING_APPROVAL、COMPLETED、FAILED、CANCELLED。version 处理客户端并发;lease_token 处理工作进程执行权;control_version 处理会话接管。source_event_id、方案版本和工单版本。本路线仅阅读以下用例及其断言,没有执行测试或模型调用。测试文件中的运行条件和断言用于理解设计边界。