企业官网建设流程全解析

“开箱即用”这四个字，放在企业级智能问答系统这个领域里，是我见过最容易被低估的承诺。RAG（检索增强生成）和LLM（大语言模型）本身并不难理解，难的是把两者真正放进一套工作流引擎里，让企业几十万份文档变成能稳定、可靠、带权限回答问题的基础设施。下文不聊概念，只聊落地方案。我会用一套基于FastAPI + LangChain + LangGraph + pgvector的组合，把实际搭建RAG+LLM工作流引擎时踩过的坑、验证过的方案、最终沉淀的架构设计完整拆开。无论你是刚准备做企业知识库的研发，还是已经在做相关项目但被效果和可维护性折磨的人，这篇文章都值得读完。

1. 先搞明白：企业级智能问答到底需要什么

1.1 从“能回答问题”到“能可靠地回答问题”

很多人一上来就急着调LLM的API，写完一个“把文档塞进向量库再提问”的demo就宣布完成了知识库问答。这个路线的问题在于：demo阶段你只需要处理三四份PDF，企业场景里是三四万份，而且格式千奇百怪——PDF、Word、PPT、扫描件、Excel、网页导出文件，甚至还有图片里的表格。更麻烦的是，文档与文档之间可能有冲突说法：“2024年版《报销制度》”说发票抬头必须是公司全称，但某部门的内部邮件里又说可以用简称。LLM看到这些内容，很容易给出一个“看起来合理但实际错误”的答案。

所以你会发现，企业级智能问答系统真正要解决的不是“能不能回答问题”，而是“能不能一直可靠地回答问题”。我习惯把需求分成五个层次来梳理：

前两层是“能用”的门槛，后三层才是“敢用”的门槛。尤其第四层权限感知，很多团队最后翻车都翻在这里。比如一家制造企业的内部问答系统上线后，一线员工提问“去年车间主任的绩效方案”，系统如果检索到了只对管理层开放的文档，即使LLM生成的回答看似正常，敏感信息也已经通过上下文泄露了。这种情况出了问题，不是技术事故，是合规事故。

1.2 工作流引擎在整个链路里的位置

搞清楚了需求层次，再来看技术结构。一个最小可用的RAG链路是这样的：

用户问题 → 向量化 → 向量检索 → 拼接Prompt → LLM生成 → 返回

但企业级链路里，这一步远不够。真实场景里，用户问题进来之后往往要经过这样一串处理：

• 判断意图：是问知识库内容，还是闲聊（“你好”“你是谁”）？
• 查询改写：用户问“上个月的报销截止时间”，系统需要理解“上个月”是哪个月；
• 分路检索：同时做向量检索和关键词检索，取并集，避免单个检索漏检；
• 元数据过滤：按部门、文档类型、权限范围过滤；
• 重排：候选文档里挑最相关的5条，而不是直接信任向量距离；
• 生成并标注引用：回答里必须带上来源文档编号；
• 答案校验：某些场景下检出的相关性太低，要触发拒答流程。

这些步骤如果全部写在Controller或者Service的一个大函数里，前期勉强能跑，但只要有一天你需要加一个分支（比如“检索结果太差时走人工转接”），或者想调整流程顺序，就会感受到什么叫牵一发而动全身。更不用提每次修改后，你根本没法知道问题是出在检索环节、重排环节还是生成环节。

工作流引擎存在的意义，就是把这一串复杂逻辑显式地建模成“节点（Node）+ 边（Edge）”。每个节点只干一件独立的事情，节点之间通过状态对象传递数据。这样带来了三个直接好处：第一，每个环节可单独测试和观测；第二，流程调整变成配置变更，而不是代码重构；第三，所有中间结果都被保留，出了问题能回溯。

1.3 落地前必须确认的优先级

在开始选型之前，还有一个容易忽略的环节——需求优先级排序。我见过不少项目，团队把“答案要非常准确”放在第一位，结果在权限隔离上偷工减料，最后上线第一周就被安全部门叫停。如果让我给一个建议的排序，会是这样：

1. 权限隔离：不可妥协。检索阶段就要做，不能等生成后再筛。
2. 答案可溯源：每条回答都能找到来源文档，这是建立信任的基础。
3. 更新不中断服务：知识库内容每天都在变，全量重建是最不能接受的做法。
4. 可观测和可评估：至少要有日志、trace、评估集，否则效果好坏全靠“感觉”。

把这个排序定下来之后，你会发现后面的技术选型其实非常顺理成章。

2. 技术栈选型的逻辑：为什么是 FastAPI + LangChain + LangGraph + pgvector

2.1 每一层的职责划分

这套方案的选型逻辑不是“哪个框架火用哪个”，而是每一层都有明确职责，换掉任何一个组件都不会牵连其他层。整体分层如下：

这个分工的核心逻辑是：FastAPI负责“门面”，LangGraph负责“大脑”，LangChain负责“手脚”，pgvector负责“记忆”。

FastAPI作为服务层的选择可能不需要多解释，异步支持好、类型校验强、自动生成OpenAPI文档，在Python生态里做AI服务几乎是首选。但有一个细节值得注意：不要把业务逻辑写在路由函数里。路由函数只做参数解析和结果返回，真正的问答流程通过调用工作流引擎完成。这样HTTP层和工作流层解耦，后续如果你要接gRPC或者消息队列，改动面很小。

2.2 pgvector 凭什么胜出

向量数据库的选择可能是这个方案里争议最大的地方。市面上有Milvus、Qdrant、Weaviate、Elasticsearch等专业向量库，为什么最终选pgvector？

我的判断标准很简单：在绝大多数企业知识库场景里，数据量是几十万到百万级chunk，这个量级下pgvector的性能完全够用，而它的运维成本比专业向量库低一个数量级。

pgvector有一个杀手级特性：向量过滤条件可以直接和SQL的WHERE条件叠加。比如权限过滤，在向量检索的同时就能指定team_id = 'finance_team'，数据库在检索向量的同时完成权限过滤。而专业向量库通常需要把权限条件作为metadata过滤参数传入，不仅写法受限，在复杂权限模型下（比如按角色、部门、项目等多个维度控制）实现起来非常别扭。

下面是这套方案里最核心的一张表结构：

CREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE document_chunks ( id BIGSERIAL PRIMARY KEY, document_id BIGINT NOT NULL, chunk_index INT NOT NULL, content TEXT NOT NULL, metadata JSONB NOT NULL DEFAULT '{}', team_id TEXT NOT NULL, doc_version INT NOT NULL DEFAULT 1, embedding vector(1024) ); CREATE INDEX ON document_chunks USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 128);

这里有几个细节值得展开说：

• embedding vector(1024)的维度取决于你选的Embedding模型，bge-large-zh-v1.5输出1024维，bge-m3同样输出1024维。维度不是越高越好，维度高意味着存储和计算成本更高，而你实际需要的可能只是768甚至512维。
• HNSW索引参数m=16, ef_construction=128是兼顾查询速度和构建速度的经验取值。如果数据量超过100万，可以适当调大ef_construction到256，但构建时间会增加不少。
• team_id字段单列出来，而不是塞进metadataJSONB里，是为了让权限过滤走普通B-tree索引。如果权限条件放在JSONB里，性能会差不少。

性能方面，用一张100万行、1024维向量的表做实测，HNSW索引下单次检索耗时大约在20到50毫秒（取决于lists和probe参数）。对于问答系统来说，这个延迟完全在可接受范围内。真正到千万级数据量、要求P99延迟低于20毫秒的时候，再考虑换Milvus也不迟，前期用pgvector把业务跑通，是性价比最高的路径。

2.3 LangGraph 在什么时候成为必须品

先说我自己的经历。最早做企业内部问答的时候，我用的是LangChain的Chain/LCEL来串流程，一个线性链路：改写 → 检索 → 生成。上线后前两个月还好，但随着业务方不断提需求，链路开始长出各种分支——用户问“报销流程”时，需要检索HR和财务两个部门的文档；检索结果质量太差时，要触发澄清反问；部分场景需要连续多轮追问，每轮都依赖上一轮的答案。LCEL对这种非线性流程的支持相当吃力，代码里全是if和else，中间状态散落得到处都是。

换到LangGraph后，整个流程被重构为一张有向图。每个节点是一个Python函数，节点之间通过共享状态对象传递数据。这个模型很接近后端工程师熟悉的有限状态机，理解成本很低。LangGraph真正值钱的地方有三个：

• 显式状态管理：整个问答链路的所有中间数据（改写后的query、检索候选、重排结果、最终答案）都定义在一个State对象里，可序列化、可追踪、可恢复。
• 条件边：一个节点可以根据返回结果动态决定下一个节点，比如“是否需要查询改写”“是否触发拒答”，这些逻辑在图上就是一条带条件的边，而不是代码里的if。
• Checkpointer断点机制：支持在某个节点暂停执行，等人工确认后再继续。这个能力在企业场景里非常实用，比如敏感操作需要审批、质检环节需要人工抽检。

当你发现问答链路里出现两个以上条件分支时，LangGraph就不是可选项，而是必选项。

3. 一次完整的问答旅程：从文档上传到答案生成

3.1 文档接入与解析：企业文件远比想象中复杂

很多团队在搭建RAG系统时严重低估了文档解析的工作量。真实企业环境里，你拿到的“资料”大概率是这样的：

• PDF里混合了文本层和图片，部分页面是扫描件；
• Word文档里嵌套表格，表格里还有合并单元格；
• PPT里文字分散在文本框和图表注释里，提取顺序经常混乱；
• Excel多个sheet结构不一致，转成纯文本后很难看懂对应关系。

我的经验是：解析阶段多花一小时，检索效果能少踩一个月的坑。如果文档是扫描件，必须先过OCR。中文场景推荐PaddleOCR或RapidOCR，识别准确率在印刷体上可以达到95%以上。解析后的内容建议转成Markdown而不是纯文本，原因是Markdown保留了标题层级、列表、表格这些结构信息，后续分块时可以利用。

下面是这套系统里负责文档解析的简化版代码：

from fastapi import UploadFile from pathlib import Path import fitz # PyMuPDF async def parse_document(file: UploadFile) -> Document: suffix = Path(file.filename).suffix.lower() if suffix == ".pdf": doc = fitz.open(stream=await file.read(), filetype="pdf") pages = [] for page in doc: text = page.get_text("text") if len(text.strip()) < 20: # 文本层内容很少，走OCR text = ocr_page(page) pages.append(text) return Document( content="\n\n".join(pages), metadata={"source": file.filename, "format": "pdf"} ) elif suffix in (".docx", ".pptx"): # 用python-docx或python-pptx提取，保留标题层级 ... else: raise UnsupportedFormatError(f"不支持的文件格式: {suffix}")

这里有一个关键判断逻辑：len(text.strip()) < 20。PDF的每一页如果文本层提取出的内容太少，大概率是扫描件。这个阈值可以根据实际文档情况调整，但不能省。如果扫描件混在正常PDF里没有走OCR，检索效果会直接断崖式下跌，因为向量里存的是一堆乱码或者空字符串。

3.2 分块策略：Chunk Size 不是随意定的

分块（Chunking）是整个RAG链路里最容易被忽略、但对最终效果影响最大的环节之一。文本切得太大，向量化后语义容易被稀释，检索定位不准；切得太小，单个chunk的上下文不足，LLM回答时缺少背景。更麻烦的是，如果切断了语义完整的段落，检索到的内容可能是“半句话”，答案自然会出问题。

我的经验是：中文场景下，chunk size取300到500个字符（不是token），overlap取50到100个字符，是一个比较稳的起点。但更重要的是结构感知——优先按照Markdown标题层级切分，其次才是字符数兜底。比如一篇文章先按H1、H2、H3切出小节，每个小节如果还太长，再用滑动窗口切分。

LangChain的MarkdownHeaderTextSplitter和RecursiveCharacterTextSplitter组合起来可以做到这一点，示例代码如下：

from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter headers_to_split_on = [ ("#", "H1"), ("##", "H2"), ("###", "H3"), ] md_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on) char_splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=80, separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", "；", ";", "，", ",", " ", ""], ) raw_doc = "# 2024版报销制度\n\n## 报销范围\n\n... \n\n## 报销流程\n\n..." # 第一步：按标题切 sections = md_splitter.split_text(raw_doc) # 第二步：过长的章节按字符切 chunks = [] for section in sections: if len(section.page_content) > 500: sub_chunks = char_splitter.split_text(section.page_content) for sub in sub_chunks: chunks.append(Chunk( content=sub, metadata={**section.metadata, "header_path": build_header_path(section)} )) else: chunks.append(Chunk( content=section.page_content, metadata={**section.metadata} ))

注意代码里我加了一个header_path字段。这个字段的用途是在检索到某个chunk时，能够告诉用户“这段内容来自《2024版报销制度》的‘第二章 报销范围’”。企业用户看到答案时如果没有来源路径，信任度会大打折扣；反过来，有明确的标题路径，即使答案稍有偏差，用户也能快速定位原文核实。

还有一个容易忽略的点：表格要单独处理。如果用字符级分块，一个跨页的表格会被拦腰切断，结构完全丢失。我的做法是在解析阶段检测表格区域，把表格整体转成Markdown表格格式后作为一个独立chunk存储，并在metadata里标注type: table。

3.3 向量化与存储：Embedding模型选型

Embedding模型的选择直接影响检索效果。OpenAI的text-embedding-3-large效果确实不错，但对企业私有化部署来说有两个绕不开的问题：数据出境合规、API稳定性。大多数企业内部知识库对数据安全有硬性要求，我在这套系统里首选本地部署的开源模型。中文场景下实测下来，BAAI的bge系列表现最稳定，尤其是bge-large-zh-v1.5和bge-m3。

bge-large-zh-v1.5输出1024维向量，在中文语义匹配上的效果很好，模型体积约1.3GB，单张16GB显存的卡就能跑推理。bge-m3的好处是多语言支持更好，如果企业内部有英文文档混合场景，优先考虑它。加载和推理的代码非常简单：

from sentence_transformers import SentenceTransformer model = SentenceTransformer("BAAI/bge-large-zh-v1.5") def embed_text(text: str) -> list[float]: return model.encode(text, normalize_embeddings=True).tolist()

有一点要注意：normalize_embeddings=True必须设置。因为pgvector的余弦距离计算不会自动归一化，如果Embedding不求模为1，余弦相似度计算会有偏差。如果你用cosine_ops索引，却忘了归一化，检索结果会莫名其妙地变差。

写入pgvector的代码如下：

import json import psycopg async def insert_chunks(chunks: list[Chunk], embeddings: list[list[float]]) -> None: with psycopg.connect(conninfo) as conn: with conn.cursor() as cur: for chunk, emb in zip(chunks, embeddings): cur.execute( """ INSERT INTO document_chunks (document_id, chunk_index, content, metadata, team_id, embedding) VALUES (%s, %s, %s, %s, %s, %s) """, ( chunk.document_id, chunk.chunk_index, chunk.content, json.dumps(chunk.metadata, ensure_ascii=False), chunk.team_id, emb, ) )

如果只是批量导入，配一个连接池就够。但如果文档量特别大（比如五万份文档、上百万chunk），建议用COPY命令批量导入，写入速度能提升一个数量级。psycopg的copy()方法支持这个操作，值得提前设计好。

3.4 检索与重排：从Top-K到Rerank

检索阶段，我采用“向量检索 + 关键词检索 + 重排”三段式架构，而不是只做纯向量检索。原因很简单：纯向量检索擅长语义匹配，但对精确词匹配（比如产品型号“V2.3.1”、工单号“INC-20250101”）反而容易漏。关键词检索可以在这些场景下兜底。

向量检索直接用pgvector：

def vector_search(query_emb, team_id, top_k=20): rows = conn.execute( """ SELECT id, content, metadata, 1 - (embedding <=> %s) AS similarity FROM document_chunks WHERE team_id = %s ORDER BY embedding <=> %s LIMIT %s """, (query_emb, team_id, query_emb, top_k) ).fetchall() return rows

关键词检索可以用PostgreSQL自带的全文搜索，也可以用单独维护的索引。数据量不大时，tsvector+GIN索引完全够用：

ALTER TABLE document_chunks ADD COLUMN content_tsv tsvector; UPDATE document_chunks SET content_tsv = to_tsvector('simple', content); CREATE INDEX idx_tsv ON document_chunks USING GIN(content_tsv);

检索时：

SELECT id, content, metadata FROM document_chunks WHERE team_id = %s AND content_tsv @@ plainto_tsquery('simple', %s) ORDER BY ts_rank(content_tsv, plainto_tsquery('simple', %s)) DESC LIMIT 20;

向量检索和关键词检索各取Top 20，合并去重后进入Rerank阶段。这里强烈建议不要省：Rerank是整个RAG链路里性价比最高的优化点。

Rerank模型的选型上，bge-reranker-v2-m3是目前中文场景下的最佳选择之一。它会把“问题和候选文本对”作为输入，输出相关性分数，然后按分数重新排序，取Top 5作为最终的上下文。示例代码：

from FlagEmbedding import FlagReranker reranker = FlagReranker("BAAI/bge-reranker-v2-m3") def rerank(query: str, candidates: list[str], top_k=5) -> list[str]: # compute_score接受[[query, doc], ...]格式 pairs = [[query, doc] for doc in candidates] scores = reranker.compute_score(pairs) ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True) return [doc for doc, score in ranked[:top_k]]

Rerank的效果提升有多大？我在一个制造业客户的知识库上做过对比：纯向量检索Top 5的准确率是61%，加上Rerank之后准确率到了82%。提升的本质是：向量检索的排序信号和真实“问答相关性”之间存在差距，Rerank模型专门解决这个gap。这个差距在专业领域的多义词场景下尤其明显。

4. 引入LangGraph工作流编排：把多分支流程变成可控状态机

4.1 核心状态定义：整个问答链路的“数据库”

LangGraph的核心理念是状态机，你首先要定义一个State类型，它相当于整个问答链路里的“数据库”。所有节点都从State里读输入，把输出写回State，下一个节点再从State里取。这个设计最大的好处是流程里的每一步发生了什么，都能通过State完整回放。

下面是这套系统里的核心State定义：

from typing import TypedDict, List, Literal class QAState(TypedDict): question: str history: List[dict] route: Literal["rag", "chitchat", "refuse"] rewritten_query: str vector_hits: list keyword_hits: list reranked_docs: list answer: str citations: list error: str

定义完State之后，每个节点就是一个接收State、返回State更新的函数。例如查询改写节点：

def rewrite_query(state: QAState) -> dict: if not state["history"]: return {"rewritten_query": state["question"]} prompt = f"""你是问答系统的查询改写器。用户可能引用历史对话中的内容，请结合对话历史把问题改写成一个独立完整的查询。 历史对话： {history_to_text(state["history"])} 当前问题：{state["question"]} 只输出改写后的查询，不要任何解释。""" rewritten = llm.invoke(prompt) return {"rewritten_query": rewritten}

4.2 节点设计与图的构建：五大节点，各司其职

这套系统里，我把整个问答链路拆成六个节点，每个节点只负责一件事：

构建成LangGraph的方式如下：

from langgraph.graph import StateGraph, END graph = StateGraph(QAState) # 添加节点 graph.add_node("route", route_node) graph.add_node("rewrite", rewrite_node) graph.add_node("retrieve", retrieve_node) graph.add_node("rerank", rerank_node) graph.add_node("generate", generate_node) graph.add_node("fallback", fallback_node) # 入口 graph.set_entry_point("route") # 定义边的逻辑 graph.add_conditional_edges( "route", route_to_node, # 根据route字段决定下一个节点 { "rag": "rewrite", "chitchat": "generate", "refuse": "fallback", } ) graph.add_edge("rewrite", "retrieve") graph.add_edge("retrieve", "rerank") graph.add_edge("rerank", "generate") graph.add_edge("generate", END) graph.add_edge("fallback", END) app = graph.compile()

这段代码最大的亮点在add_conditional_edges。当用户问“你好”时，route节点直接判定为chitchat，跳去生成一个简单问候回答，完全不走检索链路，省掉一大笔向量检索和LLM token的耗时。当route判定为refuse（比如涉及违禁话题），走fallback节点返回预设话术。

4.3 条件路由、人工介入与断点续跑

进入生产后，条件路由确实能解决绝大多数动态流程问题。我实际应用中还有一个比较特殊但非常重要的场景：人工审批节点。

企业知识库里有些文档是需要多级审核才能发布到公开知识库的。比如法务部门起草的合同模板，HR的新版员工手册，这些文档在“上传后”和“对全员可见”之间有一个审批环节。我们用LangGraph的Checkpointer实现了这个流：

• 上传新文档 → 进入ingest_workflow（单独的知识库写入工作流，不是问答工作流）
• 解析 + 分块 + 向量化完成后，写入pending_chunks表，状态为pending
• 人工在管理后台确认通过后，更新状态为published，同时把chunk插入document_chunks表

这个流程的关键之处在于：LangGraph的Checkpointer支持在某个节点暂停，把State序列化保存，等外部事件触发后再恢复执行。具体到抽象一点，你可以把一个“文档发布”的工作流定义成图，在publish_approval节点设置断点，审批人在后台点“通过”后，工作流从断点继续跑。

source code这里不贴具体实现，因为涉及业务逻辑比较多，但思路就是：

config = {"configurable": {"thread_id": document_id}} # 第一次执行，停在审批节点 app.invoke({"action": "publish_request"}, config) # 审批通过后，同一thread_id继续执行 app.invoke({"action": "approve"}, config)

LangGraph内部通过thread_id把两次invoke关联到同一个状态实例，这是代码里最核心的暗线。搞懂这个机制，很多企业级场景（比如工单流转、敏感操作复核）都能靠它实现。

5. 企业落地RAG时最容易踩的坑

5.1 权限过滤必须发生在检索阶段，而不是生成后

这是我最想强调的一条。很多团队做权限控制时，用的是“先检索，再根据权限过滤结果”，这种方案在技术上容易实现，但在RAG场景里是一个严重的安全漏洞。原因在于：LLM生成答案时，会把Prompt里的所有内容都当上下文。你即使最后不给用户展示某份文档，它可能已经被LLM“看过”了。如果文档内容被LLM记住并夹带到回答里，权限就形同虚设。

正确做法是检索SQL里做权限过滤，让无权访问的文档连进入候选集的机会都没有。用pgvector的好处在这里完全体现：普通向量库通常只能做简单的metadata过滤，但PostgreSQL里可以JOIN权限表：

SELECT c.id, c.content, c.metadata, 1 - (c.embedding <=> %s) AS similarity FROM document_chunks c INNER JOIN user_team_permissions p ON c.team_id = p.team_id WHERE p.user_id = %s ORDER BY c.embedding <=> %s LIMIT 20;

这个查询里，权限过滤是数据库层面完成的，MySQL和PostgreSQL都能做得很快。如果你用Milvus这类专业向量库，可能需要用Partition（分区）来实现，但分区的管理成本会高不少。

5.2 幻觉控制：不是靠Prompt就能解决，要靠系统设计

很多人把“LLM幻觉”归咎于Prompt没写好，实际上幻觉是整个RAG系统设计问题的综合体现。我的经验是，从三个层面同时下手：

第一个层面是Prompt的强约束。生成节点加载的提示词必须明确告诉模型：“只能使用提供的引用内容回答，引用内容不足时直接说不知道，不得猜测。”这段提示词看起来简单，但实测能显著降低幻觉率。举个例子：

system_prompt = """你是一个企业知识库问答助手。回答问题时必须遵守以下规则： 1. 只能依据"参考资料"中的内容回答； 2. 如果参考资料中没有答案，明确回答"根据现有知识库无法回答该问题"； 3. 回答时标注引用来源，格式为[编号]，编号对应参考资料中的文档标题； 4. 不要拼接、推测或编造任何信息。"""

第二个层面是降低温度参数。很多团队习惯用默认的temperature（通常是0.7到1.0），这在对话生成时效果很好，但在知识问答场景里会带来过多随机性。我实际生产配置里，LLM的temperature设为0.1到0.2之间，top_p设为0.85。这样回答内容更保守，更依赖资料，而不是自由发挥。

第三个层面是引用溯源强制。检索到的每个chunk在进入LLM前，都会加上一个编号前缀，比如[1]《2024版报销制度》第二章。LLM在生成回答时，必须引用编号。这个机制让“答案可追溯”成为系统能力，而不是模型自觉。

5.3 没有评估集，效果全靠“感觉”是最大的工程债

很多团队在调整分块策略、Embedding模型、Rerank模型时，凭的是“跑几个案例看看效果”。这个做法在demo阶段没问题，但进入生产后一定翻车。原因很简单：你看到的几个案例不具有统计意义，你根本无法判断优化方向对不对。

我在项目里强烈建议维护一个“评估集（Eval Set）”，不需要太大，50到100条真实业务问题就够了。每个问题标注：

• 期望答案来源（对应哪份文档或哪些chunk）；
• 期望答案类型（是事实性回答、流程引导、还是需要拒答）；
• 特别标注边界情况（比如“文档里没有答案，期望拒答”）。

所有链路改动（换Embedding模型、调整chunk size、换Rerank策略）都在这个评估集上跑一遍，看三个指标：检索命中率、答案准确率、拒答准确率。用分数说话，而不是“我觉得更好了”。

下面是我常用的一个极简评估脚本：

eval_set = load_eval_set("eval_questions.json") def evaluate(pipeline, eval_set): hit_count = 0 correct_count = 0 refuse_correct = 0 for item in eval_set: result = pipeline.invoke({"question": item["question"]}) # 检索命中评估 if any(doc_id in result["citations"] for doc_id in item["target_doc_ids"]): hit_count += 1 # 答案正确性评估 if result["answer"] == item["expected_answer"] or (item["expect_refuse"] and result["route"] == "refuse"): correct_count += 1 return { "recall@5": hit_count / len(eval_set), "answer_accuracy": correct_count / len(eval_set), }

这个脚本简陋，但足够支撑前期的迭代优化。等系统稳定后，还可以引入LLM-as-a-Judge的方式，让GPT或本地模型给答案打分，但初期不建议，因为“模型打分”本身的误差会干扰你对真实效果的判断。

6. 从Demo到生产：三个必须补齐的工程化拼图

6.1 可观测性：每个问题都要能回溯

RAG系统进入生产环境后，可观测性不是锦上添花，而是刚需。用户在群里反馈“答案不对”，如果系统没有trace，你根本没法定位是检索问题、Rerank问题还是LLM生成问题。

我的做法是给整个链路加上一个贯穿始终的trace_id，从FastAPI收到请求开始生成，注入到LangGraph的State里，然后在每个节点的开始和结束都打印结构化日志：

{ "trace_id": "8f3a2b9c...", "node": "retrieve", "action": "vector_search", "query": "上个月的报销截止时间", "top_k": 20, "latency_ms": 35, "hit_count": 20 }

每一条日志都包含trace_id，这样用日志查询工具就能串联起一个请求的完整生命周期。如果团队有条件，建议直接接入OpenTelemetry，LangGraph的节点和FastAPI的中间件都可以做instrumentation，后续看Grafana上的链路火焰图会非常直观。

6.2 文档更新走异步任务，增量更新不能全量重建

企业在实际使用中，知识库内容每天都在变化。有些团队偷懒的做法是每天凌晨全量重建向量库，这在数据量小、更新频率低的场景下可以容忍，但数据量一上来，全量重建不仅耗时，还要停服，用户体验极差。

正确做法是增量更新：文档上传接口只负责接收文件，真正的解析、分块、向量化等耗时代码放到异步任务队列（Celery或Arq）里执行。FastAPI接口收到文件后马上返回{"document_id": "xxx", "status": "processing"}，前端轮询状态，处理完成后再通知用户。

增量更新时的关键点是：同一个文档的新版本上线后，旧版本的chunk要能精确删除。我的表结构里有一个doc_version字段。新版本向量化完成后，先执行DELETE FROM document_chunks WHERE document_id = %s，再把新chunk插入。这里的核心是保证更新过程的原子性：不能出现旧chunk删了、新chunk还没插好的空窗期。可以用一个事务把删除和插入包起来，或者引入一个published_epoch字段来标记chunk的生效时间，查询时只挑最新的版本。

6.3 并发、限流与降级策略：LLM不可用时系统不能挂

上线初期容易忽略的一个问题是并发和依赖故障。一个问答请求涉及LLM调用、Embedding调用、Rerank调用，任何一个外部依赖抖动，都会拖垮整个服务。而且LLM API是有速率限制的，如果没有限流，系统可能被自己的用户流量打挂。

我在这套系统里做了三级防护：

• 接口层：FastAPI依赖slowapi做IP级限流，普通用户每分钟不超过20次问答请求；
• 服务层：所有LLM调用包了一层带信号量的异步客户端，控制并发数不超过API配额；
• 降级策略：当LLM调用失败时，系统不直接报错，而是返回“暂时无法生成答案”，同时把检索到的Top 5相关文档标题返回给用户。这样即使LLM挂了，用户依然能得到部分价值，而不是看到一个500错误页面。

指数退避重试也是必须的。LLM调用偶尔会超时，简单的重试可以解决，但不能无限重试，也不能固定间隔重试。我用的是指数退避加抖动：

import random, time async def llm_with_retry(prompt, max_retries=3): for attempt in range(max_retries): try: return await llm.ainvoke(prompt) except RateLimitError: if attempt == max_retries - 1: raise backoff = min(2 ** attempt, 30) + random.uniform(0, 1) await asyncio.sleep(backoff)

这套机制上线后，我们遇到过OpenAI API区域性故障、本地模型OOM、数据库死锁，但整体服务一直保持可用，只是部分请求走了降级路径。对生产系统来说，服务质量的优先级排序永远是：可用性 > 准确性 > 成本。

最后再分享一个小技巧。如果你也是在做企业知识库相关项目，一开始不要把范围铺得太大。找一个业务方怨气最重、文档相对集中的场景（比如HR政策问答、IT帮助台），先跑通一个高质量闭环，用真实效果赢得业务方的信任，再逐步扩展到其他部门。RAG系统的成败，有很大一部分取决于你对真实场景的理解深度，而不是你用了多先进的模型。