DeepSeek 实战案例分享｜附源码

本文将通过一个完整的实战案例，带你使用 DeepSeek API 搭建一个“企业知识库智能问答助手”。文章包含需求分析、技术架构、核心流程、完整源码示例、运行方式以及优化思路，适合想要将 DeepSeek 接入真实业务场景的开发者参考。

一、为什么选择 DeepSeek 做实战项目？

近几年，大语言模型在企业内部知识管理、智能客服、代码辅助、内容生成、数据分析等场景中得到了大量应用。相比传统搜索系统，大模型不仅能够理解用户问题，还可以基于上下文生成更自然、更准确的回答。

DeepSeek 作为国产大模型中的优秀代表，具备以下优势：

1. 中文理解能力强
   对中文语义、业务文本、技术文档的理解效果较好，适合国内企业场景。

2. API 接入简单
   开发者可以通过标准 HTTP 接口快速调用模型能力，接入成本低。

3. 性价比较高
   在很多文本生成、问答、总结类任务中，可以用较低成本获得不错的效果。

4. 适合二次开发
   可以结合向量数据库、RAG、Agent、工作流等技术，构建业务应用。

本文选择一个非常典型、也非常实用的案例：基于 DeepSeek 的企业知识库问答助手。

二、项目效果预览

我们要实现的系统具备以下能力：

用户输入问题，例如：

“公司年假制度是怎么规定的？”

系统会从本地知识库中检索相关内容，并调用 DeepSeek 生成回答，例如：

根据知识库内容，公司员工入职满一年后可享受带薪年假。年假天数根据工龄不同分为 5 天、10 天和 15 天。员工申请年假需提前在 OA 系统提交申请，并经直属主管审批通过后方可休假。

相比直接把问题丢给大模型，这种方案有明显优势：

• 回答更贴合企业内部资料；
• 可以降低模型胡编乱造的概率；
• 能够回答模型训练数据中不存在的私有信息；
• 后续可以持续更新知识库，而不必重新训练模型。

三、技术方案说明

本案例采用常见的 RAG 架构。

RAG 全称是 Retrieval-Augmented Generation，即“检索增强生成”。它的基本思想是：

1. 先把企业文档切分成多个文本片段；
2. 为每个片段生成向量；
3. 用户提问时，也将问题转换为向量；
4. 在向量数据库中查找最相关的文本片段；
5. 将检索到的内容作为上下文，交给 DeepSeek；
6. DeepSeek 根据上下文生成最终回答。

整体流程如下：

用户问题
   ↓
问题向量化
   ↓
向量数据库检索
   ↓
获取相关知识片段
   ↓
拼接 Prompt
   ↓
调用 DeepSeek API
   ↓
返回答案

四、项目技术栈

本项目使用 Python 实现，整体技术栈如下：

这里为了方便演示，我们使用本地 FAISS 向量库。如果是生产环境，也可以替换为 Milvus、Qdrant、Weaviate、Elasticsearch Vector 等方案。

五、项目目录结构

建议项目目录如下：

deepseek-rag-demo/
├── app.py
├── build_index.py
├── knowledge_base/
│   └── company_policy.txt
├── vector_store/
│   ├── index.faiss
│   └── chunks.json
├── requirements.txt
└── .env

各文件说明：

六、准备知识库文件

我们先创建一个简单的企业制度文档，路径为：

knowledge_base/company_policy.txt

示例内容如下：

公司考勤制度：

员工标准工作时间为周一至周五，上午 9:00 至下午 18:00，中午 12:00 至 13:30 为午休时间。员工每日应按时打卡，迟到或早退需在 OA 系统中说明原因。

公司年假制度：

员工入职满一年后可享受带薪年假。累计工作满 1 年不满 10 年的员工，每年可享受 5 天年假；累计工作满 10 年不满 20 年的员工，每年可享受 10 天年假；累计工作满 20 年及以上的员工，每年可享受 15 天年假。年假申请需至少提前 3 个工作日在 OA 系统提交，并经直属主管审批。

公司报销制度：

员工因公产生的交通费、住宿费、餐饮费等费用，可按照公司财务制度申请报销。报销时需提供真实有效的发票，并在费用发生后的 30 天内提交报销申请。超过 30 天未提交的，原则上不予报销。

公司远程办公制度：

员工因特殊情况需要远程办公的，应提前向直属主管申请，并说明远程办公原因、时间和工作安排。远程办公期间，员工应保持通讯畅通，并按时完成工作任务。

实际业务中，这里可以放置更多文档，比如：

• 公司制度；
• 产品说明；
• 技术文档；
• 售后手册；
• 合同模板；
• 培训资料；
• FAQ 文档。

七、安装依赖

创建 requirements.txt：

fastapi==0.115.0
uvicorn==0.30.6
python-dotenv==1.0.1
requests==2.32.3
faiss-cpu==1.8.0.post1
sentence-transformers==3.0.1
numpy==1.26.4

安装依赖：

pip install -r requirements.txt

如果你使用的是国内网络环境，安装 sentence-transformers 可能稍慢，可以配置镜像源：

pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

八、配置 DeepSeek API Key

创建 .env 文件：

DEEPSEEK_API_KEY=你的 DeepSeek API Key
DEEPSEEK_API_URL=https://api.deepseek.com/chat/completions
DEEPSEEK_MODEL=deepseek-chat

注意：

• 不要把真实 API Key 提交到 GitHub；
• 生产环境建议使用密钥管理服务；
• 如果项目部署到服务器，可以通过环境变量注入。

九、构建知识库向量索引源码

创建 build_index.py 文件。

该脚本负责：

1. 读取知识库文本；
2. 将长文本切分为多个 chunk；
3. 使用本地 Embedding 模型生成向量；
4. 构建 FAISS 索引；
5. 保存索引和文本片段。

完整源码如下：

import os
import json
import faiss
import numpy as np
from sentence_transformers import SentenceTransformer


KNOWLEDGE_DIR = "knowledge_base"
VECTOR_DIR = "vector_store"
INDEX_PATH = os.path.join(VECTOR_DIR, "index.faiss")
CHUNKS_PATH = os.path.join(VECTOR_DIR, "chunks.json")


def load_documents():
    """
    读取 knowledge_base 目录下的所有 txt 文件。
    """
    documents = []

    for filename in os.listdir(KNOWLEDGE_DIR):
        if filename.endswith(".txt"):
            file_path = os.path.join(KNOWLEDGE_DIR, filename)

            with open(file_path, "r", encoding="utf-8") as f:
                content = f.read()

            documents.append({
                "filename": filename,
                "content": content
            })

    return documents


def split_text(text, chunk_size=300, overlap=50):
    """
    将长文本切分为多个片段。

    参数说明：
    - chunk_size：每个片段最大长度
    - overlap：相邻片段之间的重叠长度

    加入 overlap 的好处是避免关键信息刚好被切断。
    """
    chunks = []
    start = 0
    text_length = len(text)

    while start < text_length:
        end = start + chunk_size
        chunk = text[start:end]

        if chunk.strip():
            chunks.append(chunk.strip())

        start += chunk_size - overlap

    return chunks


def build_index():
    """
    构建 FAISS 向量索引。
    """
    os.makedirs(VECTOR_DIR, exist_ok=True)

    print("正在加载 Embedding 模型...")
    model = SentenceTransformer("sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2")

    print("正在读取知识库文档...")
    documents = load_documents()

    all_chunks = []

    for doc in documents:
        chunks = split_text(doc["content"])

        for i, chunk in enumerate(chunks):
            all_chunks.append({
                "filename": doc["filename"],
                "chunk_id": i,
                "text": chunk
            })

    if not all_chunks:
        raise ValueError("知识库为空，请在 knowledge_base 目录下添加 txt 文档。")

    texts = [item["text"] for item in all_chunks]

    print(f"共切分出 {len(texts)} 个文本片段，正在生成向量...")
    embeddings = model.encode(texts, normalize_embeddings=True)

    embeddings = np.array(embeddings).astype("float32")

    dimension = embeddings.shape[1]

    print("正在构建 FAISS 索引...")
    index = faiss.IndexFlatIP(dimension)
    index.add(embeddings)

    faiss.write_index(index, INDEX_PATH)

    with open(CHUNKS_PATH, "w", encoding="utf-8") as f:
        json.dump(all_chunks, f, ensure_ascii=False, indent=2)

    print("向量索引构建完成。")
    print(f"索引文件：{INDEX_PATH}")
    print(f"文本片段文件：{CHUNKS_PATH}")


if __name__ == "__main__":
    build_index()

运行命令：

python build_index.py

如果运行成功，你会看到类似输出：

正在加载 Embedding 模型...
正在读取知识库文档...
共切分出 4 个文本片段，正在生成向量...
正在构建 FAISS 索引...
向量索引构建完成。

十、编写 DeepSeek 问答接口源码

接下来创建 app.py，实现一个 HTTP 问答接口。

该接口核心流程为：

1. 接收用户问题；
2. 加载 FAISS 索引；
3. 对用户问题做向量化；
4. 检索 Top K 相关文本；
5. 组装 Prompt；
6. 调用 DeepSeek；
7. 返回最终答案。

完整源码如下：

import os
import json
import faiss
import requests
import numpy as np

from dotenv import load_dotenv
from fastapi import FastAPI
from pydantic import BaseModel
from sentence_transformers import SentenceTransformer


load_dotenv()

DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
DEEPSEEK_API_URL = os.getenv("DEEPSEEK_API_URL", "https://api.deepseek.com/chat/completions")
DEEPSEEK_MODEL = os.getenv("DEEPSEEK_MODEL", "deepseek-chat")

INDEX_PATH = "vector_store/index.faiss"
CHUNKS_PATH = "vector_store/chunks.json"

app = FastAPI(title="DeepSeek RAG Demo", description="基于 DeepSeek 的知识库问答系统")

print("正在加载 Embedding 模型...")
embedding_model = SentenceTransformer("sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2")

print("正在加载 FAISS 索引...")
index = faiss.read_index(INDEX_PATH)

print("正在加载文本片段...")
with open(CHUNKS_PATH, "r", encoding="utf-8") as f:
    chunks = json.load(f)


class ChatRequest(BaseModel):
    question: str
    top_k: int = 3


class ChatResponse(BaseModel):
    question: str
    answer: str
    references: list


def retrieve_context(question: str, top_k: int = 3):
    """
    根据用户问题检索最相关的知识片段。
    """
    query_embedding = embedding_model.encode(
        [question],
        normalize_embeddings=True
    )

    query_embedding = np.array(query_embedding).astype("float32")

    scores, indices = index.search(query_embedding, top_k)

    results = []

    for score, idx in zip(scores[0], indices[0]):
        if idx == -1:
            continue

        item = chunks[idx]
        results.append({
            "score": float(score),
            "filename": item["filename"],
            "chunk_id": item["chunk_id"],
            "text": item["text"]
        })

    return results


def build_prompt(question: str, contexts: list):
    """
    构造提交给 DeepSeek 的 Prompt。
    """
    context_text = "\n\n".join([
        f"【资料 {i + 1}】\n{ctx['text']}"
        for i, ctx in enumerate(contexts)
    ])

    prompt = f"""
你是一个严谨的企业知识库问答助手。请根据下面提供的资料回答用户问题。

要求：
1. 只能基于资料内容进行回答；
2. 如果资料中没有明确答案，请回答“根据现有资料无法确定”；
3. 回答要清晰、简洁、准确；
4. 如果涉及制度、流程、时间要求，请尽量列出关键条件。

资料内容：
{context_text}

用户问题：
{question}

请给出答案：
"""

    return prompt.strip()


def call_deepseek(prompt: str):
    """
    调用 DeepSeek Chat Completions API。
    """
    if not DEEPSEEK_API_KEY:
        raise ValueError("请先在 .env 文件中配置 DEEPSEEK_API_KEY")

    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {DEEPSEEK_API_KEY}"
    }

    payload = {
        "model": DEEPSEEK_MODEL,
        "messages": [
            {
                "role": "system",
                "content": "你是一个专业、可靠、严谨的中文企业知识库问答助手。"
            },
            {
                "role": "user",
                "content": prompt
            }
        ],
        "temperature": 0.2,
        "stream": False
    }

    response = requests.post(
        DEEPSEEK_API_URL,
        headers=headers,
        json=payload,
        timeout=60
    )

    response.raise_for_status()

    data = response.json()

    return data["choices"][0]["message"]["content"]


@app.post("/chat", response_model=ChatResponse)
def chat(req: ChatRequest):
    """
    知识库问答接口。
    """
    contexts = retrieve_context(req.question, req.top_k)

    prompt = build_prompt(req.question, contexts)

    answer = call_deepseek(prompt)

    return ChatResponse(
        question=req.question,
        answer=answer,
        references=contexts
    )


@app.get("/")
def root():
    return {
        "message": "DeepSeek RAG Demo is running.",
        "usage": "POST /chat with JSON: {\"question\": \"公司年假制度是什么？\"}"
    }

十一、启动服务并测试

启动服务：

uvicorn app:app --reload --host 0.0.0.0 --port 8000

打开浏览器访问：

http://localhost:8000

你会看到：

{
  "message": "DeepSeek RAG Demo is running.",
  "usage": "POST /chat with JSON: {\"question\": \"公司年假制度是什么？\"}"
}

使用 curl 测试问答接口：

curl -X POST "http://localhost:8000/chat" \
  -H "Content-Type: application/json" \
  -d '{"question":"公司年假制度是怎么规定的？","top_k":3}'

可能返回：

{
  "question": "公司年假制度是怎么规定的？",
  "answer": "根据资料，公司员工入职满一年后可享受带薪年假。累计工作满1年不满10年的员工，每年可享受5天年假；累计工作满10年不满20年的员工，每年可享受10天年假；累计工作满20年及以上的员工，每年可享受15天年假。年假申请需至少提前3个工作日在OA系统提交，并经直属主管审批。",
  "references": [
    {
      "score": 0.82,
      "filename": "company_policy.txt",
      "chunk_id": 1,
      "text": "公司年假制度：员工入职满一年后可享受带薪年假..."
    }
  ]
}

十二、核心代码解析

1. 为什么需要文本切分？

企业知识库文档通常比较长，如果直接把整篇文档传给大模型，会遇到几个问题：

• Token 成本高；
• 上下文窗口有限；
• 相关信息被大量无关内容淹没；
• 响应速度变慢。

因此，我们会先将文档切分成多个文本块。用户提问时，只取最相关的几个片段交给模型。

本案例中使用了一个简单的字符长度切分方法：

def split_text(text, chunk_size=300, overlap=50):
    chunks = []
    start = 0
    text_length = len(text)

    while start < text_length:
        end = start + chunk_size
        chunk = text[start:end]

        if chunk.strip():
            chunks.append(chunk.strip())

        start += chunk_size - overlap

    return chunks

这种方式简单易懂，但在生产环境中可以进一步优化，比如：

• 按标题切分；
• 按段落切分；
• 按 Markdown 层级切分；
• 按语义边界切分；
• 对表格和代码块做特殊处理。

2. 为什么使用向量检索？

传统关键词搜索依赖词面匹配，例如用户问：

“休假要提前多久申请？”

文档中写的是：

“年假申请需至少提前 3 个工作日在 OA 系统提交。”

如果使用关键词搜索，可能因为“休假”和“年假”、“多久”和“3 个工作日”没有直接重合，导致检索效果不稳定。

向量检索的优势是可以理解语义相似度。即使用户表达方式不同，只要语义接近，也能找到相关内容。

本案例中使用：

SentenceTransformer("sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2")

这是一个支持多语言的轻量级 Embedding 模型，适合本地演示。如果追求更高效果，可以替换为效果更好的向量模型。

3. 为什么 Prompt 很重要？

RAG 系统不是简单地把资料拼接给模型，而是要通过 Prompt 约束模型行为。

例如本文中的 Prompt：

你是一个严谨的企业知识库问答助手。请根据下面提供的资料回答用户问题。

要求：
1. 只能基于资料内容进行回答；
2. 如果资料中没有明确答案，请回答“根据现有资料无法确定”；
3. 回答要清晰、简洁、准确；
4. 如果涉及制度、流程、时间要求，请尽量列出关键条件。

这段提示词的作用是：

• 降低模型编造答案的概率；
• 明确回答边界；
• 要求模型遵守资料；
• 让回答更适合企业制度类场景。

在真实项目中，Prompt 设计往往需要反复测试和迭代。

十三、加入简单前端页面

如果你不想每次都用 curl 调接口，可以加一个简单 HTML 页面。

在 app.py 中增加如下接口：

from fastapi.responses import HTMLResponse


@app.get("/ui", response_class=HTMLResponse)
def ui():
    return """



    
    
    


    
DeepSeek 企业知识库问答助手
    
    

    提交问题

    
答案会显示在这里。

    


"""

重启服务后访问：

http://localhost:8000/ui

这样就能通过网页进行测试了。

十四、生产环境优化建议

本文案例是一个适合入门和原型验证的版本，如果要上线到真实业务环境，还需要从多个方面进行增强。

1. 文档解析能力

当前只支持 TXT 文件，生产环境通常需要支持：

• PDF；
• Word；
• Excel；
• Markdown；
• HTML；
• 飞书文档；
• 语雀文档；
• Confluence 页面。

可以引入以下工具：

• pypdf;
• python-docx;
• openpyxl;
• unstructured;
• markitdown;
• 自研文档解析服务。

2. 权限控制

企业知识库往往包含敏感信息，不同员工能访问的文档范围不同。

例如：

• HR 制度所有员工可见；
• 财务数据仅财务部门可见；
• 技术架构仅研发部门可见；
• 合同信息仅销售和法务可见。

因此，检索时不能只做向量相似度，还要加入权限过滤。

常见做法是给每个 chunk 增加元数据：

{
  "text": "某段知识内容",
  "department": "finance",
  "permission": ["finance_manager", "cfo"]
}

检索时根据当前用户身份过滤结果。

3. 答案可溯源

企业场景中，用户通常不仅想知道答案，还希望知道答案来自哪里。

可以在回答中加入引用来源：

答案：
员工入职满一年后可以享受带薪年假。

参考来源：
1. company_policy.txt，第 2 段，公司年假制度

这样可以提高用户信任度，也方便后续核查。

4. 增加重排序模型

向量检索通常先召回 Top 10 或 Top 20，然后使用 rerank 模型进行重排序。

流程变为：

用户问题
  ↓
向量检索 Top 20
  ↓
Rerank 重排序
  ↓
选取 Top 3
  ↓
调用 DeepSeek 生成答案

这样能显著提升复杂问题下的检索准确率。

5. 对话历史管理

如果用户连续提问：

用户：公司年假制度是什么？
用户：那要提前多久申请？

第二个问题中的“那”依赖上一轮上下文。此时需要引入对话历史，将上下文一起传给模型，或者先让模型改写问题：

将用户的追问改写为独立问题：
历史问题：公司年假制度是什么？
当前问题：那要提前多久申请？
改写后：公司年假需要提前多久申请？

然后再进行检索。

6. 防止幻觉与越权回答

可以从以下方面控制：

• Prompt 中明确要求只基于资料回答；
• 检索结果相似度低时拒答；
• 对答案进行二次校验；
• 对敏感问题加入安全策略；
• 对模型输出进行审计和日志记录。

例如，如果最高相似度低于阈值，可以直接返回：

if contexts[0]["score"] < 0.4:
    return "根据现有资料无法确定。"

十五、常见问题排查

1. FAISS 索引文件不存在

如果启动 app.py 报错：

RuntimeError: could not open vector_store/index.faiss

说明你还没有构建索引，需要先运行：

python build_index.py

2. DeepSeek API Key 未配置

如果报错：

请先在 .env 文件中配置 DEEPSEEK_API_KEY

检查 .env 文件是否存在，并确认变量名是否正确：

DEEPSEEK_API_KEY=你的真实Key

3. Embedding 模型下载失败

首次运行会自动下载模型，如果网络不稳定，可以：

• 设置 HuggingFace 镜像；
• 手动下载模型；
• 使用国内可访问的 Embedding API；
• 将模型放到本地路径加载。

例如：

model = SentenceTransformer("./models/paraphrase-multilingual-MiniLM-L12-v2")

十六、进一步扩展：支持流式输出

在真实聊天产品中，用户更喜欢看到答案逐字输出，而不是等待全部生成完毕后一次性返回。

DeepSeek API 支持流式输出时，可以将：

"stream": False

改为：

"stream": True

然后使用 Python 的流式请求处理响应。FastAPI 中可以通过 StreamingResponse 返回流式内容。

示例思路：

from fastapi.responses import StreamingResponse


def call_deepseek_stream(prompt: str):
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {DEEPSEEK_API_KEY}"
    }

    payload = {
        "model": DEEPSEEK_MODEL,
        "messages": [
            {"role": "system", "content": "你是一个企业知识库问答助手。"},
            {"role": "user", "content": prompt}
        ],
        "temperature": 0.2,
        "stream": True
    }

    with requests.post(
        DEEPSEEK_API_URL,
        headers=headers,
        json=payload,
        stream=True,
        timeout=60
    ) as response:
        response.raise_for_status()

        for line in response.iter_lines():
            if line:
                decoded = line.decode("utf-8")
                yield decoded + "\n"


@app.post("/chat-stream")
def chat_stream(req: ChatRequest):
    contexts = retrieve_context(req.question, req.top_k)
    prompt = build_prompt(req.question, contexts)

    return StreamingResponse(
        call_deepseek_stream(prompt),
        media_type="text/event-stream"
    )

实际项目中，还需要解析 SSE 数据格式，只把增量文本返回给前端。

十七、项目完整运行步骤总结

从零开始运行本案例，只需要以下步骤：

第一步：创建项目

mkdir deepseek-rag-demo
cd deepseek-rag-demo

第二步：创建目录

mkdir knowledge_base
mkdir vector_store

第三步：安装依赖

pip install -r requirements.txt

第四步：配置 .env

DEEPSEEK_API_KEY=你的 DeepSeek API Key
DEEPSEEK_API_URL=https://api.deepseek.com/chat/completions
DEEPSEEK_MODEL=deepseek-chat

第五步：准备知识库文档

将公司制度、FAQ 或产品说明放入：

knowledge_base/

第六步：构建索引

python build_index.py

第七步：启动服务

uvicorn app:app --reload --port 8000

第八步：调用接口

curl -X POST "http://localhost:8000/chat" \
  -H "Content-Type: application/json" \
  -d '{"question":"报销需要在多久内提交？","top_k":3}'

十八、结语

本文通过一个完整案例，演示了如何使用 DeepSeek API 构建企业知识库问答助手。这个案例虽然不复杂，但已经覆盖了 RAG 应用的核心链路：

• 文档加载；
• 文本切分；
• 向量化；
• 向量检索；
• Prompt 构建；
• DeepSeek 调用；
• API 服务封装；
• 前端页面测试。

如果你正在尝试把大模型能力落地到业务系统中，RAG 是非常值得优先实践的方向。它不需要重新训练大模型，却可以让模型回答私有知识库中的内容，非常适合企业内部助手、智能客服、售后支持、技术文档问答、合同审查辅助等场景。

后续你可以基于本文源码继续扩展，例如：

• 接入数据库；
• 增加用户登录；
• 支持多知识库；
• 增加文档上传；
• 支持 PDF 和 Word 解析；
• 接入权限系统；
• 增加答案引用；
• 支持流式输出；
• 部署到 Docker 或 Kubernetes。

通过这些优化，一个简单 Demo 就可以逐步演进为真正可用的企业级 AI 应用。