# 对话消息、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 解析、图形渲染和代码引用；业务接口与生成模型调用未执行。