实战指南：基于 Apache Doris 构建企业级 RAG（检索增强生成）应用

1. 引言

在 LLM（大语言模型）应用爆发的今天，RAG（Retrieval-Augmented Generation，检索增强生成）已成为解决大模型“幻觉”和知识滞后问题的标准方案。而作为 RAG 核心组件的向量数据库，其性能直接决定了检索的效率。

GitHub 上的开源项目 [apache-doris-rag]提供了一个非常优秀的范例，展示了如何利用Apache Doris强大的向量检索能力来构建 RAG 系统。

为什么选择 Apache Doris 做 RAG？

• 极速性能：基于 C++ 实现的 HNSW 索引，查询速度极快。
• 混合检索：支持向量检索 + 标量筛选（如：WHERE category='finance' AND vector_distance(...) < 0.5），这是传统专用向量数据库的弱项，却是 Doris 的强项。
• 架构统一：无需维护单独的 Vector DB，数仓和向量库合二为一，降低运维成本。

本文将深入解析该项目的核心实现，带您从零构建一个基于 Apache Doris 的 RAG 系统。

2. 架构设计

在基于 Apache Doris 的 RAG 系统中，数据流转主要分为两个阶段：知识入库（Indexing）和问答检索（Retrieval & Generation）。

3. 环境准备

在开始代码之前，我们需要准备基础环境。

3.1 软件依赖

• Apache Doris: 建议 2.0 及以上版本（2.1+ 对向量索引支持更完善）。
• Python: 3.8+
• Python 库:pymysql(连接 Doris),langchain,openai(或sentence-transformers),pandas.

3.2 启动 Apache Doris (Docker 快速启动)

如果您还没有 Doris 环境，可以使用 Docker 快速拉取一个单机版用于测试：

# 下载 Docker 镜像dockerpull apache/doris:2.1.0-bin-x64# 启动 (需根据官方文档配置 FE/BE 端口)# 这里假设您已经有可用的 Doris 实例，IP 为 127.0.0.1，FE Query Port 为 9030

4. 数据库设计 (Doris 端)

这是最关键的一步。我们需要在 Doris 中创建一个支持向量索引的表。

4.1 创建向量表

登录 MySQL 客户端或 DBeaver 连接 Doris，执行以下 SQL：

CREATEDATABASEIFNOTEXISTSrag_demo;USErag_demo;CREATETABLEIFNOTEXISTSdocuments(idVARCHAR(64)NOTNULLCOMMENT"文档ID",contentTEXTCOMMENT"原始文本内容",metadataTEXTCOMMENT"元数据(JSON格式)",vector_col ARRAY<FLOAT>NOTNULLCOMMENT"向量数据",INDEXidx_vector(vector_col)USINGHNSW PROPERTIES("dim"="1536",-- 对应 OpenAI embedding 维度"metric_type"="cosine_similarity",-- 余弦相似度"m"="16",-- HNSW 参数"ef_construction"="200"-- HNSW 参数))ENGINE=OLAPDUPLICATEKEY(id)DISTRIBUTEDBYHASH(id)BUCKETS1PROPERTIES("replication_num"="1","enable_light_schema_change"="true");

注意：dim参数必须与你选用的 Embedding 模型维度一致。OpenAItext-embedding-3-small通常是 1536 维。

5. 代码实现 (Python 端)

参考apache-doris-rag的逻辑，我们将代码分为数据导入和问答检索两部分。

5.1 导入数据 (Ingestion)

这段代码负责读取文件、切分文本、计算向量并存入 Doris。

importosimportuuidimportjsonimportpandasaspdfromlangchain.document_loadersimportTextLoader,PyPDFLoaderfromlangchain.text_splitterimportRecursiveCharacterTextSplitterfromlangchain_openaiimportOpenAIEmbeddingsimportpymysql# 1. 配置信息DORIS_HOST='127.0.0.1'DORIS_PORT=9030DORIS_USER='root'DORIS_PASSWORD=''DORIS_DB='rag_demo'# 初始化 Embedding 模型os.environ["OPENAI_API_KEY"]="sk-xxxx..."# 替换你的 Keyembeddings=OpenAIEmbeddings(model="text-embedding-3-small")defget_doris_connection():returnpymysql.connect(host=DORIS_HOST,port=DORIS_PORT,user=DORIS_USER,password=DORIS_PASSWORD,database=DORIS_DB,charset='utf8mb4')defingest_docs(file_path):# 2. 加载与切分文档print(f"正在加载文件:{file_path}...")loader=TextLoader(file_path,encoding='utf-8')# 或者是 PyPDFLoaderdocuments=loader.load()text_splitter=RecursiveCharacterTextSplitter(chunk_size=500,chunk_overlap=50)chunks=text_splitter.split_documents(documents)print(f"文档已切分为{len(chunks)}个片段，正在生成向量...")# 3. 批量生成向量并写入 Dorisconn=get_doris_connection()cursor=conn.cursor()batch_data=[]forchunkinchunks:# 生成向量vector=embeddings.embed_query(chunk.page_content)# 准备 SQL 插入数据# Doris 接收数组格式字符串: "[0.1, 0.2, ...]"vector_str=str(vector)doc_id=str(uuid.uuid4())metadata=json.dumps(chunk.metadata)# 简单处理单引号转义content=chunk.page_content.replace("'","\\'")sql=f""" INSERT INTO documents (id, content, metadata, vector_col) VALUES ('{doc_id}', '{content}', '{metadata}',{vector_str}) """try:cursor.execute(sql)exceptExceptionase:print(f"Insert error:{e}")conn.commit()conn.close()print("数据入库完成！")# 运行导入if__name__=="__main__":# 创建一个测试文件withopen("test_knowledge.txt","w",encoding="utf-8")asf:f.write("Apache Doris 是一个现代化的 MPP 分析型数据库产品。它支持极速的实时数据分析。")ingest_docs("test_knowledge.txt")

5.2 检索与问答 (RAG)

这里我们利用 SQL 进行向量检索，然后调用 LLM 回答。

fromlangchain_openaiimportChatOpenAI# 初始化 LLMllm=ChatOpenAI(model="gpt-3.5-turbo",temperature=0)defsearch_doris(query_text,top_k=3):"""在 Doris 中进行向量搜索"""# 1. 将用户问题转为向量query_vector=embeddings.embed_query(query_text)vector_str=str(query_vector)conn=get_doris_connection()cursor=conn.cursor()# 2. 执行向量检索 SQL# 使用 cosine_distance 函数，结果越接近 0 越相似 (Doris 2.1+ 语法可能略有不同，以官方文档为准)# 若 metric_type 定义为 cosine_similarity，这里通常查出来的是距离，排序需注意sql=f""" SELECT content, cosine_distance(vector_col,{vector_str}) as dist FROM documents ORDER BY dist ASC LIMIT{top_k}"""cursor.execute(sql)results=cursor.fetchall()conn.close()# 返回纯文本列表return[row[0]forrowinresults]defrag_chat(query):# 1. 检索上下文print(f"用户提问:{query}")print("正在检索 Doris...")retrieved_docs=search_doris(query)context="\n\n".join(retrieved_docs)print(f"检索到{len(retrieved_docs)}条相关上下文。")# 2. 构建 Promptprompt=f""" 基于以下已知信息，简洁和专业地回答用户的问题。如果无法从信息中得到答案，请说"我不知道"。 已知信息:{context}问题:{query}"""# 3. LLM 生成print("正在思考...")response=llm.invoke(prompt)returnresponse.content# 运行测试if__name__=="__main__":answer=rag_chat("Apache Doris 的特点是什么？")print("-"*30)print(f"AI 回答: \n{answer}")

6. 实现细节深度解析

在apache-doris-rag及类似的实现中，有几个核心点值得注意：

6.1 向量索引的选择

Doris 提供了HNSW索引。在CREATE TABLE语句中，我们设置了ef_construction和m。

• m: 节点在图中的最大连接数。值越大，索引质量越高，但内存占用和构建时间增加。
• ef_construction: 构建索引时的搜索范围。值越大，索引质量越高，构建越慢。
• PROPERTIES: 对于生产环境，务必根据数据量调整这些参数。

6.2 混合查询 (Hybrid Search)

Doris 的一大杀手锏是可以在 SQL 中直接混合标量过滤。例如，我们只想在“2023年之后”的文档中搜索：

# 代码片段演示sql=f""" SELECT content, cosine_distance(vector_col,{vector_str}) as dist FROM documents WHERE metadata LIKE '%"year": 2024%' -- 混合过滤 ORDER BY dist ASC LIMIT 3 """

这种能力使得 Doris 在处理带有复杂业务逻辑的 RAG 场景（如多租户隔离、权限控制）时，比纯向量数据库（如 Milvus, Chroma）更具优势。

6.3 性能优化

• Stream Load: 在 Python 代码中我们使用了INSERT，这适合小批量数据。如果是海量数据导入，建议使用 Doris 的Stream LoadHTTP 接口，吞吐量更高。
• 分区 (Partitioning): 如果数据量达到亿级，建议按时间或业务ID对表进行PARTITION BY，进一步提升查询效率。

7. 总结

通过参考apache-doris-rag的实现思路，我们发现基于 Apache Doris 构建 RAG 系统非常直观。它不需要引入复杂的新的向量数据库组件，仅仅利用现有的数仓设施，就能实现SQL + Vector的强大检索能力。

核心优势总结：

1. 架构极简：ETL 处理完的数据直接生成向量存入 Doris，数据不搬家。
2. 查询灵活：标准 SQL 接口，运维人员零门槛上手。
3. 高并发：Doris 原生的 MPP 架构能够支撑高并发的 RAG 请求。

参考资料:

• Apache Doris 官方文档: Vector Search
• GitHub 项目: GeiTe168/apache-doris-rag