1. 写在本系列第四篇前面为什么还要聊架构本系列前三篇我们聊得比较基础把 RAG 的完整链路拆开过了一遍文档加载、文本切分、向量化、召回、基础提示词设计也分享了不少关于 embedding 选型和重排的踩坑记录。到第四篇我决定把主题收窄到真正影响系统能不能上生产的三件事用 Haystack 稳定住检索流水用 LangGraph 编排决策路径再用工具合约和上下文工程把模型的行为框在可控边界里。很多同学本地跑通了一个 LangGraph 教程里的 Demo就以为万事大吉结果数据量一上来幻觉、超时、工具调用错乱全来了。这篇的目的就是帮你把这些坑提前填上。我会假定你已经大概知道 LangGraph 的 StateGraph 是什么也见过 Haystack 的 Pipeline 长什么样。如果你第一次听说这两个东西建议先看一下官方示例再回来。这篇文章不是入门教程而是一份从 Demo 走向生产级流水的经验总结。写的过程中我会尽量把每一步“为什么这么做”讲清楚而不是只给你一段能跑就完事的代码。1.1 从“能跑”到“生产级”差在哪太多人把“用 FastAPI 包一层接口”当成生产级。我承认这是第一步但离真正的生产级还差得很远。我一般会用三个维度来快速判断一个 RAG 项目是不是成熟错误处理是否有兜底、运行过程是否可观测、架构能否在不重写的前提下演进。这三个维度缺一个后面都会很难受。错误处理不是给每个函数套个 try-except 就完了。你需要考虑模型返回非法 JSON、向量库连接突然断开、工具调用超过超时时间这些高并发下必然会出现的情况。我的习惯是给每条链路预设失败路径检索失败就降级成纯语言模型回答并明确告诉用户“当前没有检索到资料”这比直接抛 500 要安全得多工具调用失败就把错误原因写进状态让生成节点基于错误信息做澄清回答而不是静默吞掉。这些兜底逻辑必须在架构设计时留出位置而不是出了问题再补。可观测性是生产级的硬指标。没有 trace 的 AI 服务线上出问题只能靠猜。我要求每一步都留下记录检索耗时、文档命中数、模型生成 token 数、工具调用参数和结果。至少要能回答三个问题这个请求为什么回答成这样哪个环节最慢用户问法变了之后效果为什么波动这么大前面加了再强的模型如果回答不了这三个问题依然是盲人摸象。架构能否持续演进则是很多项目最容易忽略的。我见过把文档切分、向量化、检索、对话生成全部写在一个 service 类里的项目一开始只有五百条数据看起来还行到后面要换向量库、加新工具、支持多租户每一步都像在拆雷。生产级系统一定是有边界的模块组合每个模块能被独立替换。这也是我坚持把 Haystack 和 LangGraph 分开的原因。1.2 Haystack 和 LangGraph 的定位拆解先说结论这两个工具不是同一个赛道上的竞品它们更像是供应链里的两段工序。Haystack 的核心优势是组件化 Pipeline尤其擅长文档处理、索引和检索。它自带一堆预处理组件比如 HTML 清理、按语义切分、去重、元数据过滤而且节点替换成本很低。今天用 OpenAI 的 embedding明天换成国产模型只需要换一个 Embedder 节点后面的检索、存储逻辑一概不用动。LangGraph 的核心优势是状态机式的图编排。它把一次对话或一次任务定义为状态在节点之间流转节点可以决定走哪条边也可以并行执行。这对于工具调用、多轮对话、条件分支这种复杂流程非常友好。比如一个客服机器人用户问完天气又查日历可能还要调用工单系统这种流程用固定顺序的 Pipeline 会写得很痛苦但在 LangGraph 里就是几个节点和条件边的事。我当前推荐的分工是离线索引、在线检索这些偏确定性、高吞吐的任务交给 Haystack意图识别、工具选择、多步推理这些偏决策、高灵活性的任务交给 LangGraph。两个框架通过普通的数据结构衔接Haystack 返回的Documents列表可以直接存进 LangGraph 状态也可以经过一个压缩节点变成摘要再进入提示词。下面揣上一张简单的分工表方便还没想清楚的同学对照能力维度HaystackLangGraph文档清洗与切分强组件丰富弱需自己实现检索与相似度召回强天然组件化弱需外接意图判断与条件分支弱不擅长强状态图天然支持工具调用与外部交互弱偏批处理强节点编排灵活多轮对话状态管理弱强状态持久化可替换性和可测试性强节点独立中等但图结构清晰表格只是一个参考。如果你的业务场景主要是“给定问题固定检索并生成”只用 Haystack 也完全够如果是 Agent 型应用多工具、多轮、需要决策那 LangGraph 几乎是必选项。两者结合起来基本能覆盖绝大多数生产级 RAG 和智能体场景。2. 生产级 RAG 的上下文工程不只是把文档塞进 prompt上下文工程这个词最近很热但我发现大多数人对它的理解还停留在“多塞几篇文档进去”。实际上上下文工程的核心是在有限的 token 预算里把“模型身份、用户意图、证据材料”三件事以正确的结构和权重组织起来。做不好上下文工程换更大的模型也只能缓解表象不能解决根因。2.1 先定规则context 的 key / query / value 三段式我习惯把一次 LLM 请求的上下文拆成三个角色key、query、value。这个思路跟向量数据库里的 key-query-value 有一点像但这里对应的是 prompt 里的不同信息源。key 是模型应该扮演的角色和遵守的边界比如“你是企业客服助手只回答员工手册范围内的问题不知道就说不知道”query 是当前用户的问题经过必要改写后填入value 是检索回来的文档、工具返回的结果等外部证据。为什么要分这么清楚因为实战里我踩过太多“角色混淆”的坑。比如把检索文档和对话历史放在一起模型就开始分不清哪些是用户说的、哪些是资料说的再比如 system prompt 里混入了大量业务规则用户问题在后半段出现模型在长上下文里很容易“迷失中间”。把三者分隔开并放在合适的位置比调任何 prompt 技巧都管用。具体到排版我通常会把 system 内容放在最前检索文档放在用户消息前的独立区块用户 query 放最后。如果检索文档很多宁可截断也不要让它们淹没用户问题。一份典型的 prompt 结构可以理解成最前面是身份证中间是证据卷宗最后是传票。模型先明确身份再看证据最后响应问题这样它的回答天然就有据可依。举个例子假设用户问“本月的项目周报什么时候提交”。system 里写的身份是“项目管理助手只依据知识库回答”documents 里放了三篇关于周报制度的文档query 区保留原始问题。模型看到的结构是我是项目管理助手 → 我的知识库里有三份规定 → 用户想知道提交时间。它大概率会用文档里的时间条款回答而不是凭训练记忆编一个时间。这就是三段式的作用。2.2 用 Chat Template 固化上下文策略上下文策略不是每次动态发散而是应该像代码规范一样固定下来。我自己写 RAG 服务时一定会把 prompt 模板抽成独立的.j2或.jinja2文件而不是散落在业务代码里手拼字符串。用模板的好处是格式统一、可测试、可做版本管理换模型时也只需要换渲染逻辑不用改业务代码。一个简单但实用的模板会包含三个区域system、documents、query。其中documents里的每个文档用索引标号包起来方便后续在回答中引用。下面是我在项目里常用的一种结构system {{ system_prompt }} /system {% if documents %} documents {% for doc in documents %} doc id{{ loop.index0 }} {{ doc.content }} /doc {% endfor %} /documents {% endif %} query {{ query }} /query这个模板在公司内部知识库问答里跑得很稳最主要原因是模型可以清楚看到文档边界。后来我又往上加了meta区块用来放业务线标识、置信度阈值、引用格式要求效果也不错。模板一旦固定后面做评测、做回归测试就方便很多因为输入输出结构是确定的。使用模板时还有一个细节如果检索结果为空就不要渲染documents区块而是让它完全消失。很多模型在空标签面前会强行脑补内容倒不如干脆不给它机会。同样system prompt 也不能写“如果文档为空请告诉用户”这种模板化话术很容易让模型在本来有文档的情况下也画蛇添足。2.3 上下文窗口预算token 的计算与裁剪策略大模型上下文窗口看起来很大但真正能高效利用的区域其实有限。我一般会在生成之前做一个 token 预算控制避免把窗口塞满。假设模型窗口是 8192我会先预留 512 到 1024 token 给回答生成再预留一部分给必要的格式开销真正能用于文档内容的大概只有 6000 左右。如果检索回来 30 篇文档每篇都很长全塞进去不仅浪费资源还会让模型混淆优先级。所以我通常会在检索节点和生成节点之间加一个“上下文压缩”步骤。压缩策略按优先级执行先按重排分数或相似度保留前 5 到 8 篇文档再根据 token 数从低优先级的文档开始裁剪每篇只保留与 query 最相关的一个段落而不是全文对话历史只保留最近两轮system prompt 原则上不动。这个策略听起来简单但对回答质量的影响非常明显。如果需要程序化实现可以用模型自己的 tokenizer 统计并迭代截断。下面是我的一个简化版本from transformers import AutoTokenizer tokenizer AutoTokenizer.from_pretrained(your-model-id) def fit_context(docs, query, system, max_context_tokens6000): budget max_context_tokens parts [(system, system), (query, query)] for doc in docs: current len(tokenizer.encode(doc)) if budget - current 200: continue parts.append((doc, doc)) budget - current return \n.join(f{name}\n{content}\n/{name} for name, content in parts)这个函数没有做细粒度的段落截断但已经能保证每次请求不会超限。真正进生产时我还会加一个“最高优先文档永远放最后”的小技巧因为很多模型对最后一段内容的注意力更高。把这些规则沉淀成代码比每次靠感觉调 prompt 要可靠得多。3. 工具合约让模型安全、可预期地调用外部能力工具调用是 Agent 时代的核心能力但也是引入不确定性的最大来源。模型输出一个函数名和几个参数系统就真的去执行一个外部动作这在生产环境里是有风险的事。工具合约的价值就是把这种不确定性约束在一个可校验、可审计的范围内。3.1 工具合约到底是什么工具合约是对“模型可调用函数”的结构化描述不只是 OpenAI 工具调用里的 JSON Schema。它至少包含四部分函数名和用途声明、参数 Schema、执行前置条件、返回结果描述。前置条件往往最容易被忽略比如“发送邮件”这个工具必须校验收件人域名是否在白名单比如“查询订单”工具必须校验用户有没有权限。这些规则如果在合约里写清楚模型就不会在参数里编造一个越权身份去调用。我一直把工具合约比作前后端接口文档。你不希望前端随意给后端传一个千奇百怪的对象同样你也不希望 LLM 拿着自由生成的参数去调外部服务。合约就像一层防波堤模型生成参数后系统要先做 schema 校验、权限校验、格式规范再审慎执行。这也是为什么我会坚持在代码里用 Pydantic 定义工具入参而不是让模型直接返回一个字典。一个完整的工具合约大概长这样class SendMailArgs(BaseModel): to: str subject: str body: str urgent: bool False看起来简单但配合字段描述后价值就出来了。我会把参数描述写得很口语化比如to字段写成“收件人邮箱必须是企业通讯录内的邮箱否则校验失败”这比只写“邮箱地址”好一万倍因为模型会参考描述来生成合理的值。合约越清晰模型越不容易自由发挥。这里再给一份对应的 JSON Schema 示例方便你用原生方式接入模型服务商{ type: object, properties: { to: { type: string, description: 收件人邮箱必须是企业通讯录内的邮箱否则校验失败 }, subject: { type: string, description: 邮件主题不超过 100 字 }, body: { type: string, description: 邮件正文内容 }, urgent: { type: boolean, description: 是否标记为紧急邮件 } }, required: [to, subject, body] }记住一个原则永远不要在合约里写模型无法理解或无法验证的自由文本。比如“根据当前时间决定”这种描述除非你真的在参数里传了当前时间否则模型会瞎猜。合约里的每个字都可以被模型读取并影响行为所以描述质量直接决定工具调用质量。3.2 在 LangGraph 中落地工具节点LangGraph 里落地工具合约我习惯把“工具注册表”和“工具执行节点”分开。注册表维护每个工具的名称、描述、参数 Schema、执行函数和前置条件验证函数。执行节点读取 LLM 生成的工具调用列表逐一查注册表、校验参数、执行、回填状态。这样每次新增工具都只需要往注册表里加一项不需要改动图结构。下面是一个简化但完整的工具节点示例import json from typing import TypedDict from langgraph.graph import StateGraph, START, END from pydantic import BaseModel, ValidationError class ToolRegistry: def __init__(self): self._tools {} def register(self, name, description, schema, handler, validateNone): self._tools[name] { description: description, schema: schema, handler: handler, validate: validate, } def execute(self, name, raw_args): spec self._tools.get(name) if not spec: return {status: error, error: funknown tool {name}} try: args spec[schema].model_validate(raw_args) if spec[validate]: spec[validate](args) result spec[handler](args) return {status: ok, result: result} except ValidationError as e: return {status: error, error: finvalid args: {e}} except Exception as e: return {status: error, error: str(e)} registry ToolRegistry() def tool_node(state): tool_calls state.get(tool_calls, []) results [] for call in tool_calls: name call.get(name) raw_args call.get(arguments, {}) results.append(registry.execute(name, raw_args)) return {**state, tool_results: results}这个节点的关键是即使参数校验失败也不会直接让程序崩溃而是把错误信息返回给后续生成节点。模型看到错误后可以重新生成修正后的工具调用。这个循环往复的过程正是 LangGraph 的长项因为状态里始终保留着历史工具调用和错误信息模型能看清自己刚才到底错在哪。如果你想给 LLM 透出工具列表可以参考这样转成 schemadef to_tool_spec(name, registry): spec registry._tools[name] model spec[schema] props {} required [] for field_name, field in model.model_fields.items(): props[field_name] {type: string} if field.description: props[field_name][description] field.description if field.is_required(): required.append(field_name) return { type: function, function: { name: name, description: spec[description], parameters: {type: object, properties: props, required: required}, }, }这段代码省略了 int、boolean 等类型的转换真实场景里需要根据field.annotation做映射。但只要把to_tool_spec写好后面加多少工具都不怕它始终能保证发给模型的 schema 是服务商能接受的格式。3.3 常见坑provider rejected the request schema or tool payload跑 LangGraph 或 LangChain 时接入某些模型服务偶尔会看到provider rejected the request schema or tool payload这一类报错。初看像模型接口挂了其实绝大多数是工具 Schema 不符合服务商要求。我遇到最多的情况有三个。第一直接把 Pydantic 模型序列化作为工具 schema 传给服务商结果里面带上了$defs或default服务商不支持。第二数组参数没有写items导致参数结构不完整。第三一次性塞了二三十个工具服务商限制了单次请求的工具数量。解决办法也很套路自己写一个简单的 schema 转换函数把 Pydantic 模型转化成服务商能接受的简化 JSON Schema只保留type、properties、required、description并且不要使用default。同时控制单个请求的工具数量超过 10 个就考虑分组。实测下来这个转换函数能消掉九成的“provider rejected”问题剩下的要么是 API key 权限要么是工具描述里带了服务商禁用的字符。如果你的项目里确实需要很多工具我建议先做一个“工具选择器”节点。这个节点先用一个轻量模型看到用户的 query从工具注册表里选出最相关的 5 个工具再把它们传给真正执行工具调用的模型。这样既绕过了工具数量限制也能提高主模型的选择准确率因为工具列表变短了。4. 实战Haystack LangGraph 的最小生产骨架前两章讲的是方法和原理这一章给一套可以直接拿来改的最小骨架。我不会贴一个动辄几百行的完整项目而是把每个关键环节的代码片段和责任边界说清楚。你可以把它当成脚手架按业务需要往里添加节点。4.1 依赖与项目结构我建议的基础依赖如下放在requirements.txt里haystack-ai2.0 langgraph0.2 fastapi0.115 uvicorn0.30 pydantic2.7 transformers4.40 tiktoken版本号只是参考实际以你当前项目的兼容性为准。LangGraph 的 API 迭代特别快有时候官方示例的写法过几周就变了所以如果你要用我列出的写法最好固定住大版本。项目目录我一般按下面这样拆任务边界清晰project/ ├── app.py # FastAPI 入口创建 app 并加载编译后的 graph ├── indexer.py # Haystack 索引流水 ├── retriever.py # Haystack 检索服务封装 ├── graph.py # LangGraph 图定义和编译入口 ├── context.py # 上下文组装与 token 预算控制 ├── tools/ │ ├── __init__.py │ ├── registry.py # 工具注册表 │ ├── calendar.py │ └── search.py └── config.py # 模型名、密钥、向量库连接等配置这个结构不是死的但至少保证了“索引”“检索”“编排”“工具”“上下文”各自有家。后面做测试、加监控、换模型时都只需要动对应模块。4.2 索引流水Haystack 负责离线清洗和向量化Haystack 的 Pipeline 非常适合离线文档处理。我可以把一批 PDF、Markdown、HTML 丢进去经过清理、切分、向量化、写入向量库整个过程可以定时跑也可以在文档更新时触发。如果文档里有很多图片我通常会让一个多模态模型先生成图片描述再和正文一起走后续步骤。这就回答了很多人问的“RAG 知识库能否存储图片”的问题不能直接存二进制但可以通过图文映射把图片变成文本和 URL 存入元数据检索时再按 URL 回显图片。下面是一个简化版索引流水from haystack import Pipeline from haystack.components.preprocessors import DocumentCleaner, DocumentSplitter from haystack.components.embedders import OpenAITextEmbedder from haystack.components.writers import DocumentWriter from haystack_integrations.document_stores.pgvector import PgvectorDocumentStore store PgvectorDocumentStore( embedding_dim1536, table_namedoc_embeddings, ) index_pipe Pipeline() index_pipe.add_component(cleaner, DocumentCleaner()) index_pipe.add_component(splitter, DocumentSplitter(split_bysentence, split_length8, split_overlap2)) index_pipe.add_component(embedder, OpenAITextEmbedder(modeltext-embedding-3-small)) index_pipe.add_component(writer, DocumentWriter(document_storestore)) index_pipe.connect(cleaner.documents, splitter.documents) index_pipe.connect(splitter.documents, embedder.documents) index_pipe.connect(embedder.documents, writer.documents)这里我特意设置了split_overlap2也就是相邻切块之间保留两句重叠。这个参数很重要它能让跨句子边界的语义不会因为切分而断裂。另一个关键点是 writer 节点会把splitter传来的文档连同元数据一起写进向量库。元数据里我至少存三个字段原始文件名、来源链接、文档类型这样生成节点可以在回答里带出处。索引流水跑完后我会单独写一个检索封装模块。它不直接调用 Haystack 的 Pipeline而是把向量检索、关键词检索、重排封装成一个函数方便 LangGraph 节点调用。例如def haystack_retrieve(query: str, top_k: int 8): query_embedding embedder.run(query) vector_hits store.search(query_embedding, top_ktop_k, filters{tenant: current_tenant}) keyword_hits bm25_search(query, top_ktop_k) # 合并去重并排序 return rerank_and_merge(vector_hits, keyword_hits, query)这样可以随时替换检索实现而不用改图代码。4.3 在线编排LangGraph 构建 RAG 工具调用 Agent在线编排层我用 LangGraph 的 StateGraph 把智能体的流程定义成一张图。最基本的两个节点是检索和生成但如果要支持工具调用我还会加一个意图路由节点。它的工作很简单把用户问题喂给 LLM让它输出一个结构化的意图比如need_search、need_tool、direct_answer然后根据意图走不同分支。这里给出一个带条件路由的示例from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list, add_messages] retrieved_docs: list tool_results: list def route(state: State): # 调用一个小模型判断是否需要检索/工具 decision llm_judge(state[messages][-1].content) if decision retrieve: return retrieve elif decision tool: return tools else: return respond graph StateGraph(State) graph.add_node(retrieve, retrieve_node) graph.add_node(tools, tool_node) graph.add_node(respond, respond_node) graph.add_conditional_edges(START, route) graph.add_edge(retrieve, respond) graph.add_edge(tools, respond) graph.add_edge(respond, END) app graph.compile()这里的add_messages是为了让消息列表自动累积不会互相覆盖。实际项目中retrieve_node通常会调用我们在 retriever.py 里封装好的 Haystack 检索服务把结果转换成上下文区块。respond_node则负责把系统提示词、文档、工具结果和用户问题一起交给最终模型。这个过程在外人看来有点绕但在图上却非常直观。如果要对外提供服务我会用 FastAPI 把编译后的app包一层这样 HTTP 进来后调用一次app.invoke()再把最终messages里的回答返回给前端。注意 FastAPI 层不要放任何业务逻辑只做参数校验和响应格式转换避免把服务层也搞成一团乱麻。4.4 观测Trace、延迟与成本生产级离不开观测。用 LangGraph 本身可以接 LangSmith但很多企业数据隐私不允许把请求发到第三方所以我通常自己写一个简单的 trace 装饰器。每个节点执行时记录名称、耗时、输入摘要和输出摘要。最终答案生成后再统一计算 token 消耗和成本估算。一个轻量做法是在节点函数外面包一个通用 wrapperimport time, logging def traced(node_func): def wrapper(state): start time.perf_counter() try: result node_func(state) return result finally: elapsed time.time() - start logging.info(node%s elapsed%.3f input_keys%s, node_func.__name__, elapsed, list(state.keys())) return wrapper日志采集到位后我建议每天做一个聚合报表检索节点平均耗时、生成节点 P95 耗时、工具调用成功率、上下文裁剪掉了多少文档。这些数字能直观告诉你系统哪里需要优化而不是靠感觉调参。成本方面把 input token 和 output token 分别乘单价再加向量库检索费用就能得到单次请求的真实成本。很多项目做到后面成本瓶颈往往不是模型贵而是上下文里塞了太多无效内容裁剪策略反而成了省钱利器。5. 常见问题与排查技巧这一章是我自己趟过雷之后沉淀下来的速查表也是被问得最多的一些问题。如果你正在跑 LangGraph 或 RAG 项目大概率会碰到下面的某个坑。5.1 LangGraph 状态更新的并发问题LangGraph 在并行节点或工具同时返回时状态字段覆盖问题非常常见。比如并行调用“查询日历”和“查询邮件”两个节点都返回字典默认情况下后执行完成的节点会直接覆盖前一个节点对results字段的写入。如果你的生成节点只读results那等于是丢了数据。解决办法有两个一是给每个工具调用独立结果字段比如calendar_result、mail_result二是把状态字段定义成可合并类型用Annotated[list, add_messages]这类自定义 reducer。我在生产里更倾向于前者因为逻辑更直白排障时一眼就能看到是哪个工具返回失败。如果你用了后者注意 reducer 要放在状态定义里而不是节点函数里。5.2 检索结果不够精准怎么办很多团队遇到 RAG 效果差第一反应是换更好的 embedding 模型。但我在排查时一定会先把问题定位在“没召回”还是“召回没用上”。做法是拿一个样本问题打印出检索返回的前十篇文档标题和片段人工判断里面有没有答案。如果没有答案问题在索引侧比如切分太碎、关键词被拆开、文档元数据过滤条件太严、向量库索引参数不对。如果有答案但模型答错那问题在上下文组装和生成策略比如文档顺序不对、上下文被无关内容稀释、prompt 没有强调引用来源。这两个方向排查思路完全不同不要一上来就换模型。我常用的一个补救技巧是混合检索把向量召回和 BM25 关键词召回的 union 或 rank 融合对于专有名词和型号这种纯文本场景特别有效。一个极简的混合召回策略可以写成把向量检索的结果和 BM25 的结果合并到同一列表按分数归一化后相加取前 top_k。很多向量库已经内置了带 BM25 的 hybrid search直接用即可。如果没有就自己维护一个倒排索引量小的时候效果也不差。5.3 工具调用不稳定的原因如果模型经常不调工具或者调错参数我首先看的是模型本身的工具调用能力其次才是提示词。不同模型在 Function Calling 上的表现差异很大有些模型非常适合工具调用有些则更适合把工具描述写进普通文本让模型自由生成 JSON。后者需要你自己做输出解析和错误重试复杂度会高不少。在模型能力没问题的情况下工具描述和参数描述是最大的变量。我见过一个失败的案例一个update_order工具的notes字段描述写的是“备注信息”模型就经常把用户无关的话填进去。改成“仅用于记录本次更新操作的人工审核备注不得包含用户原始诉求”后错误率立刻下降。工具数量也要克制一次请求最多给模型 5 到 8 个工具再多就分组。分组方式可以放在同一个模型请求里让模型先选方向再选具体工具。5.4 缓存与随机性让问题可复现生产环境最怕的是一会儿对一会儿不对。我强烈建议把 LLM 的温度设为 0并开启可视化的缓存策略。语义缓存可以用向量相似度做直接把用户问题 embedding 后去缓存库找如果命中了就不需要再走完整检索和生成链路。缓存 key 里最好包含改写后的 query、检索命中文档的哈希、上下文模板版本。这样只要这几个要素没变回答就是稳定可复现的。缓存策略还能顺带降低成本和延迟。在一些查询密集型场景命中率能到 30% 以上那省下来的钱和响应时间非常可观。当然如果业务要求回答必须是最新数据缓存过期时间要设得很短或者干脆不做缓存只做 trace 复现。我自己的实现是先在 Redis 里存语义向量和回答查询时用一个similarity_threshold去卡。命中后直接返回答案并打上一个cache_hittrue的标记。这个标记很重要因为评测时需要排除缓存命中的样本只看真正走完整链路的请求效果。6. 我的几条实践建议最后聊点虚的但很可能对你有用。技术方案可以照抄但工程观可能才是决定项目能做多远的因素。6.1 先画状态图再写代码我从第二版 LangGraph 项目开始养成了个习惯写任何编排代码之前先在白板上把状态图画出来。不需要画得规范只需要标明输入从哪进、有哪些节点、失败路径怎么走、状态里存什么。这个动作帮我省掉至少三分之一的重构时间。因为一旦你把图画出来你就会发现很多逻辑在设计阶段就能推翻。比如“用户问题进来永远先调用一次检索”这个直觉在图上走一遍你会发现很多问题根本不需要检索纯调用工具或者直接回答更便宜更快。边界和异常也看得更清楚。6.2 把合约当文档而不是代码工具合约不是写完了就完事的代码它更像是前后端之间的接口文档。我会在新增工具时先在合约描述里写清楚这个工具是干什么的、什么条件下调用、参数边界是什么。这些描述不仅给 LLM 看也给人看。半年后回来维护的人包括我自己都能靠这份合约快速接手。如果你用 Pydantic 定义合约记得把每个字段的注释写完整。这个注释在很多模型服务商那里会自动转成工具 schema 的一部分直接影响模型对参数的理解。你多写两句边界条件比反复调 prompt 有效得多。6.3 这条路线后续还能怎么玩这套 Haystack LangGraph 的骨架我已经在好几个真实项目里复用后续还有几个方向值得探索。一个是多模态 RAG用多模态模型生成图像和表格的描述再统一进入检索解决非结构化数据占比越来越高的现实问题。另一个是用反馈数据驱动工具选择把历史请求里的工具调用成功率、用户满意度变成 low-level 排序信号。还有就是把状态图持久化到数据库让 Agent 在跨会话、跨设备的情况下保留长期记忆而不是每次请求都从零开始。这些方向都不会推翻现有架构只会在节点和状态层面增加内容。我自己的体会是AI 应用工程化最难的不是会用某个框架而是能设计出稳定的、可观测的、可演进的结构。希望这一篇能帮你把结构搭稳少踩几个我踩过的坑。
1. 写在本系列第四篇前面为什么还要聊架构本系列前三篇我们聊得比较基础把 RAG 的完整链路拆开过了一遍文档加载、文本切分、向量化、召回、基础提示词设计也分享了不少关于 embedding 选型和重排的踩坑记录。到第四篇我决定把主题收窄到真正影响系统能不能上生产的三件事用 Haystack 稳定住检索流水用 LangGraph 编排决策路径再用工具合约和上下文工程把模型的行为框在可控边界里。很多同学本地跑通了一个 LangGraph 教程里的 Demo就以为万事大吉结果数据量一上来幻觉、超时、工具调用错乱全来了。这篇的目的就是帮你把这些坑提前填上。我会假定你已经大概知道 LangGraph 的 StateGraph 是什么也见过 Haystack 的 Pipeline 长什么样。如果你第一次听说这两个东西建议先看一下官方示例再回来。这篇文章不是入门教程而是一份从 Demo 走向生产级流水的经验总结。写的过程中我会尽量把每一步“为什么这么做”讲清楚而不是只给你一段能跑就完事的代码。1.1 从“能跑”到“生产级”差在哪太多人把“用 FastAPI 包一层接口”当成生产级。我承认这是第一步但离真正的生产级还差得很远。我一般会用三个维度来快速判断一个 RAG 项目是不是成熟错误处理是否有兜底、运行过程是否可观测、架构能否在不重写的前提下演进。这三个维度缺一个后面都会很难受。错误处理不是给每个函数套个 try-except 就完了。你需要考虑模型返回非法 JSON、向量库连接突然断开、工具调用超过超时时间这些高并发下必然会出现的情况。我的习惯是给每条链路预设失败路径检索失败就降级成纯语言模型回答并明确告诉用户“当前没有检索到资料”这比直接抛 500 要安全得多工具调用失败就把错误原因写进状态让生成节点基于错误信息做澄清回答而不是静默吞掉。这些兜底逻辑必须在架构设计时留出位置而不是出了问题再补。可观测性是生产级的硬指标。没有 trace 的 AI 服务线上出问题只能靠猜。我要求每一步都留下记录检索耗时、文档命中数、模型生成 token 数、工具调用参数和结果。至少要能回答三个问题这个请求为什么回答成这样哪个环节最慢用户问法变了之后效果为什么波动这么大前面加了再强的模型如果回答不了这三个问题依然是盲人摸象。架构能否持续演进则是很多项目最容易忽略的。我见过把文档切分、向量化、检索、对话生成全部写在一个 service 类里的项目一开始只有五百条数据看起来还行到后面要换向量库、加新工具、支持多租户每一步都像在拆雷。生产级系统一定是有边界的模块组合每个模块能被独立替换。这也是我坚持把 Haystack 和 LangGraph 分开的原因。1.2 Haystack 和 LangGraph 的定位拆解先说结论这两个工具不是同一个赛道上的竞品它们更像是供应链里的两段工序。Haystack 的核心优势是组件化 Pipeline尤其擅长文档处理、索引和检索。它自带一堆预处理组件比如 HTML 清理、按语义切分、去重、元数据过滤而且节点替换成本很低。今天用 OpenAI 的 embedding明天换成国产模型只需要换一个 Embedder 节点后面的检索、存储逻辑一概不用动。LangGraph 的核心优势是状态机式的图编排。它把一次对话或一次任务定义为状态在节点之间流转节点可以决定走哪条边也可以并行执行。这对于工具调用、多轮对话、条件分支这种复杂流程非常友好。比如一个客服机器人用户问完天气又查日历可能还要调用工单系统这种流程用固定顺序的 Pipeline 会写得很痛苦但在 LangGraph 里就是几个节点和条件边的事。我当前推荐的分工是离线索引、在线检索这些偏确定性、高吞吐的任务交给 Haystack意图识别、工具选择、多步推理这些偏决策、高灵活性的任务交给 LangGraph。两个框架通过普通的数据结构衔接Haystack 返回的Documents列表可以直接存进 LangGraph 状态也可以经过一个压缩节点变成摘要再进入提示词。下面揣上一张简单的分工表方便还没想清楚的同学对照能力维度HaystackLangGraph文档清洗与切分强组件丰富弱需自己实现检索与相似度召回强天然组件化弱需外接意图判断与条件分支弱不擅长强状态图天然支持工具调用与外部交互弱偏批处理强节点编排灵活多轮对话状态管理弱强状态持久化可替换性和可测试性强节点独立中等但图结构清晰表格只是一个参考。如果你的业务场景主要是“给定问题固定检索并生成”只用 Haystack 也完全够如果是 Agent 型应用多工具、多轮、需要决策那 LangGraph 几乎是必选项。两者结合起来基本能覆盖绝大多数生产级 RAG 和智能体场景。2. 生产级 RAG 的上下文工程不只是把文档塞进 prompt上下文工程这个词最近很热但我发现大多数人对它的理解还停留在“多塞几篇文档进去”。实际上上下文工程的核心是在有限的 token 预算里把“模型身份、用户意图、证据材料”三件事以正确的结构和权重组织起来。做不好上下文工程换更大的模型也只能缓解表象不能解决根因。2.1 先定规则context 的 key / query / value 三段式我习惯把一次 LLM 请求的上下文拆成三个角色key、query、value。这个思路跟向量数据库里的 key-query-value 有一点像但这里对应的是 prompt 里的不同信息源。key 是模型应该扮演的角色和遵守的边界比如“你是企业客服助手只回答员工手册范围内的问题不知道就说不知道”query 是当前用户的问题经过必要改写后填入value 是检索回来的文档、工具返回的结果等外部证据。为什么要分这么清楚因为实战里我踩过太多“角色混淆”的坑。比如把检索文档和对话历史放在一起模型就开始分不清哪些是用户说的、哪些是资料说的再比如 system prompt 里混入了大量业务规则用户问题在后半段出现模型在长上下文里很容易“迷失中间”。把三者分隔开并放在合适的位置比调任何 prompt 技巧都管用。具体到排版我通常会把 system 内容放在最前检索文档放在用户消息前的独立区块用户 query 放最后。如果检索文档很多宁可截断也不要让它们淹没用户问题。一份典型的 prompt 结构可以理解成最前面是身份证中间是证据卷宗最后是传票。模型先明确身份再看证据最后响应问题这样它的回答天然就有据可依。举个例子假设用户问“本月的项目周报什么时候提交”。system 里写的身份是“项目管理助手只依据知识库回答”documents 里放了三篇关于周报制度的文档query 区保留原始问题。模型看到的结构是我是项目管理助手 → 我的知识库里有三份规定 → 用户想知道提交时间。它大概率会用文档里的时间条款回答而不是凭训练记忆编一个时间。这就是三段式的作用。2.2 用 Chat Template 固化上下文策略上下文策略不是每次动态发散而是应该像代码规范一样固定下来。我自己写 RAG 服务时一定会把 prompt 模板抽成独立的.j2或.jinja2文件而不是散落在业务代码里手拼字符串。用模板的好处是格式统一、可测试、可做版本管理换模型时也只需要换渲染逻辑不用改业务代码。一个简单但实用的模板会包含三个区域system、documents、query。其中documents里的每个文档用索引标号包起来方便后续在回答中引用。下面是我在项目里常用的一种结构system {{ system_prompt }} /system {% if documents %} documents {% for doc in documents %} doc id{{ loop.index0 }} {{ doc.content }} /doc {% endfor %} /documents {% endif %} query {{ query }} /query这个模板在公司内部知识库问答里跑得很稳最主要原因是模型可以清楚看到文档边界。后来我又往上加了meta区块用来放业务线标识、置信度阈值、引用格式要求效果也不错。模板一旦固定后面做评测、做回归测试就方便很多因为输入输出结构是确定的。使用模板时还有一个细节如果检索结果为空就不要渲染documents区块而是让它完全消失。很多模型在空标签面前会强行脑补内容倒不如干脆不给它机会。同样system prompt 也不能写“如果文档为空请告诉用户”这种模板化话术很容易让模型在本来有文档的情况下也画蛇添足。2.3 上下文窗口预算token 的计算与裁剪策略大模型上下文窗口看起来很大但真正能高效利用的区域其实有限。我一般会在生成之前做一个 token 预算控制避免把窗口塞满。假设模型窗口是 8192我会先预留 512 到 1024 token 给回答生成再预留一部分给必要的格式开销真正能用于文档内容的大概只有 6000 左右。如果检索回来 30 篇文档每篇都很长全塞进去不仅浪费资源还会让模型混淆优先级。所以我通常会在检索节点和生成节点之间加一个“上下文压缩”步骤。压缩策略按优先级执行先按重排分数或相似度保留前 5 到 8 篇文档再根据 token 数从低优先级的文档开始裁剪每篇只保留与 query 最相关的一个段落而不是全文对话历史只保留最近两轮system prompt 原则上不动。这个策略听起来简单但对回答质量的影响非常明显。如果需要程序化实现可以用模型自己的 tokenizer 统计并迭代截断。下面是我的一个简化版本from transformers import AutoTokenizer tokenizer AutoTokenizer.from_pretrained(your-model-id) def fit_context(docs, query, system, max_context_tokens6000): budget max_context_tokens parts [(system, system), (query, query)] for doc in docs: current len(tokenizer.encode(doc)) if budget - current 200: continue parts.append((doc, doc)) budget - current return \n.join(f{name}\n{content}\n/{name} for name, content in parts)这个函数没有做细粒度的段落截断但已经能保证每次请求不会超限。真正进生产时我还会加一个“最高优先文档永远放最后”的小技巧因为很多模型对最后一段内容的注意力更高。把这些规则沉淀成代码比每次靠感觉调 prompt 要可靠得多。3. 工具合约让模型安全、可预期地调用外部能力工具调用是 Agent 时代的核心能力但也是引入不确定性的最大来源。模型输出一个函数名和几个参数系统就真的去执行一个外部动作这在生产环境里是有风险的事。工具合约的价值就是把这种不确定性约束在一个可校验、可审计的范围内。3.1 工具合约到底是什么工具合约是对“模型可调用函数”的结构化描述不只是 OpenAI 工具调用里的 JSON Schema。它至少包含四部分函数名和用途声明、参数 Schema、执行前置条件、返回结果描述。前置条件往往最容易被忽略比如“发送邮件”这个工具必须校验收件人域名是否在白名单比如“查询订单”工具必须校验用户有没有权限。这些规则如果在合约里写清楚模型就不会在参数里编造一个越权身份去调用。我一直把工具合约比作前后端接口文档。你不希望前端随意给后端传一个千奇百怪的对象同样你也不希望 LLM 拿着自由生成的参数去调外部服务。合约就像一层防波堤模型生成参数后系统要先做 schema 校验、权限校验、格式规范再审慎执行。这也是为什么我会坚持在代码里用 Pydantic 定义工具入参而不是让模型直接返回一个字典。一个完整的工具合约大概长这样class SendMailArgs(BaseModel): to: str subject: str body: str urgent: bool False看起来简单但配合字段描述后价值就出来了。我会把参数描述写得很口语化比如to字段写成“收件人邮箱必须是企业通讯录内的邮箱否则校验失败”这比只写“邮箱地址”好一万倍因为模型会参考描述来生成合理的值。合约越清晰模型越不容易自由发挥。这里再给一份对应的 JSON Schema 示例方便你用原生方式接入模型服务商{ type: object, properties: { to: { type: string, description: 收件人邮箱必须是企业通讯录内的邮箱否则校验失败 }, subject: { type: string, description: 邮件主题不超过 100 字 }, body: { type: string, description: 邮件正文内容 }, urgent: { type: boolean, description: 是否标记为紧急邮件 } }, required: [to, subject, body] }记住一个原则永远不要在合约里写模型无法理解或无法验证的自由文本。比如“根据当前时间决定”这种描述除非你真的在参数里传了当前时间否则模型会瞎猜。合约里的每个字都可以被模型读取并影响行为所以描述质量直接决定工具调用质量。3.2 在 LangGraph 中落地工具节点LangGraph 里落地工具合约我习惯把“工具注册表”和“工具执行节点”分开。注册表维护每个工具的名称、描述、参数 Schema、执行函数和前置条件验证函数。执行节点读取 LLM 生成的工具调用列表逐一查注册表、校验参数、执行、回填状态。这样每次新增工具都只需要往注册表里加一项不需要改动图结构。下面是一个简化但完整的工具节点示例import json from typing import TypedDict from langgraph.graph import StateGraph, START, END from pydantic import BaseModel, ValidationError class ToolRegistry: def __init__(self): self._tools {} def register(self, name, description, schema, handler, validateNone): self._tools[name] { description: description, schema: schema, handler: handler, validate: validate, } def execute(self, name, raw_args): spec self._tools.get(name) if not spec: return {status: error, error: funknown tool {name}} try: args spec[schema].model_validate(raw_args) if spec[validate]: spec[validate](args) result spec[handler](args) return {status: ok, result: result} except ValidationError as e: return {status: error, error: finvalid args: {e}} except Exception as e: return {status: error, error: str(e)} registry ToolRegistry() def tool_node(state): tool_calls state.get(tool_calls, []) results [] for call in tool_calls: name call.get(name) raw_args call.get(arguments, {}) results.append(registry.execute(name, raw_args)) return {**state, tool_results: results}这个节点的关键是即使参数校验失败也不会直接让程序崩溃而是把错误信息返回给后续生成节点。模型看到错误后可以重新生成修正后的工具调用。这个循环往复的过程正是 LangGraph 的长项因为状态里始终保留着历史工具调用和错误信息模型能看清自己刚才到底错在哪。如果你想给 LLM 透出工具列表可以参考这样转成 schemadef to_tool_spec(name, registry): spec registry._tools[name] model spec[schema] props {} required [] for field_name, field in model.model_fields.items(): props[field_name] {type: string} if field.description: props[field_name][description] field.description if field.is_required(): required.append(field_name) return { type: function, function: { name: name, description: spec[description], parameters: {type: object, properties: props, required: required}, }, }这段代码省略了 int、boolean 等类型的转换真实场景里需要根据field.annotation做映射。但只要把to_tool_spec写好后面加多少工具都不怕它始终能保证发给模型的 schema 是服务商能接受的格式。3.3 常见坑provider rejected the request schema or tool payload跑 LangGraph 或 LangChain 时接入某些模型服务偶尔会看到provider rejected the request schema or tool payload这一类报错。初看像模型接口挂了其实绝大多数是工具 Schema 不符合服务商要求。我遇到最多的情况有三个。第一直接把 Pydantic 模型序列化作为工具 schema 传给服务商结果里面带上了$defs或default服务商不支持。第二数组参数没有写items导致参数结构不完整。第三一次性塞了二三十个工具服务商限制了单次请求的工具数量。解决办法也很套路自己写一个简单的 schema 转换函数把 Pydantic 模型转化成服务商能接受的简化 JSON Schema只保留type、properties、required、description并且不要使用default。同时控制单个请求的工具数量超过 10 个就考虑分组。实测下来这个转换函数能消掉九成的“provider rejected”问题剩下的要么是 API key 权限要么是工具描述里带了服务商禁用的字符。如果你的项目里确实需要很多工具我建议先做一个“工具选择器”节点。这个节点先用一个轻量模型看到用户的 query从工具注册表里选出最相关的 5 个工具再把它们传给真正执行工具调用的模型。这样既绕过了工具数量限制也能提高主模型的选择准确率因为工具列表变短了。4. 实战Haystack LangGraph 的最小生产骨架前两章讲的是方法和原理这一章给一套可以直接拿来改的最小骨架。我不会贴一个动辄几百行的完整项目而是把每个关键环节的代码片段和责任边界说清楚。你可以把它当成脚手架按业务需要往里添加节点。4.1 依赖与项目结构我建议的基础依赖如下放在requirements.txt里haystack-ai2.0 langgraph0.2 fastapi0.115 uvicorn0.30 pydantic2.7 transformers4.40 tiktoken版本号只是参考实际以你当前项目的兼容性为准。LangGraph 的 API 迭代特别快有时候官方示例的写法过几周就变了所以如果你要用我列出的写法最好固定住大版本。项目目录我一般按下面这样拆任务边界清晰project/ ├── app.py # FastAPI 入口创建 app 并加载编译后的 graph ├── indexer.py # Haystack 索引流水 ├── retriever.py # Haystack 检索服务封装 ├── graph.py # LangGraph 图定义和编译入口 ├── context.py # 上下文组装与 token 预算控制 ├── tools/ │ ├── __init__.py │ ├── registry.py # 工具注册表 │ ├── calendar.py │ └── search.py └── config.py # 模型名、密钥、向量库连接等配置这个结构不是死的但至少保证了“索引”“检索”“编排”“工具”“上下文”各自有家。后面做测试、加监控、换模型时都只需要动对应模块。4.2 索引流水Haystack 负责离线清洗和向量化Haystack 的 Pipeline 非常适合离线文档处理。我可以把一批 PDF、Markdown、HTML 丢进去经过清理、切分、向量化、写入向量库整个过程可以定时跑也可以在文档更新时触发。如果文档里有很多图片我通常会让一个多模态模型先生成图片描述再和正文一起走后续步骤。这就回答了很多人问的“RAG 知识库能否存储图片”的问题不能直接存二进制但可以通过图文映射把图片变成文本和 URL 存入元数据检索时再按 URL 回显图片。下面是一个简化版索引流水from haystack import Pipeline from haystack.components.preprocessors import DocumentCleaner, DocumentSplitter from haystack.components.embedders import OpenAITextEmbedder from haystack.components.writers import DocumentWriter from haystack_integrations.document_stores.pgvector import PgvectorDocumentStore store PgvectorDocumentStore( embedding_dim1536, table_namedoc_embeddings, ) index_pipe Pipeline() index_pipe.add_component(cleaner, DocumentCleaner()) index_pipe.add_component(splitter, DocumentSplitter(split_bysentence, split_length8, split_overlap2)) index_pipe.add_component(embedder, OpenAITextEmbedder(modeltext-embedding-3-small)) index_pipe.add_component(writer, DocumentWriter(document_storestore)) index_pipe.connect(cleaner.documents, splitter.documents) index_pipe.connect(splitter.documents, embedder.documents) index_pipe.connect(embedder.documents, writer.documents)这里我特意设置了split_overlap2也就是相邻切块之间保留两句重叠。这个参数很重要它能让跨句子边界的语义不会因为切分而断裂。另一个关键点是 writer 节点会把splitter传来的文档连同元数据一起写进向量库。元数据里我至少存三个字段原始文件名、来源链接、文档类型这样生成节点可以在回答里带出处。索引流水跑完后我会单独写一个检索封装模块。它不直接调用 Haystack 的 Pipeline而是把向量检索、关键词检索、重排封装成一个函数方便 LangGraph 节点调用。例如def haystack_retrieve(query: str, top_k: int 8): query_embedding embedder.run(query) vector_hits store.search(query_embedding, top_ktop_k, filters{tenant: current_tenant}) keyword_hits bm25_search(query, top_ktop_k) # 合并去重并排序 return rerank_and_merge(vector_hits, keyword_hits, query)这样可以随时替换检索实现而不用改图代码。4.3 在线编排LangGraph 构建 RAG 工具调用 Agent在线编排层我用 LangGraph 的 StateGraph 把智能体的流程定义成一张图。最基本的两个节点是检索和生成但如果要支持工具调用我还会加一个意图路由节点。它的工作很简单把用户问题喂给 LLM让它输出一个结构化的意图比如need_search、need_tool、direct_answer然后根据意图走不同分支。这里给出一个带条件路由的示例from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list, add_messages] retrieved_docs: list tool_results: list def route(state: State): # 调用一个小模型判断是否需要检索/工具 decision llm_judge(state[messages][-1].content) if decision retrieve: return retrieve elif decision tool: return tools else: return respond graph StateGraph(State) graph.add_node(retrieve, retrieve_node) graph.add_node(tools, tool_node) graph.add_node(respond, respond_node) graph.add_conditional_edges(START, route) graph.add_edge(retrieve, respond) graph.add_edge(tools, respond) graph.add_edge(respond, END) app graph.compile()这里的add_messages是为了让消息列表自动累积不会互相覆盖。实际项目中retrieve_node通常会调用我们在 retriever.py 里封装好的 Haystack 检索服务把结果转换成上下文区块。respond_node则负责把系统提示词、文档、工具结果和用户问题一起交给最终模型。这个过程在外人看来有点绕但在图上却非常直观。如果要对外提供服务我会用 FastAPI 把编译后的app包一层这样 HTTP 进来后调用一次app.invoke()再把最终messages里的回答返回给前端。注意 FastAPI 层不要放任何业务逻辑只做参数校验和响应格式转换避免把服务层也搞成一团乱麻。4.4 观测Trace、延迟与成本生产级离不开观测。用 LangGraph 本身可以接 LangSmith但很多企业数据隐私不允许把请求发到第三方所以我通常自己写一个简单的 trace 装饰器。每个节点执行时记录名称、耗时、输入摘要和输出摘要。最终答案生成后再统一计算 token 消耗和成本估算。一个轻量做法是在节点函数外面包一个通用 wrapperimport time, logging def traced(node_func): def wrapper(state): start time.perf_counter() try: result node_func(state) return result finally: elapsed time.time() - start logging.info(node%s elapsed%.3f input_keys%s, node_func.__name__, elapsed, list(state.keys())) return wrapper日志采集到位后我建议每天做一个聚合报表检索节点平均耗时、生成节点 P95 耗时、工具调用成功率、上下文裁剪掉了多少文档。这些数字能直观告诉你系统哪里需要优化而不是靠感觉调参。成本方面把 input token 和 output token 分别乘单价再加向量库检索费用就能得到单次请求的真实成本。很多项目做到后面成本瓶颈往往不是模型贵而是上下文里塞了太多无效内容裁剪策略反而成了省钱利器。5. 常见问题与排查技巧这一章是我自己趟过雷之后沉淀下来的速查表也是被问得最多的一些问题。如果你正在跑 LangGraph 或 RAG 项目大概率会碰到下面的某个坑。5.1 LangGraph 状态更新的并发问题LangGraph 在并行节点或工具同时返回时状态字段覆盖问题非常常见。比如并行调用“查询日历”和“查询邮件”两个节点都返回字典默认情况下后执行完成的节点会直接覆盖前一个节点对results字段的写入。如果你的生成节点只读results那等于是丢了数据。解决办法有两个一是给每个工具调用独立结果字段比如calendar_result、mail_result二是把状态字段定义成可合并类型用Annotated[list, add_messages]这类自定义 reducer。我在生产里更倾向于前者因为逻辑更直白排障时一眼就能看到是哪个工具返回失败。如果你用了后者注意 reducer 要放在状态定义里而不是节点函数里。5.2 检索结果不够精准怎么办很多团队遇到 RAG 效果差第一反应是换更好的 embedding 模型。但我在排查时一定会先把问题定位在“没召回”还是“召回没用上”。做法是拿一个样本问题打印出检索返回的前十篇文档标题和片段人工判断里面有没有答案。如果没有答案问题在索引侧比如切分太碎、关键词被拆开、文档元数据过滤条件太严、向量库索引参数不对。如果有答案但模型答错那问题在上下文组装和生成策略比如文档顺序不对、上下文被无关内容稀释、prompt 没有强调引用来源。这两个方向排查思路完全不同不要一上来就换模型。我常用的一个补救技巧是混合检索把向量召回和 BM25 关键词召回的 union 或 rank 融合对于专有名词和型号这种纯文本场景特别有效。一个极简的混合召回策略可以写成把向量检索的结果和 BM25 的结果合并到同一列表按分数归一化后相加取前 top_k。很多向量库已经内置了带 BM25 的 hybrid search直接用即可。如果没有就自己维护一个倒排索引量小的时候效果也不差。5.3 工具调用不稳定的原因如果模型经常不调工具或者调错参数我首先看的是模型本身的工具调用能力其次才是提示词。不同模型在 Function Calling 上的表现差异很大有些模型非常适合工具调用有些则更适合把工具描述写进普通文本让模型自由生成 JSON。后者需要你自己做输出解析和错误重试复杂度会高不少。在模型能力没问题的情况下工具描述和参数描述是最大的变量。我见过一个失败的案例一个update_order工具的notes字段描述写的是“备注信息”模型就经常把用户无关的话填进去。改成“仅用于记录本次更新操作的人工审核备注不得包含用户原始诉求”后错误率立刻下降。工具数量也要克制一次请求最多给模型 5 到 8 个工具再多就分组。分组方式可以放在同一个模型请求里让模型先选方向再选具体工具。5.4 缓存与随机性让问题可复现生产环境最怕的是一会儿对一会儿不对。我强烈建议把 LLM 的温度设为 0并开启可视化的缓存策略。语义缓存可以用向量相似度做直接把用户问题 embedding 后去缓存库找如果命中了就不需要再走完整检索和生成链路。缓存 key 里最好包含改写后的 query、检索命中文档的哈希、上下文模板版本。这样只要这几个要素没变回答就是稳定可复现的。缓存策略还能顺带降低成本和延迟。在一些查询密集型场景命中率能到 30% 以上那省下来的钱和响应时间非常可观。当然如果业务要求回答必须是最新数据缓存过期时间要设得很短或者干脆不做缓存只做 trace 复现。我自己的实现是先在 Redis 里存语义向量和回答查询时用一个similarity_threshold去卡。命中后直接返回答案并打上一个cache_hittrue的标记。这个标记很重要因为评测时需要排除缓存命中的样本只看真正走完整链路的请求效果。6. 我的几条实践建议最后聊点虚的但很可能对你有用。技术方案可以照抄但工程观可能才是决定项目能做多远的因素。6.1 先画状态图再写代码我从第二版 LangGraph 项目开始养成了个习惯写任何编排代码之前先在白板上把状态图画出来。不需要画得规范只需要标明输入从哪进、有哪些节点、失败路径怎么走、状态里存什么。这个动作帮我省掉至少三分之一的重构时间。因为一旦你把图画出来你就会发现很多逻辑在设计阶段就能推翻。比如“用户问题进来永远先调用一次检索”这个直觉在图上走一遍你会发现很多问题根本不需要检索纯调用工具或者直接回答更便宜更快。边界和异常也看得更清楚。6.2 把合约当文档而不是代码工具合约不是写完了就完事的代码它更像是前后端之间的接口文档。我会在新增工具时先在合约描述里写清楚这个工具是干什么的、什么条件下调用、参数边界是什么。这些描述不仅给 LLM 看也给人看。半年后回来维护的人包括我自己都能靠这份合约快速接手。如果你用 Pydantic 定义合约记得把每个字段的注释写完整。这个注释在很多模型服务商那里会自动转成工具 schema 的一部分直接影响模型对参数的理解。你多写两句边界条件比反复调 prompt 有效得多。6.3 这条路线后续还能怎么玩这套 Haystack LangGraph 的骨架我已经在好几个真实项目里复用后续还有几个方向值得探索。一个是多模态 RAG用多模态模型生成图像和表格的描述再统一进入检索解决非结构化数据占比越来越高的现实问题。另一个是用反馈数据驱动工具选择把历史请求里的工具调用成功率、用户满意度变成 low-level 排序信号。还有就是把状态图持久化到数据库让 Agent 在跨会话、跨设备的情况下保留长期记忆而不是每次请求都从零开始。这些方向都不会推翻现有架构只会在节点和状态层面增加内容。我自己的体会是AI 应用工程化最难的不是会用某个框架而是能设计出稳定的、可观测的、可演进的结构。希望这一篇能帮你把结构搭稳少踩几个我踩过的坑。
