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 状态设计：别把整个对话历史塞进 State
LangGraph 的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 定义，再自动导出 schema：
from pydantic import BaseModel, Field from typing import Literal class QueryOrderArgs(BaseModel): order_id: str = Field( ..., pattern=r"^\d{12}$", description="12位纯数字订单号，例如 202405120001。若用户未提供，禁止猜测。" ) detail_level: Literal["summary", "full"] = Field( default="summary", description="summary 只返回状态，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% | 固定，尽量精简 |
| 工具 schema | 15% | 工具多时按需动态注入 |
| 对话历史 | 20% | 滑动窗口 + 摘要 |
| 检索 chunk | 40% | 按 rerank 分数动态裁剪 |
| 工具结果 | 10% | 超长结果先摘要 |
| 输出预留 | 5% | 防止生成被截断 |
这个比例不是死的，但思路是固定的：先扣掉固定开销，再按优先级分配剩余预算。检索 chunk 占大头是因为它直接决定回答质量，但也要按分数裁剪，低分 chunk 该丢就丢。
4.2 检索 chunk 的动态裁剪：按分数断崖，不按固定条数
新手常写top_k=5，不管分数高低都塞 5 条。问题是有些查询只有 1 条真正相关，另外 4 条是噪声，塞进去反而干扰模型。我的做法是：
- 先取
top_k=10做 rerank。 - 计算相邻 chunk 的分数差，找到第一个“断崖”（分数骤降超过阈值）。
- 断崖之前的所有 chunk 保留，之后全部丢弃。
- 如果断崖不明显，最多保留 5 条。
def dynamic_cut(chunks, max_keep=5, drop_ratio=0.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_recent=3): 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_ascii=False) 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_by="sentence"和split_length控制。对于结构化文档（Markdown、HTML），先按标题切，再按段落切，效果比纯字数切好很多。
5.3 混合检索：BM25 和向量检索的权重怎么调
纯向量检索对专有名词、编号、代码不敏感，纯 BM25 对语义改写不敏感。生产环境我基本都用混合检索，权重从 0.5:0.5 起步，再根据评测集调。
| 场景 | BM25 权重 | 向量权重 | 说明 |
|---|---|---|---|
| 客服 FAQ | 0.3 | 0.7 | 用户问法多样，语义为主 |
| 技术文档 | 0.5 | 0.5 | 术语和语义都重要 |
| 订单/编号查询 | 0.7 | 0.3 | 精确匹配为主 |
调权重一定要有评测集，别凭感觉。我一般准备 50 到 100 条“问题-正确文档”对，跑一遍看 hit rate，再微调。
5.4 查询改写：把用户的话翻译成检索友好的话
用户问“我上周买的东西咋还没到”，直接拿去检索大概率召回一堆无关内容。查询改写要做两件事：一是补全指代（“那个东西”变成具体商品或订单），二是拆解意图（“没到”对应“物流延迟”）。
LangGraph 的analyze节点里，我会先做一次轻量改写，再决定检索策略。改写用便宜模型，prompt 里明确要求“输出检索用的关键词，不要回答”。
REWRITE_PROMPT = """将用户问题改写为适合检索的查询。 要求： 1. 补全指代，保留关键实体（订单号、商品名、日期）。 2. 输出 1-3 个检索查询，每行一个。 3. 不要回答问题，只输出查询。 用户问题：{query} """5.5 命中率评测：没有评测集，调优就是盲人摸象
我见过太多团队凭感觉调 RAG，改完不知道是变好还是变坏。最低成本的评测方法是：从真实日志里抽 100 条问题，人工标注正确文档，然后每次改动跑一遍，记录 hit rate@5 和 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_k=10)) pipe.add_component("ranker", TransformersSimilarityRanker(top_k=5)) pipe.connect("embedder.embedding", "retriever.query_embedding") pipe.connect("retriever.documents", "ranker.documents") return pipe这个流水线输入查询文本，输出 rerank 后的 chunk。注意top_k分两段：检索阶段取 10，rerank 阶段取 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 结果喂给 LangGraph
def 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 或本体增强——没有评测基线，加什么都是玄学。
