路线 04：知识检索、引用校验与客户记忆

学习场景

商家发布键盘清洁说明，客户询问清洁方法，并授权系统以后使用简洁回复。沿着资料发布、回答引用和沟通习惯保存三个过程，理解知识正文、检索索引、客户授权和模型上下文各自的作用。

图中后端路径相对 src/zhifu/。连续箭头表示直接函数调用，箭头文字注明调用条件；间断箭头标明 HTTP、框架调度、异步队列或持久数据关系。框架节点、第三方库节点和存储节点单独标明类型。execute、model 分别是 Agent 图中的工具节点与模型节点内部函数。返回路线目录。

图 1：知识草稿、向量索引与发布版本

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 维。

图 1 源码导航

knowledge/api.py：revise、transition：核对修订、发布、撤回各自更新哪些字段。
knowledge/vector.py：RetrievalClient.index_document、embeddings：追踪每批最多 32 个分段、向量维度检查与索引主键。
knowledge/inference.py：lifespan、embed、knowledge/config.py：KnowledgeSettings：查看独立向量服务的模型加载、接口入口和配置。
knowledge/batching.py：Batcher.submit、Batcher.run：查看队列容量、等待超时、取消请求与结果顺序处理。
knowledge/models.py：KnowledgeDocument、knowledge/vector.py：setup：对照 MySQL 正文记录与 Milvus 字段、中文分析器和索引定义。

图 2：混合检索、候选校验与回答引用

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，工作进程保存失败状态，最终答复不会提交。检索正文和已经产生的模型用量有各自的保存位置。

图 2 源码导航

knowledge/api.py：retrieve、agent_runtime/runtime.py：execute：分别找到 HTTP 检索入口和 Agent 工具入口。
knowledge/vector.py：RetrievalClient.search、stable_candidates：追踪检索过滤、RRF 排序与相同分数的处理。
knowledge/api.py：verified_results：查看候选标识怎样对应 MySQL 的当前正文与版本。
agent_runtime/runtime.py：validate_citations、model：找到模型调用前后的两次引用检查，以及用量独立提交的位置。
app.py：retrieval_client 初始化、agent_runtime/worker.py：main：确认 HTTP 进程与工作进程分别管理自己的检索客户端。

图 3：客户授权、记忆提取与后续使用

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。

图 3 源码导航

memory/api.py：consent、change、clear：核对每种客户操作的授权与版本变化。
conversations/api.py：accept_message、memory/service.py：enqueue：找到消息事件与提取任务共同提交的位置。
memory/worker.py：execute_one、execute_claimed、claim：阅读后台容量、客户任务串行、会话等待及过期租约。
memory/worker.py：process、memory/service.py：Communication：查看 LangMem 的字段限制、提取提示、用量保存与提交检查。
memory/service.py：namespace、snapshot、read、write、erase：追踪 Redis 命名空间、内容有效期与删除对象。
agent_runtime/runtime.py：model：找到记忆进入系统提示以及返回后的版本复核。

关键数据与状态

KnowledgeDocument：source_text 保存编辑原文；chunks 保存检索分段；version、status 决定引用是否有效。状态包括 DRAFT、PUBLISHED、REVOKED。
知识索引主键包含租户、文档、版本和分段位置；每条引用保留 document_id、version、position。知识访问范围为当前租户与店铺。
MemoryProfile：consent 控制授权；epoch 使旧任务失效；revision 控制并发操作；value_key 指向当前可读内容。客户记忆以租户和客户为范围，可以用于该客户的后续会话。
MemoryJob：conversation_id 与 sequence 指向来源消息，状态包括 QUEUED、RUNNING、COMPLETED、CANCELLED、FAILED，lease_token 防止旧工作进程覆盖新执行结果。
记忆在 MySQL 中的可读有效期为 29 天，Redis 内容 TTL 为 30 天。未授权或已经过期时返回空内容；有效 MySQL 记录对应的 Redis 内容缺失时返回 MEMORY_UNAVAILABLE。

阅读检查问题

Milvus 仍保存某个旧版本向量时，为什么知识检索不能继续返回已撤回正文？模型正在生成时撤回资料，哪一次检查处理这个变化？
客户在记忆提取过程中清空内容，旧任务已经产生模型费用并写入新 Redis 内容键。代码分别怎样处理用量、内容键和任务状态？
店铺知识与客户记忆分别使用哪些字段隔离访问范围？客户沟通习惯通过什么位置进入模型请求，订单事实又通过什么入口查询？

测试文件阅读入口

本路线仅阅读以下用例及其断言，没有执行测试、向量计算或生成模型调用。

test_knowledge_memory.py：test_publication_and_revocation：阅读发布前后、撤回后、跨租户访问及旧向量清理的断言。
test_knowledge_memory.py：test_memory_consent_isolation_and_clear、test_memory_clear_invalidates_queued_extraction：阅读授权、客户隔离、并发版本和旧提取任务取消。
knowledge_memory_model_acceptance.py：test_actual_model_retrieves_published_document、test_actual_langmem_extraction_and_cross_conversation：阅读真实模型验收文件中工具检索和跨会话记忆的检查方法。
knowledge_memory_model_acceptance.py：test_clear_during_actual_extraction、test_revoke_during_actual_answer：阅读模型调用期间清空记忆、撤回资料的并发场景。

04-knowledge-memory.md 18 KB 히스토리 Raw