企业官网建设流程全解析

1. 从零搭建生产级 RAG 的整体设计思路

1.1 为什么单靠向量检索撑不起生产环境

很多人第一次接触 RAG，脑子里想的都是“把文档切块、丢进向量库、检索 Top-K、拼进 Prompt”，跑个 Demo 感觉效果还行，就以为大功告成。但真到了生产环境，问题会一个接一个冒出来：用户问“上季度的退货政策跟这季度有什么区别”，向量检索返回的全是单段政策文本，模型根本没法做对比；用户问“帮我查一下订单 12345 的物流状态”，检索器压根不知道要去调接口，只会从知识库里瞎找一段相似文本糊弄过去。

这就是朴素 RAG 的瓶颈：它把“检索”等同于“语义相似度匹配”，但真实业务里的信息需求远不止“找一段相似的话”。有些问题需要跨文档聚合，有些需要实时数据，有些需要多步推理，有些需要精确匹配结构化字段。单靠一个向量索引，就像拿一把螺丝刀去修整辆车——不是不能用，是场景一复杂就歇菜。

所以生产级 RAG 的设计思路必须从“单点检索”升级为“流水线编排”。Haystack 负责把检索、排序、生成这些环节标准化成可插拔的组件，LangGraph 负责把这些组件编排成有状态、有分支、有循环的图结构。两者配合，才能覆盖从简单问答到复杂 Agent 推理的全谱系需求。

1.2 Haystack 和 LangGraph 各自扮演什么角色

Haystack 的定位是检索流水线的标准化框架。它把文档存储、检索器、排序器、Prompt 构建、生成器这些环节抽象成统一的组件接口，你可以像搭积木一样替换其中任何一块。比如今天用 BM25 做稀疏检索，明天换成 Embedding 做稠密检索，后天加一个 Cross-Encoder 做重排序，Pipeline 的骨架不用动，只换组件就行。这种设计在需要快速迭代检索策略的阶段特别省事。

LangGraph 的定位是有状态 Agent 流程的编排引擎。它把整个 RAG 流程建模成一张图，节点是具体的操作（检索、生成、工具调用、条件判断），边是节点之间的流转逻辑。关键在于它支持状态传递和条件分支：你可以在图里维护一个共享的 State 对象，每个节点读写这个 State，根据当前 State 的内容决定下一步走哪条路。这就让“先判断问题类型，再决定走检索还是走工具调用”这种逻辑变得非常自然。

两者结合的方式通常是：用 Haystack 构建底层的检索和生成组件，用 LangGraph 把这些组件包装成节点，编排成完整的 Agent 流程。Haystack 管“怎么查得准”，LangGraph 管“什么时候查、查完之后干什么”。

1.3 生产级 RAG 的四个核心模块拆解

把生产级 RAG 拆开来看，核心模块可以归为四块：

• 检索层：负责从知识库中召回候选文档。生产环境通常需要混合检索（稀疏 + 稠密），再加一层重排序来提升精度。
• 工具层：负责处理检索解决不了的问题，比如实时数据查询、结构化计算、外部 API 调用。工具层的关键是工具合约的设计——每个工具接受什么参数、返回什么格式、什么条件下触发，都要定义清楚。
• 上下文工程层：负责把检索结果、工具返回、对话历史、系统指令组装成最终送给 LLM 的上下文。这一步直接决定生成质量，但很多人恰恰在这里偷懒。
• 编排层：负责根据用户输入动态决定走哪条路径。简单问题直接检索生成，复杂问题可能需要多轮检索、工具调用、甚至自我反思。

下面这张表可以帮你快速判断自己的 RAG 系统目前处于哪个阶段：

2. 检索层的深度优化与 Haystack 组件选型

2.1 混合检索的落地细节：稀疏与稠密怎么配合

纯向量检索有个致命弱点：对精确匹配不敏感。用户搜“RFC 7231”，向量模型可能返回一堆讲 HTTP 协议的文档，但就是找不到那个编号对应的具体章节。反过来，纯 BM25 又对语义改写无能为力，用户问“怎么让网页加载更快”，BM25 可能匹配不到“前端性能优化”这种文档。

混合检索的思路是两路并行召回，然后融合排序。Haystack 里实现起来不复杂：

from haystack import Pipeline from haystack.components.retrievers import InMemoryBM25Retriever, InMemoryEmbeddingRetriever from haystack.components.joiners import DocumentJoiner pipeline = Pipeline() pipeline.add_component("bm25_retriever", InMemoryBM25Retriever(document_store=store)) pipeline.add_component("embedding_retriever", InMemoryEmbeddingRetriever(document_store=store)) pipeline.add_component("joiner", DocumentJoiner(sort_by="score", join_mode="reciprocal_rank_fusion")) pipeline.connect("bm25_retriever.documents", "joiner.documents") pipeline.connect("embedding_retriever.documents", "joiner.documents")

这里的关键参数是join_mode。reciprocal_rank_fusion（RRF）是我实测下来最稳的融合策略，它不依赖两路检索的原始分数可比性，只看排名。具体公式是score = Σ 1/(k + rank)，k 通常取 60。这样即使 BM25 的分数范围是 0-20，向量相似度是 0-1，融合后也不会出现某一路被另一路压制的情况。

注意：两路检索的 Top-K 不要设成一样的。BM25 建议取 20-30，向量检索取 10-15。原因是 BM25 的召回率高但精度低，多召回一些让后面的重排序去筛；向量检索精度相对高，取太多反而引入噪声。

2.2 重排序模型的选择与性能权衡

混合检索之后，候选文档可能有 30-50 篇，直接塞给 LLM 既浪费 Token 又稀释关键信息。重排序的作用就是在这批候选里挑出真正相关的 3-5 篇。

Haystack 支持的重排序模型主要有两类：

• Cross-Encoder 类：比如cross-encoder/ms-marco-MiniLM-L-6-v2，把 query 和 document 拼在一起送进模型打分。精度高，但速度慢，适合候选集不大的场景。
• Late Interaction 类：比如 ColBERT 风格的模型，预先计算文档的 token 级向量，查询时做 MaxSim 操作。速度快，但需要额外的索引存储。

我一般建议：如果候选集在 50 篇以内，直接用 Cross-Encoder，延迟增加 100-200ms 但精度提升明显。如果候选集上百，考虑先用轻量模型粗筛到 20 篇，再用 Cross-Encoder 精排。

from haystack.components.rankers import TransformersSimilarityRanker ranker = TransformersSimilarityRanker( model="cross-encoder/ms-marco-MiniLM-L-6-v2", top_k=5, score_threshold=0.3 )

score_threshold这个参数值得说一下。设太低会引入不相关文档，设太高可能把边缘相关的也过滤掉。我的经验是先在验证集上画一条 Precision-Recall 曲线，找到 F1 最高的阈值，然后稍微往下调 0.05，给召回留点余量。

2.3 文档切分策略：固定长度 vs 语义切分

文档切分看似简单，实则影响巨大。固定长度切分（比如每 512 token 一刀切）实现简单，但容易把一段完整的论述拦腰截断，检索时两半都不完整。

语义切分的思想是沿着文档的自然边界切，比如按段落、按标题层级、按句子边界。Haystack 提供了DocumentSplitter组件，支持按 word、sentence、passage 等粒度切分：

from haystack.components.preprocessors import DocumentSplitter splitter = DocumentSplitter( split_by="sentence", split_length=5, split_overlap=1 )

split_overlap是重叠窗口，设成 1 表示相邻块之间有一句话的重叠。这个重叠很重要，因为很多问题的答案恰好跨越两个块的边界，有重叠才能保证至少有一个块包含完整信息。

对于结构化文档（比如 Markdown 或 HTML），更好的做法是按标题层级切分，把每个小节作为一个独立的块，同时保留标题路径作为元数据。这样检索时可以用标题路径做过滤，比如“只在‘退款政策’这一节里搜”。

3. 工具合约设计：让 LLM 知道什么时候该调工具

3.1 工具合约的三要素：Key、Query、Value

工具合约的本质是告诉 LLM 三件事：我是谁（Key）、我在找什么（Query）、我能提供什么（Value）。这三者缺一不可。

很多人在定义工具时只写了功能描述，比如“查询订单状态”，但没告诉 LLM 这个工具需要什么参数、参数格式是什么、返回结果长什么样。结果 LLM 要么不敢调，要么调了之后不知道怎么处理返回值。

一个完整的工具合约应该包含：

• 工具名称：简短、唯一、语义明确，比如query_order_status而不是tool_1。
• 功能描述：一句话说明这个工具做什么，什么场景下应该用。
• 参数定义：每个参数的名称、类型、是否必填、取值范围、示例值。
• 返回格式：返回值的结构，最好给出示例。
• 触发条件：明确什么情况下应该调用这个工具，什么情况下不应该。

在 LangGraph 里，工具通常定义成带类型注解的函数，然后用@tool装饰器包装：

from langchain_core.tools import tool @tool def query_order_status(order_id: str) -> dict: """查询指定订单的当前物流状态。 适用场景：用户询问某个具体订单的配送进度、预计到达时间。 不适用场景：用户询问退货政策、支付问题等非物流问题。 Args: order_id: 订单编号，格式为 10 位数字字符串，例如 "1234567890" Returns: {"status": "运输中", "location": "杭州转运中心", "eta": "2024-01-15"} """ # 实际调用内部 API return internal_api.get_order_status(order_id)

3.2 工具调用的触发判断：规则、模型还是混合

LLM 判断是否调用工具，有三种常见策略：

纯规则匹配：用关键词或正则判断。比如用户输入包含“订单号”就触发订单查询工具。优点是快、可控，缺点是覆盖不全，用户说“我买的东西到哪了”就匹配不到。

纯模型判断：把工具列表和用户输入一起送给 LLM，让 LLM 输出该调哪个工具。优点是灵活，缺点是可能误判，而且每次都要消耗 Token。

混合策略：先用规则做粗筛，缩小工具候选集，再让模型在候选集里做精判。这是我在生产环境最常用的方式。比如先判断用户输入是否包含数字 ID，如果有，就把所有需要 ID 参数的工具筛出来，再让模型选具体调哪个。

在 LangGraph 里，这个判断逻辑可以做成一个独立节点：

def route_decision(state): user_input = state["user_input"] # 粗筛：是否包含订单号模式 if re.search(r'\d{10}', user_input): return "order_tools" # 粗筛：是否涉及政策类问题 if any(kw in user_input for kw in ["政策", "规则", "条款"]): return "retrieval" # 默认走检索 return "retrieval"

3.3 工具返回结果的格式化与注入

工具返回的结果不能直接塞进上下文，需要做格式化。原因有两个：一是原始返回可能包含大量无关字段，浪费 Token；二是 LLM 对结构化数据的理解能力有限，需要转成自然语言或简洁的 JSON。

我的做法是给每个工具定义一个format_output函数，把原始返回转成适合 LLM 阅读的格式：

def format_order_status(raw: dict) -> str: return ( f"订单当前状态：{raw['status']}\n" f"最新位置：{raw['location']}\n" f"预计到达：{raw['eta']}" )

然后在工具节点里调用这个格式化函数，把结果写入 State 的tool_results字段。后续的生成节点从 State 里读取格式化后的结果，拼进 Prompt。

实操心得：工具返回结果里如果有时间戳、ID 这类信息，建议保留原始值的同时加一个自然语言解释。比如"eta": "2024-01-15"可以格式化成"预计到达：2024年1月15日"，LLM 生成回答时不容易搞错格式。

4. 上下文工程：把正确的东西放在正确的位置

4.1 上下文窗口的分配策略

LLM 的上下文窗口是有限资源，怎么分配直接决定生成质量。一个典型的 RAG 请求，上下文里通常包含四部分：系统指令、对话历史、检索结果、用户当前问题。

我的分配原则是：

• 系统指令：固定占用 10-15%，放在最前面。这部分包含角色定义、输出格式要求、安全约束等。
• 对话历史：动态占用 20-30%，只保留最近 N 轮。如果历史太长，用摘要压缩。
• 检索结果：占用 40-50%，这是核心信息，不能省。
• 用户问题：占用 5-10%，放在最后面，紧挨着生成位置。

这个分配不是死的，要根据任务类型调整。比如工具调用场景，检索结果可以少一些，给工具返回留空间；纯问答场景，检索结果可以占到 60%。

4.2 检索结果的去重、压缩与排序

检索回来的文档块经常有重复内容，比如同一段话在不同块里各出现一次。直接拼进去不仅浪费 Token，还会让 LLM 误以为这个信息特别重要。

去重的简单做法是用 MinHash 或 SimHash 计算文档块的指纹，相似度超过阈值的只保留一个。Haystack 的DocumentJoiner其实已经做了一部分去重，但它是基于文档 ID 的，对内容重复无能为力。

压缩的思路是抽取式压缩：对每个文档块，只保留与 query 最相关的句子。可以用一个轻量模型（比如 MiniLM）给每个句子打分，取 Top-3 句子拼成压缩后的块。这样能把 500 token 的块压到 150 token 左右，信息密度大幅提升。

排序方面，除了相关性分数，还可以考虑多样性。如果 Top-5 文档全部来自同一份文件，信息覆盖面可能不够。可以用 MMR（Maximal Marginal Relevance）算法在相关性和多样性之间做平衡：

def mmr_select(docs, query_embedding, lambda_param=0.7, top_k=5): selected = [] candidates = docs.copy() while len(selected) < top_k and candidates: mmr_scores = [] for doc in candidates: relevance = cosine_sim(doc.embedding, query_embedding) redundancy = max([cosine_sim(doc.embedding, s.embedding) for s in selected], default=0) mmr_scores.append(lambda_param * relevance - (1 - lambda_param) * redundancy) best_idx = np.argmax(mmr_scores) selected.append(candidates.pop(best_idx)) return selected

lambda_param设成 0.7 表示更看重相关性，设成 0.5 表示相关性和多样性各占一半。具体取值要看业务场景，政策问答类可以偏相关性，调研类可以偏多样性。

4.3 对话历史的管理：截断、摘要还是向量化

多轮对话里，历史信息的管理是个头疼问题。全保留会爆窗口，全丢弃会丢失上下文。

三种策略各有适用场景：

• 截断：只保留最近 N 轮。简单粗暴，适合历史信息价值不高的场景，比如客服问答。
• 摘要：用 LLM 把历史对话压缩成一段摘要。保留关键信息，但增加一次 LLM 调用，有延迟成本。
• 向量化：把历史对话存进向量库，每轮根据当前问题检索相关历史。适合长对话场景，但实现复杂度最高。

我通常用滑动窗口 + 摘要的混合策略：最近 3 轮保留原文，更早的对话用 LLM 压缩成一段 200 字以内的摘要，放在系统指令后面。这样既保留了近期细节，又不至于丢失远期关键信息。

def manage_history(history, max_recent=3): if len(history) <= max_recent: return history recent = history[-max_recent:] older = history[:-max_recent] summary = llm_summarize(older) return [{"role": "system", "content": f"历史对话摘要：{summary}"}] + recent

5. LangGraph 编排：把检索、工具、生成串成有状态的图

5.1 状态定义与节点划分

LangGraph 的核心是 State。整个图共享一个 State 对象，每个节点读取 State、执行操作、写回 State。State 的定义决定了图的表达能力。

一个典型的 RAG Agent State 包含这些字段：

from typing import TypedDict, Annotated from langgraph.graph import add_messages class RAGState(TypedDict): messages: Annotated[list, add_messages] # 对话消息 user_input: str # 当前用户输入 retrieved_docs: list # 检索到的文档 tool_results: list # 工具调用结果 route: str # 路由决策 final_answer: str # 最终回答

Annotated[list, add_messages]这个写法表示 messages 字段用add_messages函数做更新，新消息会追加而不是覆盖。这是 LangGraph 里管理对话历史的推荐方式。

节点划分的原则是单一职责：每个节点只做一件事。常见的节点包括：

• classify：判断用户意图，决定路由
• retrieve：执行检索
• call_tool：执行工具调用
• generate：生成最终回答
• reflect：检查生成结果是否需要修正

5.2 条件边与循环：让流程会拐弯

LangGraph 的条件边是实现动态路由的关键。你可以在节点执行完后，根据 State 的内容决定下一步走哪个节点：

from langgraph.graph import StateGraph, END def route_after_classify(state): if state["route"] == "tool": return "call_tool" elif state["route"] == "retrieve": return "retrieve" else: return "generate" graph = StateGraph(RAGState) graph.add_node("classify", classify_node) graph.add_node("retrieve", retrieve_node) graph.add_node("call_tool", tool_node) graph.add_node("generate", generate_node) graph.add_conditional_edges("classify", route_after_classify) graph.add_edge("retrieve", "generate") graph.add_edge("call_tool", "generate") graph.add_edge("generate", END)

循环的典型场景是自我反思：生成节点输出答案后，用一个检查节点判断答案是否完整、是否引用了不存在的来源。如果不合格，回到检索节点重新检索，最多循环 2-3 次。

def should_retry(state): if state.get("retry_count", 0) >= 2: return "end" if state.get("quality_check") == "fail": return "retrieve" return "end" graph.add_conditional_edges("check", should_retry, {"retrieve": "retrieve", "end": END})

注意：循环一定要设上限，否则可能陷入死循环。我一般设 2 次重试，超过就直接返回当前最好的结果，并在回答里注明“信息可能不完整”。

5.3 流式输出与中间状态的可观测性

生产环境里，用户等 5 秒才看到完整回答是不可接受的。LangGraph 支持流式输出，可以在每个节点执行完后立即推送中间状态：

for event in graph.stream({"user_input": "查询订单1234567890"}, stream_mode="updates"): for node_name, output in event.items(): print(f"[{node_name}] {output}")

stream_mode="updates"表示每个节点执行完后推送增量更新。你可以根据节点名称给用户展示不同的提示，比如“正在检索知识库...”、“正在查询订单系统...”、“正在生成回答...”。这种反馈能显著提升用户体验。

可观测性方面，建议在每个节点里加日志记录，把输入、输出、耗时都打出来。LangGraph 配合 LangSmith 可以做全链路追踪，但即使不用 LangSmith，自己写个简单的日志装饰器也能满足基本需求：

import time def log_node(func): def wrapper(state): start = time.time() result = func(state) elapsed = time.time() - start print(f"{func.__name__} 耗时 {elapsed:.2f}s, 输入: {state.get('user_input', '')[:50]}") return result return wrapper

6. 常见问题与排查技巧实录

6.1 检索召回率低：从 Query 改写入手

用户的问题往往和文档里的表述不一致。用户问“怎么退钱”，文档里写的是“退款流程”。这种词汇鸿沟是召回率低的主要原因。

解决办法是Query 改写：在检索之前，先用 LLM 把用户问题改写成多个不同表述的查询，分别检索后合并结果。

def rewrite_query(user_input): prompt = f"""把下面的问题改写成3个不同表述的检索查询，每行一个： 原问题：{user_input} 要求：保持语义不变，使用不同的关键词和句式。""" return llm.generate(prompt).split("\n")

实测下来，Query 改写能把召回率提升 15-25%，代价是增加一次 LLM 调用和多次检索。如果延迟敏感，可以只改写一次，生成 2-3 个变体。

6.2 工具调用误触发：阈值与白名单

LLM 有时候会“过度热情”，明明不需要调工具也去调。比如用户只是问“你们支持退货吗”，LLM 可能触发订单查询工具。

控制误触发的手段有几个：

• 提高触发阈值：在工具描述里明确写“仅当用户提供了具体订单号时才调用此工具”。
• 加白名单：某些工具只在特定路由下才暴露给 LLM。比如订单工具只在route == "order"时才加入工具列表。
• 后置校验：工具调用前检查参数是否合法，比如订单号是否符合格式，不符合就直接返回错误提示而不是真的调接口。

def validate_tool_call(tool_name, args): if tool_name == "query_order_status": if not re.match(r'^\d{10}$', args.get("order_id", "")): return False, "订单号格式不正确，请提供10位数字订单号" return True, None

6.3 上下文超长：截断策略与优先级

上下文超长是 RAG 系统最常见的报错。处理策略的核心是优先级排序：哪些内容必须保留，哪些可以丢。

我的优先级顺序是：

1. 系统指令（不可丢）
2. 用户当前问题（不可丢）
3. 工具返回结果（如果本轮有工具调用）
4. 检索结果中分数最高的 2-3 篇
5. 最近 2 轮对话历史
6. 更早的对话摘要
7. 检索结果中分数较低的篇目（可丢）

实现上，可以先计算各部分 Token 数，然后从低优先级开始砍，直到总 Token 数低于模型上限的 80%（留 20% 给生成）。

6.4 生成质量不稳定：温度、Prompt 与 Few-shot

同样的检索结果，LLM 有时生成得很好，有时胡言乱语。这通常和三个因素有关：

温度参数：RAG 场景建议温度设 0.1-0.3，太高容易发挥，太低容易死板。我一般用 0.2。

Prompt 结构：把检索结果放在 Prompt 中间，用户问题放在最后，系统指令放在最前。这种“三明治”结构比把所有内容混在一起效果好。

Few-shot 示例：在 Prompt 里加 1-2 个“问题-检索结果-回答”的示例，能显著提升输出格式的稳定性。示例要选有代表性的，覆盖不同的问题类型。

PROMPT_TEMPLATE = """你是一个知识库助手。根据下面的参考资料回答用户问题。 如果参考资料中没有相关信息，直接说“根据现有资料无法回答”，不要编造。 参考资料： {context} 用户问题：{question} 回答要求： 1. 只使用参考资料中的信息 2. 引用来源时注明文档标题 3. 回答简洁，不超过200字 示例： 问题：退货需要几天？ 参考资料：[退款政策] 退货申请审核通过后，3-5个工作日内退款到账。 回答：根据退款政策，退货申请审核通过后，3-5个工作日内退款到账。 """

7. 从可用到好用：几个容易被忽略的优化点

7.1 缓存策略：哪些环节可以缓存

RAG 流水线里，检索和生成是两个最耗时的环节。合理的缓存能大幅降低延迟。

• Embedding 缓存：同一段文本的 Embedding 结果不会变，可以缓存。用文本的哈希值做 key，避免重复计算。
• 检索结果缓存：如果两个用户问了相似的问题，检索结果可能高度重叠。可以用 query 的语义哈希做 key，缓存 Top-K 文档 ID。
• 生成结果缓存：完全相同的 query + context 组合可以直接返回缓存答案。但要注意 context 可能因为文档更新而变化，缓存要设 TTL。

实操心得：缓存粒度不要太细，否则命中率低；也不要太粗，否则容易返回过期结果。我一般对 Embedding 做永久缓存，对检索结果做 1 小时缓存，对生成结果做 10 分钟缓存。

7.2 降级方案：当检索或工具不可用时怎么办

生产环境必须有降级方案。检索服务挂了、工具 API 超时了，系统不能直接报错，要给用户一个合理的回复。

• 检索降级：如果向量检索不可用，自动切到 BM25；如果都不可用，返回“知识库暂时不可用，请稍后重试”。
• 工具降级：如果工具调用超时，返回“查询超时，请稍后重试或联系人工客服”。
• 生成降级：如果 LLM 服务不可用，返回检索到的原始文档片段，让用户自己看。

降级逻辑可以用 LangGraph 的条件边实现：在节点里捕获异常，把错误信息写入 State，然后路由到降级节点。

7.3 评测体系：怎么知道系统变好了还是变坏了

没有评测的优化都是瞎猜。RAG 系统的评测至少要看三个指标：

• 检索命中率：正确答案是否在检索结果里。可以用人工标注的问答对来测。
• 生成忠实度：生成的答案是否完全基于检索结果，有没有编造。可以用 NLI 模型自动判断。
• 端到端满意度：用户对最终回答的评分。可以用点赞/点踩按钮收集。

我习惯在每次修改检索策略或 Prompt 后，跑一遍固定的评测集（50-100 个问题），对比修改前后的指标变化。评测集要覆盖不同类型的问题：事实型、对比型、多跳推理型、工具调用型。

这套东西搭起来之后，RAG 系统才算真正从“能跑”变成“能扛”。Haystack 和 LangGraph 的组合给了足够的灵活性，但灵活性也意味着更多的决策点。每个决策点都需要根据业务场景做权衡，没有一刀切的最优解。我自己的经验是：先把检索做扎实，再把工具合约定义清楚，最后在编排层做精细化控制。顺序反了，后面会越调越乱。