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

知识库AI智能化是一个永无止境的进化过程。随着大模型技术的发展，未来可以探索更多优化方向：• 实时更新：支持增量同步，保持知识时效性• 知识推理：引入知识图谱，提升问答准确性• 多模态支持：除文本外，支持图片、音视频等富媒体内容技术在飞速发展，但打造一个好的知识库系统不仅仅是技术问题。建议从小规模试点开始，在实践中不断优化，让系统真正服务于业务需求。

爱编程的小辞

1109人浏览 · 2025-03-13 10:10:27

爱编程的小辞 · 2025-03-13 10:10:27 发布

术语名词

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

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真正掌握业务精髓，还需：

知识注入：将个人/企业专属的《XXX》存入Doris数据库，相当于给AI装上"行业大脑"

智能关联：遇到复杂问题时，DeepSeek AI会自动关联多个知识库的章节内容，像资深专家般综合分析，妥妥的AI助手

精准应答：基于实时更新的个人/企业知识库，DeepSeek 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️⃣ 连接管理

• 基于 DorisConfig 配置创建数据库连接

• 使用 @contextmanager 实现安全的连接上下文管理

• 自动处理连接异常和资源释放

2️⃣ 数据操作

• 执行指定 SQL 查询（当前实现为获取全部 document 字段，即直接获取全部内容喂给DeepSeek；暂时没用到embedding值，后续可以结合起来用，效果更佳）

• 将查询结果拼接为字符串返回

• 为上层 QASystem 提供原始数据输入

3️⃣ 资源隔离

• 封装所有数据库相关操作细节

• 隔离 SQL 语法与业务逻辑

• 为后续扩展更多数据库操作提供统一入口

DocumentLoader

主要实现三个关键功能：

1️⃣ 文件加载

• 自动扫描指定目录下的 Markdown 文件（**/*.md）

• 使用 UnstructuredMarkdownLoader 解析文件内容

2️⃣ 文本分割

• 基于 token 数量进行智能分割

• 通过重叠 token 保持上下文连贯性

• 输出格式化的 Document 对象列表

3️⃣ 参数化配置

• 允许灵活调整块大小（默认 400 tokens）

• 可设置块间重叠量（默认 50 tokens）

• 为后续向量化存储提供标准化输入

QASystem

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

1️⃣ 系统整合

• 协调 Doris 数据库操作

• 管理文档向量化存储

• 桥接大语言模型（deepseek-r1）

2️⃣ 处理流程

• 知识库初始化（文档向量化存储到Doris）

• 问答处理链构建（LangChain 流水线）

• 用户查询执行（端到端响应生成）

3️⃣ 参数管理

• 集中管理嵌入模型（embedding_model本次选用的是bge-m3）和 LLM 选择（llm本次选用的是deepseek-r1）

• 控制知识库文件存储路径

• 维护数据库连接配置

Client

主要作为入口交互：

1️⃣ 系统配置

• 定义 Doris 数据库连接参数

• 指定文档存储表名（langchain）

• 设置认证信息（用户名/密码）

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。

1. from_messages：把系统指令、用户提问等不同角色的对话片段，像搭积木一样组合，3秒构建多轮对话模板。

2. from_template：一键复制现有模板，好比PS图层编辑般自由修改，快速生成变体对话方案。本次选用的是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大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】