图_4RAG工作流模块
对应 [[4_代码实现_RAG Pipeline]]。根据问题复杂度和证据质量，决定返回证据、改写一次、请求澄清或报告无知识。
正常入口与两条执行路径
run_rag_graph() 初始化状态并执行 build_rag_graph() 编译出的图：
- 简单问题：首次检索 → Grader → 返回证据 / 改写一次再检索再评分 / HITL / 无知识。
- 复杂问题：复杂度规划时一并得到子问题 →
Send并行分发 → 每题固定“检索 → 评分” →synthesis()合并证据与 Trace。子问题不会单独生成回答，也不走改写循环。
HITL 的实际实现
图执行结束时只生成最小恢复快照。[[图_7Agent对话编排模块]] 把快照放进会话 metadata 的 pending_hitl；下一轮经 resume_rag_from_hitl() 拼接补充、重新检索并评分，跳过外层 Agent 和原始复杂度分解。这里没有 Checkpointer / interrupt 暂停一个仍在内存里的图。
模块边界
RAGState 属于单次图执行；ChatRequestContext 属于单次请求；PendingHitlState 才负责跨轮等待。Pipeline 输出真实 Evidence 和路由状态，不承担最终答案生成。
图形约定
绿色是目录，蓝色是文件，紫色是类，橙色是函数；标题中的名称与职责之间保留 12 个空格。类框中的编号通常对应其方法；函数框的编号描述内部步骤。没有组合，矩形宽高可分别调整。
导航
[[项目实现|查看项目实现总图]] · [[图_1数据存储模块|数据存储模块]]
Excalidraw Data
Text Elements
04 RAG工作流模块 · 实现结构 ^a8a28540
目录 → 文件 → 类 / 函数；框内编号表示局部职责，箭头表示主要调用。 ^5bc2698f
backend/ 承载本模块的后端代码 ^092fa1ad
rag/ 在检索之上组织证据工作流 ^15dd6ba3
[[4_代码实现_RAG Pipeline#2.1 pipeline.py|pipeline.py]] LangGraph 路由、评分、合成与 HITL 恢复 ^d066a5f7
RAGState 保存单次图执行的业务状态
- question / docs / route / rewrite_count / rag_trace / request_context
- sub_results 用 operator.add 汇聚并行结果 ^ae109df3
EvidenceGrade 约束 Grader 的结构化输出
- 相关性、可回答性、歧义、路由、置信度与 HITL 提示 ^6bac6eb3
ComplexityResult 约束复杂度规划的结果
- simple / complex、原因和最多 4 个子问题 ^d00bfbb9
run_rag_graph() 正常 RAG 的对外入口
- _initial_state()：建立本轮状态，注入 ChatRequestContext
- rag_graph.invoke()：执行已编译图
- 若需 HITL，附带 _build_hitl_resume_state() 最小恢复快照 ^761094bc
build_rag_graph() 把函数注册为图节点与条件边
- 入口 classify_complexity；简单 / 复杂两条分支
- 评分后仅 rewrite 才回路，其余 route 都结束
- compile()：不配置 Checkpointer，不使用 interrupt ^95378a73
路由与格式辅助： _route_after_complexity()：简单 / 复杂 _route_after_grade()：rewrite / end _format_docs()：构造 Evidence 上下文 _emit()：委托 ctx 上报 rag_step ^2d0e77a3
resume_rag_from_hitl() 用户补充后的针对性检索入口
- _state_from_resume()：校验快照，原问题拼接用户补充
- _retrieve_resume_query()：重新检索，再调用评分节点
- 仍需澄清则生成新快照；不重走复杂度与子问题规划
- 恢复快照来自 schemas/chat.py，跨轮由 Service 保存 ^40555e52
快照边界： _build_hitl_resume_state() 保存问题、路由、 检索状态、改写次数和复杂度规划信息。 原始 docs 不放进恢复快照，恢复时重新检索。 ^cbeb31bf
classify_complexity() 确定本次走哪条路径
- 明确简单意图先走本地规则，否则 FAST_MODEL 结构化判断
- 复杂问题同时规划子问题，避免再单独调用一次 LLM ^9e035c59
retrieve_initial() 简单问题首次检索
- 调用 utils.retrieve_documents() → docs / context / Trace ^716fce9d
grade_documents_node() 评估证据并确定 route
- 无 docs 直接判无知识；否则 GRADE_MODEL 结构化评分
- _resolve_route()：answer / rewrite / clarify / scope_select / no_knowledge
- 歧义触发 HITL；终止失败路径清空对外 docs / context ^b122cf60
rewrite_question_node() 用一次改写机会纠错
- 预算最多 1 次；调用 utils.rewrite_query_once()
- 记录 Step-back / HyDE 方法与新查询 ^9d8649f5
retrieve_rewritten() 用新查询重新检索
- 调用 utils.retrieve_documents()，更新证据与 Trace
- 回到 grade_documents_node()；不再无限改写 ^8b73e63a
非 rewrite 的 route → 图结束。 answer：返回证据；clarify / scope_select：请求补充； no_knowledge：返回无知识状态。 部分证据：首次可改写；预算用尽后转澄清。 ^741a6a7e
prepare_sub_questions() 使用规划器已产出的子问题
- 逐条 emit 展示子问题；不再次进行 LLM 分解 ^94eed9c2
_fanout_sub_questions() 并行分发子问题
- LangGraph Send：每题一份独立 RAGState，共享请求 ctx ^9d5f6939
rag_sub_agent() 执行每个子问题的固定短链
- retrieve_initial() → grade_documents_node()
- 保留可用 / 部分证据；不做查询改写，不生成子答案
- 把 {question, docs, status, trace} 放入 sub_results ^e061cbaa
synthesis() 汇总并行证据与状态
- 只取 answerable / partial 子结果，dedupe_documents() 去重
- 有证据返回 answer；全部无证据时汇总 HITL 或 no_knowledge
- 汇总 sub_traces；这里合成 Evidence，不生成最终答案 ^8db7a0e4
simple ^4925e46f
complex ^d10b8e00
rewrite ^017a2911
再评分 ^f5e2088b
并行结果汇合 ^b0a388d3
输出边界：docs + rag_trace + 可选 hitl_resume_state；最终回答由外层 Agent / Service 生成。 ^d8ff1f52
本图只展开当前源码可达路径；_route_after_initial() 未接入 build_rag_graph()，不画成运行节点。 ^1ceef116
Element Links
d066a5f7: [[4_代码实现_RAG Pipeline#2.1 pipeline.py]]
%%
Drawing
1 | N4IgLgngDgpiBcIYA8DGBDANgSwCYCd0B3EAGhADcZ8BnbAewDsEAmcm+gV31TkQAswYKDXgB6MSgw4CxAHSp6AWzJJMMJTEZgaCANqg8CEOgAc6FqYCsAFgAMqyLGNgUYVcgQ2r5CKx8gRHhg/F4AbACc5Pww2ADmgl5WchF2aekZ6WHk6Ixx6gh27GD49ADWMADC9Jj0+MYAxACMLADMdjYA7KoARuioZXGlnIy41bX1iCW5NFDo+Fru5ABm2J |
%%
