在 LangChain / LangGraph 里做到逐段溯源的本质是把元数据当作一等公民从入库一直传递到最终回答的每一个句子——而不是等回答生成完再去反查。一旦链路中任何一环丢了chunk_id/page/source溯源就断了后面再怎么补都补不回来。 核心设计思路溯源系统的本质是建立一条不可断裂的引用链原始文档 → Chunk带稳定ID元数据→ 向量库 → 检索 → 上下文编号 → LLM 生成带 [N] 标记的回答 → 后解析 → 映射回原文片段其中有两个致命细节决定了溯源能否真正落地元数据必须在切分时继承到每个 chunk且带start_index和稳定chunk_id这是事后定位的物理坐标LLM 生成时必须边答边标[1][2]而不是答完再贴来源——后者必然错位⚠️ 经验之谈单纯依赖 LLM 自发的引用是不可信的。生产环境中必须把引用关系做在应用层应用层维护 chunk→编号的映射LLM 只负责在合适的位置输出[N]标记。 步骤一入库阶段 —— 给每个 chunk 打上物理坐标溯源的最底层是切分这一步。LangChain 的RecursiveCharacterTextSplitter会把父文档的 metadata 继承到每个子 chunk同时建议你手动加上稳定chunk_id。fromlangchain_core.documentsimportDocumentfromlangchain.text_splitterimportRecursiveCharacterTextSplitterfromlangchain_community.document_loadersimportPyPDFLoaderdefingest_with_full_metadata(pdf_path:str)-list[Document]:# 1. 加载文档loaderPyPDFLoader(pdf_path)pagesloader.load()# 2. 注入业务元数据项目/部门/密级等fori,docinenumerate(pages):doc.metadata.update({source:pdf_path,page:doc.metadata.get(page,i),project:ProjectAlpha,department:Legal,})# 3. 切分 —— 关键点add_start_indexTrue 会记录每个 chunk 在原文档中的起始偏移splitterRecursiveCharacterTextSplitter(chunk_size512,chunk_overlap50,add_start_indexTrue,# ← 溯源的关键记录字符偏移length_functionlen,)splitssplitter.split_documents(pages)# 4. 给每个 chunk 分配全局稳定 ID# 生产建议tenant_id kb_id file_hash chunk_indexsource_counts:dict[str,int]{}forchunkinsplits:srcchunk.metadata.get(source,unknown)idxsource_counts.get(src,0)source_counts[src]idx1chunk.metadata[chunk_id]f{src}::{idx:04d}# 此时 chunk.metadata 示例# {# source: contract_2024.pdf,# page: 3,# chunk_id: contract_2024.pdf::0001,# start_index: 2048,# project: ProjectAlpha,# department: Legal# }returnsplits为什么要chunk_idstart_index双保险chunk_id跨会话、跨查询的稳定引用句柄可用于缓存/去重/增量更新start_indexchunk 在原文件中的字符偏移可在前端做点击引用跳转到原文位置 步骤二检索阶段 —— 元数据必须跟着向量一起返回向量库存入时带着 metadata检索时也必须原样返回。最容易踩的坑是检索函数返回了{content, source}但进 LLM prompt 时只拼接了contentmetadata 留在检索器内部——溯源链在这一环断了。fromlangchain_community.vectorstoresimportChromafromlangchain_openaiimportOpenAIEmbeddings embeddingsOpenAIEmbeddings(modeltext-embedding-3-small)vectorstoreChroma.from_documents(documentssplits,embeddingembeddings,collection_namekb_with_trace,persist_directory./chroma_store,)defretrieve_with_evidence(query:str,top_k:int4)-list[dict]:检索并返回带完整元数据的 evidence 列表resultsvectorstore.similarity_search_with_score(query,ktop_k)evidence[]fori,(doc,score)inenumerate(results):evidence.append({ref_id:i1,# 给 LLM 看的引用编号 [1][2]...chunk_id:doc.metadata[chunk_id],source:doc.metadata[source],page:doc.metadata[page],start_index:doc.metadata.get(start_index,-1),score:float(score),content:doc.page_content,metadata:doc.metadata,# 完整元数据透传})returnevidence 步骤三上下文构造 —— 把 evidence 编号后喂给 LLM这是溯源的桥evidence 的顺序就是引用编号[1][2][3]的顺序。LLM 只需在生成时输出[1]这样的标记应用层就能 O(1) 映射回原文。defbuild_context_prompt(evidence:list[dict])-str:把 evidence 渲染成带编号的上下文块blocks[]forevinevidence:blocks.append(f[{ev[ref_id]}]{ev[content]}\nf └─ 来源:{ev[source]}| 页码:{ev[page]}f| chunk_id:{ev[chunk_id]})return\n\n.join(blocks)✍️ 步骤四提示词设计 —— 强制边答边标提示词必须非常明确地要求 LLM在生成每一句事实陈述时就地标注[N]。这是溯源是否可用的关键。fromlangchain_core.promptsimportChatPromptTemplate CITATION_PROMPTChatPromptTemplate.from_messages([(system, 你是企业知识问答助手。仅基于【上下文】中的证据作答遵循以下规则 1. 每条事实陈述必须以 [N] 形式标注其来源编号例如根据公司规定[D1]员工… 2. 如果一句话综合了多个来源标注所有相关编号如 [D1][D2] 3. 如果上下文中没有相关信息回答未在提供的资料中找到依据禁止编造 4. 回答结构 - 一句话结论 - 分点事实依据每点带引用 - 风险与边界 5. 禁止输出 [N] 以外的来源编号N 必须是上下文中真实存在的编号 【上下文】 {context} ),(human,用户问题{question}),]) 这个提示词模板借鉴了生产级可追溯 Agent 的设计——关键不是答完再贴来源而是生成过程中逐条标注。⚙️ 步骤五在 LangGraph 中编排 —— 把 evidence 放进 StateLangGraph 的优势在于State 是所有节点共享的黑板。我们把retrieved_evidence放进 State保证 retrieve 节点 → answer 节点之间的溯源链不会断。fromtypingimportAnnotated,TypedDictfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.graph.messageimportadd_messagesclassQAState(TypedDict):question:strplanned_terms:list[str]retrieval_context:list[dict]# ← 关键检索证据存进 Stateanswer:strcitations:list[dict]# ← 解析后的引用映射trace:list[dict]# 调试用每步的快照defnode_retrieve(state:QAState)-dict:检索节点生成 evidence 并存入 Statequery .join(state.get(planned_terms,[]))orstate[question]evidenceretrieve_with_evidence(query,top_k4)return{retrieval_context:evidence,trace:[{step:retrieve,evidence_count:len(evidence)}]}defnode_answer(state:QAState)-dict:回答节点基于 evidence 生成带引用的答案evidencestate[retrieval_context]context_strbuild_context_prompt(evidence)chainCITATION_PROMPT|llm|StrOutputParser()raw_answerchain.invoke({context:context_str,question:state[question]})# 解析 [N] 标记建立句子 → 原文 chunk的映射citationsparse_citations(raw_answer,evidence)return{answer:raw_answer,citations:citations,trace:state.get(trace,[])[{step:answer,cited_chunk_ids:[c[chunk_id]forcincitations]}]}# 编排builderStateGraph(QAState)builder.add_node(retrieve,node_retrieve)builder.add_node(answer,node_answer)builder.add_edge(START,retrieve)builder.add_edge(retrieve,answer)builder.add_edge(answer,END)appbuilder.compile() 步骤六后解析 —— 把[N]映射回原文片段LLM 输出[1][2]后应用层必须做后解析建立最终的句子 ↔ 原文映射表。这一步是排查错误的关键。importredefparse_citations(answer:str,evidence:list[dict])-list[dict]:解析回答中的 [N] 标记映射回 evidence# 建立 ref_id → evidence 的字典ref_map{ev[ref_id]:evforevinevidence}# 按句子切分逐句找出它引用了哪些来源sentencesre.split(r(?[。]),answer)citation_records[]forsentinsentences:sentsent.strip()ifnotsent:continue# 找出该句中出现的所有 [N]refs[int(m)forminre.findall(r\[(\d)\],sent)]forref_idinrefs:evref_map.get(ref_id)ifev:citation_records.append({sentence:sent,chunk_id:ev[chunk_id],source:ev[source],page:ev[page],start_index:ev[start_index],score:ev[score],original_text:ev[content],})returncitation_records解析完成后citation_records就是一张完整的溯源表回答中的句子原文 chunk_id文件页码原文片段“根据规定[D1]员工…”contract.pdf::0001contract.pdf15“员工离职需提前30天…”“年假政策[S2]规定…”policy.pdf::0007policy.pdf3“正式员工年假不少于5天…”一旦用户反馈这句话不对你可以直接定位到原文片段核实而不是盲猜。 进阶多 Agent 场景下的溯源如果你的系统是多 Agent 协作如 ServiceNow 的客户成功 Agent、VC 研究 Agent溯源会更复杂——每个子 Agent 都可能调用不同工具、检索不同知识源。关键模式来自 LangChain 官方可审计 Agent 案例每个研究节点只基于自己检索到的 evidence 写作并把### Citations列表附在输出末尾合成节点synthesizer无工具、只组装——它组合各节点的输出因此每句引用都能追到某个节点的检索结果State 中用 reducer 合并并发写入避免并行节点写冲突# 多 Agent 场景下的 State 设计要点classMemoState(TypedDict):company:strresearch_output:Annotated[dict,lambdaa,b:{**a,**b}]# reducer 合并citations:list[dict]# 全局引用汇总LangSmith 的 trace 能力可以在这种多 Agent 场景下提供逐步的 input/output/context 快照帮你把成品 memo 中的任意一句话追溯到具体的搜索结果。 最终总结要把 LangChain/LangGraph Agent 的回答做到逐段溯源本质上是在全链路维护一条元数据生命线️ 架构层面的三个铁律入库即打源chunk_id稳定 IDstart_index偏移source/page物理坐标三者缺一不可State 中传 evidenceLangGraph 的 State 是溯源链载体retrieve 节点的输出必须原样传给 answer 节点应用层管引用映射LLM 只输出[N]标记由应用层维护编号 ↔ chunk的字典——不信任 LLM 自发生成的引用 实施层面的六个步骤步骤关键动作易错点1. 入库add_start_indexTrue 注入chunk_id忘了加chunk_id后续无法稳定引用2. 检索返回完整 metadata不只是 content只取 text 导致 metadata 丢失3. 上下文evidence 按顺序编号[1][2][3]编号顺序与 evidence 列表不一致4. 提示词强制边答边标[N]提示词含糊LLM 答完才补引用5. 编排retrieval_context存入 LangGraph State节点间用消息传递 evidence断了链6. 后解析正则提取[N]建立句子↔原文映射表没做后解析前端无法展示引用 生产环境的三条建议评估与调试用 LangSmith它能在每次 invoke 时记录每个节点、每次工具调用、每次 LLM 交互的 input/output/context是 Agent 调试的X 光机多源回答要做引用粒度控制UI 上不要把所有 chunk 都堆给用户按引用频率/score 排序展示** tenant 隔离**多租户场景下 metadata 必须带tenant_id避免跨租户溯源污染 最后一句经验溯源系统不是功能而是基础设施。它在系统正确时毫无存在感但在出事时是唯一的救命稻草——所以宁可过度设计不可缺失。