3分钟！教会你用Doris+DeepSeek搭建RAG知识库（喂饭级教程）

清晨，我泡上一杯98年的咖啡，打开电脑，收到一条来自Doris用户的消息： 

"请问有 将doris数据库数据作为deepseek大模型的知识库的方法吗" 

"好滴，安排！" 

于是，和这位Doris用户聊了聊，发现这类需求随着DeepSeek的爆火被无限放大了，但也确实是一个‌引申的机遇与挑战。 

也罢，今天就来一探究竟，一起学习学习如何基于Doris+DeepSeek打造你的专属AI助手！

术语名词

在开始之前，先用大白话，介绍下几个术语名词：

1️⃣ AGI：通用人工智能，AI界的"全能学霸"。致力于像人类一样能学数学、写诗、看病，甚至自己发明工具，目前还是科幻级存在。

2️⃣ AIGC：人工智能生成，AI的"创作工厂"。输入一句话，自动生成文章/插画/音乐。DeepSeek写情书，Midjourney画星空，都是它的拿手好戏。

3️⃣ LLM：大语言模型，AI专属"语言魔术师"。吃下全网文本，练就聊天/编程/翻译神技。DeepSeek、ChatGPT、Grok都是这类超级话痨。

3️⃣ LangChain：AI基础平台，AI应用的"乐高积木"。把聊天机器人、文档分析、LLM、Doris等功能模块化，开发者像搭积木一样快速拼出智能应用。

4️⃣ RAG：检索增强生成，AI的"外接大脑"。给模型配个随身百科，回答前先翻书查资料。比如让Doris+DeepSeek用你公司的产品手册回答问题。

5️⃣ Fine-tuning：微调，AI的"私人教练"。用行业秘籍给通用模型开小灶。比如用Doris最佳实践集训练，让DeepSeek秒变Doris专家。这块需要注意：越是微调，专用性越强，通用性越弱。

小贴士：

Fine-tuning 微调如同给DeepSeek做定制手术——通过专业训练让模型精通某个领域，但越专精就越难跨界（比如医疗AI看不懂法律条文）。

RAG则是给DeepSeek装智能U盘（外挂）——不修改模型本身，回答时自动调取知识库内容（好比律师查法典办案），通用性无损还能随时更新知识。

Doris+DeepSeek的结合，其实就是RAG应用了。

一分钟定方案

Apache Doris作为一款基于 MPP 架构的高性能、实时分析型开源数据库，具备优秀的计算检索能力。既能支持高并发的点查询场景，也能支持高吞吐的复杂分析场景，还能进行PB级日志检索，并完美融合湖仓一体。

DeepSeek则是新一代的开源大语言模型，具有强大的语义理解和推理能力，以高性价比和开源著称。

两者结合，如何解决个人和企业知识库的难题？

正当我思考时，灵光一闪：

方案主要分为4个模块：

1. 数据源端（Sources）

接入本地知识库的文件数据，做加载、清洗和分割处理。

2. 数据导入端（Embedding）

Vector Embedding（向量嵌入）是将文本、图像等数据转换为数值向量的机器学习技术，其核心是通过高维空间中的向量位置来捕捉数据的语义信息。

实现向量化的就是Embedding，一段文本向量化会转化为一堆浮点数，这些浮点数排列在一起就像一个数组，例如某段话向量化：

[0.30633423, 0.32576653, -1.4459475, 0.8461606, 0.015846528...]

这一步主要是基于这类技术将数据源进行向量化存到Doris。

3. 数据融合端（LLM）

大模型的知识库存在两大局限：时效性天花板（如deepseek-r1:14b的知识截至2024年7月）和个人/企业知识盲区（无法获取内部资料）。

要让AI真正掌握业务精髓，还需：

简单而已一句话：将自己的知识内容喂给LLM，让LLM结合我们给出的知识去回答问题。

4. 数据客户端（Processing）

进行常规QA，提问入口。

一分钟撕代码

确定好方案后，即刻开撕代码。

环境准备

Doris环境

🔗 Doris部署：https://doris.apache.org/zh-CN/docs/dev/gettingStarted/quick-start

如果已经有Doris集群，直接用即可。

若当前还没Doris环境，可以参考Doris官方文档，基于Docker或本地化快速部署搭建一套Doris集群：

# Docker为例
# 1. 创建 docker-compose.yaml 文件
# 复制以下内容到 docker-compose.yaml 文件中，并将 DORIS_QUICK_START_VERSION 替换为指定的 Doris 版本，例如 2.1.8。
version: "3"
services:
  fe:
    image: apache/doris.fe-ubuntu:${DORIS_QUICK_START_VERSION}
    hostname: fe
    environment:
     - FE_SERVERS=fe1:127.0.0.1:9010
     - FE_ID=1
    network_mode: host
  be:
    image: apache/doris.be-ubuntu:${DORIS_QUICK_START_VERSION}
    hostname: be
    environment:
     - FE_SERVERS=fe1:127.0.0.1:9010
     - BE_ADDR=127.0.0.1:9050
    depends_on:
      - fe
    network_mode: host

# 2. 启动Doris集群
docker-compose -f ./docker-compose.yaml up -d

DeepSeeek环境

本地部署的Ollama+DeepSeek，网上教程非常多，本文还是就不单独介绍了。

简而言之，参考一臻整理的如下简图（两步完成不带WebUI）：

Python环境

本次使用的Python版本是3.8.17，相关的包都可以畅通无阻的install（建议用conda管理py环境）：

pip install langchain
pip install langchain-community
pip install -qU langchain-community
pip install  sqlalchemy
pip install --upgrade --quiet  pymysql
pip install markdown

代码实现

经过一臻七七四十九秒的编写调试，完整代码如下：

from dataclasses import dataclass
from contextlib import contextmanager

from langchain_community.llms.ollama import Ollama
from langchain_community.embeddings.ollama import OllamaEmbeddings
from langchain_community.document_loaders import DirectoryLoader, UnstructuredMarkdownLoader
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_text_splitters import TokenTextSplitter
from langchain_community.vectorstores.apache_doris import ApacheDorisSettings, ApacheDoris
from pymysql import connect
from pymysql.connections import Connection

@dataclass
class DorisConfig:
    """Apache Doris configuration"""
    host: str
    port: int
    username: str
    password: str
    database: str
    table: str

    def to_settings(self) -> ApacheDorisSettings:
        """Convert config to ApacheDorisSettings"""
        settings = ApacheDorisSettings()
        settings.host = self.host
        settings.port = self.port
        settings.username = self.username
        settings.password = self.password
        settings.database = self.database
        settings.table = self.table
        return settings

class DocumentLoader:
    """Document loading and processing"""
    def __init__(self, docs_dir: str, chunk_size: int = 400, chunk_overlap: int = 50):
        self.docs_dir = docs_dir
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap

    def load_and_split(self):
        """Load and split documents into chunks"""
        loader = DirectoryLoader(
            self.docs_dir,
            glob="**/*.md",
            loader_cls=UnstructuredMarkdownLoader
        )
        docs = loader.load()

        splitter = TokenTextSplitter(
            chunk_size=self.chunk_size,
            chunk_overlap=self.chunk_overlap
        )
        return splitter.split_documents(docs)

class Doris:
    """Apache Doris database operations"""
    def __init__(self, config: DorisConfig):
        self.config = config

    @contextmanager
    def get_connection(self) -> Connection:
        """Get database connection with context management"""
        conn = connect(
            host=self.config.host,
            port=self.config.port,
            user=self.config.username,
            password=self.config.password,
            database=self.config.database
        )
        try:
            yield conn
        finally:
            conn.close()

    def get_context(self) -> str:
        """Get context from database"""
        with self.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(f"SELECT document FROM `{self.config.table}`")
            rows = cursor.fetchall()
            return"\n".join(row[0] for row in rows)

class QASystem:
    """Question Answering System"""
    def __init__(
            self,
            config: DorisConfig,
            docs_dir: str = "./docs",
            embedding_model: str = "bge-m3",
            llm_model: str = "deepseek-r1"
    ):
        self.config = config
        self.docs_dir = docs_dir
        self.embedding_model = embedding_model
        self.llm_model = llm_model
        self.db = Doris(config)

    def initialize_vectorstore(self):
        """Initialize vector store with documents"""
        loader = DocumentLoader(self.docs_dir)
        docs = loader.load_and_split()

        embeddings = OllamaEmbeddings(model=self.embedding_model)
        ApacheDoris.from_documents(
            docs,
            embeddings,
            config=self.config.to_settings()
        )

    def get_qa_chain(self, context: str):
        """Create QA chain"""
        template = """
        Answer the question based only on the following context:{context}
        Question: 请严格结合{context}的内容回答{question}的问题，别发散，说人话。
        """
        prompt = ChatPromptTemplate.from_template(template)
        llm = Ollama(model=self.llm_model)

        return (
                {
                    "context": RunnableLambda(lambda d: context),
                    "question": RunnablePassthrough()
                }
                | prompt
                | llm
                | StrOutputParser()
        )

    def query(self, question: str) -> str:
        """Query the QA system"""
        context = self.db.get_context()
        qa_chain = self.get_qa_chain(context)
        return qa_chain.invoke(question)

def main():
    # Configuration
    config = DorisConfig(
        host="{按需补齐}",
        port=9030,
        username="{按需补齐}",
        password="{按需补齐}",
        database="{按需补齐}",
        table="{按需补齐，默认langchain}"
    )

    # Initialize QA system
    qa_system = QASystem(config)

    # Initialize vector store (only needed once)
    qa_system.initialize_vectorstore()

    # Query example
    question = "请介绍下Apache Doris,以及当前最新的Release版本是多少"
    response = qa_system.query(question)
    print("----- Response -----")
    print(response)

if __name__ == "__main__":
    main()

代码解析

代码经过精简，很多block没有进行过多地细化深入。主要是为了让大家能够快速熟悉Doris+DeepSeek的完整流程，后续可以结合自己需求，按模块进行调整应用。

代码主流程如下：

从里到外，每一层的解析：

DorisConfig

主要功能分为2个方面：

1️⃣ 配置集中管理

封装 Doris 数据库连接所需的 6 个关键参数。使用类型注解确保配置数据的正确性。

2️⃣ 配置格式转换

to_settings() 方法将配置转换为 LangChain 官方库 ApacheDoris 模块需要的参数格式，方便与 LangChain 的 Apache Doris进行集成。

Doris

主要职责分为三个层次：

1️⃣ 连接管理

2️⃣ 数据操作

3️⃣ 资源隔离

DocumentLoader

主要实现三个关键功能：

1️⃣ 文件加载

2️⃣ 文本分割

3️⃣ 参数化配置

QASystem

主要承担三个层面的职责：

1️⃣ 系统整合

2️⃣ 处理流程

3️⃣ 参数管理

Client

主要作为入口交互：

1️⃣ 系统配置

2️⃣ 初始化流程

3️⃣ 演示流程

其它说明

1. 测试文件

主要导入了3个md文件，分别是：

🔗 Doris README：https://github.com/apache/doris/edit/master/README.md

🔗 Doris简介文档：https://github.com/apache/doris-website/blob/master/i18n/zh-CN/docusaurus-plugin-content-docs/current/gettingStarted/what-is-apache-doris.md

🔗 Doris版本最新发布文档：https://github.com/apache/doris-website/blob/master/i18n/zh-CN/docusaurus-plugin-content-docs/current/releasenotes/all-release.md

2. Doris表

在进行ApacheDoris.from_documents时，Doris会自动建如下表：

CREATE TABLE`langchain` (
`id`varchar(50) NULL,
`document`textNULL,
`embedding`array<float> NULL,
`metadata`textNULL
) ENGINE=OLAP
UNIQUEKEY(`id`)

主要存储文档切割后的原文、embedding值和文件元数据：

由于Doris当前还没支持vector类型，所以还没法直接转为Retriever检索器，否则会报如下异常：

Syntax error in line 4:\n...    cosine_distance(array<float>

因此这块的流程改为将向量化后的Doris数据，直接读取（可以结合Doris倒排索引加速）外挂为上下文context。

预计7月将完成Doris向量化能力大统一，敬请期待！

3. 提示词

构建QA chain时，prompt提示词这块值得一提。大家结合应用时可以个性化调整：

template = """
        Answer the question based only on the following context:{context}
        Question: 请严格结合{context}的内容回答{question}的问题，别发散，说人话。
        """
prompt = ChatPromptTemplate.from_template(template)

在langchain中，ChatPromptTemplate是一个定义提问模版的方法，主要有两个功能：from_messages 和 from_template。

一分钟看结果

代码编写调试完后，来对比下结果。

首先，本地部署的是ollama+deepseek r1:14b，直接问相同的问题，回答：

结果有几个明显的错误点，比如Doris是2017年正式在GitHub上开源，最新发布的 release 版本是 v2.1.8（仅按发布时间排）。

再基于Doris+DeepSeek，问相同的问题，回答：

结果准确率明显上升，且完美结合了prompt和Doris知识数据进行回答。

通过DeepSeek 结合 Doris 打造的 RAG 知识库，既发挥了 DeepSeek 强大的模型推理能力，又利用了 Doris 高效的存储与检索优势，大幅提升回答准确度，有效避免AI幻觉，为个人/企业带来高质量的本地知识库。

结语

知识库AI智能化是一个永无止境的进化过程。随着大模型技术的发展，未来可以探索更多优化方向：

技术在飞速发展，但打造一个好的知识库系统不仅仅是技术问题。建议从小规模试点开始，在实践中不断优化，让系统真正服务于业务需求。

最后，近期看到很多小伙伴因为AI而感到危机。

一臻表示：AI不会替代人，但会用AI的人大概率会替代不会用AI的人。好比‌工业时代到互联网时代的转折期，会上网的人，替换了很多不会上网的人。互联网会把工业时代的产业重写一遍，AI也可以重写以往的产业模式。共勉！