项目实现
基于 SuperMew-0.1.0 与博客第 1～7 篇的实现部分，将七张原生子图按主调用关系拼接。全部是可编辑矩形、文字和箭头，子图没有缩小字号。右侧保留整片空白扩展区，后续接入第 8 篇之后的模块。
总图怎么读
- 上排从左到右：对话编排 → 状态 / Runtime → 工具。
- 右侧从上到下：工具 → RAG 工作流 → 检索。
- 检索向左回到数据存储；左下文档入库也进入同一存储模块。
- 箭头连接模块边界，箭头旁点名实际调用入口。它表示列出的接口发生调用，不表示框内所有函数都调用对方。模块内部的文件、类和函数仍按子图原样展开。
- 蓝色实线是问答相关调用，蓝色虚线表示返回文本或状态，绿色是文档入库，紫色是 HITL 直接恢复，灰色是会话持久化依赖。没有接点的线交叉不表示合流。
两条业务流
用户问答
chat_with_agent()/chat_with_agent_stream()从 ConversationStorage 加载历史与 metadata，建立本轮 ctx。- Runtime 通过
create_agent_for_request(ctx)装配模型、提示词和工具。Service 构造长期笔记 + 最近 6 条历史 + 当前问题，调用 Agent。 - 需要知识时调用闭包工具 →
run_rag_graph()。Pipeline 做复杂度判断、证据评估和至多一次查询改写；复杂问题并行检索后合成 Evidence。 - 所有检索入口复用
utils.retrieve_documents()，由 Milvus 召回 L3，再根据 parent_chunk_id 经 Redis / PG 获取父块，重排和过滤。 - Pipeline 返回证据及 Trace；工具给模型带来源的片段，同时把 Trace / 快照写到 ctx。Agent 组织回答，Service 取走 Trace，并保存 AI 消息和 metadata。
- 若要求用户澄清，Service 保存 pending_hitl。下一轮只有具备有效恢复快照时，才直接进入
resume_rag_from_hitl()，绕过外层 Agent 和复杂度规划；无快照走兼容回退。恢复后再回答或继续追问。
天气是独立 Tool 分支，不走 Milvus。流式分支通过 ctx 的 Queue 合并 rag_step 与文本；[DONE] 发出后仍执行笔记更新和落库。
文档入库
- DocumentLoader 解析支持的格式；HTML 单独清理结构，然后仍回到统一逐页三级分块。
- 调用方拆出 L1/L2 与 L3。L1/L2 先写 ParentChunkStore（PG + Redis）；L3 交给 MilvusWriter。
- Writer 复用 EmbeddingService 编码 Dense 向量，再由 MilvusStore 写入全文、向量及关联 ID；BM25 在 Milvus 集合内计算。
这里的“调用方”对应第 2 篇完整入库测试所展示的编排，不额外虚构 IndexingPipeline 类，也不提前画第 8 篇之后的上传 / 任务接口。检索侧编码 Query 复用同一个 EmbeddingService；这项复用在检索子图中说明，总图不为每一条 import 再加横贯画布的线。
子图导航
- [[图_1数据存储模块]]：PG / Redis / Milvus 的底层连接、模型与存取。
- [[图_2文档入库模块]]：原文 → 三级 Chunk → Embedding → 分层存储。
- [[图_3检索模块]]：召回 → Auto-Merging → Rerank → 阈值过滤。
- [[图_4RAG工作流模块]]：简单 / 复杂问题路由、Grader、改写、合成、HITL 恢复。
- [[图_5工具层模块]]：知识工具闭包与天气工具包装。
- [[图_6Agent状态管理模块]]：Schema / 请求 ctx / ConversationStorage / Runtime。
- [[图_7Agent对话编排模块]]：同步问答、SSE、跨轮 HITL 和长期笔记。
同一个 backend/rag 或 backend/chat 目录在不同模块视图中按文件职责展开，并非工程里存在多份目录。标题保留 Obsidian 双链，点击总图模块标题可回到独立子图；文字与框未组合，矩形宽高可单独调整。
Excalidraw Data
Text Elements
项目实现 SuperMew 0.1.0 · 第 1～7 篇 ^4edf5644
七张子图保留原字号拼接；点击模块标题可打开独立子图。右侧留给后续模块，当前不展开第 8 篇之后的实现。 ^8c65a2a9
蓝色：正常问答 / 证据调用 绿色：文档入库 紫色：HITL 恢复 蓝色虚线：结果返回 灰色：持久化依赖 ^489b86b1
主链：对话编排 → Agent 装配 → 工具 → RAG 工作流 → 检索 → 数据存储；文档入库从左下接入相同存储。 ^d6e483ee
[[图_7Agent对话编排模块|07 Agent对话编排模块 · 实现结构]] ^1774d592
目录 → 文件 → 类 / 函数；框内编号表示局部职责，箭头表示主要调用。 ^ed54796b
backend/ 承载本模块的后端代码 ^82d835bf
chat/ 承载会话问答的业务编排 ^99193efb
[[7_代码实现_Agent 对话编排#2. 模块实现|service.py]] 同步 / 流式主入口及其辅助流程 ^066e5b55
chat_with_agent() 执行一轮同步问答
- storage.load_with_meta()：取历史、笔记、pending_hitl
- 创建 ctx、重置预算；先保存当前 Human 消息
- 有有效恢复快照 → 直接续接 RAG；否则创建 Agent.invoke()
- 读取 Trace；若需澄清，用标准追问覆盖模型文本并建 pending_hitl
- 更新标题 / 笔记 / 待处理状态，保存 AI 消息与 rag_trace
- 返回 {response, rag_trace}；finally ctx.close() ^a0b54d23
chat_with_agent_stream() 产生本轮 SSE 事件
- 先发 rag_step；加载历史，创建 Queue 与流式 ctx，保存 Human
- 正常分支：启动 _agent_worker()，消费 Queue 并逐条 yield
- HITL 恢复：线程执行 RAG 并转发步骤，再 model.astream() 回答
- 输出 trace / hitl_request 和 [DONE]；断开时取消 Agent 任务
- [DONE] 后继续更新笔记与落库；finally ctx.close() ^810d3bf1
_agent_worker() 流式入口内部的生产协程
- Agent.astream(stream_mode="messages", recursion_limit=8)
- 只取 AI 文本；跳过工具调用片段，HITL 时抑制普通回答
- 向 Queue 写 content / error，最后写内部 None 哨兵 ^cdbda1f1
同步 invoke 同样设置 recursion_limit=8。 [DONE] 是应用事件；不代表后续笔记更新和 PG 提交已完成。 ^46591645
_build_context_messages() 控制本轮送给 Agent 的上下文
- 长期 persistent_note → SystemMessage
- 只取最近 6 条历史消息，再追加当前 HumanMessage ^485674da
_build_pending_hitl() 构造跨轮等待状态
- 组合原问题、追问、选项、已补充答案和恢复快照
- PendingHitlState 校验后写入 metadata.pending_hitl ^68f0135f
_resume_rag_from_hitl_sync() 绕过外层 Agent 续接检索
- 从 pending_hitl 取快照，调用 pipeline.resume_rag_from_hitl()
- 缺快照时返回 None，由主入口保留兼容回退路径 ^62b17c2e
_current_pending_hitl() 读取并校验已有追问
- 校验 PendingHitlState；无效值按未等待处理
- _pending_resume_state()：提取快照
- _build_hitl_resume_query()：无快照兼容路径拼接补充 ^b694d095
_build_hitl_event() 生成流式追问事件
- 从 pending_hitl 选择前端所需字段
- _format_hitl_message()：生成同步返回与历史保存的追问正文 ^f1b44637
_answer_resumed_rag_sync() 基于恢复后的真实证据回答
- 无证据 → _no_knowledge_response() 固定提示
- 有证据 → _build_resume_answer_messages() → model.invoke()
- 仍需 HITL 时由主入口继续追问，不调用本回答分支 ^eac8dd4b
_should_update_persistent_note() 判断是否需要更新长期笔记
- 已有笔记，或历史消息数超过 6 条时更新 ^f7486aa5
_update_persistent_note_sync() 压缩保留长期关键信息
- FAST_MODEL：旧笔记 + 本轮问答；首次纳入更早历史
- 提示限制约 500 中文字；调用失败时保留旧笔记 ^9ddcb530
update_persistent_note() 让流式收尾异步等待笔记更新
- run_in_executor() 调用同步摘要函数，避免阻塞事件循环 ^f40d13c0
generate_session_title() 为首次会话生成短标题
- 本地截取首条用户文本，最多 16 字；不请求模型 ^71c9221c
其余小工具合并阅读： _is_hitl_trace() / _hitl_route_from_trace() / _hitl_prompt_from_trace() / _hitl_options_from_trace()：识别追问 _existing_hitl_answers()：保留补充历史；_extract_ai_content() / _format_retrieved_chunks()：整理模型输入输出 ^b60d2ec7
主链依赖：图 6 的 Storage / ctx / Runtime → 图 5 工具 → 图 4 RAG；有效 HITL 恢复直接进入图 4。 ^fd192aac
[[图_6Agent状态管理模块|06 Agent状态管理模块 · 实现结构]] ^8e7ac74c
目录 → 文件 → 类 / 函数；框内编号表示局部职责，箭头表示主要调用。 ^6b0bad62
backend/ 承载本模块的后端代码 ^9e12ab0e
schemas/ 声明并校验交换数据 ^e84bae57
[[6_代码实现_Agent 状态管理#2.1 数据结构实现|chat.py]] 统一 Chat / Trace / HITL 数据协议 ^4361fa6e
StrictSchema：Pydantic 基类，extra="forbid"。 下列类只定义 / 校验数据，不直接执行检索或落库。 ^3954b2f2
RetrievedChunk 规范化一条真实检索证据
- 正文、来源、页码、Chunk 关联 ID 与分数字段 ^eaa38342
RagTraceFields 定义主 / 子 Trace 共用字段
- 候选与合并、重排、评分、改写、HITL、复杂度信息 ^89fbfb39
RagSubTrace 继承共用字段，表示一个子问题轨迹
- 复用 RagTraceFields 的结构，供父级 Trace 汇总 ^ff2c6b18
RagTrace 表示本轮主问题的完整轨迹
- 继承 RagTraceFields，增加 sub_traces 列表 ^1be200d0
HitlResumeState 校验可跨轮保存的最小恢复快照
- 原问题、route / status、改写次数和复杂度规划信息
- 不含检索全文；恢复时再查知识库 ^a8f26296
PendingHitlState 描述等待用户补充的会话任务
- 追问 ID、原问题、提示 / 选项、已补充答案、创建时间
- 嵌套可选 resume_state，保存到会话 metadata ^6e958f93
normalize_rag_sub_trace() 清理并校验子问题 Trace
- _normalize_trace_fields() 过滤字段；_normalize_chunks() 校验证据 ^75e1dfdf
normalize_rag_trace() 统一对外或保存的主 Trace
- 清理主字段与证据，逐个规范化 sub_traces，再做模型校验 ^40979b8a
同文件其余协议类（此处不展开字段表）： ChatRequest / ChatResponse：问答请求与结果 MessageInfo / SessionMessagesResponse：历史消息 SessionInfo / SessionListResponse：会话摘要与列表 SessionDeleteResponse：删除结果 ^1940ae1b
这些是数据结构；第 8 篇之后的 HTTP 路由尚未纳入。 ^a2051aa5
chat/ 管理请求生命周期、会话历史与模型装配 ^628670b7
[[6_代码实现_Agent 状态管理#对话上下文数据结构|request_context.py]] 承载本次请求的临时状态 ^4e7d847f
ChatRequestContext 隔离预算、Trace 和流式事件
- for_sync() / for_stream()：创建同步 / 流式上下文
- reset_knowledge_tool_budget() / acquire_knowledge_tool_slot()：知识工具一次预算
- emit_rag_step()：计算耗时，经事件循环线程安全地推入 Queue
- store_rag_trace()：规范化 Trace，校验并暂存恢复快照
- peek_rag_trace() / take_rag_trace()：查看 / 取走后清空
- close()：停用上下文并释放 Queue / loop 引用 ^7adf5147
存活范围：Service 创建 → Tool / Pipeline 共用 → Service 读取 → finally close()。 RLock 保护共享字段；流式 Queue 只属于本次请求，不负责跨轮持久化。 ^2e40f2ad
[[6_代码实现_Agent 状态管理#2.2 会话存储管理|storage.py]] 把会话与消息保存到关系库并缓存 ^c3525680
ConversationStorage 按用户与 session_id 组织会话
- save()：查建会话 → 合并 metadata → 删除旧消息并重写全量历史 → PG 提交
- 提交后刷新 Redis 消息缓存，并失效会话列表缓存
- load()：Redis 命中直接返回；未命中查 PG 并回填
- load_with_meta()：读取消息及 PG 会话 metadata
- list_sessions() / list_session_infos()：缓存优先的会话列表
- get_session_messages()：带 Trace 的历史；delete_session()：删除并清缓存 ^95347d0c
storage = ConversationStorage()；依赖图 1 的 SessionLocal、ORM 模型与 cache。 metadata 保存 title / persistent_note / pending_hitl；消息附加数据保存 rag_trace。 ^64defc2f
[[6_代码实现_Agent 状态管理#2.3 Agent Runtime|runtime.py]] 创建共享模型与每请求 Agent ^33c7a03d
model = init_chat_model(MODEL)：外层回答模型 fast_model = init_chat_model(FAST_MODEL)：记忆摘要模型 SYSTEM_PROMPT：工具使用、证据引用和无知识约束 ^3e05f320
create_agent_for_request(ctx) 将模型、提示词与工具装配成 Agent
- create_agent(model, tools=[天气工具, make_search_knowledge_base(ctx)])
- 每次请求创建 Agent；知识工具闭包绑定当前 ctx ^3533a22e
四种边界：Schema 是协议；ctx 是本次请求；Storage 是跨轮数据；Runtime 是运行装配。 ^0fc8d9c4
[[图_5工具层模块|05 工具层模块 · 实现结构]] ^ffd68381
目录 → 文件 → 类 / 函数；框内编号表示局部职责，箭头表示主要调用。 ^6753ed5f
backend/ 承载本模块的后端代码 ^8cf53a22
tools/ 为 Agent 暴露业务工具 ^d935a66d
[[5_代码实现_工具层#2.1 knowledge.py|knowledge.py]] 将请求 ctx 绑定到知识检索工具 ^c53e4d71
make_search_knowledge_base(ctx) 为每个请求生成工具
- 闭包持有当前请求的 ChatRequestContext
- 返回内部 @tool("search_knowledge_base") 函数 ^7f85a930
search_knowledge_base(query) 运行证据工作流并格式化返回
- ctx.acquire_knowledge_tool_slot()：每轮最多执行一次知识工具
- 延迟导入并调用 run_rag_graph(query, ctx)
- ctx.store_rag_trace()：暂存 Trace 和可选恢复快照
- 返回 NEEDS_CLARIFICATION / NEEDS_SCOPE_SELECTION / NO_KNOWLEDGE
- 有证据则返回带编号、来源、页码的 Retrieved Chunks ^e557cc2a
返回闭包工具 ^2fbf6f9b
工具只返回证据或控制信息，回答仍交给 Agent。 这是一轮知识工具的硬预算；不是对天气工具的通用次数限制。 ^48c182b5
[[5_代码实现_工具层#2.2 weather.py|weather.py]] 提供实时天气与预报 ^69b1205d
get_current_weather_tool() 对外的 @tool 包装
- 转发 location / extensions 参数到普通 Python 函数 ^fb72b1c7
get_current_weather() 请求高德天气 HTTP API
- 校验 AMAP_API_KEY、城市和 extensions
- base 返回实时天气；all 返回预报
- 处理请求超时、HTTP 错误和业务失败 ^b51c0593
这是独立工具分支，不经过 Milvus 或 RAG Pipeline。 ^5bd20cf4
init.py 整理包级导出名称
- 导出 make_search_knowledge_base
- 将装饰后的 get_current_weather_tool 导出为 get_current_weather ^7c9d8590
Agent 装配入口见 [[图_6Agent状态管理模块]] 的 runtime.create_agent_for_request()。 ^f98afd00
[[图_2文档入库模块|02 文档入库模块 · 实现结构]] ^8f066a98
目录 → 文件 → 类 / 函数；框内编号表示局部职责，箭头表示主要调用。 ^9f1b2a88
backend/ 承载本模块的后端代码 ^bc52cf3e
indexing/ 解析文档、分块并写入索引 ^fee133fa
[[2_代码实现_文档入库#2.1 document_loader.py|document_loader.py]] 统一格式入口与三级分块 ^dff7cf18
sanitize_text() 去除不可用字符
- 清理 NUL、控制字符、零宽字符，保留换行等排版字符 ^dd1c7826
DocumentLoader 把原文转成统一 Chunk 列表
- init()：建立 L1 / L2 / L3 三个递归分块器
- load_documents_from_folder()：遍历支持的文件，逐个加载
- load_document()：分派 PDF / Word / Excel / HTML 加载器
- _load_from_langchain_docs()：清洗正文、补齐来源，逐页分块
- _split_page_to_three_levels()：页内 L1 → L2 → L3 逐级切分
- _build_chunk_id()：文件名 :: 页码 :: 层级 :: 层内序号 ^f2c5395a
默认 chunk_size / overlap：L1 2400 / 400；L2 1600 / 200；L3 800 / 100。 返回三层全文和 chunk_id / parent_chunk_id / root_chunk_id；不会跨页合并。 ^95edf2a4
入库调用方负责分流：
- L1 / L2 → ParentChunkStore.upsert_documents()
- L3 → MilvusWriter.write_documents() 先保存父块，再写叶块；这不是本文件中的独立 Pipeline 类。 ^50c0784b
[[2_代码实现_文档入库#2.2 html_processor.py|html_processor.py]] 把网页正文整理为可分块文档 ^3db70a60
load_html_for_document_loader() 接入统一加载流程
- 调用 parse_html_file_to_sections() 获取章节
- 将章节包装为 LangChain Document，附带标题 / 页码元数据 ^936a1d05
parse_html_file_to_sections() 提取并切分网页正文
- _read_html_text()：识别 BOM / charset，尝试 UTF-8 / GB18030
- _strip_noise() / _pick_root()：去脚本导航，选择正文
- _doc_title()：提取页面标题；_br_to_newlines()：保留换行
- _linearize_blocks()：把标题、段落、列表、表格转为有序文本
- _split_into_sections()：按二级标题组织章节 ^35aa493c
HTML 结构化结果仍回到 DocumentLoader，复用相同三级分块。 ^714c0e74
HTML 分支 ^4b26ed69
[[2_代码实现_文档入库#2.3 embedding.py|embedding.py]] 统一入库与查询使用的密集向量模型 ^a5484ae5
_create_dense_embedder() 读取配置并建立本地模型
- HuggingFaceEmbeddings；默认 BAAI/bge-m3、CPU
- normalize_embeddings=True；不在 Python 侧计算 BM25 稀疏向量 ^8f4b3dff
EmbeddingService 提供共享的文本向量化接口
- init()：持有一个本地 Dense 模型
- get_embeddings(texts)：批量编码；空输入返回 []，失败抛异常 ^733f95f5
embedding_service = EmbeddingService()：进程内共享实例。 查询端复用它，避免索引与查询使用不同的向量空间。 ^3496e13e
[[2_代码实现_文档入库#2.4 milvus_writer.py|milvus_writer.py]] 批量编码并写入 L3 叶块 ^f4fc7263
MilvusWriter 把调用方传入的 Chunk 写入向量库
- init()：复用 embedding_service 和 MilvusStore
- write_documents()：初始化集合；默认每批 50 条
- 提取 text → get_embeddings() → 组装全文 / 向量 / 关联 ID
- 调用 MilvusStore.insert()，成功后累计条数并回调进度
- 调用方只传 L3；本类不保存 L1 / L2，不负责文件上传 ^82771e31
存储依赖见 [[图_1数据存储模块]]： L1 / L2 → ParentChunkStore → PostgreSQL + Redis L3 → MilvusStore → Milvus（BM25 由集合内置 Function 生成） ^03e5fa12
get_embeddings(texts) ^92a86b24
文档 §3 验证示例：test/2_test_indexing.py；test_full_indexing_pipeline 演示父块 / 叶块分流。 ^35d8a20a
[[图_1数据存储模块|01 数据存储模块 · 实现结构]] ^8c50dca8
[[1_项目设计#1）env.py|env.py]] 统一加载项目环境变量
- load_env()：读取根目录 .env，只加载一次
- 在导入其他存储模块前调用 ^723caaa3
infra/ 提供数据库连接与缓存访问 ^2c6d8abe
db/ 声明关系数据库的 ORM 模型 ^d85a446f
indexing/ 提供文档索引与知识库存取 ^41d440d5
[[1_项目设计#1）database.py|database.py]] 配置关系库连接与建表能力
- engine：配置 PostgreSQL 引擎
- SessionLocal：创建数据库会话
- Base：提供 ORM 声明基类
- init_db()：导入模型并创建数据表
- before_cursor_execute()：SQL 执行前净化参数
- _clean_nul_chars()：递归清洗字符串 ^60082795
[[1_项目设计#2）cache.py|cache.py]] 封装 Redis 缓存访问 ^8cf86ae0
RedisCache 统一缓存读写与清理
- init()：读取连接、键前缀与 TTL
- _get_client()：按需创建 Redis 客户端
- get_json()：读 JSON；未命中 / 异常返回 None
- set_json()：写 JSON 并设置过期时间
- delete() / delete_pattern()：按键 / 模式删除 ^37fceea4
cache = RedisCache()：供其他模块共享 ^1863ea9e
[[1_项目设计#1）models.py|models.py]] 声明四类持久化数据模型 ^8c31231f
User 定义账号与角色
- 声明 users 表：账号、密码哈希、角色
- sessions：关联用户的会话 ^65025fe0
ChatSession 定义会话及用户、消息关联
- 声明会话及 metadata_json
- 限定同一用户下 session_id 唯一
- 关联所属用户和消息列表 ^0a391481
ChatMessage 定义消息正文与检索轨迹
- 声明消息正文、类型、时间与 rag_trace
- 关联所属会话；随会话级联删除 ^282f7f65
ParentChunk 定义 L1/L2 父级文本块
- 声明 L1/L2 父块全文与来源
- 保存块层级及父子关联 ID ^68c4303a
四个类均继承 database.py 中的 Base。 ^c512fdcd
[[1_项目设计#1）milvus_client.py|milvus_client.py]] 封装 Milvus 集合与检索操作 ^8d78ad09
MilvusSettings 读取并保存 Milvus 连接配置
- from_env()：读取地址、Collection 与超时配置 ^dbb636b7
milvus_client_session() 管理一次客户端连接的创建与关闭
- 创建客户端 → yield 供调用 → finally 关闭连接 ^c100886c
MilvusStore 提供集合管理、数据读写与检索
- _run() / session()：通过上述短连接执行操作
- init_collection() → ensure_collection()：创建字段、BM25 Function 与索引
- insert()：写入 L3 数据；query() / query_all()：条件 / 分页查询
- get_chunks_by_ids()：按 chunk_id 批量取块
- dense_retrieve()：Dense 检索
- hybrid_retrieve()：Dense + BM25 → RRF → 返回候选
- delete()：删除匹配记录；has_collection()：检查集合
- drop_collection()：删除整个集合 ^ddb79cc8
get_milvus_store() 获取可复用的 MilvusStore 实例
- 按需创建并复用 Store 实例；不缓存客户端连接 ^c2001fcc
[[1_项目设计#2）parent_chunk_store.py|parent_chunk_store.py]] 封装父块持久化与缓存 ^6ea2c68c
ParentChunkStore 管理父块的持久化与缓存
- upsert_documents()：更新 / 插入父块，写 Redis，最后提交 PG
- get_documents_by_ids()：先查 Redis；未命中批查 PG 并回填缓存
- delete_by_filename()：删除 PG 父块，再清除对应缓存
- _to_dict() / _cache_key()：转换 ORM 记录 / 生成父块缓存键 ^a9edcc2f
依赖：SessionLocal + ParentChunk + cache 本文件只负责父块存取，不负责文档分块或向量化。 ^9f4e8ce5
Base ^9e3a61c7
主要依赖：models.py → database.Base；ParentChunkStore → SessionLocal / ParentChunk / cache。 ^1c9486df
辅助：_normalize_filter() 将空过滤条件规范为 id >= 0。 ^b75b0301
[[图_4RAG工作流模块|04 RAG工作流模块 · 实现结构]] ^9dbc7095
目录 → 文件 → 类 / 函数；框内编号表示局部职责，箭头表示主要调用。 ^176490be
backend/ 承载本模块的后端代码 ^ac7d681f
rag/ 在检索之上组织证据工作流 ^7effacaa
[[4_代码实现_RAG Pipeline#2.1 pipeline.py|pipeline.py]] LangGraph 路由、评分、合成与 HITL 恢复 ^00cb1f57
RAGState 保存单次图执行的业务状态
- question / docs / route / rewrite_count / rag_trace / request_context
- sub_results 用 operator.add 汇聚并行结果 ^ca0d4889
EvidenceGrade 约束 Grader 的结构化输出
- 相关性、可回答性、歧义、路由、置信度与 HITL 提示 ^05c261d8
ComplexityResult 约束复杂度规划的结果
- simple / complex、原因和最多 4 个子问题 ^2f7861fd
run_rag_graph() 正常 RAG 的对外入口
- _initial_state()：建立本轮状态，注入 ChatRequestContext
- rag_graph.invoke()：执行已编译图
- 若需 HITL，附带 _build_hitl_resume_state() 最小恢复快照 ^f57c9dfe
build_rag_graph() 把函数注册为图节点与条件边
- 入口 classify_complexity；简单 / 复杂两条分支
- 评分后仅 rewrite 才回路，其余 route 都结束
- compile()：不配置 Checkpointer，不使用 interrupt ^2d6a4412
路由与格式辅助： _route_after_complexity()：简单 / 复杂 _route_after_grade()：rewrite / end _format_docs()：构造 Evidence 上下文 _emit()：委托 ctx 上报 rag_step ^642ecdce
resume_rag_from_hitl() 用户补充后的针对性检索入口
- _state_from_resume()：校验快照，原问题拼接用户补充
- _retrieve_resume_query()：重新检索，再调用评分节点
- 仍需澄清则生成新快照；不重走复杂度与子问题规划
- 恢复快照来自 schemas/chat.py，跨轮由 Service 保存 ^1b15612f
快照边界： _build_hitl_resume_state() 保存问题、路由、 检索状态、改写次数和复杂度规划信息。 原始 docs 不放进恢复快照，恢复时重新检索。 ^9380c9f2
classify_complexity() 确定本次走哪条路径
- 明确简单意图先走本地规则，否则 FAST_MODEL 结构化判断
- 复杂问题同时规划子问题，避免再单独调用一次 LLM ^feb895ee
retrieve_initial() 简单问题首次检索
- 调用 utils.retrieve_documents() → docs / context / Trace ^edc834f9
grade_documents_node() 评估证据并确定 route
- 无 docs 直接判无知识；否则 GRADE_MODEL 结构化评分
- _resolve_route()：answer / rewrite / clarify / scope_select / no_knowledge
- 歧义触发 HITL；终止失败路径清空对外 docs / context ^07de307d
rewrite_question_node() 用一次改写机会纠错
- 预算最多 1 次；调用 utils.rewrite_query_once()
- 记录 Step-back / HyDE 方法与新查询 ^06fcc1c9
retrieve_rewritten() 用新查询重新检索
- 调用 utils.retrieve_documents()，更新证据与 Trace
- 回到 grade_documents_node()；不再无限改写 ^14d793a0
非 rewrite 的 route → 图结束。 answer：返回证据；clarify / scope_select：请求补充； no_knowledge：返回无知识状态。 部分证据：首次可改写；预算用尽后转澄清。 ^95cbf3d1
prepare_sub_questions() 使用规划器已产出的子问题
- 逐条 emit 展示子问题；不再次进行 LLM 分解 ^e24ef921
_fanout_sub_questions() 并行分发子问题
- LangGraph Send：每题一份独立 RAGState，共享请求 ctx ^1243ab73
rag_sub_agent() 执行每个子问题的固定短链
- retrieve_initial() → grade_documents_node()
- 保留可用 / 部分证据；不做查询改写，不生成子答案
- 把 {question, docs, status, trace} 放入 sub_results ^adf2a96d
synthesis() 汇总并行证据与状态
- 只取 answerable / partial 子结果，dedupe_documents() 去重
- 有证据返回 answer；全部无证据时汇总 HITL 或 no_knowledge
- 汇总 sub_traces；这里合成 Evidence，不生成最终答案 ^89d13428
simple ^8a468c62
complex ^a439b6d0
rewrite ^17867063
再评分 ^69064315
并行结果汇合 ^420faa40
输出边界：docs + rag_trace + 可选 hitl_resume_state；最终回答由外层 Agent / Service 生成。 ^d3c3efd3
本图只展开当前源码可达路径；_route_after_initial() 未接入 build_rag_graph()，不画成运行节点。 ^d8d8c8c5
[[图_3检索模块|03 检索模块 · 实现结构]] ^f68fe7a4
目录 → 文件 → 类 / 函数；框内编号表示局部职责，箭头表示主要调用。 ^75001298
backend/ 承载本模块的后端代码 ^56c80cba
rag/ 提供检索与 RAG 工作流 ^800e19d5
[[3_代码实现_检索模块#2.1 utils.py|utils.py]] 检索主链与工作流辅助函数 ^c281b186
retrieve_documents() 统一检索入口
- resolve_candidate_k()：计算召回候选数
- embedding_service.get_embeddings([query])：编码查询
- 仅查 L3；MilvusStore.hybrid_retrieve()：Dense + BM25 → RRF
- 召回或后处理异常时尝试 dense_retrieve() 降级
- _finalize_retrieval()：统一处理；返回 {docs, meta}
- 编码失败或两种检索均失败：返回空证据与失败元数据 ^a5aa9632
_finalize_retrieval() 固定顺序处理候选证据
- _auto_merge_candidates()：父块替换
- _rerank_documents()：选取 top_k
- _meets_rerank_min_score()：阈值过滤，再统计 meta ^44400558
先 Top-K 再过滤；过滤后可以少于 K，不补齐。 Hybrid 异常的 try 块包含后处理；降级不只针对 Milvus 调用失败。 ^3abf454d
resolve_candidate_k() 确定召回规模
- 优先读 RETRIEVAL_CANDIDATE_K；否则 top_k × 配置倍数 ^18d2caf3
_auto_merge_candidates() 沿父子关系逐级扩大上下文
- _merge_to_parent_level()：先 L3 → L2，再 L2 → L1
- 按 parent_chunk_id 分组；命中数达到门槛才尝试替换
- 每层记录应用情况、命中父块与替换片段数 ^3e3c7a88
_merge_to_parent_level() 批量取父块并替换子块
- ParentChunkStore.get_documents_by_ids()：Redis → PG
- 找到父块才替换；父块继承子块中的最高有效分数 ^1f9dd901
_rerank_documents() 重排候选并限制条数
- HOST / MODEL / KEY 配齐后请求外部 Rerank API
- 返回 rerank_score；保持来源与层级等元数据
- 未配置或调用失败：按原检索分数排序并取 Top-K ^21c0b20a
候选结果 ^3524b86b
① 合并 ^bfe918d6
辅助能力：供 Pipeline 调用，不属于每次检索必经步骤。 ^4e123934
RewritePlan 约束单次查询改写的输出结构
- method 只能是 step_back 或 hyde
- 分别携带抽象问题或假设文档，后续再做互斥校验 ^6deff92a
rewrite_query_once() 用 FAST_MODEL 选择一种改写方式
- with_structured_output(RewritePlan) 生成结构化结果
- 校验方法与内容；原查询 + 抽象问题 / 假设文档组成新查询
- 只构造检索词；HyDE 文本不作为真实证据 ^ee2c1661
dedupe_documents() 合并多路证据时去重
- 优先按 chunk_id；缺失时按来源 / 页码 / 全文识别重复
- 重复项保留较高有效分数，输出有序证据 ^b567d672
retrieval_trace_fields() 选择要向上层暴露的检索元数据
- 提取候选量、层级、合并、Rerank、阈值等 Trace 字段 ^9d312ed5
配置 / 小工具合并阅读： _optional_env() / _read_positive_int_env() / _read_float_env()：读配置 _effective_score() / _sort_by_rank_score()：统一排序 _get_rerank_endpoint() / _meets_rerank_min_score()：端点与阈值 ^995dc098
输入：query + top_k；输出：docs（真实文档片段）+ meta（过程信息）。 ^37aaf3df
① 准备本轮运行 load_with_meta() / for_sync|for_stream() create_agent_for_request(ctx) ^cf669781
② 装配并由 Agent 选择工具 make_search_knowledge_base(ctx) 或高德天气工具 ^6aff543a
③ 知识工具执行 RAG run_rag_graph(query, ctx) ^10c66d17
④ 检索原问题 / 改写问题 / 子问题 retrieve_documents() → docs + meta ^7d5acbc3
⑤ L3 召回与父块回查 MilvusStore.hybrid_retrieve() ParentChunkStore.get_documents_by_ids() ^2f86fcdb
文档入库：调用方先父后叶 L1 / L2 → ParentChunkStore.upsert_documents() L3 → MilvusWriter → MilvusStore.insert() ^3baf23eb
会话持久化 ConversationStorage → SessionLocal / ORM / cache 消息与 Trace；title / 笔记 / pending_hitl ^80bfe3df
证据或控制文本 → Agent Trace / 恢复快照 → ctx.store_rag_trace() ^003ea8b0
回答 / 流式文本返回 Service 取 Trace → 更新 metadata → storage.save() ^f44a4a8e
HITL 下一轮：有效 resume_state + 用户补充 → resume_rag_from_hitl() 跳过外层 Agent / 复杂度规划 → 针对性检索与评分 → 回到 Service 回答或继续追问 ^7f783dd8
后续模块预留区 ^28080fe7
暂不展开第 8 篇之后的代码 后续直接在此接入新子图 ^45515c07
当前实现范围：第 1～7 篇；普通博客文章与已完成的数据存储子图保持原文件。 ^9e80bf3e
Element Links
1774d592: [[图_7Agent对话编排模块]]
066e5b55: [[7_代码实现_Agent 对话编排#2. 模块实现]]
8e7ac74c: [[图_6Agent状态管理模块]]
4361fa6e: [[6_代码实现_Agent 状态管理#2.1 数据结构实现]]
4e7d847f: [[6_代码实现_Agent 状态管理#对话上下文数据结构]]
c3525680: [[6_代码实现_Agent 状态管理#2.2 会话存储管理]]
33c7a03d: [[6_代码实现_Agent 状态管理#2.3 Agent Runtime]]
ffd68381: [[图_5工具层模块]]
c53e4d71: [[5_代码实现_工具层#2.1 knowledge.py]]
69b1205d: [[5_代码实现_工具层#2.2 weather.py]]
f98afd00: [[图_6Agent状态管理模块]]
8f066a98: [[图_2文档入库模块]]
dff7cf18: [[2_代码实现_文档入库#2.1 document_loader.py]]
3db70a60: [[2_代码实现_文档入库#2.2 html_processor.py]]
a5484ae5: [[2_代码实现_文档入库#2.3 embedding.py]]
f4fc7263: [[2_代码实现_文档入库#2.4 milvus_writer.py]]
03e5fa12: [[图_1数据存储模块]]
8c50dca8: [[图_1数据存储模块]]
723caaa3: [[1_项目设计#1）env.py]]
60082795: [[1_项目设计#1）database.py]]
8cf86ae0: [[1_项目设计#2）cache.py]]
8c31231f: [[1_项目设计#1）models.py]]
8d78ad09: [[1_项目设计#1）milvus_client.py]]
6ea2c68c: [[1_项目设计#2）parent_chunk_store.py]]
9dbc7095: [[图_4RAG工作流模块]]
00cb1f57: [[4_代码实现_RAG Pipeline#2.1 pipeline.py]]
f68fe7a4: [[图_3检索模块]]
c281b186: [[3_代码实现_检索模块#2.1 utils.py]]
%%
Drawing
1 | N4IgLgngDgpiBcIYA8DGBDANgSwCYCd0B3EAGhADcZ8BnbAewDsEAmcm+gV31TkQAswYKDXgB6MSgw4CxAHSp6AWzJJMMJTEZgaCANqg8CEABYYuAGYBWAGwmTqyLGNgUYVcgRWADOQgIAZityIjwwfgQADh9yfhhsAHNBLxs5AHZydEYE9QRfEBowfHoAaxgAYXpMenxjAGIARhYA7xM01QAjdFQShOLORlxK6trEIqyaKHR8LXdyC2xMTABlSF |
%%
