|
|
@@ -0,0 +1,527 @@
|
|
|
+# 对话消息、Agent worker 与前端展示流程
|
|
|
+
|
|
|
+代码依据：2026-09-30 当前工作区文件。主流程使用消费者 H5 的 `chat` 页面，人工客服流程见第 8 节。图中的函数名、事件名、表名与状态名均取自当前代码。
|
|
|
+
|
|
|
+## 1. 一次普通对话的调用顺序
|
|
|
+
|
|
|
+```mermaid
|
|
|
+sequenceDiagram
|
|
|
+ autonumber
|
|
|
+ actor User as 客户
|
|
|
+ participant UI as H5 chat 页面
|
|
|
+ participant API as FastAPI
|
|
|
+ participant DB as MySQL
|
|
|
+ participant Worker as Agent worker
|
|
|
+ participant Redis as Redis state
|
|
|
+ participant Model as 模型服务
|
|
|
+ UI->>API: 查询或创建会话，读取历史和当前运行
|
|
|
+ API->>DB: 查询 conversations / conversation_events / agent_runs
|
|
|
+ API-->>UI: 会话状态、消息历史、当前运行
|
|
|
+ UI->>API: GET /conversations/{id}/events，Last-Event-ID
|
|
|
+ API-->>UI: SSE connected 注释，保持连接
|
|
|
+ User->>UI: 输入消息并点击发送
|
|
|
+ UI->>API: POST /conversations/{id}/messages<br/>text + expected_control_version + Idempotency-Key
|
|
|
+ API->>DB: 开始事务，校验身份并锁定会话
|
|
|
+ API->>DB: 检查幂等键、控制版本与活动运行
|
|
|
+ API->>DB: INSERT agent_runs，status=QUEUED
|
|
|
+ API->>DB: UPDATE conversations.active_run_id 和 sequence
|
|
|
+ API->>DB: INSERT conversation_events，type=message.user
|
|
|
+ API->>DB: 提交事务
|
|
|
+ API-->>UI: HTTP 201，EventView，payload.run_id
|
|
|
+ Note over UI: 返回序号较新时保存运行标识，清空输入<br/>消息列表由服务端事件更新
|
|
|
+ Worker->>DB: execute_one → reconcile → claim<br/>查询任务，锁定租户、会话与运行
|
|
|
+ Worker->>Redis: acquire 申请模型并发额度
|
|
|
+ Redis-->>Worker: 额度准入结果
|
|
|
+ Worker->>DB: RUNNING，lease_token + 1<br/>lease_until=当前时间+125秒，提交事务
|
|
|
+ Worker->>DB: load → model<br/>保存 ModelAttempt 与 agent.model_started
|
|
|
+ Worker->>DB: BudgetCallback 预占 token<br/>已配置 MoneyPolicy 时预占金额
|
|
|
+ Worker->>Model: ChatOpenAI.bind_tools(...).ainvoke(...)
|
|
|
+ Model-->>Worker: 完整 AIMessage 与 usage
|
|
|
+ Worker->>DB: 预算结算；独立保存 ModelAttempt 用量
|
|
|
+ opt 模型要求调用工具
|
|
|
+ Worker->>DB: tools / followup 执行业务并保存 ToolMessage
|
|
|
+ Worker->>DB: 保存 agent.tools_completed 与 next_node
|
|
|
+ Worker->>Model: 使用工具结果继续模型调用
|
|
|
+ Model-->>Worker: 完整最终答复与 usage
|
|
|
+ end
|
|
|
+ Worker->>DB: 校验租约、权限、控制权、引用与记忆版本
|
|
|
+ Worker->>DB: 同一事务保存 COMPLETED 和 message.assistant<br/>清除 active_run_id，提交事务
|
|
|
+ Worker->>Redis: finally 释放并发额度
|
|
|
+ loop SSE 每轮查询后等待0.5秒
|
|
|
+ API->>DB: read_events 查询 sequence 大于当前游标的事件
|
|
|
+ DB-->>API: 已提交事件，每批最多50条
|
|
|
+ API-->>UI: SSE id / event / data
|
|
|
+ end
|
|
|
+ UI->>UI: receiveEvent 按序号去重、核验控制版本、更新状态
|
|
|
+ UI->>UI: shared.messages → messageItems → Vue 消息气泡
|
|
|
+ UI-->>User: 显示完整答复和原始消息时间
|
|
|
+```
|
|
|
+
|
|
|
+图中接口路径省略统一前缀 `/openapi/v1`。SSE 从建立连接后便持续读取，客户消息和运行事件可以在执行过程中先后到达。数据库提交后，HTTP 响应、worker 领取和 SSE 读取由各自进程调度，三者的先后顺序没有固定保证。
|
|
|
+
|
|
|
+普通成功运行通过 `message.assistant` 提交最终答复，同时数据库状态变为 `COMPLETED`。前端据此清除当前活动运行并展示正文。
|
|
|
+
|
|
|
+代码入口：[customer.js:254](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/pages/customer.js:254)、[conversations/api.py:185](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/api.py:185)、[worker.py:137](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/worker.py:137)、[runtime.py:284](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/runtime.py:284)、[stream.py:131](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/stream.py:131)。
|
|
|
+
|
|
|
+## 2. 前端如何建立会话并发送消息
|
|
|
+
|
|
|
+```mermaid
|
|
|
+flowchart TD
|
|
|
+ A[进入消费者 chat 页面] --> B[loadPage 建立本次页面 generation]
|
|
|
+ B --> C[ensureConversation]
|
|
|
+ C --> D{已有 conversationId？}
|
|
|
+ D -->|已有| E[GET 当前会话]
|
|
|
+ D -->|没有| F[GET 当前店铺本人会话列表]
|
|
|
+ F --> G{列表存在会话？}
|
|
|
+ G -->|存在| H[使用返回列表第一项]
|
|
|
+ G -->|没有| I[POST 创建会话，保存聊天与控制状态]
|
|
|
+ E --> J[保存 conversation 和 control_owner]
|
|
|
+ H --> J
|
|
|
+ I --> J
|
|
|
+ J --> K[refreshMessages 按 after 读取历史<br/>每页100条，按 sequence 去重]
|
|
|
+ K --> L[重新读取会话<br/>存在 active_run_id 时读取运行]
|
|
|
+ L --> M[读取人工协助状态<br/>connectMessages 建立 SSE]
|
|
|
+ M --> N[connected=true]
|
|
|
+ N --> O[send 检查登录、输入、连接和 busy]
|
|
|
+ O --> P{存在活动运行？}
|
|
|
+ P -->|没有| Q[ensureConversation 获取当前控制版本]
|
|
|
+ P -->|等待 clarification| Q
|
|
|
+ P -->|其他活动状态| S[发送入口停止本次操作]
|
|
|
+ Q --> T[组合 aiContext 与问题文本<br/>检查总长度不超过4000字符]
|
|
|
+ T --> Clarify{正在等待 clarification？}
|
|
|
+ Clarify -->|是的| R[POST /agent/runs/id/inputs<br/>等待标识、运行版本、输入框回答]
|
|
|
+ Clarify -->|其他情况| U[生成或复用本次消息 Idempotency-Key]
|
|
|
+ U --> V[api.send → api.request → fetch POST]
|
|
|
+ V --> W[HTTP 201 返回 EventView]
|
|
|
+ R --> X[HTTP 202 返回输入收据和原运行状态]
|
|
|
+ W --> Y[存在 run_id 且返回序号较新时更新 active_run_id<br/>移除本次页面幂等键]
|
|
|
+ X --> InputResult[interrupt_id 仍匹配时更新 activeRun<br/>移除本次输入幂等键]
|
|
|
+ InputResult --> Z[清空输入，showLatestMessages<br/>SSE 事件到达后展示消息]
|
|
|
+ Y --> Z
|
|
|
+```
|
|
|
+
|
|
|
+`ensureConversation()` 使用已保存的会话 ID；没有 ID 时使用列表第一项，列表为空才创建。后端会话列表按照 UUID 升序返回。`consult(context)` 从商品或订单入口保存 `state.aiContext`，然后进入 `#/customer/chat`。
|
|
|
+
|
|
|
+| 请求位置 | 字段 | 用途与约束 |
|
|
|
+| --- | --- | --- |
|
|
|
+| Header | `Authorization: Bearer ...` | 消费者当前会话令牌 |
|
|
|
+| Header | `Idempotency-Key` | 1～64 字符；同一次提交结果未知时复用原键 |
|
|
|
+| JSON | `text` | 后端限制 1～4000 字符；页面会加入商品或订单咨询上下文 |
|
|
|
+| JSON | `expected_control_version` | 最近读取的会话控制版本，至少为 1 |
|
|
|
+
|
|
|
+页面输入控件最多 2000 字符，加入咨询对象后再次检查 4000 字符上限。`send()` 成功后更新运行标识并清空输入，服务端事件负责向 `shared.messages` 添加消息。`result.sequence > state.after` 检查处理 SSE 提前送达的情况。
|
|
|
+
|
|
|
+页面的 `key(action, body)` 将幂等键保存在 `state.commandKeys`。普通 HTTP 请求默认超时为 20 秒；网络结果未知时保留页面键，客户重试相同提交时继续使用。`createApi.request()` 同时通过 `pendingKeys` 管理请求键，页面显式传入的键优先生效。
|
|
|
+
|
|
|
+代码：[会话与历史读取](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/pages/customer.js:136)、[页面加载](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/pages/customer.js:373)、[消费者接口映射](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/services/customer.js:15)、[HTTP 客户端](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/api.js:44)。
|
|
|
+
|
|
|
+## 3. 后端如何将消息与任务一起入库
|
|
|
+
|
|
|
+```mermaid
|
|
|
+flowchart TD
|
|
|
+ A[POST messages<br/>FastAPI 校验 Message 与 Idempotency-Key] --> B[sessions.begin 开始事务]
|
|
|
+ B --> C[authenticate 校验令牌、租户、账户<br/>access 锁定并检查会话访问权限]
|
|
|
+ C --> D[accept_message 计算请求 SHA-256]
|
|
|
+ D --> E[prior 查询同会话、同操作者、同键的事件]
|
|
|
+ E --> F{已有事件？}
|
|
|
+ F -->|同摘要| G[返回原 EventView]
|
|
|
+ F -->|摘要不同| X[抛出 BusinessError<br/>事务撤销本次未提交修改]
|
|
|
+ F -->|没有| H[检查 expected_control_version<br/>客服还需已接管且为 assigned_staff_id]
|
|
|
+ H --> I{CUSTOMER 且 control_owner=AGENT？}
|
|
|
+ I -->|符合| J[确认 active_run_id 为空<br/>读取最近12条历史消息<br/>连同本次文本限制24000字符]
|
|
|
+ J --> K[queue_run 读取发布版本和模型配置<br/>检查账户活动任务少于3个]
|
|
|
+ K --> L[session.add AgentRun<br/>session.flush 执行 INSERT，status=QUEUED]
|
|
|
+ L --> M[设置 run.conversation_id、control_version<br/>设置 conversation.active_run_id]
|
|
|
+ I -->|客户由人工接待或客服回复| N[run_id=null]
|
|
|
+ M --> O[emit：conversation.sequence + 1<br/>session.add ConversationEvent]
|
|
|
+ N --> O
|
|
|
+ O --> P[客户消息按授权情况添加 MemoryJob<br/>已分配客服时添加通知]
|
|
|
+ P --> Q[构造 EventView]
|
|
|
+ Q --> R[正常退出 sessions.begin<br/>flush 剩余变更并 COMMIT]
|
|
|
+ G --> R
|
|
|
+ R --> S[HTTP 201 返回持久事件]
|
|
|
+ H -. 校验失败 .-> X
|
|
|
+ J -. 存在活动运行 .-> X
|
|
|
+ K -. 配置或数量不符合 .-> X
|
|
|
+```
|
|
|
+
|
|
|
+`message()` 将这些操作放在同一个 `async with sessions.begin()` 中。`session.add()` 把对象交给当前 SQLAlchemy Session；`flush()` 将 SQL 发送给数据库，事务在正常退出上下文时提交。任何一步抛出异常，本次事务中尚未提交的修改一起撤销。
|
|
|
+
|
|
|
+`queue_run()` 中的 `flush()` 发生在调用方设置 `conversation_id` 和 `control_version` 之前。运行插入、后续关联字段、会话活动标识和用户事件最终一起提交。worker 和 SSE 使用各自 Session 查询已提交数据；数据库连接使用 `READ COMMITTED`。
|
|
|
+
|
|
|
+| 数据表 | 本次写入或更新 | 后续读取方 |
|
|
|
+| --- | --- | --- |
|
|
|
+| `conversations` | 保存客户、店铺和控制状态；`active_run_id` 指向运行；`sequence` 分配下一个事件序号 | 页面、worker、SSE |
|
|
|
+| `conversation_events` | `message.user` 或 `message.staff`；`payload={text, run_id}`；请求键、摘要、控制版本、创建时间 | 历史接口、SSE、下一次对话上下文 |
|
|
|
+| `agent_runs` | 客户处于 AGENT 模式时创建：`QUEUED/model`、上下文、调用身份、模型参数、发布快照、期限 | Agent worker |
|
|
|
+| `memory_jobs` | 已授权记忆的客户消息创建任务，关联会话与事件序号 | memory worker |
|
|
|
+| `notifications` | 已分配客服时，由 `notify()` 为仍有权限的员工保存客户消息通知 | 商家通知接口 |
|
|
|
+
|
|
|
+`conversation_events` 保存页面可读取的会话事件；`agent_runs.messages` 保存 LangChain 格式的模型输入、助手工具调用和工具结果。
|
|
|
+
|
|
|
+- `authenticate()` 校验 AccessSession、租户与账户状态；写入请求锁定操作者记录。`access()` 锁定会话，校验客户本人或客服的店铺权限。
|
|
|
+- `prior()` 位于控制版本检查之前。同键同内容返回原事件；同键不同内容返回 `409 IDEMPOTENCY_CONFLICT`。
|
|
|
+- 会话行锁保护 `active_run_id` 与 `sequence`。同一会话已有活动运行时返回 `409 RUN_IN_PROGRESS`。
|
|
|
+- 最近最多 12 条历史消息恢复正序后追加当前消息，正文合计最多 24000 字符。客户历史使用 `HumanMessage`，助手及客服历史使用 `AIMessage`。
|
|
|
+- `queue_run()` 在接受新运行时读取店铺当前发布，已有会话的新运行也使用此时的配置。模型名、地址、凭据环境变量名、超时、输出限制、提示词及 `graph_version` 保存在运行快照中。进行中的运行、补充输入、审批等待及进程恢复继续使用原运行快照。当前代码登记的最新图版本为 `after-sales-v10`。
|
|
|
+- 新运行期限为 10 分钟。账户中未过期的 `QUEUED`、`RUNNING`、`WAITING_USER`、`WAITING_APPROVAL` 合计达到 3 个时，返回 `429 AGENT_QUEUE_FULL`。
|
|
|
+
|
|
|
+代码：[message 与 accept_message](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/api.py:185)、[queue_run](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/service.py:20)、[emit](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/service.py:31)、[会话数据模型](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/models.py:10)、[运行数据模型](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/models.py:10)、[数据库连接](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/core/database.py:20)。
|
|
|
+
|
|
|
+## 4. worker 如何启动、领取和循环处理任务
|
|
|
+
|
|
|
+容器 `agent` 服务使用 `python -m zhifu.operations.worker agent`。开发环境也支持 `python -m zhifu.modules.agent_runtime.worker`；该入口提供 `--once`，执行一次领取和处理后退出。
|
|
|
+
|
|
|
+```mermaid
|
|
|
+flowchart TD
|
|
|
+ A[operations.worker.run agent] --> B[登记停止信号<br/>进入 run_monitored]
|
|
|
+ B --> C[heartbeat agent<br/>每5秒写 Redis，记录20秒有效]
|
|
|
+ C --> D[启动 agent_runtime.worker.main 任务<br/>读取 Settings 与 AgentSettings]
|
|
|
+ D --> E[创建数据库 engine 和 sessions<br/>连接 Redis state<br/>建立知识检索 HTTP 与 Milvus 客户端]
|
|
|
+ E --> F[execute_one]
|
|
|
+ F --> G[reconcile 核对等待中的业务运行<br/>每轮最多100个]
|
|
|
+ G --> H[生成 capacity_id<br/>10秒期限内执行 claim]
|
|
|
+ H --> I[开始领取事务<br/>锁定 CapacityPolicy 并读取租户预留额度]
|
|
|
+ I --> J[筛选存在任务且未满并发的租户<br/>优先未用完预留额度的租户<br/>随后按 selected_at 与 Tenant.id 排序]
|
|
|
+ J --> K{取得租户且重新检查额度通过？}
|
|
|
+ K -->|没有通过| N[本次没有领取任务]
|
|
|
+ K -->|通过| L[更新 tenant_dispatch.selected_at<br/>按 created_at、id 选择任务]
|
|
|
+ L --> M[锁定 Conversation<br/>重新锁定并检查 AgentRun]
|
|
|
+ M --> O{任务已超过 deadline_at？}
|
|
|
+ O -->|已经超过| P[FAILED，RUN_DEADLINE<br/>finish_run 保存失败事件并释放会话]
|
|
|
+ P --> N
|
|
|
+ O -->|仍然有效| Q[Redis acquire 原子检查<br/>全局、交互、后台及租户预留额度]
|
|
|
+ Q --> R{取得并发额度？}
|
|
|
+ R -->|没有取得| N
|
|
|
+ R -->|取得| S[旧 STARTED 尝试改为 UNKNOWN<br/>RUNNING，lease_token + 1<br/>lease_until=当前时间+125秒]
|
|
|
+ S --> T[提交领取事务<br/>返回 run_id 与 lease_token]
|
|
|
+ T --> U[核对旧模型尝试的预算<br/>build_graph，执行期限120秒]
|
|
|
+ U --> V[finally 释放 Redis 并发额度]
|
|
|
+ N --> V
|
|
|
+ V --> W{本次是否处理运行？}
|
|
|
+ W -->|已经处理| F
|
|
|
+ W -->|没有处理| X[等待0.5秒]
|
|
|
+ X --> F
|
|
|
+ B -. 收到停止信号 .-> Y[取消业务任务并等待清理<br/>关闭客户端，删除进程心跳]
|
|
|
+```
|
|
|
+
|
|
|
+`claim()` 接受三类候选：
|
|
|
+
|
|
|
+1. `QUEUED`：尚待执行或重新排队的运行。
|
|
|
+2. `RUNNING` 且 `lease_until < now()`：执行租约已经过期的运行。
|
|
|
+3. `WAITING_USER`、`WAITING_APPROVAL` 且 `deadline_at <= now()`：需要结束的过期等待运行。
|
|
|
+
|
|
|
+领取使用 MySQL 行锁协调多个 worker，在取得租户锁后重新统计仍有有效租约的运行数量。默认租户并发为 2，有 `TenantQuota.interactive_limit` 时使用该租户配置。全局模型并发默认 4，交互预留默认 2，后台预留默认 1；具体部署值由配置确定。
|
|
|
+
|
|
|
+Redis `acquire()` 使用 Lua 一次检查并更新 `agent:active:global`、`:interactive`、`:background` 与 `:owners`。额度记录有效期为 150 秒；`execute_one()` 的 `finally` 主动释放本次 `capacity_id`。Redis 更新和 MySQL 提交各自执行，释放逻辑和有效期限负责退出后的额度清理。
|
|
|
+
|
|
|
+每个 `main()` 循环等待当前 `execute_one()` 完成后才领取下一项。因此单个进程一次执行一个运行，多个进程通过共同的数据库记录和 Redis 额度协调并发。
|
|
|
+
|
|
|
+`heartbeat()` 的 Redis 记录表示进程存活。`AgentRun.lease_until` 是单次领取的数据库执行期限，当前实现领取时设置一次。每次领取增加 `lease_token`，节点提交必须携带相同 token；旧进程的迟到结果无法通过租约检查。
|
|
|
+
|
|
|
+图中的循环表示正常处理路径。一般执行异常会在记录运行失败后继续抛出，结束当前进程；预算准入错误记录失败后允许循环继续。容器配置使用 `restart: unless-stopped`。
|
|
|
+
|
|
|
+代码：[容器配置](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/compose.runtime.yaml:50)、[统一进程入口](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/operations/worker.py:49)、[进程心跳](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/core/telemetry.py:31)、[claim](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/worker.py:28)、[execute_one](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/worker.py:137)、[main](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/worker.py:210)、[并发额度](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/capacity.py:75)。
|
|
|
+
|
|
|
+## 5. worker 如何执行模型、工具和最终提交
|
|
|
+
|
|
|
+```mermaid
|
|
|
+flowchart TD
|
|
|
+ A[graph.ainvoke<br/>run_id、token、recursion_limit=24] --> B[load<br/>locked_run 后读取数据库 next_node]
|
|
|
+ B --> C{next_node}
|
|
|
+ C -->|model| D[事务一：locked_run<br/>核验引用与记忆快照<br/>检查模型次数、step 与上下文长度]
|
|
|
+ D --> E[保存 STARTED ModelAttempt<br/>model_calls + 1<br/>emit agent.model_started，提交事务]
|
|
|
+ E --> F[事务外读取已授权记忆<br/>构造 ChatOpenAI 与工具定义]
|
|
|
+ F --> G[BudgetCallback.on_chat_model_start<br/>预占 token；配置 MoneyPolicy 时预占金额]
|
|
|
+ G --> H[模型网络调用<br/>max_retries=0，返回完整 AIMessage]
|
|
|
+ H --> I[预算回调处理 usage<br/>独立事务保存 ModelAttempt RECEIVED 与 usage]
|
|
|
+ I --> J[事务二：重新 locked_run<br/>检查引用、记忆版本和工具参数]
|
|
|
+ J --> K[追加 AIMessage，step + 1]
|
|
|
+ K --> L{存在 tool_calls？}
|
|
|
+ L -->|没有| M[COMPLETED，next_node=done<br/>version + 1，清除 lease_until]
|
|
|
+ M --> N[finish_run<br/>emit message.assistant<br/>清除 conversation.active_run_id]
|
|
|
+ N --> O[提交事务，图执行结束]
|
|
|
+ L -->|存在| P[保存 next_node=tools 并提交]
|
|
|
+ P --> Q[工具节点：检查工具名称和参数<br/>每轮最多4个工具]
|
|
|
+ C -->|tools| Q
|
|
|
+ Q --> R{工具种类}
|
|
|
+ R -->|search_knowledge| S[事务外查询知识服务<br/>事务内再次校验结果权限和版本]
|
|
|
+ R -->|普通业务工具| T[invoke_tool<br/>执行业务检查与读写]
|
|
|
+ R -->|request_clarification| U[pause_for_input 保存等待事件<br/>随后保存 ToolMessage 和 step]
|
|
|
+ R -->|create_exchange_proposal| V[创建方案；保存 ToolMessage<br/>step 加一]
|
|
|
+ R -->|create_followup_request| W[先提交 next_node=followup]
|
|
|
+ C -->|followup| W2[followup 节点执行业务登记]
|
|
|
+ W --> W2
|
|
|
+ W2 --> X[同一事务保存业务收据、通知、ToolMessage<br/>next_node=model，emit agent.tools_completed]
|
|
|
+ S --> Y[保存 ToolMessage，step + 1<br/>next_node=model，emit agent.tools_completed]
|
|
|
+ T --> Y
|
|
|
+ X --> D
|
|
|
+ Y --> D
|
|
|
+ U --> Z[追加助手问题；emit agent.tools_completed<br/>WAITING_USER 或 WAITING_APPROVAL，next_node=wait]
|
|
|
+ V --> BusinessWait{发布版本启用 business_wait？}
|
|
|
+ BusinessWait -->|已经启用| PauseBusiness[pause_for_business 保存等待事件]
|
|
|
+ PauseBusiness --> Z
|
|
|
+ BusinessWait -->|没有启用| ContinueModel[next_node=model<br/>emit agent.tools_completed]
|
|
|
+ ContinueModel --> D
|
|
|
+ Z --> Z2[提交事务，图执行结束<br/>保留活动运行等待后续输入或业务事件]
|
|
|
+ C -->|done 或 wait| Z3[图执行结束]
|
|
|
+```
|
|
|
+
|
|
|
+`build_graph()` 先进入 `load`，按照数据库中的 `run.next_node` 继续。工具执行结果、`step`、`next_node` 和模型上下文保存在 MySQL，重新领取后继续读取这些记录。
|
|
|
+
|
|
|
+`locked_run()` 在关键节点读取与结果提交时检查：运行身份有效、租户和账户相符、店铺有效、权限有效、会话由 AGENT 控制、控制版本一致、活动运行相同、状态为 `RUNNING`、token 相同、租约和任务期限均未到期。
|
|
|
+
|
|
|
+模型调用的提交位置：
|
|
|
+
|
|
|
+1. 调用前事务写入 `ModelAttempt(status=STARTED)` 和 `agent.model_started`，增加持久化的 `model_calls`。
|
|
|
+2. 模型请求在业务事务外执行；`BudgetCallback` 在发送前预占 token，租户配置 `MoneyPolicy` 时同时预占金额，返回后处理真实 `usage`。
|
|
|
+3. 独立事务保存 `ModelAttempt(status=RECEIVED, usage=...)`，保留已经发生的用量。
|
|
|
+4. 重新获得运行锁并通过校验后，保存模型消息和后续节点。最终答复、`COMPLETED` 和清除 `active_run_id` 在同一事务提交。
|
|
|
+
|
|
|
+普通工具由 `invoke_tool()` 调用业务服务。工具命令标识为 `agent:{UUID(bytes=run.id).hex}:{run.step}:{index}`，相同持久化步骤重新执行时保持一致，供业务服务核验幂等收据。`request_clarification` 必须单独调用；`create_followup_request` 也必须单独调用，并使用 `followup` 节点。方案等待取决于 `GRAPH_RELEASES[run.graph_version]["business_wait"]`；关闭此能力的历史发布版本按普通工具路径继续模型节点。
|
|
|
+
|
|
|
+| 限制 | 当前代码中的值 |
|
|
|
+| --- | --- |
|
|
|
+| 单次领取的图执行期限 | 120 秒 |
|
|
|
+| 数据库执行租约 | 125 秒 |
|
|
|
+| Redis 并发额度有效期 | 150 秒 |
|
|
|
+| 单次模型超时 | 发布快照的 `model_timeout`；配置默认 30 秒，上限 30 秒，并受剩余任务期限约束 |
|
|
|
+| 模型 SDK 自动重试 | `max_retries=0` |
|
|
|
+| 每个运行模型调用上限 | 6 次，通过 `model_calls < 6` 检查 |
|
|
|
+| 节点步骤限制 | 通过 `step < 20` 检查 |
|
|
|
+| 单轮工具数量 | 最多 4 个 |
|
|
|
+| 运行上下文长度 | `json.dumps(run.messages, ensure_ascii=False)` 最多 32000 字符 |
|
|
|
+| 默认最大输出 | 1024 token；配置允许 128～4096，运行使用发布快照 |
|
|
|
+
|
|
|
+代码：[locked_run](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/runtime.py:47)、[工具节点](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/runtime.py:95)、[build_graph](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/runtime.py:244)、[模型节点](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/runtime.py:284)、[预算回调](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/budget_callback.py:23)、[finish_run](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/service.py:96)。
|
|
|
+
|
|
|
+## 6. SSE 如何读取数据库并让前端显示消息
|
|
|
+
|
|
|
+```mermaid
|
|
|
+flowchart TD
|
|
|
+ A[connectMessages<br/>GET events，携带 Authorization 和 Last-Event-ID] --> B[API 认证、会话权限和游标检查]
|
|
|
+ B --> C[Redis Lua 申请全局与账户 SSE 额度]
|
|
|
+ C --> D[返回 text/event-stream<br/>浏览器验证响应后调用 onOpen]
|
|
|
+ D --> E[generate 立即发送 connected 注释]
|
|
|
+ E --> F[read_events：独立短事务<br/>重新验证身份、会话权限和游标]
|
|
|
+ F --> G[查询 sequence 大于 cursor 的事件<br/>升序返回，每批最多50条]
|
|
|
+ G --> H[逐项推进服务端 cursor<br/>过滤当前控制版本不适用的 Agent 事件]
|
|
|
+ H --> I[发送 id、event 和 data]
|
|
|
+ I --> J[fetch 正文流 → TextDecoderStream<br/>EventSourceParserStream → JSON.parse]
|
|
|
+ J --> K[核验 sequence、id、type、control_version]
|
|
|
+ K --> L[receiveEvent：sequence 去重<br/>更新 state.after]
|
|
|
+ L --> M[过滤旧控制版本<br/>更新 owner、active_run_id 和等待状态]
|
|
|
+ M --> N[shared.messages.push]
|
|
|
+ N --> O[messageItems 生成正文或服务提示]
|
|
|
+ O --> P[Vue 以 sequence 为 key 更新页面<br/>ExpandableText 显示正文]
|
|
|
+ P --> Q[查看最新消息时自动滚动<br/>阅读历史时增加未读数量]
|
|
|
+ I --> R{连接达到约50秒？}
|
|
|
+ R -->|尚未达到| S[等待0.5秒]
|
|
|
+ S --> F
|
|
|
+ R -->|已经达到| T[结束响应，释放 SSE 额度]
|
|
|
+ T --> U[浏览器正常读完后<br/>携带最新 state.after 再次订阅]
|
|
|
+ U --> A
|
|
|
+```
|
|
|
+
|
|
|
+SSE 从 `conversation_events` 查询消息。worker 保存结果后，正在运行的 `generate()` 在后续查询中读取已提交事件，再通过 HTTP 响应正文发送给浏览器。Redis 在此流程中管理连接额度；会话事件的内容和序号保存在 MySQL。
|
|
|
+
|
|
|
+请求路径为 `GET /openapi/v1/conversations/{conversation_id}/events`。初次订阅默认 `Last-Event-ID=0`；当前页面已经读取历史时使用其 `state.after`。每轮查询在独立事务中验证 token、账户、租户与会话权限，查询完成后释放数据库连接。API 使用 `asyncio.Semaphore(settings.pool_size)` 限制同时进行的 SSE 数据库读取。
|
|
|
+
|
|
|
+持久事件结构如下，标识和正文仅说明字段形式。
|
|
|
+
|
|
|
+```text
|
|
|
+id: 7
|
|
|
+event: message.assistant
|
|
|
+data: {"sequence":7,"type":"message.assistant","control_version":1,"created_at":"2026-09-30T00:00:00+00:00","payload":{"text":"已查询到订单状态。","run_id":"{run_id}"}}
|
|
|
+
|
|
|
+```
|
|
|
+
|
|
|
+`id` 等于 `sequence`，`event` 等于 `type`，`data` 为完整 `EventView`。服务端立即发送 `connected` 注释，每 15 秒发送心跳，约 50 秒结束连接，发送超时为 5 秒。前端单次连接总期限为 65 秒，使用 `EventSourceParserStream` 处理 SSE 分段与心跳，解析缓冲上限为 1 MiB。
|
|
|
+
|
|
|
+默认全局最多 200 条 SSE 连接，每账户最多 3 条；部署可修改。Redis 连接额度有效期为 60 秒，连接结束通过 `release()` 清除。服务端每轮还检查续传游标：小于 `stream_cursor` 返回 `410 EVENT_CURSOR_EXPIRED`，大于会话最新 `sequence` 返回 `422 INVALID_EVENT_CURSOR`。
|
|
|
+
|
|
|
+前端 `receiveEvent()` 的执行顺序为：检查 `sequence <= state.after` 并跳过重复事件；新事件立即更新 `state.after`；过滤旧控制版本；修改会话与运行状态；最后向 `shared.messages` 添加事件。`messageItems` 重新计算后，Vue 更新页面。
|
|
|
+
|
|
|
+| 持久事件 | 前端处理与显示 |
|
|
|
+| --- | --- |
|
|
|
+| `message.user` | 从 `payload.text` 或 `payload.answer` 生成客户气泡，显示“已发送”；携带 `run_id` 时登记活动运行。 |
|
|
|
+| `agent.model_started` | 保存进度事件；页面根据活动运行显示查询提示。该事件没有专门把本地 `activeRun.status` 改为 `RUNNING` 的分支。 |
|
|
|
+| `agent.tools_completed` | 保存工具进度事件。普通消息区域根据消息类型生成正文。 |
|
|
|
+| `message.assistant` | 显示完整 AI 答复；携带 `waiting` 时保存等待信息，普通完成答复清除适用的活动运行。 |
|
|
|
+| `run.waiting` | 保存 `interrupt_id`、运行版本、问题及等待类型；显示补充信息或业务操作入口。 |
|
|
|
+| `run.resumed` | 更新活动运行，清除等待信息；存在 `proposal_id` 时重新读取业务卡片。 |
|
|
|
+| `message.staff` | 显示人工客服正文。 |
|
|
|
+| `control.changed` | 更新控制权、控制版本、当前运行及人工协助状态，显示服务方变化提示。 |
|
|
|
+| `handoff.requested` | 保存已申请人工协助状态，并显示提示。 |
|
|
|
+| `run.failed` / `run.cancelled` | 清除适用的活动运行，显示失败或停止提示。 |
|
|
|
+
|
|
|
+`messageItems` 将客户、助手和客服正文交给 `ExpandableText`；时间使用服务端事件的原始 `created_at`。浏览器距离消息列表底部小于 48 px 时，新消息自动滚动到最新位置；阅读历史时保留当前位置并增加未读数量。用户发送消息或点击新消息按钮时显示最新消息。
|
|
|
+
|
|
|
+| 连接情况 | 处理规则 |
|
|
|
+| --- | --- |
|
|
|
+| 服务端正常结束响应 | `connectMessages()` 自动继续循环，以最后事件序号订阅。 |
|
|
|
+| `410 EVENT_CURSOR_EXPIRED` 或流内 `stream.expired` | 本次连接循环自动恢复一次：清空本地历史和游标，重新读取消息历史、当前会话和活动运行，再订阅。历史接口仍可读取持久事件。 |
|
|
|
+| `401 / 403 / 404` | 停止连接并清理适用的当前会话内容；认证失效时清除对应登录会话。 |
|
|
|
+| `429 SSE_LIMIT` | 根据 `Retry-After` 显示等待秒数，之后允许用户重新连接。 |
|
|
|
+| 网络异常或连接超时 | 显示错误，用户点击“重新连接”触发页面重新加载与订阅。 |
|
|
|
+| 页面切换或组件卸载 | 中止旧连接；`generation / valid()` 限制旧页面的请求修改当前页面。 |
|
|
|
+
|
|
|
+服务端过滤 `agent.*` 和 `message.assistant` 的过期控制版本；前端实时处理另外过滤 `run.*` 的旧控制版本。历史读取保留原始事件及原始控制版本。
|
|
|
+
|
|
|
+代码：[SSE 查询](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/stream.py:50)、[SSE 生成器](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/stream.py:131)、[前端 SSE 读取](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/api.js:11)、[receiveEvent 与重连](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/pages/customer.js:177)、[气泡与滚动](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/pages/customer.js:343)、[页面模板](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/pages/customer.js:466)。
|
|
|
+
|
|
|
+## 7. 等待输入、业务确认、失败和任务恢复
|
|
|
+
|
|
|
+```mermaid
|
|
|
+stateDiagram-v2
|
|
|
+ [*] --> QUEUED: 消息与运行提交成功
|
|
|
+ QUEUED --> RUNNING: worker 领取，取得新 lease_token
|
|
|
+ RUNNING --> COMPLETED: 提交完整助手答复
|
|
|
+ RUNNING --> WAITING_USER: 保存补充问题或待确认方案
|
|
|
+ WAITING_USER --> QUEUED: 接受 clarification 输入
|
|
|
+ WAITING_USER --> WAITING_APPROVAL: 核对客户已确认方案
|
|
|
+ WAITING_USER --> QUEUED: 已提交业务结果允许继续生成
|
|
|
+ WAITING_APPROVAL --> WAITING_USER: 方案需要修订
|
|
|
+ WAITING_APPROVAL --> QUEUED: 核对有效业务结果
|
|
|
+ RUNNING --> RUNNING: 租约到期后重新领取
|
|
|
+ QUEUED --> FAILED: 超过执行期限
|
|
|
+ RUNNING --> FAILED: 执行、预算或权限检查失败
|
|
|
+ WAITING_USER --> FAILED: 超过等待期限或权限失效
|
|
|
+ WAITING_APPROVAL --> FAILED: 超过等待期限或权限失效
|
|
|
+ QUEUED --> CANCELLED: 用户停止或客服接管
|
|
|
+ RUNNING --> CANCELLED: 用户停止或客服接管
|
|
|
+ WAITING_USER --> CANCELLED: 用户停止或客服接管
|
|
|
+ WAITING_APPROVAL --> CANCELLED: 用户停止或客服接管
|
|
|
+ COMPLETED --> [*]
|
|
|
+ FAILED --> [*]
|
|
|
+ CANCELLED --> [*]
|
|
|
+```
|
|
|
+
|
|
|
+状态转换发生在对应处理函数的身份、状态、版本和幂等检查通过以后。
|
|
|
+
|
|
|
+**补充信息。** `pause_for_input()` 将运行改为 `WAITING_USER / wait`，增加 `version`，清除 `lease_until`，等待期限设置为 24 小时。`waiting` 保存 `interrupt_id`、`run_version`、问题和候选信息；同事务保存带 `waiting` 的 `message.assistant` 和 `run.waiting`。本次 graph 结束后释放并发额度，会话保留原 `active_run_id`。
|
|
|
+
|
|
|
+前端发现等待类型为 `clarification` 时，发送 `POST /openapi/v1/agent/runs/{run_id}/inputs`：
|
|
|
+
|
|
|
+```json
|
|
|
+{
|
|
|
+ "interrupt_id": "{current_interrupt_id}",
|
|
|
+ "expected_run_version": 2,
|
|
|
+ "answer": "客户补充的信息"
|
|
|
+}
|
|
|
+```
|
|
|
+
|
|
|
+`accept_input()` 检查幂等收据、等待状态、运行版本、等待标识、期限和会话控制权。成功时追加 `HumanMessage`，保存 `RunInputEvent`，清空等待信息，将原运行恢复为 `QUEUED / model`，更新本次访问 token，重新设置 10 分钟执行期限；同事务保存 `message.user` 和 `run.resumed`。worker 随后领取原 `run_id`。
|
|
|
+
|
|
|
+**方案确认与主管审批。** `pause_for_business()` 保存 `business_ticket_id`、业务版本及等待信息。草稿和需要修订的方案进入 `WAITING_USER`，已提交审批的方案进入 `WAITING_APPROVAL`。客户确认和主管决定通过业务接口保存。每轮 worker 开始时，`reconcile()` 最多核对 100 个未过期的业务等待运行，验证持久业务事件、客户确认、主管权限及方案版本，再保存输入收据和更新后的等待状态，或者重新排队并保存 `run.resumed`。
|
|
|
+
|
|
|
+**失败记录。** graph 执行异常后，`execute_one()` 在独立事务中检查运行仍为 `RUNNING` 且 `lease_token` 相同，然后保存 `FAILED`、`error_code`，增加版本，清除等待与租约，并保存 `run.failed`、清除会话活动运行。仍为 `STARTED` 的尝试按情况改为 `NOT_SENT` 或 `UNKNOWN`。预算准入拒绝允许 worker 继续循环；其他执行异常继续传播。上述处理完成后，`finally` 释放 Redis 额度。
|
|
|
+
|
|
|
+**进程恢复。** 进程异常终止后仍为 `RUNNING` 的记录，在执行租约到期后满足重新领取条件。新领取增加 `lease_token`，旧 `STARTED` 尝试改为 `UNKNOWN` 并核对预算，graph 从 MySQL 的 `next_node` 和 `messages` 继续。模型结果尚未保存时可能产生新的模型尝试，各次尝试保留独立计量记录。已经保存为 `FAILED` 的任务不在 `claim()` 的候选条件中。
|
|
|
+
|
|
|
+**停止与清理。** 用户停止调用 `/agent/runs/{id}/cancel`，保存 `CANCELLED` 和 `run.cancelled`。统一进程入口管理停止信号与心跳；`main()` 的 `finally` 关闭 Redis 和数据库 engine，`retrieval_client()` 上下文关闭知识服务客户端。容器按配置处理进程重新启动。
|
|
|
+
|
|
|
+代码：[补充信息](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/inputs.py:63)、[输入 API](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/api.py:182)、[业务等待](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/business_wait.py:30)、[业务核对](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/business_wait.py:157)、[失败处理](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/worker.py:173)、[用户停止](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/api.py:155)。
|
|
|
+
|
|
|
+## 8. 人工客服接管和回复如何返回消费者页面
|
|
|
+
|
|
|
+```mermaid
|
|
|
+sequenceDiagram
|
|
|
+ autonumber
|
|
|
+ participant Staff as 商家后台
|
|
|
+ participant API as FastAPI
|
|
|
+ participant DB as MySQL
|
|
|
+ participant H5 as 消费者 H5
|
|
|
+ Staff->>API: POST conversations/id/takeover，控制版本和幂等键
|
|
|
+ API->>DB: 事务：权限和控制版本检查，锁定会话
|
|
|
+ API->>DB: 取消活动运行，清除 active_run_id
|
|
|
+ API->>DB: HUMAN，指定客服，control_version 加一
|
|
|
+ API->>DB: 保存 control.changed，提交
|
|
|
+ API-->>Staff: 返回控制事件
|
|
|
+ Staff->>API: GET conversation 和 messages
|
|
|
+ API-->>Staff: 当前控制状态与历史
|
|
|
+ API->>DB: 消费者 SSE 读取新事件
|
|
|
+ API-->>H5: control.changed
|
|
|
+ H5->>API: POST messages，客户在 HUMAN 模式发送
|
|
|
+ API->>DB: 保存 message.user，run_id 为 null，提交
|
|
|
+ API-->>H5: HTTP 201
|
|
|
+ API-->>H5: SSE message.user
|
|
|
+ Staff->>API: 打开会话或点击刷新消息
|
|
|
+ API->>DB: GET 会话状态和消息历史
|
|
|
+ API-->>Staff: 客户消息
|
|
|
+ Staff->>API: POST messages，客服回复
|
|
|
+ API->>DB: 检查当前接管员工，保存 message.staff，提交
|
|
|
+ API-->>Staff: HTTP 201
|
|
|
+ Staff->>API: 重新读取会话与消息
|
|
|
+ API-->>Staff: 显示已发送回复
|
|
|
+ API->>DB: 消费者 SSE 读取 message.staff
|
|
|
+ API-->>H5: SSE message.staff
|
|
|
+ H5->>H5: messageItems 生成客服气泡
|
|
|
+```
|
|
|
+
|
|
|
+商家后台 `control(action)` 显示确认窗口，再发送 `takeover` 或 `release`。后端要求员工具备 `conversations:manage`；交还控制权要求当前员工是已接管客服。成功后增加 `control_version` 并保存 `control.changed`。交还后客户再次发言，按 `AGENT` 模式创建新运行。
|
|
|
+
|
|
|
+客服回复要求 `HUMAN`、`assigned_staff_id` 等于当前员工，且具有 `conversations:reply`。回复使用同一个消息接口、幂等键和控制版本检查，保存 `message.staff`。模型结果提交时的 `locked_run()` 继续检查会话控制权和版本。
|
|
|
+
|
|
|
+商家会话页通过 `conversation()` 查询详情和前 100 个历史事件，“读取后续消息”调用 `moreMessages()`，使用 `after=messageSequence` 继续。打开会话、点击刷新，以及本人回复或控制操作成功后重新读取消息。商家页面每 30 秒更新的是员工通知和未读数量；聊天内容按上述页面操作读取。
|
|
|
+
|
|
|
+商家 `presentEvent()` 将客户、客服和助手正文转为消息条目，将模型开始、工具完成等信息放入“服务处理记录”。消费者 H5 持续使用第 6 节的 SSE 流程接收人工回复。
|
|
|
+
|
|
|
+代码：[后端控制权操作](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/api.py:264)、[商家读取、控制和回复接口](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/services/merchant.js:43)、[商家页面操作](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/pages/merchant.js:125)、[商家通知定时器](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/pages/merchant.js:117)。
|
|
|
+
|
|
|
+## 9. 接口、版本字段和事务查询位置
|
|
|
+
|
|
|
+普通消息请求如下，标识为占位符。
|
|
|
+
|
|
|
+```http
|
|
|
+POST /openapi/v1/conversations/{conversation_id}/messages
|
|
|
+Authorization: Bearer {access_token}
|
|
|
+Content-Type: application/json
|
|
|
+Idempotency-Key: {same_operation_uuid}
|
|
|
+
|
|
|
+{
|
|
|
+ "text": "请帮我查询订单状态",
|
|
|
+ "expected_control_version": 1
|
|
|
+}
|
|
|
+```
|
|
|
+
|
|
|
+成功响应为 `201 EventView`，主要字段为 `sequence`、`type`、`control_version`、`created_at` 和 `payload`。`payload.run_id` 关联异步运行；`payload.text` 是已保存正文。字段校验会去除请求字符串首尾空白，并拒绝未定义字段。
|
|
|
+
|
|
|
+| 接口或处理阶段 | 结果与处理规则 |
|
|
|
+| --- | --- |
|
|
|
+| `POST /conversations` | 客户在具备 `orders:read` 的店铺创建会话；初始控制方为 `AGENT`；幂等作用域为租户与客户。 |
|
|
|
+| `POST /conversations/{id}/messages` | 成功保存消息返回 `201`；`CONTROL_CONFLICT`、`RUN_IN_PROGRESS`、`IDEMPOTENCY_CONFLICT` 返回 `409`。 |
|
|
|
+| 消息受理的任务上限 | 账户活动任务达到上限时返回 `429 AGENT_QUEUE_FULL`；默认重试等待信息由服务端返回。 |
|
|
|
+| 消息受理的模型配置检查 | 模型名称或地址不可用时返回 `503 MODEL_NOT_CONFIGURED`；发布版本检查也可能拒绝受理。 |
|
|
|
+| `GET /conversations/{id}/messages` | `after >= 0`，`limit=1–100`，响应 `{items, sequence}`；分页使用实际最后一项的序号。 |
|
|
|
+| `GET /conversations/{id}/events` | SSE 通过 Bearer 认证；窗口过期返回 `410`，游标超过范围返回 `422`，连接限额返回 `429`。 |
|
|
|
+| `GET /agent/runs/{id}` | 读取当前运行状态、版本和等待信息，并执行运行访问权限检查。 |
|
|
|
+| `POST /agent/runs/{id}/inputs` | 接收等待中的补充信息，成功返回 `202` 和原运行更新状态；检查等待标识、运行版本和幂等键。 |
|
|
|
+| worker 模型预算检查 | token 或适用金额预算不足、缺少价格配置时保存 `run.failed`；消息受理的 `201` 已表示客户正文保存成功。 |
|
|
|
+
|
|
|
+表中接口路径统一带 `/openapi/v1` 前缀；身份、权限或参数不符合时返回相应 `401 / 403 / 404 / 422`。数据库幂等记录和版本检查由对应接口在事务中执行。
|
|
|
+
|
|
|
+| 标识或字段 | 控制的内容 |
|
|
|
+| --- | --- |
|
|
|
+| `conversation_id` | 当前客户会话及其全部持久事件。 |
|
|
|
+| `run_id` | 某次异步运行，包括后续补充信息和业务等待。 |
|
|
|
+| `sequence` | 会话事件顺序，供历史分页、SSE 续传和前端去重使用。 |
|
|
|
+| `control_version` | 会话控制权版本，接管和交还时增加。 |
|
|
|
+| `AgentRun.version` / `run_version` | 运行状态版本，补充信息与业务恢复时检查。 |
|
|
|
+| `lease_token` | 某次 worker 领取的执行权，重新领取时增加。 |
|
|
|
+| `interrupt_id` | 当前等待步骤标识，限定输入所恢复的等待。 |
|
|
|
+| `Idempotency-Key` | 同一操作重试标识，作用域和内容检查由具体接口确定。 |
|
|
|
+
|
|
|
+| 事务边界 | 一起提交的主要数据 |
|
|
|
+| --- | --- |
|
|
|
+| 接受客户消息 | 新运行、会话活动运行指针、事件序号、`message.user`，以及适用的记忆任务和通知。 |
|
|
|
+| worker 领取 | 租户调度时间、旧模型尝试状态、`RUNNING`、新执行 token 及租约。Redis 额度通过独立 Lua 调用取得。 |
|
|
|
+| 模型开始 | `ModelAttempt STARTED`、模型调用计数、`agent.model_started`。 |
|
|
|
+| 预算与用量 | 预算预留、结算及尝试用量使用各自短事务，保留已经发生的消耗。 |
|
|
|
+| 工具完成 | 工具业务变更、`ToolMessage`、步骤、后续节点和适用进度或等待事件。 |
|
|
|
+| 最终答复 | 完整 `AIMessage`、`COMPLETED`、运行版本、清除租约、`message.assistant`、清除 `active_run_id`。 |
|
|
|
+| 补充信息 | 输入收据、上下文、原运行重新排队、客户消息、`run.resumed`。 |
|
|
|
+
|
|
|
+查询单次对话时，可从 `conversation_id` 找到 `conversation_events.payload.run_id`，据此读取 `agent_runs` 和 `model_attempts`，再用事件 `sequence` 核对浏览器的 `state.after`。
|
|
|
+
|
|
|
+本文依据当前源码说明执行过程。文档检查覆盖 Mermaid 解析、图形渲染和代码引用；业务接口与生成模型调用未执行。
|
