路线 01：前后端咨询与人工协作

学习场景

客户在消费者 H5 发送问题，页面持续接收处理状态与答复；客服在商家后台接管会话并发送人工回复。沿着消息、事件序号和控制版本，追踪浏览器与后端如何共同维护同一场会话。

图中后端路径相对 src/zhifu/；frontend/、scripts/ 等路径相对仓库根目录。连续箭头表示函数直接调用；间断箭头注明 HTTP、框架回调或持久数据传递。框架节点、配置节点和数据节点单独标明。图中只展示本路线需要追踪的分支。返回路线目录。

图 1：客户消息进入会话事务

flowchart TD
    UI["frontend/src/pages/customer.js<br/>Vue 模板：van-form @submit"]
    SEND["frontend/src/pages/customer.js<br/>send() / perform(action)<br/>校验发送条件并执行 action"]
    ENSURE["frontend/src/pages/customer.js<br/>ensureConversation()<br/>读取或创建当前会话"]
    SERVICE["frontend/src/services/customer.js<br/>send() → request()<br/>附带正文与控制版本"]
    HTTP["frontend/src/api.js<br/>request()<br/>添加认证、幂等键并 fetch"]
    API["modules/conversations/api.py<br/>message()<br/>开启消息事务"]
    AUTH["modules/identity/auth.py<br/>authenticate()<br/>验证访问会话"]
    ACCESS["modules/conversations/service.py<br/>access()<br/>检查会话权限并锁定会话"]
    ACCEPT["modules/conversations/api.py<br/>accept_message() / prior()<br/>检查重复请求与控制版本"]
    QUEUE["modules/agent_runtime/service.py<br/>queue_run()<br/>创建待执行运行"]
    EMIT["modules/conversations/service.py<br/>emit()<br/>分配 sequence 并登记事件"]
    DB[("MySQL 持久数据<br/>AgentRun、Conversation<br/>ConversationEvent")]
    WORKER["modules/agent_runtime/worker.py<br/>claim()<br/>独立工作进程领取任务"]
    UI -. "Vue 调用提交处理函数" .-> SEND
    SEND -->|action 回调：查询当前状态| ENSURE
    SEND -->|action 回调：普通消息分支| SERVICE
    SERVICE --> HTTP
    HTTP -. "POST /openapi/v1/conversations/{id}/messages；FastAPI 路由" .-> API
    API --> AUTH
    API --> ACCESS
    API --> ACCEPT
    ACCEPT -->|客户处于 AGENT 模式| QUEUE
    ACCEPT --> EMIT
    QUEUE -. "同一事务登记运行" .-> DB
    EMIT -. "message 事务提交消息及会话变化" .-> DB
    DB -. "读取已提交的待执行运行" .-> WORKER

send() 的普通消息分支向会话接口提交消息。ensureConversation() 根据本地是否已有 conversationId，调用会话详情，或者查询会话列表并按需创建会话。accept_message() 先检查幂等记录，已有记录直接返回原事件；新请求继续检查 expected_control_version。

客户在 AGENT 模式下发送普通消息时，queue_run() 登记运行，当前会话保存 active_run_id。消息、运行和会话变化在 message() 的同一事务内提交。HTTP 201 表示消息已被接受，后续答复通过图 2 的事件连接到达。HUMAN 模式客户消息与客服消息直接登记会话事件。

等待补充信息时，send() 调用 services/customer.js 的 resume()，请求 /openapi/v1/agent/runs/{run_id}/inputs。补充信息、模型调用和工具执行继续阅读路线 02：Agent 运行过程。

图 1 源码导航

send()、perform()、ensureConversation()：确认按钮条件、普通发送与补充信息分支。
createCustomerService()、request()：观察接口路径、Bearer token、幂等键与错误处理。
message()、accept_message()、prior()：定位事务、控制版本与幂等记录。
authenticate()、access()、queue_run()、emit()、claim()：区分消息事务和后续工作进程。

图 2：历史读取、SSE 订阅与断线续传

flowchart TD
    WATCH["frontend/src/pages/customer.js<br/>Vue watch：props.page 变化"]
    LOAD["frontend/src/pages/customer.js<br/>loadPage()<br/>建立页面请求代次"]
    HISTORY["frontend/src/pages/customer.js<br/>refreshMessages()<br/>按 after 读取历史与当前运行"]
    CONNECT["frontend/src/pages/customer.js<br/>connectMessages()<br/>维护连接及续传循环"]
    CLIENT["frontend/src/services/customer.js<br/>events()<br/>选择当前会话事件地址"]
    FETCH["frontend/src/api.js<br/>events()<br/>fetch 与 EventSourceParserStream"]
    API["modules/conversations/stream.py<br/>events()<br/>检查权限、游标与 Redis 额度"]
    RESPONSE["modules/conversations/stream.py<br/>LeasedEventSourceResponse.__call__()<br/>管理响应及额度清理"]
    GENERATE["modules/conversations/stream.py<br/>generate()<br/>循环读取并生成 SSE 事件"]
    READ["modules/conversations/stream.py<br/>read_events()<br/>每轮重新验证身份与访问权限"]
    DB[("MySQL 持久数据<br/>ConversationEvent、Conversation<br/>会话身份及授权记录")]
    RECEIVE["frontend/src/pages/customer.js<br/>receiveEvent()<br/>处理序号、控制版本与页面状态"]
    RELEASE["modules/conversations/stream.py<br/>release()<br/>释放 Redis 连接额度"]
    WATCH -. "Vue 调用监听函数" .-> LOAD
    LOAD --> HISTORY
    LOAD -->|有效聊天页面且读取成功| CONNECT
    CONNECT --> CLIENT
    CLIENT --> FETCH
    FETCH -. "GET events；Last-Event-ID；FastAPI 路由" .-> API
    API -. "返回响应对象；ASGI 调用 __call__" .-> RESPONSE
    RESPONSE -. "EventSourceResponse 消费异步生成器" .-> GENERATE
    GENERATE --> READ
    DB -. "每轮短事务读取已提交事件" .-> READ
    GENERATE -. "HTTP SSE 正文：完整事件" .-> FETCH
    FETCH -->|调用传入的 onEvent 回调| RECEIVE
    CONNECT -->|410：本轮连接循环恢复一次| HISTORY
    GENERATE -->|finally| RELEASE
    RESPONSE -->|响应结束时 finally| RELEASE

loadPage() 关闭上一条连接，以 generation 和 valid() 限定当前页面请求。refreshMessages() 通过历史接口分页读取事件，并查询会话和活动运行；成功后 connectMessages() 建立 SSE。api.events() 使用依赖 EventSourceParserStream 读取完整事件，随后调用页面传入的 onEvent。

服务端每轮 read_events() 使用独立事务验证 token、账户、租户和会话权限，查询完成后释放数据库连接。generate() 推送事件的 id、event 和 JSON data；当前传输单位为完整持久事件。每次连接约 50 秒结束，客户端正常结束后携带 state.after 继续订阅。

receiveEvent() 跳过已处理序号，再更新 state.after；旧控制版本的助手、Agent 进度和运行事件随后被过滤。游标过期时，连接循环自动重新读取历史和当前状态一次。网络异常显示重新连接入口；页面切换或组件卸载会中止旧连接。release() 同时由生成器与响应生命周期保护，重复调用保持幂等。

图 2 源码导航

loadPage() 与 watch 注册、refreshMessages()：跟踪初始化、历史分页和页面代次。
connectMessages()、客户端 events()、receiveEvent()：跟踪续传游标及 UI 状态。
服务端 events()、generate()、read_events()：检查订阅与每轮查询的边界。
LeasedEventSourceResponse.call()、release()、history()：核对提前断开、额度清理和历史记录读取。

图 3：客服接管、人工回复与消费者页面更新

flowchart TD
    CONTROL["frontend/src/pages/merchant.js<br/>control(action)<br/>请求确认接管或交还"]
    CONFIRM["frontend/src/main.js<br/>confirm()<br/>保存确认窗口与操作回调"]
    ACCEPT["frontend/src/App.js<br/>accept()<br/>确认后执行已保存回调"]
    SERVICE["frontend/src/services/merchant.js<br/>control() / command() / request()<br/>附带 expected_control_version"]
    HTTP["frontend/src/api.js<br/>request()<br/>发送商家身份请求"]
    API["modules/conversations/api.py<br/>control()<br/>权限、幂等和控制版本检查"]
    EMIT["modules/conversations/service.py<br/>emit()<br/>登记 control.changed"]
    REPLY["frontend/src/pages/merchant.js<br/>reply()<br/>发送人工回复并刷新会话"]
    REPLY_SERVICE["frontend/src/services/merchant.js<br/>reply() / command() / request()<br/>附带会话控制版本"]
    MESSAGE["modules/conversations/api.py<br/>message() / accept_message()<br/>确认当前接管客服并登记回复"]
    REFRESH["frontend/src/services/merchant.js<br/>conversation()<br/>重新查询详情与消息历史"]
    QUERY["modules/conversations/api.py<br/>get_conversation() / history()<br/>读取可访问的会话与事件"]
    DB[("MySQL 持久数据<br/>Conversation、AgentRun<br/>ConversationEvent")]
    SSE["modules/conversations/stream.py<br/>read_events() / generate()<br/>已有 SSE 连接读取新事件"]
    CUSTOMER["frontend/src/pages/customer.js<br/>receiveEvent()<br/>显示控制变化与人工消息"]
    CONTROL --> CONFIRM
    CONFIRM -. "Vue 展示窗口；用户确认触发 accept" .-> ACCEPT
    ACCEPT -->|执行回调中的 run 与 action| SERVICE
    SERVICE --> HTTP
    HTTP -. "POST conversations/{id}/{action}；FastAPI 路由" .-> API
    API --> EMIT
    EMIT -. "事务提交控制变化与适用的运行取消" .-> DB
    REPLY -->|run 的 action 回调| REPLY_SERVICE
    REPLY_SERVICE --> HTTP
    HTTP -. "POST conversations/{id}/messages；FastAPI 路由" .-> MESSAGE
    MESSAGE -. "同一事务提交 message.staff" .-> DB
    ACCEPT -->|控制请求成功后，回调继续调用| REFRESH
    REPLY -->|回复成功后，action 回调继续调用| REFRESH
    REFRESH -->|经服务 request 包装| HTTP
    HTTP -. "GET 会话详情与 messages；FastAPI 路由" .-> QUERY
    DB -. "事务查询已提交记录" .-> QUERY
    DB -. "读取已提交的控制变化与人工消息" .-> SSE
    SSE -. "HTTP SSE，经图 2 客户端处理" .-> CUSTOMER

control() 要求 conversations:manage 权限；接管成功后将控制方设为 HUMAN，增加 control_version，记录 assigned_staff_id，并取消适用的活动运行。交还要求当前员工就是已接管客服。交还之后，客户下一次普通发言创建新的 Agent 运行。

客服回复要求 conversations:reply 权限、HUMAN 控制方，以及匹配的 assigned_staff_id。商家页面打开会话、执行刷新、回复成功或控制操作成功时读取详情与历史。消费者通过持续 SSE 接收 message.staff；商家通知的 30 秒轮询负责通知和未读数量。

图 3 源码导航

页面 control() 与 reply()、confirm()、accept()、服务 control() 与 reply()：核对确认步骤、请求字段和成功后刷新。
后端 control()、accept_message()：定位接管、交还和人工回复的权限检查。
conversation()、get_conversation()、history()：核对商家页面读取的数据。
finish_run()：理解运行提交时对控制方、控制版本和活动运行的检查。

关键数据与状态

conversation_id 标识会话；active_run_id 指向当前活动运行。客户普通消息被接受后，后端可能创建新的 AgentRun。
sequence 为会话持久事件序号；state.after 为前端已经处理到的序号。历史分页、SSE 续传和消息去重使用这些字段。
control_owner 为 AGENT 或 HUMAN；control_version 在接管、交还时增加。请求中的 expected_control_version 保护当前操作，事件中的版本帮助页面识别旧回复。
Idempotency-Key 与消息摘要共同限定同一请求。会话消息及控制操作的作用域为租户、会话和操作者。
stream_cursor 表示可续传窗口边界；游标过期使用 410 EVENT_CURSOR_EXPIRED。持久历史继续由 messages 接口提供。
WAITING_USER、WAITING_APPROVAL、run.waiting、run.resumed 支持补充信息与业务等待；run.failed、run.cancelled 使页面清理对应活动运行。

阅读检查问题

同一条消息由于网络原因重复提交时，前后端分别在哪里保留或检查幂等标识？为何返回 201 后页面仍可能收到 run.failed？
state.after、sequence、stream_cursor 各自控制什么？游标过期与普通连接结束分别经过哪些函数？
客服接管时，后端怎样取消已有运行，消费者页面怎样处理迟到的助手事件，客服回复又受到哪些身份条件限制？

测试文件阅读入口

本路线只阅读用例及其断言；编写本文未执行这些用例，也未调用生成模型。

test_conversations.py：接管与交还、并发接管、消息幂等、SSE 续传、权限失效与保留窗口。
test_sse_limits.py：Redis 原子连接额度与共享限制。
customer-sse.mjs：浏览器接收人工消息、正常续传、断网恢复、页面离开及游标过期。
merchant-conversation.mjs：商家队列进入、人工接管、回复与客户历史可见性。

01-conversation.md 16 KB Permanentný odkaz História Raw