【DeepSeek实战】基于 V4 的企业级 RAG 系统：私有知识库问答实战

pip install langchain deepseek-sdk chromadb unstructured[pdf] tiktoken

from langchain.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
def load_and_split_documents(docs_dir: str = "./docs"):
    """
    加载文档并进行智能切片
    """
    # 加载 Markdown 文档
    loader = DirectoryLoader(docs_dir, glob="**/*.md")
    documents = loader.load()
    # 智能切片：保持段落完整性
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,      # 每块 1000 字符
        chunk_overlap=200,    # 重叠 200 字符，保持上下文连贯
        separators=["\n\n", "\n", "。", "！", "？", " ", ""]
    )
    chunks = text_splitter.split_documents(documents)
    print(f"共切分为 {len(chunks)} 个文档块")
    return chunks

import chromadb
from chromadb.utils import embedding_functions
from langchain.embeddings import HuggingFaceEmbeddings
class VectorStoreManager:
    def __init__(self, persist_directory: str = "./chroma_db"):
        self.client = chromadb.PersistentClient(path=persist_directory)
        # 使用本地嵌入模型（也可替换为 OpenAI Embedding）
        self.embedding_func = HuggingFaceEmbeddings(
            model_name="BAAI/bge-m3",
            model_kwargs={'device': 'cpu'}
        )
        # 创建或获取集合
        self.collection = self.client.get_or_create_collection(
            name="company_docs",
            embedding_function=self.embedding_func
        )
    def add_documents(self, chunks):
        """
        将文档块存入向量数据库
        """
        for i, chunk in enumerate(chunks):
            self.collection.add(
                ids=[f"doc_{i}"],
                documents=[chunk.page_content],
                metadatas=[{
                    "source": chunk.metadata.get("source", ""),
                    "page": chunk.metadata.get("page", 0)
                }]
            )
        print(f"成功存入 {len(chunks)} 个文档块")
    def search(self, query: str, top_k: int = 5):
        """
        语义检索相关文档
        """
        results = self.collection.query(
            query_texts=[query],
            n_results=top_k
        )
        return results['documents'][0], results['metadatas'][0]

from deepseek import AsyncDeepSeek
import asyncio
class RAGChatBot:
    def __init__(self, api_key: str):
        self.client = AsyncDeepSeek(api_key=api_key)
        self.vector_store = VectorStoreManager()
    async def answer_question(self, question: str) -> dict:
        """
        回答用户问题
        :return: {"answer": "...", "sources": [...]}
        """
        # 1. 检索相关文档
        contexts, metadatas = self.vector_store.search(question, top_k=3)
        # 2. 构建 Prompt
        context_text = "\n\n".join(contexts)
        prompt = f"""
        你是一个专业的技术助手。请根据以下参考资料回答问题。
        参考资料:
{context_text}
        问题: {question}
        要求:
        1. 答案必须基于参考资料，不要编造
        2. 如果资料中没有相关信息，请明确说明
        3. 在答案末尾列出引用的文档来源
        4. 语言简洁专业，不超过 300 字
        答案:
        """
        # 3. 调用 DeepSeek V4 生成答案
        response = await self.client.chat.completions.create(
            model="deepseek-chat",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3
        )
        answer = response.choices[0].message.content
        # 4. 提取来源链接
        sources = [meta.get("source", "") for meta in metadatas]
        return {
            "answer": answer,
            "sources": list(set(sources))
        }
# 使用示例
async def main():
    bot = RAGChatBot(api_key=os.getenv("DEEPSEEK_API_KEY"))
    result = await bot.answer_question("如何配置公司的 CI/CD 流水线？")
    print(f"答案: {result['answer']}")
    print(f"来源: {result['sources']}")
asyncio.run(main())

传统人工客服年度成本:
├── 人力成本: 10人 × ¥15,000/月 × 12 = ¥1,800,000
├── 培训费用: ¥50,000/年
├── 管理成本: ¥100,000/年
└── 总计:       ¥1,950,000
RAG 智能问答年度成本:
├── 服务器费用: ¥8,000/月 × 12 = ¥96,000
├── API 费用:   500次 × 250天 × ¥0.3/次 = ¥37,500
├── 维护人力:   1人 × ¥15,000/月 × 12 = ¥180,000
└── 总计:       ¥313,500
🎉 年度节省: ¥1,636,500 (约 164 万元)