1. 从玩具到产线为什么第三篇才聊上下文工程前两篇我们把 Haystack 的检索管线和 LangGraph 的状态机骨架搭起来了能跑通一个“能回答问题”的 RAG Demo。但 Demo 和产线之间隔着一道很深的沟这道沟的名字就叫上下文工程。我见过太多团队卡在这一步检索召回率看着还行模型回答却总是差口气要么答非所问要么把检索到的无关片段硬塞进答案里要么多轮对话之后彻底失忆。问题往往不在模型本身而在你喂给模型的上下文到底长什么样。这一篇的核心就一件事把 Haystack 的检索能力和 LangGraph 的流程编排能力拧成一股绳构建一套生产级的 RAG 系统。所谓生产级不是指能跑就行而是指上下文可控、工具调用可约束、状态可追踪、失败可回滚。关键词里的 Haystack、LangGraph、RAG、LLM、上下文工程这五个词其实是一条链Haystack 负责把知识捞出来LangGraph 负责决定什么时候捞、捞完怎么用、用完怎么记LLM 是最终的执行者而上下文工程就是贯穿始终的那根线。这篇文章适合谁看如果你已经写过基础的 RAG 流程知道 embedding 和向量库是怎么回事但一到多轮对话、工具调用、上下文裁剪就抓瞎那这篇就是写给你的。如果你还在纠结 RAG 是什么建议先翻前两篇把基础管线跑通。下面所有内容都基于我实际项目中的踩坑经验代码和配置可以直接抄但参数你得根据自己的数据调。2. 整体架构设计Haystack 管检索LangGraph 管决策2.1 为什么不让 Haystack 一条路走到黑Haystack 的 Pipeline 很强大检索、排序、生成可以串成一条线。但生产环境里用户的问题不是每次都走同一条路。有的问题需要查知识库有的问题需要调外部 API有的问题需要先澄清再检索有的问题需要多轮检索逐步逼近。这种条件分支和循环用 Haystack 的 Pipeline 硬写会非常别扭因为它的设计哲学是线性数据流。LangGraph 恰好补上这块。它把流程建模成状态图节点是处理步骤边是跳转条件状态在节点之间传递和累积。你可以把它理解成一个专门为 LLM 应用设计的状态机。Haystack 的检索组件可以封装成 LangGraph 的一个节点检索结果写进状态下一个节点根据状态决定是直接生成还是继续检索。我试过两种方案一种是纯 Haystack 加自定义组件做分支代码写出来像意大利面另一种是 Haystack 只做检索LangGraph 做全流程编排逻辑清晰得多。实测下来第二种方案在需求变更时的修改成本低一个数量级。2.2 状态设计整个系统的记忆中枢LangGraph 的核心是状态State。状态定义得好不好直接决定后面写起来顺不顺。我的经验是状态字段要分三类输入类用户原始问题、对话历史、会话 ID中间类检索到的文档、工具调用结果、重试次数、当前步骤标记输出类最终答案、引用来源、置信度用 Python 的 TypedDict 定义状态时要注意 LangGraph 对状态更新的合并策略。默认是覆盖但对话历史这种需要追加的字段得用Annotated配合operator.add来声明累加行为。这个细节不搞清楚多轮对话会莫名其妙丢历史。from typing import TypedDict, Annotated import operator class RAGState(TypedDict): question: str chat_history: Annotated[list, operator.add] documents: list tool_results: list answer: str retry_count: int route: strretry_count这个字段看着不起眼但它是防止死循环的关键。后面讲工具调用的时候会详细说。2.3 节点划分每个节点只做一件事节点划分的原则是单一职责。我见过有人把检索和生成塞进一个节点结果调试的时候根本不知道是检索错了还是生成错了。我的划分方式是这样的路由节点判断用户意图决定走知识库检索、工具调用还是直接回答检索节点调用 Haystack 的检索管线返回文档列表相关性评估节点判断检索结果是否足够回答问题不够就触发重试或改写查询工具调用节点执行外部工具比如查天气、算数学、调 API生成节点把上下文组装好调用 LLM 生成最终答案兜底节点处理无法回答的情况给出礼貌的拒绝或转人工提示每个节点都是纯函数输入状态、输出状态更新。这样单元测试也好写单独喂一个状态进去看输出对不对就行。3. 上下文工程的核心让 LLM 看到它该看的3.1 上下文窗口不是垃圾桶很多人做 RAG 的思路是检索 Top-K 文档全部塞进 prompt让模型自己挑。这在 K 小的时候还行K 一大就出问题。模型的注意力是有限的无关信息越多关键信息被淹没的概率越大。我做过一个对比实验同样的问题塞 10 篇文档和塞 3 篇精排后的文档答案准确率差了将近 20 个百分点。上下文工程的第一原则宁缺毋滥。Haystack 的检索管线里可以加一个 Reranker用交叉编码器对初筛结果精排只保留最相关的几篇。Reranker 的推理成本比 embedding 检索高但比让 LLM 处理一堆垃圾上下文便宜多了。from haystack.components.rankers import TransformersSimilarityRanker ranker TransformersSimilarityRanker( modelcross-encoder/ms-marco-MiniLM-L-6-v2, top_k3 )top_k3是我在多数场景下的默认值。如果你的文档片段很短可以放宽到 5如果片段很长压到 2 甚至 1。这个参数没有万能值得拿你的真实 query 去测。3.2 上下文组装顺序和格式都有讲究检索到的文档怎么拼进 prompt直接影响模型的理解。我的做法是按相关性从高到低排列每篇文档前面加一个编号和来源标记方便模型引用也方便后面做溯源。[文档1] 来源产品手册.pdf 内容... [文档2] 来源FAQ.md 内容...为什么按相关性降序因为 LLM 对上下文开头和结尾的内容注意力更强中间部分容易被忽略。把最相关的放前面是顺应模型的注意力分布。还有一个细节文档之间要加明确的分隔符。我试过用空行分隔模型有时候会把两篇文档的内容混在一起。后来改用---加编号混淆的情况少了很多。3.3 对话历史的裁剪策略多轮对话场景下历史消息会越积越多迟早撑爆上下文窗口。全量保留不现实全丢又失忆。我的策略是滑动窗口加摘要最近 N 轮对话保留原文N 一般取 3 到 5更早的对话用 LLM 压缩成一段摘要放在系统提示里摘要的更新频率不用每轮都做可以每 5 轮触发一次这个策略在 LangGraph 里实现起来很自然加一个节点专门管历史裁剪在生成节点之前执行。摘要生成可以用便宜的小模型没必要上大模型。注意摘要会丢失细节如果业务对历史细节要求高比如法律咨询、医疗问诊建议保留原文但做更激进的截断或者用向量库把历史也索引起来按需检索。4. 工具合约让 LLM 调工具不翻车4.1 工具调用的本质是合约LLM 调用工具本质上是一次合约交互模型输出一个结构化的调用请求系统执行后返回结果模型再基于结果继续。这个合约要成立需要三个条件模型知道有哪些工具可用、模型知道每个工具的参数格式、系统能正确解析模型的输出。LangGraph 里工具调用通常配合 LangChain 的 Tool 抽象来做。定义一个工具就是写一个函数加一个 schemafrom langchain_core.tools import tool tool def search_knowledge_base(query: str, top_k: int 3) - list: 搜索内部知识库返回相关文档片段。 Args: query: 搜索关键词 top_k: 返回文档数量默认3 # 实际检索逻辑 return resultsdocstring 不是装饰它是给模型看的工具说明。写得越清楚模型调用越准确。我见过有人 docstring 写“搜索”模型根本不知道搜什么、怎么搜调用成功率极低。4.2 参数校验别信模型的输出模型输出的工具调用参数永远不要直接信任。它可能传错类型、漏传必填项、传超出范围的值。我的做法是在工具函数入口加一层校验tool def search_knowledge_base(query: str, top_k: int 3) - list: if not query or len(query.strip()) 0: return [{error: 查询词不能为空}] top_k max(1, min(top_k, 10)) # ...返回错误信息而不是抛异常是因为模型看到错误信息后有机会自我修正。抛异常会直接中断流程体验很差。4.3 工具调用的循环控制LangGraph 的工具调用通常是一个循环模型决定调工具执行工具结果回传给模型模型再决定是继续调还是生成最终答案。这个循环必须有终止条件否则模型可能陷入无限调用。我的做法是设一个max_iterations比如 5 次。超过就强制走生成节点把已有的工具结果汇总成答案。同时在状态里记retry_count每次工具调用加一路由节点检查这个值。def should_continue(state: RAGState) - str: if state[retry_count] 5: return generate if state.get(tool_results) and not state.get(needs_more_tools): return generate return tools这个路由函数是 LangGraph 条件边的核心返回的字符串对应下一个节点的名字。4.4 工具结果的上下文注入工具返回的结果怎么塞回上下文也有讲究。我的做法是给工具结果加一个明确的标记和检索文档区分开[工具结果] search_knowledge_base 参数{query: 退货政策, top_k: 3} 返回...这样模型能清楚知道哪些信息来自知识库哪些来自工具调用。在生成答案时如果两者冲突模型可以优先采信工具结果因为工具结果通常是实时的。5. 实操全流程从零搭一套可跑的管线5.1 环境准备与依赖安装先把依赖装齐。Haystack 和 LangGraph 的版本更新很快建议锁版本避免 API 变动导致代码跑不起来。pip install haystack-ai2.8.0 pip install langgraph0.2.60 pip install langchain-core0.3.30 pip install transformers4.47.0 pip install sentence-transformers3.3.1向量库我用的 Qdrant本地跑用 Docker 起一个就行docker run -p 6333:6333 qdrant/qdrant如果你数据量小用内存向量库也行但生产环境还是建议上独立的向量数据库持久化和并发都更好。5.2 Haystack 检索管线搭建检索管线分三步文档预处理、索引、查询。预处理阶段要把文档切块切块大小直接影响检索效果。我的经验是中文文档按 300 到 500 字切英文按 200 到 300 词切块之间留 10% 到 20% 的重叠避免关键信息被切断。from haystack import Pipeline from haystack.components.preprocessors import DocumentSplitter from haystack.components.embedders import SentenceTransformersDocumentEmbedder from haystack.components.writers import DocumentWriter from haystack.document_stores.in_memory import InMemoryDocumentStore document_store InMemoryDocumentStore() splitter DocumentSplitter(split_length400, split_overlap50) embedder SentenceTransformersDocumentEmbedder( modelBAAI/bge-small-zh-v1.5 ) writer DocumentWriter(document_storedocument_store) indexing_pipeline Pipeline() indexing_pipeline.add_component(splitter, splitter) indexing_pipeline.add_component(embedder, embedder) indexing_pipeline.add_component(writer, writer) indexing_pipeline.connect(splitter, embedder) indexing_pipeline.connect(embedder, writer)embedding 模型选bge-small-zh-v1.5是因为它在中文检索任务上表现稳定模型体积小本地跑没压力。如果你的文档以英文为主换成all-MiniLM-L6-v2就行。查询管线加一个 Rerankerfrom haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever from haystack.components.embedders import SentenceTransformersTextEmbedder from haystack.components.rankers import TransformersSimilarityRanker query_pipeline Pipeline() query_pipeline.add_component(text_embedder, SentenceTransformersTextEmbedder( modelBAAI/bge-small-zh-v1.5 )) query_pipeline.add_component(retriever, InMemoryEmbeddingRetriever( document_storedocument_store, top_k10 )) query_pipeline.add_component(ranker, TransformersSimilarityRanker( modelcross-encoder/ms-marco-MiniLM-L-6-v2, top_k3 )) query_pipeline.connect(text_embedder.embedding, retriever.query_embedding) query_pipeline.connect(retriever.documents, ranker.documents)初筛 top_k10精排后留 3这个组合在多数场景下召回和精度的平衡比较好。5.3 LangGraph 状态图组装把检索管线封装成一个节点函数def retrieve_node(state: RAGState) - dict: result query_pipeline.run({ text_embedder: {text: state[question]}, ranker: {query: state[question]} }) return {documents: result[ranker][documents]}路由节点判断意图def route_node(state: RAGState) - dict: question state[question] # 简单规则包含计算走工具否则走检索 if any(kw in question for kw in [计算, 换算, 查询天气]): return {route: tools} return {route: retrieve}生成节点组装上下文def generate_node(state: RAGState) - dict: docs state.get(documents, []) context \n\n.join([ f[文档{i1}] 来源{d.meta.get(source, 未知)}\n内容{d.content} for i, d in enumerate(docs) ]) prompt f基于以下上下文回答问题。如果上下文不足以回答请明确说明。 上下文 {context} 问题{state[question]} # 调用 LLM answer llm.invoke(prompt) return {answer: answer}组装状态图from langgraph.graph import StateGraph, END workflow StateGraph(RAGState) workflow.add_node(route, route_node) workflow.add_node(retrieve, retrieve_node) workflow.add_node(tools, tool_node) workflow.add_node(generate, generate_node) workflow.set_entry_point(route) workflow.add_conditional_edges( route, lambda s: s[route], {retrieve: retrieve, tools: tools} ) workflow.add_edge(retrieve, generate) workflow.add_edge(tools, generate) workflow.add_edge(generate, END) app workflow.compile()跑起来result app.invoke({ question: 退货政策是什么, chat_history: [], retry_count: 0 }) print(result[answer])5.4 参数调优的实操记录上面这套跑通之后接下来就是调参。我拿一个 200 条问题的测试集做了几轮对比记录如下参数初始值调整后效果变化切块长度400350召回率 3%切块重叠5070边界问题减少初筛 top_k1015召回 5%延迟 80ms精排 top_k34答案完整度提升Reranker 模型MiniLM-L6MiniLM-L12精度 2%延迟翻倍最终我选的组合是切块 350、重叠 70、初筛 15、精排 4、Reranker 用 L6。延迟和精度的平衡点因业务而异金融、医疗这种对准确性要求高的场景可以牺牲延迟换精度。6. 常见问题与排查技巧实录6.1 检索到了但模型不用这是最常见的抱怨。检索结果明明包含答案模型却视而不见。原因通常有三个一是上下文太长关键信息被淹没二是文档格式混乱模型没识别出这是有效信息三是 prompt 指令不明确模型不知道要用上下文。排查顺序先看检索结果的相关性分数如果分数很低说明检索本身有问题如果分数高但模型不用检查 prompt 里有没有明确说“基于以下上下文回答”如果 prompt 没问题试着把最相关的文档挪到上下文最前面。6.2 工具调用参数格式错误模型输出的参数 JSON 解析失败通常是这几个原因模型输出了多余的解释文字、参数类型不对、必填参数缺失。我的处理方式是在解析层加容错import json import re def parse_tool_call(text: str) - dict: # 提取 JSON 块 match re.search(r\{.*\}, text, re.DOTALL) if not match: return {error: 未找到有效参数} try: return json.loads(match.group()) except json.JSONDecodeError: return {error: 参数格式错误}返回错误而不是崩溃让模型有机会重试。6.3 多轮对话历史丢失LangGraph 的状态默认是覆盖更新如果chat_history字段没用Annotated声明累加每轮都会被新值覆盖。这个坑我踩过排查了半天才发现是状态定义的问题。检查方法很简单打印每轮的状态看chat_history长度有没有增长。6.4 常见问题速查表现象可能原因排查方向答案与问题无关检索结果不相关检查 embedding 模型和切块策略答案不完整精排 top_k 太小增大 top_k 或换更强的 Reranker工具调用死循环缺少终止条件检查 retry_count 和路由逻辑多轮对话失忆状态未累加检查 Annotated 声明响应延迟高检索或 Reranker 太慢换小模型或减少 top_k模型忽略上下文prompt 指令不清明确要求基于上下文回答6.5 几个我踩过的坑第一个坑切块的时候没考虑文档结构把表格切成了两半检索出来的片段根本没法用。后来在切块前加了一个按标题分段的预处理表格和代码块单独处理效果好很多。第二个坑Reranker 模型和 embedding 模型的语言不匹配。我用中文 embedding 配英文 Reranker精排结果惨不忍睹。后来统一用中文模型问题解决。第三个坑工具调用的返回结果太长塞进上下文后把检索文档挤没了。后来给工具结果加了长度限制超过 500 字就截断只保留关键信息。提示所有参数都不要拍脑袋定拿真实数据测。我见过有人照搬网上的配置结果在自己的数据上效果一塌糊涂。RAG 没有万能配置只有适合你数据的配置。7. 上下文工程的进阶思路7.1 动态上下文预算上下文窗口是有限资源不同问题需要的上下文量不一样。简单问题可能一篇文档就够复杂问题需要多篇。我的做法是根据问题的复杂度动态调整 top_k先用一个小模型判断问题类型事实型问题 top_k2分析型问题 top_k5对比型问题 top_k8。这个判断本身也可以用一个轻量分类器做没必要上大模型。分类器的训练数据可以从历史日志里挖标注成本不高。7.2 上下文压缩检索到的文档片段里真正有用的可能只有一两句话。把整段塞进去是浪费。可以用一个小模型做抽取式压缩只保留和问题相关的句子。Haystack 里有现成的DocumentCompressor组件也可以自己写一个基于规则或模型的压缩器。压缩的代价是可能丢信息所以压缩率不要设太高。我的经验是压缩到原文的 50% 到 70% 比较安全再低就有风险了。7.3 上下文溯源生产级 RAG 必须能溯源答案里的每句话来自哪篇文档。这不仅是合规要求也是调试的刚需。我的做法是在生成 prompt 里要求模型标注引用编号生成后再用规则解析出来映射回原始文档。prompt 基于以下上下文回答问题并在答案中用[1][2]标注引用来源。 上下文 [1] ... [2] ... 问题... 模型有时候会标错所以解析后还要做一次校验看引用的编号是否在有效范围内。7.4 上下文缓存相同或相似的问题反复检索是浪费。可以在检索节点前加一层缓存用问题 embedding 做相似度匹配命中缓存就直接返回历史结果。缓存的有效期根据知识库更新频率定更新频繁的设短一点比如 1 小时更新少的可以设 1 天。缓存要注意失效策略。知识库更新后相关缓存必须清掉否则会返回过时信息。我的做法是给每个缓存条目打上文档版本号版本变了就失效。8. 生产部署的几点经验8.1 监控指标上线之后要盯这几个指标检索召回率、答案准确率、工具调用成功率、平均响应延迟、上下文 token 数分布。召回率和准确率需要人工标注样本可以每周抽一批做评估。延迟和 token 数可以实时监控设阈值告警。我习惯在 LangGraph 的每个节点加埋点记录进入时间、退出时间、状态变化。这些数据积累下来调优的时候就有依据了。8.2 降级策略LLM 服务可能超时向量库可能挂掉工具 API 可能限流。生产系统必须有降级方案。我的做法是检索失败时返回缓存结果或提示用户稍后重试LLM 超时时返回检索到的原文片段让用户自己看工具调用失败时跳过工具只用知识库回答。降级不是失败是保证核心功能可用。用户宁愿看到一个不完美的答案也不愿意看到报错页面。8.3 灰度发布RAG 系统的改动影响面很大prompt 改一句话可能让答案风格全变。所以任何改动都要灰度发布先放 10% 流量观察指标没异常再全量。灰度期间要对比新旧版本的答案质量不能只看延迟。我一般会准备一个评估集每次改动都跑一遍看准确率有没有下降。评估集不用很大100 到 200 条就够但要覆盖主要场景。8.4 成本控制LLM 调用是主要成本。控制成本的手段有几个用便宜模型做路由和压缩只让贵模型做最终生成缓存高频问题的答案限制上下文长度避免不必要的 token 消耗。我算过一笔账一个中等规模的 RAG 系统如果不做任何优化每月 LLM 成本可能上万。加上缓存和模型分级之后能压到三分之一左右。9. 写在最后这套 Haystack 加 LangGraph 的组合我在三个项目里用过从内部知识库到客服机器人都有。最大的体会是RAG 的瓶颈从来不在模型而在上下文。模型再强喂进去的是垃圾出来的也是垃圾。上下文工程做得好小模型也能给出靠谱答案做得差大模型也救不了。如果你正在搭 RAG 系统建议先把检索质量做扎实再考虑工具调用和多轮对话。检索是地基地基不稳上面盖什么都会塌。工具调用和上下文裁剪是锦上添花不是雪中送炭。最后分享一个小技巧每次调完参数别只看一两个 case 的结果跑一遍完整的评估集。我吃过亏改了一个参数某个 case 效果好了整体准确率却掉了。单点优化不等于全局优化数据说话最靠谱。