生产级RAG实战：工具合约与上下文工程，让Agent敢上线

1. 从玩具到产线为什么第二篇要死磕“工具合约”和“上下文工程”如果你已经跟着第一篇把 Haystack 的Pipeline和 LangGraph 的StateGraph跑通了大概率会经历一个很典型的心理落差demo 里问“公司年假多少天”检索、拼 prompt、调 LLM答案漂漂亮亮可一旦把知识库换成真实业务文档把工具从“计算器”换成“查订单”“发工单”“改地址”系统立刻开始胡说八道——要么该调工具的时候不调要么参数传错要么把三段互相矛盾的历史对话全塞进上下文token 烧得飞快答案还越来越离谱。这不是模型不行而是生产级 RAG 和 demo 级 RAG 之间隔着的两道墙工具合约Tool Contract和上下文工程Context Engineering。第一篇解决的是“能跑”这一篇解决的是“敢上线”。Haystack 负责把检索、排序、生成这些环节做成可替换、可观测的组件流水线LangGraph 负责把“什么时候检索、什么时候调工具、什么时候追问用户、什么时候兜底”这套控制流显式地画成图。两者拼在一起才是一个能扛住真实流量的 Agentic RAG 骨架。这篇文章适合三类人一是已经把基础 RAG 跑通、但被工具调用和上下文爆炸折磨过的工程师二是正在选型 LLM 应用框架、想搞清楚 Haystack 和 LangGraph 各自边界的技术负责人三是做知识库产品、需要把“检索命中率”和“回答可解释性”同时抓起来的从业者。全文围绕工具合约设计、上下文预算分配、状态机编排、检索瓶颈排查四条主线展开所有代码和参数都给到能直接抄的程度也会把我在实际项目里踩过的坑原样摊开。2. 整体架构再校准Haystack 管“取”LangGraph 管“断”2.1 两个框架的职责边界别混着用很多人上手就把 LangGraph 当成万能胶把检索逻辑也写进节点里结果图越画越大检索组件没法单独替换、没法单独评测。我的做法是划一条清晰的线Haystack 负责“确定性数据流”文档清洗、切分、embedding、向量检索、BM25 混合检索、rerank、prompt 组装。这些环节输入输出明确适合用Pipeline串起来每个组件可以单独换、单独测。LangGraph 负责“不确定性控制流”判断是否需要检索、判断是否需要调工具、工具调用失败后重试还是追问、多轮对话里哪些历史该保留。这些是带分支、带循环、带状态的决策用图来表达最清楚。这么分的直接好处是检索效果不好时我只动 Haystack 那一侧不用碰图控制逻辑要改时我只动 LangGraph不用重新评测检索。两边通过一个薄薄的适配层通信——把 Haystack 的检索结果封装成 LangGraph 节点能消费的结构仅此而已。2.2 状态设计别把整个对话历史塞进 StateLangGraph 的State是整个图的共享内存新手最容易犯的错就是messages: list一路 append几十轮之后 state 里堆了几万 token。我的状态结构是这样的from typing import TypedDict, Annotated, Literal from langgraph.graph.message import add_messages class RAGState(TypedDict): messages: Annotated[list, add_messages] # 只保留最近 N 轮 摘要 query: str # 当前轮改写后的查询 retrieved: list[dict] # 本轮检索到的 chunk tool_calls: list[dict] # 本轮待执行的工具调用 tool_results: list[dict] # 工具返回结果 context_budget: int # 本轮剩余 token 预算 route: Literal[retrieve, tool, answer, clarify]关键点在于context_budget这个字段。它不是一个装饰而是整个上下文工程的“总闸”。每次进入节点前先算预算超了就压缩、截断或丢弃低优先级内容而不是无脑拼 prompt。messages用add_messages做增量合并但我在写回时会做一次“滑动窗口 摘要”处理保证它不会无限膨胀。2.3 图的主干四个节点撑起一轮问答主干图我固定成四个节点加一个条件路由analyze分析用户输入判断意图决定走检索、工具、直接回答还是追问。retrieve调用 Haystack 检索流水线拿回带分数的 chunk。tool_exec执行工具调用处理参数校验和错误。generate组装最终上下文调 LLM 生成回答。route_after_analyze条件边根据analyze的输出决定下一跳。这套结构看起来朴素但覆盖了 90% 的生产场景。复杂的是每个节点内部的细节下面逐个拆。3. 工具合约让 LLM 调工具不再靠“祈祷”3.1 什么是工具合约为什么它是生产级的分水岭工具合约就是一份机器可读、人类可审、运行时强校验的工具说明书。它至少包含四部分工具名、用途描述、参数 schema、返回结构。很多人只写了前两部分就丢给 LLM结果模型凭感觉编参数后端一调就 500。我见过最典型的翻车一个“查询订单状态”的工具参数是order_id描述只写了“查询订单”。模型在用户说“我上周买的那个东西到哪了”时直接把整句话塞进order_id后端解析失败整个图崩掉。如果合约里写清楚order_id必须是 12 位纯数字、格式示例、以及“如果用户没提供订单号应先追问”模型的行为会稳定得多。3.2 用 Pydantic 定义参数 schema别手写 JSON手写 JSON Schema 容易漏字段、漏约束。我用 Pydantic 定义再自动导出 schemafrom pydantic import BaseModel, Field from typing import Literal class QueryOrderArgs(BaseModel): order_id: str Field( ..., patternr^\d{12}$, description12位纯数字订单号例如 202405120001。若用户未提供禁止猜测。 ) detail_level: Literal[summary, full] Field( defaultsummary, descriptionsummary 只返回状态full 返回物流明细。 ) class QueryOrderResult(BaseModel): status: str updated_at: str logistics: list[dict] | None Nonepattern和description是给模型看的约束Literal限定枚举值default减少模型必须填的字段。实测下来加了pattern之后模型乱填订单号的概率从三成降到几乎为零。3.3 工具描述怎么写三个关键信息缺一不可工具描述不是写给人看的文档是写给模型看的“使用条件”。我总结了一个模板每个工具描述必须回答三个问题什么时候用明确触发条件比如“当用户询问订单物流状态且已提供订单号时使用”。什么时候不用明确排除条件比如“用户只是咨询退货政策时不要调用本工具”。失败怎么办比如“若返回 not_found应提示用户核对订单号而不是重试”。这三条写清楚模型在路由节点里的判断准确率会明显提升。我做过对比同一批测试用例描述完整的工具调用准确率比只写“查询订单”高出 40% 以上。3.4 参数校验放在图里不要只靠模型自觉模型再听话也会出错所以工具执行节点必须做二次校验。我的tool_exec节点逻辑是def tool_exec(state: RAGState): results [] for call in state[tool_calls]: tool TOOL_REGISTRY.get(call[name]) if not tool: results.append({error: unknown_tool, name: call[name]}) continue try: args tool.args_model(**call[args]) # Pydantic 强校验 except ValidationError as e: results.append({error: invalid_args, detail: str(e)}) continue results.append(tool.run(args)) return {tool_results: results}校验失败不抛异常而是把错误信息作为工具结果写回 state让generate节点决定是追问用户还是换个方式回答。这样图不会因为一次参数错误就中断用户体验是连贯的。注意工具执行一定要设超时和重试上限。我一般给外部 API 工具设 3 秒超时、最多重试 1 次超过就返回降级结果绝不让图卡死。4. 上下文工程token 预算怎么花比检索本身更影响体验4.1 上下文不是越多越好先算清楚预算一个 8K 上下文的模型实际可用空间远没有 8K。系统提示词、工具 schema、对话历史、检索 chunk、工具结果、输出预留每一项都在抢预算。我的分配策略是这样的以 8K 为例内容类型预算占比说明系统提示词10%固定尽量精简工具 schema15%工具多时按需动态注入对话历史20%滑动窗口 摘要检索 chunk40%按 rerank 分数动态裁剪工具结果10%超长结果先摘要输出预留5%防止生成被截断这个比例不是死的但思路是固定的先扣掉固定开销再按优先级分配剩余预算。检索 chunk 占大头是因为它直接决定回答质量但也要按分数裁剪低分 chunk 该丢就丢。4.2 检索 chunk 的动态裁剪按分数断崖不按固定条数新手常写top_k5不管分数高低都塞 5 条。问题是有些查询只有 1 条真正相关另外 4 条是噪声塞进去反而干扰模型。我的做法是先取top_k10做 rerank。计算相邻 chunk 的分数差找到第一个“断崖”分数骤降超过阈值。断崖之前的所有 chunk 保留之后全部丢弃。如果断崖不明显最多保留 5 条。def dynamic_cut(chunks, max_keep5, drop_ratio0.5): if not chunks: return [] kept [chunks[0]] for prev, cur in zip(chunks, chunks[1:]): if len(kept) max_keep: break if cur[score] prev[score] * (1 - drop_ratio): break kept.append(cur) return kept这个逻辑实测比固定 top_k 稳得多尤其在知识库文档质量参差不齐的时候。4.3 对话历史的压缩摘要 关键实体保留多轮对话里历史不能全留也不能全丢。我的策略是保留最近 3 轮原文更早的做摘要同时把关键实体订单号、用户 ID、日期单独抽出来放进 state不依赖模型从历史里回忆。def compress_history(messages, keep_recent3): if len(messages) keep_recent * 2: return messages old messages[:-keep_recent * 2] recent messages[-keep_recent * 2:] summary llm_summarize(old) # 用便宜的小模型做摘要 return [{role: system, content: f历史摘要{summary}}] recent摘要用便宜的小模型做别用主力模型成本差好几倍。关键实体抽取可以用规则 小模型抽出来存进 state 的独立字段生成时直接注入比让模型从摘要里找靠谱得多。4.4 工具结果的注入先摘要再进上下文工具返回的结果经常很长比如物流明细几十条。直接塞进上下文会挤爆预算。我的做法是工具结果先经过一个“结果处理器”按detail_level决定返回摘要还是明细明细超过阈值就先做结构化摘要。def process_tool_result(result, budget): text json.dumps(result, ensure_asciiFalse) if len(text) budget: return text return summarize_tool_result(result) # 保留关键字段丢弃冗余这一步很多人省掉结果就是工具一调上下文直接爆模型开始丢前面的检索内容回答质量断崖式下跌。5. 检索瓶颈排查命中率上不去先别怪模型5.1 RAG 的瓶颈通常不在生成在检索我做过统计RAG 回答错误里超过六成是检索没召回正确内容而不是模型不会答。所以排查顺序永远是先看检索再看 prompt最后才怀疑模型。检索瓶颈常见的有四类切分粒度不对、embedding 模型不匹配、查询和文档语义鸿沟、混合检索权重失衡。下面逐个说排查方法。5.2 切分粒度按语义切别按固定字数切固定 512 字切分是最省事也最容易出问题的做法。它会把一个完整段落从中间切断导致 chunk 语义不完整。我优先用语义切分按标题层级、按段落、按句子边界切再控制单 chunk 在 300 到 800 字之间。Haystack 里可以用DocumentSplitter配合split_bysentence和split_length控制。对于结构化文档Markdown、HTML先按标题切再按段落切效果比纯字数切好很多。5.3 混合检索BM25 和向量检索的权重怎么调纯向量检索对专有名词、编号、代码不敏感纯 BM25 对语义改写不敏感。生产环境我基本都用混合检索权重从 0.5:0.5 起步再根据评测集调。场景BM25 权重向量权重说明客服 FAQ0.30.7用户问法多样语义为主技术文档0.50.5术语和语义都重要订单/编号查询0.70.3精确匹配为主调权重一定要有评测集别凭感觉。我一般准备 50 到 100 条“问题-正确文档”对跑一遍看 hit rate再微调。5.4 查询改写把用户的话翻译成检索友好的话用户问“我上周买的东西咋还没到”直接拿去检索大概率召回一堆无关内容。查询改写要做两件事一是补全指代“那个东西”变成具体商品或订单二是拆解意图“没到”对应“物流延迟”。LangGraph 的analyze节点里我会先做一次轻量改写再决定检索策略。改写用便宜模型prompt 里明确要求“输出检索用的关键词不要回答”。REWRITE_PROMPT 将用户问题改写为适合检索的查询。 要求 1. 补全指代保留关键实体订单号、商品名、日期。 2. 输出 1-3 个检索查询每行一个。 3. 不要回答问题只输出查询。 用户问题{query} 5.5 命中率评测没有评测集调优就是盲人摸象我见过太多团队凭感觉调 RAG改完不知道是变好还是变坏。最低成本的评测方法是从真实日志里抽 100 条问题人工标注正确文档然后每次改动跑一遍记录 hit rate5 和 MRR。这两个指标比“感觉回答变好了”靠谱一万倍。6. 常见问题与排查技巧实录6.1 工具调用相关的高频问题现象可能原因排查方法解决该调工具时不调工具描述触发条件不清看 analyze 节点输出补全“什么时候用”参数格式错误schema 约束缺失看校验错误日志加 pattern、枚举重复调用同一工具图里没有终止条件看 state 里 tool_calls加最大调用次数工具结果被忽略结果没进 generate 上下文打印最终 prompt检查注入逻辑6.2 上下文相关的高频问题上下文爆炸的典型症状是回答开始丢前面的信息、token 消耗异常高、生成被截断。排查顺序是先打印最终 prompt 看长度再看各部分占比找到超预算的那一块。我遇到最多的是工具结果没摘要其次是历史没压缩。另一个隐蔽问题是“上下文污染”检索回来的 chunk 里混入了和当前问题无关但语义相近的内容模型被带偏。解决办法是提高 rerank 阈值宁可少召回也别召回错的。6.3 我踩过的三个坑第一个坑把工具 schema 全量注入每次请求工具一多光 schema 就吃掉几千 token。后来改成按analyze节点判断的意图动态注入相关工具token 省了一半。第二个坑检索 chunk 不做去重同一段内容因为切分重叠被召回多次浪费预算还干扰模型。后来在 Haystack 流水线里加了去重组件按内容哈希去重。第三个坑图里没有全局超时某个工具卡住整个请求就挂起。后来给每个节点加了超时超时就走降级分支返回“稍后再试”而不是一直转圈。提示上线前一定要做一轮“恶意输入”测试比如空查询、超长查询、纯符号查询看图和检索流水线会不会崩。这些边界情况在真实流量里一定会出现。7. 把两条流水线接起来一个可复现的最小骨架7.1 Haystack 检索流水线的组装from haystack import Pipeline from haystack.components.retrievers import InMemoryEmbeddingRetriever from haystack.components.rankers import TransformersSimilarityRanker def build_retrieval_pipeline(document_store, embedder): pipe Pipeline() pipe.add_component(embedder, embedder) pipe.add_component(retriever, InMemoryEmbeddingRetriever(document_store, top_k10)) pipe.add_component(ranker, TransformersSimilarityRanker(top_k5)) pipe.connect(embedder.embedding, retriever.query_embedding) pipe.connect(retriever.documents, ranker.documents) return pipe这个流水线输入查询文本输出 rerank 后的 chunk。注意top_k分两段检索阶段取 10rerank 阶段取 5给动态裁剪留空间。7.2 LangGraph 图的组装from langgraph.graph import StateGraph, END def build_graph(): g StateGraph(RAGState) g.add_node(analyze, analyze_node) g.add_node(retrieve, retrieve_node) g.add_node(tool_exec, tool_exec_node) g.add_node(generate, generate_node) g.set_entry_point(analyze) g.add_conditional_edges(analyze, route_after_analyze, { retrieve: retrieve, tool: tool_exec, answer: generate, clarify: generate, }) g.add_edge(retrieve, generate) g.add_edge(tool_exec, generate) g.add_edge(generate, END) return g.compile()route_after_analyze返回字符串决定下一跳。clarify和answer都走generate但 prompt 不同由 state 里的route字段区分。7.3 适配层把 Haystack 结果喂给 LangGraphdef retrieve_node(state: RAGState): result retrieval_pipeline.run({embedder: {text: state[query]}}) chunks result[ranker][documents] kept dynamic_cut([{text: d.content, score: d.score} for d in chunks]) return {retrieved: kept}适配层只做两件事调用流水线、把结果转成 state 能消费的结构。保持薄别在这里塞业务逻辑。8. 上线前必须做的三件事第一件是压测上下文预算。用真实日志里的长对话、长文档跑一遍看有没有超预算的情况超了是截断还是报错行为要明确。第二件是工具调用的幂等性检查。查询类工具无所谓但写操作类工具发工单、改地址必须幂等否则重试会导致重复操作。我的做法是给每个写操作生成一个请求 ID后端按 ID 去重。第三件是降级路径演练。检索挂了怎么办、工具超时怎么办、模型限流怎么办每条降级路径都要手动触发一次确认返回的是用户能理解的提示而不是堆栈信息。这三件事做完这套 RAG 才算真正具备上生产的资格。至于后续怎么扩展我个人的做法是先把评测集建起来再考虑加 GraphRAG 或本体增强——没有评测基线加什么都是玄学。