一、背景与目标

很多团队拿到大模型 API 之后，会遇到两个典型问题：

1. 只能闲聊，无法回答公司内部知识（制度、产品、技术文档等）
2. 幻觉严重：模型“一本正经胡说八道”，难以直接用于生产

业界主流的解决方案是 RAG（Retrieval-Augmented Generation，检索增强生成）​：
先用向量检索找到相关文档片段，再把这些文档作为上下文交给大模型生成答案。

本文用一个完整的实战案例，带你用 DeepSeek 推理模型 + 本地向量检索，从零实现一个可落地的 企业知识库问答系统：

• 支持用自然语言问“公司用的 DeepSeek 是什么版本？怎么调用？限流是多少？”
• 知识来源于你的 PDF/Word/内部 Wiki 导出的文本
• 支持本地向量检索，后端只调 DeepSeek 的 Chat Completion 接口

你可以直接改成自己公司的知识库，就能跑起来。

---

二、整体架构设计

我们要做的是一个典型的 RAG 应用，流程如下：

1. 离线阶段（构建知识库）​

   • 收集公司文档（产品说明书、API 文档、SOP 等）
   • 切分成段落
   • 用本地向量模型生成 embeddings
   • 建立向量索引（例如用 sklearn 的 NearestNeighbors）

2. 在线阶段（回答用户问题）​

   • 用户输入问题
   • 用同一个本地向量模型把问题转成向量
   • 在向量索引中检索最相似的若干文档片段
   • 把这些片段 + 问题 组合成 Prompt，发给 DeepSeek Chat 接口
   • 返回结构化、可读的答案

架构示意：

用户问题 → 向量检索（本地） → 取出 Top-K 文档 → 构造 Prompt → 调用 DeepSeek → 返回答案

这套架构的好处：

• 数据可控：知识全在你本地，DeepSeek 只看到被选出来的内容
• 可解释：你可以把“引用的文档片段”一起展示给用户
• 成本可控：只有 DeepSeek 的调用需要付费，向量检索本地完成

---

三、环境准备

3.1 依赖安装

假设你已经有 Python 3.10+ 环境，安装依赖：

pip install openai sentence-transformers scikit-learn numpy
# 可选：Web UI
pip install streamlit

说明：

• openai：用 OpenAI 兼容的 SDK 调 DeepSeek Chat API
• sentence-transformers：本地向量模型，用来做文本嵌入
• scikit-learn：用 NearestNeighbors 做相似度检索
• streamlit：如果你想快速搭一个 Web Demo

3.2 DeepSeek API Key

1. 去 DeepSeek 平台申请 API Key
2. 在本地设置环境变量：

export DEEPSEEK_API_KEY="your_api_key_here"

---

四、准备一份简单的知识库

先准备一个最小可用的 JSON 知识库，放在 data/knowledge_base.json：

[
  {
    "title": "DeepSeek 模型简介",
    "content": "DeepSeek 是由深度求索开发的一系列大语言模型，包括 Coder、Math、Reasoner 等。其中 DeepSeek-Reasoner 专注于多步推理和逻辑分析，适合构建问答和决策支持类应用。"
  },
  {
    "title": "DeepSeek API 使用说明",
    "content": "DeepSeek 提供兼容 OpenAI 风格的 API 接口，基础 URL 为 https://api.deepseek.com/v1，主要包含 chat.completions 和 embeddings 等。调用时需在 Header 中携带 Authorization: Bearer YOUR_API_KEY。"
  },
  {
    "title": "RAG 技术架构简介",
    "content": "RAG（检索增强生成）通过向量检索选出相关文档片段，再将其与用户问题一起输入大模型，从而生成更符合事实的回答。典型步骤包括：文档向量化、构建索引、查询检索、构造 Prompt 并调用 LLM。"
  },
  {
    "title": "Python 实现 RAG 的常见做法",
    "content": "在 Python 中实现 RAG 常用 sentence-transformers 来生成向量，使用 scikit-learn 或 FAISS 做近邻检索，再调用 OpenAI/DeepSeek 等模型进行生成。关键点在于分段粒度、向量模型选择和 Prompt 设计。"
  }
]

真实项目里，你可以写个脚本把 PDF/Markdown 拆分成一段段文本后写成这个格式。

---

五、核心代码实现：RAG + DeepSeek

下面这份代码演示了一个从加载知识库 → 构建向量索引 → 调用 DeepSeek 回答问题的完整闭环，你可以保存为 rag_deepseek_demo.py。

5.1 完整代码

# -*- coding: utf-8 -*-
"""
大模型实战：基于 DeepSeek 的 RAG 知识库问答 Demo
"""

import os
import json
import logging
from typing import List, Dict, Any, Optional

from openai import OpenAI          # 用 OpenAI SDK 调 DeepSeek
from sentence_transformers import SentenceTransformer
from sklearn.neighbors import NearestNeighbors

# ========= 日志配置 =========
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# ========= DeepSeek 配置 =========
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY", "YOUR_API_KEY")
DEEPSEEK_BASE_URL = "https://api.deepseek.com/v1"
DEEPSEEK_MODEL = "deepseek-reasoner-671b"   # 根据实际可用模型替换

# 初始化 DeepSeek 客户端（兼容 OpenAI 风格）
client = OpenAI(api_key=DEEPSEEK_API_KEY, base_url=DEEPSEEK_BASE_URL)

# ========= 向量检索相关 =========
EMBEDDING_MODEL = "all-MiniLM-L6-v2"  # 通用中文/英文向量模型

def load_knowledge_base(path: str) -> List[Dict[str, str]]:
    """加载 JSON 格式的知识库"""
    with open(path, "r", encoding="utf-8") as f:
        docs = json.load(f)
    if not isinstance(docs, list):
        raise ValueError("知识库文件必须是 JSON 数组")
    for d in docs:
        if "content" not in d:
            raise ValueError("每条记录必须包含 content 字段")
    logger.info("成功加载 %d 条文档", len(docs))
    return docs

class VectorDB:
    """简单的向量数据库封装"""

    def __init__(self, docs: List[Dict[str, str]], model_name: str = EMBEDDING_MODEL):
        self.docs = docs
        self.model_name = model_name
        self._embed_model: Optional[SentenceTransformer] = None
        self.embeddings = None
        self.nn = None

    def _get_model(self) -> SentenceTransformer:
        if self._embed_model is None:
            logger.info("加载向量模型：%s", self.model_name)
            self._embed_model = SentenceTransformer(self.model_name)
        return self._embed_model

    def build(self, n_neighbors: int = 5):
        """生成所有文档向量并构建近邻索引"""
        model = self._get_model()
        texts = [d["content"] for d in self.docs]
        logger.info("开始生成 %d 条文档向量...", len(texts))
        self.embeddings = model.encode(
            texts,
            convert_to_tensor=True,
            show_progress_bar=True
        )
        # n_neighbors 不能超过样本数
        n_neighbors = min(n_neighbors, len(self.docs))
        self.nn = NearestNeighbors(
            n_neighbors=n_neighbors,
            metric="cosine"
        )
        self.nn.fit(self.embeddings)
        logger.info("向量索引构建完成（n_neighbors=%d）", n_neighbors)

    def search(self, query: str, top_k: int = 3) -> List[Dict[str, Any]]:
        """按语义相似度搜索最相关的文档"""
        if self.nn is None or self.embeddings is None:
            raise RuntimeError("请先调用 build() 构建索引")

        model = self._get_model()
        q_emb = model.encode([query], convert_to_tensor=True)
        distances, indices = self.nn.kneighbors(q_emb)

        results = []
        for idx, dist in zip(indices[0], distances[0]):
            doc = self.docs[int(idx)]
            results.append({
                "doc_id": int(idx),
                "score": float(1 - dist),  # 余弦距离转相似度
                "title": doc.get("title", ""),
                "content": doc["content"]
            })
        # 根据相似度排序并截断
        results.sort(key=lambda x: x["score"], reverse=True)
        return results[:top_k]

# ========= DeepSeek 调用封装 =========

def call_deepseek(prompt: str,
                  model: str = DEEPSEEK_MODEL,
                  temperature: float = 0.2,
                  max_tokens: int = 1024) -> str:
    """调用 DeepSeek Chat 接口"""
    try:
        resp = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            temperature=temperature,
            max_tokens=max_tokens
        )
        return resp.choices[0].message.content.strip()
    except Exception as e:
        logger.error("调用 DeepSeek 失败：%s", e)
        return f"调用 DeepSeek 失败：{e}"

class RAGWithDeepSeek:
    """RAG 主流程封装"""

    def __init__(self, vector_db: VectorDB):
        self.vector_db = vector_db

    def build_prompt(self, question: str, docs: List[Dict[str, Any]]) -> str:
        """将检索到的文档 + 问题拼成 Prompt"""
        context_blocks = []
        for i, d in enumerate(docs, start=1):
            title = d["title"] or f"文档{i}"
            context_blocks.append(f"【{title}】\n{d['content']}")
        context_text = "\n\n".join(context_blocks)

        prompt = f"""你是一个专业的企业知识库问答助手，请严格依据给定的“上下文”回答用户问题，避免凭空编造。如果上下文中没有相关信息，请明确说明“从现有文档中无法找到答案”。

[上下文文档]
{context_text}

[用户问题]
{question}

请给出：
1. 直接可读的答案
2. 如有必要，简单说明你参考了哪些文档（用标题概述即可）
"""
        return prompt

    def answer(self, question: str, top_k: int = 3) -> Dict[str, Any]:
        """完整流程：检索 → 构造 Prompt → 调用 DeepSeek"""
        retrieved = self.vector_db.search(question, top_k=top_k)
        prompt = self.build_prompt(question, retrieved)
        answer = call_deepseek(prompt)
        return {
            "question": question,
            "answer": answer,
            "retrieved_docs": retrieved
        }

# ========= demo 入口 =========

def main():
    # 1. 加载知识库
    kb_path = "data/knowledge_base.json"
    docs = load_knowledge_base(kb_path)

    # 2. 构建向量索引
    vdb = VectorDB(docs)
    vdb.build(n_neighbors=4)

    # 3. 初始化 RAG 系统
    rag = RAGWithDeepSeek(vdb)

    # 4. 测试几个典型问题
    questions = [
        "DeepSeek-Reasoner 适合做什么类型的应用？",
        "我应该如何调用 DeepSeek 的 API？",
        "什么是 RAG 架构？大概流程是怎样的？"
    ]

    for q in questions:
        print("=" * 60)
        print("问题：", q)
        result = rag.answer(q)
        print("\n【答案】\n", result["answer"])
        print("\n【命中的文档标题】")
        for d in result["retrieved_docs"]:
            print(f"- {d['title']} (score={d['score']:.3f})")
        print("=" * 60, "\n")

if __name__ == "__main__":
    main()

提示：把 YOUR_API_KEY 换成真实 Key，把 knowledge_base.json 换成你自己的文档，就可以跑通一个最小可用的 DeepSeek RAG Demo。

---

六、运行与验证效果

6.1 运行

python rag_deepseek_demo.py

终端里你会看到类似输出（示意）：

============================================================
问题： DeepSeek-Reasoner 适合做什么类型的应用？

【答案】
DeepSeek-Reasoner 是 DeepSeek 系列中专注多步推理和逻辑分析的模型，适合用于构建问答系统、
决策支持、知识库检索增强（RAG）等需要较强推理能力的场景……

【命中的文档标题】
- DeepSeek 模型简介 (score=0.89)
- RAG 技术架构简介 (score=0.75)
============================================================

可以看到：

• 模型的回答会明显参考你知识库中的描述；
• 同时你还能看到它参考了哪些文档，方便排查和解释。

---

七、落地实战中的关键细节与踩坑点

7.1 文档切分粒度

• 太长：一条记录 3000+ 字，向量会“稀释”语义，检索不到关键点
• 太短：一句话就一条记录，向量库膨胀、检索噪声增多

经验值：

• 中文文档：200～500 字为一段比较合适
• 代码/接口文档：按“一个接口/一个配置项说明”为粒度分段

7.2 向量模型选择

示例中用了 all-MiniLM-L6-v2，原因：

• 开源免费，推理快（CPU 就能用）
• 中英文都还可以，适合 Demo 和很多中小团队的场景

如果对中文语义检索要求更高，可以换成 中文优化的 sentence-transformers 模型，改一下：

python复制

EMBEDDING_MODEL = "your-chinese-embedding-model-name"

7.3 DeepSeek 模型选择与参数

• 场景：知识库问答 / 多步推理 → 推荐使用 Reasoner 系列
• temperature：
  • 0.1～0.3：更稳、更少瞎编，适合知识问答

  • 0.7：更有创造力，但幻觉多，一般不推荐给企业知识库

代码里的：

call_deepseek(prompt, temperature=0.2)

已经是偏稳的设置，可以根据实际情况微调。

7.4 降低“幻觉”的实战技巧

1. Prompt 中明确要求“严格依据上下文”“没有就说不知道”
2. 在业务侧加一个简单规则：
   • 如果检索相似度都很低（比如 score < 0.4），直接告诉用户“目前知识库里没有相关内容”
3. 可以把命中的文档标题 + 片段一并展示给用户，让用户自己判断可信度

---

八、如何扩展成真正的企业应用？

在上面的 Demo 基础上，你可以很自然地往下扩展：

1. 接公司内部 SSO
   对不同部门，只暴露不同子集的文档（权限控制）。

2. 支持文档在线更新
   定期从 Confluence/Wiki/Git 仓库同步文档，自动重新向量化和建索引。

3. 加一个简单 Web 界面
   用 streamlit 或 Vue + FastAPI 做一个提问界面和“引用文档展示”。

4. 加反馈闭环

   • 用户可以点“有用/没用”
   • 记录问题/答案/点击行为
   • 之后用这些数据微调提示词或精调检索模型

---

九、总结

我把DeepSeek 应用案例”完整走了一遍：

• 用本地向量模型 + DeepSeek 推理模型 搭建 RAG 应用
• 从 知识库 JSON 格式 → 向量索引 → Prompt 设计 → DeepSeek 调用 全流程给出了可运行代码
• 给出了在真实场景中需要注意的 切分粒度、向量模型选择、避免幻觉 等实践经验

可以直接做下面几件事，把 Demo 变成自己的应用：

1. 把 knowledge_base.json 换成你们自己的内部文档；
2. 把 DEEPSEEK_API_KEY 换成真实的 Key，确认用的模型名称正确；
3. 在此基础上加一个简单 Web 界面，就可以给同事试用了。