商家发布键盘清洁说明,客户询问清洁方法,并授权系统以后使用简洁回复。沿着资料发布、回答引用和沟通习惯保存三个过程,理解知识正文、检索索引、客户授权和模型上下文各自的作用。
图中后端路径相对 src/zhifu/。连续箭头表示直接函数调用,箭头文字注明调用条件;间断箭头标明 HTTP、框架调度、异步队列或持久数据关系。框架节点、第三方库节点和存储节点单独标明类型。execute、model 分别是 Agent 图中的工具节点与模型节点内部函数。返回路线目录。
flowchart TD
H["HTTP / 商家知识管理<br/>创建或修订、发布、撤回"]
A["modules/knowledge/api.py<br/>revise<br/>正文分段、版本检查并保存草稿"]
S["第三方库<br/>RecursiveCharacterTextSplitter.split_text<br/>生成带重叠内容的分段"]
B["modules/knowledge/api.py<br/>transition<br/>处理 publish 或 revoke"]
D[("MySQL<br/>modules/knowledge/models.py<br/>KnowledgeDocument:正文、分段、版本、状态")]
I["modules/knowledge/vector.py<br/>RetrievalClient.index_document<br/>分批生成向量并写入索引"]
E["modules/knowledge/vector.py<br/>RetrievalClient.embeddings<br/>请求内部向量服务"]
F["modules/knowledge/inference.py<br/>embed<br/>检查文本数量和长度"]
Q["modules/knowledge/batching.py<br/>Batcher.submit<br/>加入队列并等待 Future"]
W["modules/knowledge/batching.py<br/>Batcher.run<br/>合并文本并调用 model.encode"]
V[("Milvus<br/>document + version + position<br/>dense、sparse 与检索字段")]
H -. "HTTP / FastAPI:PUT 知识资料" .-> A
H -. "HTTP / FastAPI:POST publish 或 revoke" .-> B
A --> S
A -. "事务提交:version 增加,status=DRAFT" .-> D
D -. "读取并锁定当前版本" .-> B
B -->|publish 且当前为 DRAFT| I
I --> E
E -. "HTTP POST /embeddings" .-> F
F --> Q
Q -. "进程内异步队列:等待批处理结果" .-> W
I -. "向量返回后:MilvusClient.upsert" .-> V
B -. "索引完成后重新检查权限和版本:PUBLISHED" .-> D
B -. "revoke:REVOKED,version 增加" .-> D
revise 使用 chunk_size=500、chunk_overlap=50 生成最多 64 个分段,同时保留 source_text。当前版本与 expected_version 不符时拒绝写入;相同标题、正文及上一版本的紧邻重试可以返回已有结果。
transition 在索引写入前后分别检查权限与资料版本。向量生成和 Milvus 写入发生在数据库事务之外,随后重新读取当前资料并设置 PUBLISHED。同版本已经发布时直接返回。撤回通过 MySQL 状态和版本立即影响检索可用性,检索流程在图 2 中再次核验这些数据。
向量服务由 inference.py 的 lifespan 加载 SentenceTransformer,并通过 TaskGroup.create_task 启动 Batcher.run。Batcher.submit 等待每个请求自己的 Future;Batcher.run 通过 asyncio.to_thread 调用 model.encode,再按请求顺序交付向量。默认使用固定 revision 的 BAAI/bge-m3,每个向量为 1024 维。
flowchart TD
H["HTTP<br/>POST /openapi/v1/stores/{store_id}/knowledge/search"]
A["modules/knowledge/api.py<br/>retrieve<br/>检查身份与店铺权限"]
G["LangGraph 框架<br/>调度 tools 子图或 model 节点"]
T["modules/agent_runtime/runtime.py<br/>execute<br/>处理 search_knowledge 工具"]
S["modules/knowledge/vector.py<br/>RetrievalClient.search<br/>店铺过滤、向量与 BM25 混合检索"]
E["modules/knowledge/vector.py<br/>RetrievalClient.embeddings<br/>取得查询文本向量,见图 1"]
V[("Milvus<br/>dense + sparse<br/>hybrid_search / RRFRanker")]
O["modules/knowledge/vector.py<br/>stable_candidates<br/>相同分数按正文摘要等字段排序"]
F["modules/knowledge/api.py<br/>verified_results<br/>核对当前发布版本并读取正文"]
D[("MySQL<br/>modules/knowledge/models.py<br/>KnowledgeDocument:当前有效资料")]
R[("MySQL<br/>modules/agent_runtime/models.py<br/>AgentRun.messages:ToolMessage 引用")]
M["modules/agent_runtime/runtime.py<br/>model → validate_citations<br/>模型调用前后核验引用"]
H -. "HTTP / FastAPI 路由分发" .-> A
G -. "框架调度工具子图" .-> T
A --> S
T -->|通过 search_knowledge 参数调用| S
S --> E
S -. "SDK 请求:每种检索 30 个候选,RRF 合并" .-> V
S -->|收到候选后| O
A -->|检索返回后重新认证| F
T -->|检索返回后重新检查执行权| F
D -. "读取当前 PUBLISHED 版本和正文" .-> F
T -. "保存 F 返回的 document_id、version、position 和正文" .-> R
G -. "框架调度 model 节点" .-> M
R -. "读取消息中的引用" .-> M
D -. "再次检查店铺、发布状态与引用版本" .-> M
应用启动时,app.py 将复用的 RetrievalClient 保存在 app.state.knowledge;Agent 工作进程的 main 将同类客户端的 knowledge.search 传给 execute_one。图中的两条入口因此使用同一检索实现。
RetrievalClient.search 同时过滤 tenant 和 store,分别发起向量相似度与中文 BM25 检索,再由 RRFRanker 合并候选。verified_results 重新检查访问权限,从 MySQL 读取当前 PUBLISHED 文档,仅接受版本一致的候选,返回正文、标题和引用标识。HTTP 接口默认返回最多 5 个分段,允许 limit 为 1 至 10;Agent 工具分支调用 verified_results 时使用其默认值 5。
validate_citations 检查 ToolMessage 中的资料引用,分别在生成模型调用前、模型返回后执行。资料已撤回、改版或不属于当前店铺时抛出 KNOWLEDGE_CHANGED,工作进程保存失败状态,最终答复不会提交。检索正文和已经产生的模型用量有各自的保存位置。
flowchart TD
A["modules/memory/api.py<br/>consent / change / clear<br/>客户授权、主动修改或清空"]
P[("MySQL<br/>modules/memory/models.py<br/>MemoryProfile:consent、epoch、revision、value_key")]
C["modules/conversations/api.py<br/>accept_message<br/>接受客户消息并保存事件"]
E["modules/memory/service.py<br/>enqueue<br/>已授权时创建提取任务"]
J[("MySQL<br/>modules/memory/models.py<br/>MemoryJob:消息位置、epoch、执行租约")]
X["modules/memory/worker.py<br/>execute_one → execute_claimed<br/>申请后台额度后处理任务"]
K["modules/memory/worker.py<br/>claim<br/>检查授权代次、会话状态和并发额度"]
W["modules/memory/worker.py<br/>process<br/>提取沟通习惯并复核版本与租约"]
L["LangMem / LangChain / 模型服务<br/>create_memory_manager → manager.ainvoke<br/>按 Communication 提取明确表达的习惯"]
S["modules/memory/service.py<br/>write / erase<br/>写入独立内容键或删除指定内容键"]
R[("Redis<br/>AsyncRedisStore<br/>租户、客户、epoch 对应的记忆内容")]
M["modules/agent_runtime/runtime.py<br/>model<br/>调用前读取,提交答复前复核版本"]
U["modules/memory/service.py<br/>snapshot / read<br/>仅返回当前授权且仍有效的内容"]
A -. "事务提交:更新授权、epoch、revision 和可读内容键" .-> P
A -->|change 写入新内容;三种成功操作均删除旧内容| S
C -->|客户消息| E
P -. "读取 consent 和 epoch" .-> E
E -. "同会话消息事务:创建 QUEUED 任务" .-> J
X -->|execute_claimed 调用| K
X -->|领取成功后经内部 invoke 调用| W
J -. "跨进程:领取任务,增加 lease_token" .-> K
P -. "授权关闭或 epoch 不同:任务 CANCELLED" .-> K
W -. "LangMem 调用与模型 HTTP 请求" .-> L
W --> S
S -. "AsyncRedisStore.aput / adelete" .-> R
P -. "提交前读取当前状态,与领取时快照核对" .-> W
W -. "校验通过:更新 value_key 和 revision" .-> P
W -. "校验通过为 COMPLETED;版本变化时取消本次任务" .-> J
M --> U
P -. "读取授权、有效期及内容键" .-> M
R -. "AsyncRedisStore.aget:读取指定内容键" .-> U
记忆接口要求 CUSTOMER 身份。consent、change、clear 都以 expected_revision 检查并发;成功操作增加 epoch 与 revision。consent 每次成功调用均清空已有内容,包括 enabled 与当前值相同的情况;clear 清空内容并保留当前授权选择。主动修改通过 Communication 校验字段;source=user 表明内容由客户主动设置。
enqueue 在客户消息事务内创建 MemoryJob。claim 检查客户当前授权与任务的 epoch 是否一致,同一客户存在运行中的提取任务、会话仍有 active_run_id 或租户后台额度不足时暂不领取。process 读取任务引用的那条客户消息,使用 LangMem 提取明确表达的长期沟通习惯,字段限于 response_style、language、contact_channel。
process 写入新的 Redis 内容键后保存模型用量,再核对 MemoryProfile 快照与任务租约。校验通过才把新键设为可读;授权变化或客户修改造成版本不一致时,任务取消并删除本次新内容。有效执行权已经被其他工作进程取得时,当前进程不能覆盖其任务状态。
Agent 的 model 通过 snapshot 与 read 取得授权记忆,将 communication 添加到系统提示的回复形式说明中;模型返回后再次检查快照。调用前后 snapshot 不一致时返回 MEMORY_CHANGED,当前答复不能提交。已授权客户修改或清空内容会改变快照;未授权状态的快照为 None,在该状态执行 clear 后仍为 None。
source_text 保存编辑原文;chunks 保存检索分段;version、status 决定引用是否有效。状态包括 DRAFT、PUBLISHED、REVOKED。document_id、version、position。知识访问范围为当前租户与店铺。consent 控制授权;epoch 使旧任务失效;revision 控制并发操作;value_key 指向当前可读内容。客户记忆以租户和客户为范围,可以用于该客户的后续会话。conversation_id 与 sequence 指向来源消息,状态包括 QUEUED、RUNNING、COMPLETED、CANCELLED、FAILED,lease_token 防止旧工作进程覆盖新执行结果。MEMORY_UNAVAILABLE。本路线仅阅读以下用例及其断言,没有执行测试、向量计算或生成模型调用。