手把手教你用DeepSeek+LangGraph搭建企业级RAG知识库

1. 环境准备与模型选型对比

先跑通最小链路，再替换组件。需要这些：

• Python 3.11+
• DeepSeek API Key（开放平台创建，模型名 deepseek-chat、deepseek-reasoner）
• LangGraph + LangChain
• 一个向量库，本地开发用 Qdrant 或 Chroma，生产用 Milvus/pgvector
• Embedding 模型，中文场景常用 BGE-M3 或 bge-large-zh-v1.5
• Docker（部署阶段）

pip install -U langgraph langchain langchain-openai langchain-community
pip install -U qdrant-client sentence-transformers

DeepSeek 兼容 OpenAI 接口，用 ChatOpenAI 指向 https://api.deepseek.com 即可。

import os
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
    model='deepseek-chat',
    api_key=os.environ['DEEPSEEK_API_KEY'],
    base_url='https://api.deepseek.com',
    temperature=0,
)

模型选型对比：

LangGraph 最小状态图：

from typing import TypedDict, List
from langgraph.graph import StateGraph, END
class RAGState(TypedDict):
    question: str
    docs: List[dict]
    answer: str
def retrieve(state: RAGState):
    # 调向量库，返回 docs
    return {'docs': []}
def generate(state: RAGState):
    # 拼上下文，调 DeepSeek
    return {'answer': ''}
graph = StateGraph(RAGState)
graph.add_node('retrieve', retrieve)
graph.add_node('generate', generate)
graph.set_entry_point('retrieve')
graph.add_edge('retrieve', 'generate')
graph.add_edge('generate', END)
app = graph.compile()

这段先不求跑通，后面把 retrieve 和 generate 填实。

2. 文档切分、Embedding与向量库选型

切分决定检索上限。企业文档常见 PDF、Word、Confluence、飞书导出。先做解析，再切。

切分策略：

• 按标题层级切：#、##、### 作为边界，保留章节路径。
• 段落太长再递归切：chunk 500–1000 tokens，overlap 100–200。
• 表格、代码、FAQ 单独处理，不要按字符硬切。
• 每个 chunk 带元数据：source、page、section、version、access_level。

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,
    chunk_overlap=150,
    separators=['## ', '### ', '\n\n', '\n', '。', ' '],
)
chunks = splitter.split_text(raw_text)

Embedding 选型：

入库代码示意：

from qdrant_client import QdrantClient
from sentence_transformers import SentenceTransformer
client = QdrantClient(path='./qdrant_data')
model = SentenceTransformer('BAAI/bge-m3')
def embed(texts):
    return model.encode(texts, normalize_embeddings=True).tolist()
client.recreate_collection(
    collection_name='kb',
    vectors_config={'size': 1024, 'distance': 'Cosine'},
)

3. 检索增强与重排序策略

单路向量检索会漏关键词。企业文档里型号、错误码、合同条款，必须用混合检索。

流程：

查询改写用 deepseek-chat，输出 3–5 个变体。多跳问题拆子查询。改写不是为了好看，是为了覆盖不同表述。

混合检索：向量 top 20 + BM25 top 20，用 RRF 融合。RRF 公式简单：

def rrf(rank_lists, k=60):
    scores = {}
    for ranks in rank_lists:
        for i, doc_id in enumerate(ranks):
            scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + i + 1)
    return sorted(scores, key=scores.get, reverse=True)

重排序：取融合后 top 20，送交叉编码器。常用 BAAI/bge-reranker-v2-m3，云端可用 Cohere Rerank、Jina Reranker。重排后取 top 5–8 给模型。重排能明显减少“检索到了但排后面”的问题，代价是延迟。

from sentence_transformers import CrossEncoder
reranker = CrossEncoder('BAAI/bge-reranker-v2-m3')
def rerank(question, docs, top_k=6):
    pairs = [(question, d['text']) for d in docs]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(docs, scores), key=lambda x: x[1], reverse=True)
    return [d for d, _ in ranked[:top_k]]

LangGraph 节点拆法：

• rewrite_query：改写问题，保留原问题。
• retrieve：并行走向量和 BM25。
• fuse：RRF。
• rerank：交叉编码器。
• compress：只留与问题相关的句子，减少 token。
• generate：DeepSeek 生成，要求带引用编号。
• verify：检查每个引用是否支持对应句子。

4. 幻觉抑制与评测指标

幻觉抑制靠流程，不靠一句“不要编造”。

系统提示模板：

你是企业知识库助手。只允许根据 <context> 回答。
每条结论后标注来源编号，如 [1]。
如果 <context> 没有答案，直接说“当前知识库没有找到依据”，不要推测。
如果用户问题涉及权限，先检查 access_level。

生成后做引用校验。用 deepseek-chat 判断：答案中的每个事实句是否能被对应引用支持。不支持的句子删掉或标“未验证”。

verify_prompt = '''
上下文：
{context}
答案：
{answer}
逐句检查答案是否被上下文支持。输出 JSON：
{"supported": true/false, "unsupported_sentences": []}
'''

评测分两层：检索和生成。

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision
result = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy, context_precision],
)
print(result)

没有黄金集就先标 50 条。覆盖三类：单文档事实、跨文档对比、无答案拒答。每次改切分、Embedding、重排、提示词，都跑同一套集。只看几个问答的体感，会把回归漏掉。

拒答也是指标。无答案集上的正确拒答率要单独统计。企业场景里，编造一条错误政策比答不出更贵。

5. 部署上线与成本优化

上线架构：

FastAPI 暴露 /chat，请求体带 user_id、question、tenant_id。权限过滤必须在检索阶段做，不能只在生成阶段藏答案。

from fastapi import FastAPI
from pydantic import BaseModel
class ChatReq(BaseModel):
    question: str
    user_id: str
    tenant_id: str
app = FastAPI()
@app.post('/chat')
def chat(req: ChatReq):
    state = {'question': req.question, 'docs': [], 'answer': ''}
    result = app_graph.invoke(state)
    return {'answer': result['answer'], 'sources': result['docs']}

LangGraph 的持久化用 Postgres 或 Redis 做 checkpointer。多轮对话、人工审核、失败重试都需要状态。别把状态放进程内存，扩副本会丢。

成本控制清单：

• Embedding 缓存：文档不变就不重复算。
• 查询缓存：相同问题走缓存，设 TTL。
• 批处理：离线入库用 batch，不要一条一条调。
• 分层检索：先用小模型/BM25 粗筛，再向量精排。
• 上下文压缩：只留相关句，减少输入 token。
• 模型分流：简单问答走 deepseek-chat，复杂推理才用 deepseek-reasoner。
• 输出限制：max_tokens 按业务设上限。
• 监控 token：按租户、接口、模型记录消耗。
• 缓存命中：DeepSeek 提供上下文缓存，命中后输入价格更低，具体折扣看官方定价页。
• 冷热分离：热数据留向量库，冷数据归档。

• 多租户隔离：collection 或 filter 必须带 tenant_id。
• PII 脱敏：入库前识别手机号、身份证、银行卡。
• 审计日志：记录谁在何时问了什么、命中了哪些文档。
• 权限同步：文档权限变更后，向量库元数据要跟着更新。

可操作下一步：拿你手头 50 份文档，按本文切分入库，标 30 条问答，记录 Hit Rate@5 和 Faithfulness。如果 Hit Rate 低于预期，先查切分和 Embedding；如果 Faithfulness 低，先查重排和引用校验提示词。

推荐视频：B 站搜索“LangGraph RAG 教程”“DeepSeek API 实战”，腾讯视频、优酷、抖音也有相关开发者分享。视频内容随版本变化，请以 LangChain、LangGraph、DeepSeek 官方文档为准。本文不绑定具体视频链接，避免失效后误导。免责声明：视频为第三方内容，不代表本文验证其准确性。