本文基于仓库https://github.com/freemandealer/apache-doris-rag，梳理 Apache Doris 在 RAG 流程中的核心实现与关键代码。覆盖向量入库、查询增强、检索、生成、服务端与 CLI，以及知识图谱实验模块。

整体架构

1. 配置 (conf.ini): Doris 连接、Embedding/LLM 选择、文档路径、界面语言。
2. 离线索引 (index_md_to_doris.py): 扫描 markdown → 清洗切分 → 生成向量 → doris_vector_search 入库并建 HNSW 索引。
3. 检索与生成 (rag_lib.py): 基于查询向量检索 Doris，拼上下文后调用 LLM 生成回答，可选查询增强。
4. 服务层 (rag_service.py): FastAPI 提供 /api/chat 与简易 Web UI，返回答案与引用源。
5. CLI (rag_cli.py): 终端交互版。
6. 知识图谱实验 (build_knowledge_graph.py): 以向量搜索实体 + SQL 取关系的混合方式在 Doris 中管理图数据。

配置要点 (conf.ini.template)

• Doris 连接: host/query_port/http_port/user/password/db_name/table_name，还包含图模式表名。
• Embedding: type=ollama|openai，模型名、base_url 或 api_key。
• LLM: type=openai 协议兼容，模型/api_key/base_url/temperature。
• 文档根目录: doc_root。
• 应用语言: language=zh|en。

离线索引：文档入 Doris

核心流程在 index_md_to_doris.py:

# 1) 收集与清洗文档
files = collect_markdown_files(DOC_ROOT)
cleaned = clean_text(raw)  # 去掉 frontmatter 并修剪行尾
chunks = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=100).split_text(cleaned)

# 2) 构造记录
{
    "id": cur_id, "path": rel_path, "title": title, "content": chunk,
}

# 3) 生成向量
vectors = embeddings.embed_documents(texts)

# 4) 写入 Doris 并建索引
index_options = IndexOptions(index_type="hnsw", metric_type="inner_product")
table = client.create_table(table_name, df, index_options=index_options)

要点：

• 使用 doris_vector_search.DorisVectorClient；若表存在则 open_table().insert(df)。
• 列包含 id/path/title/content/embedding，索引类型 HNSW，metric 为 inner_product。

检索与生成：RAG 主链路

rag_lib.py：

def retrieve_context(query: str, top_k: int = 5) -> pd.DataFrame:
    query_vec = get_embedding_model().embed_query(query)
    table = DorisVectorClient(db, auth).open_table(table_name)
    return table.search(query_vec).limit(top_k).select(["id","path","title","content"]).to_pandas()

def query_augment(query, history=None):
    # 使用 LLM 做历史消歧/关键词补全，提示词在 i18n 文本中
    prompt = get_message(...)
    return get_llm().invoke(prompt).content.strip()

• Embedding 与 LLM 由配置决定，默认 Ollama + OpenAI 协议模型。
• 检索结果以 DataFrame 返回，供服务层拼接上下文。

服务端：FastAPI (rag_service.py)

@app.post("/api/chat")
async def chat(req: ChatRequest):
    augmented_query = query_augment(req.query, req.history)
    context_df = retrieve_context(augmented_query, top_k=5)
    context_text = "\n\n---\n\n".join([...])
    prompt = template.format(history=..., context=context_text, question=req.query)
    answer = get_llm().invoke(prompt).content
    return ChatResponse(answer=answer, sources=sources)

• 在前端 HTML 中直接调用 /api/chat，展示答案与来源 path/title。
• 历史轮次作为 prompt 一部分；UI 极简、无持久化。

CLI (rag_cli.py)

• 同样使用 query_augment → retrieve_context → llm.invoke(prompt)。
• 将检索到的上下文拼成提示词，打印答案并把问答追加到历史。

知识图谱实验 (build_knowledge_graph.py)

• 使用异步 LLM (llm_adapters.get_async_chat_llm) 抽取实体/关系为三元组，解析后构图。
• Doris 端建两张表：
  • graph_chunks (DUPLICATE KEY，含向量索引 INDEX idx_embedding (embedding) USING ANN ... dim=1024)
  • doc_status (UNIQUE KEY，用于文档增量状态)
• 入库：实体/关系/子图分别作为 chunk，文本生成向量后 table.add()。
• 检索：search_entities 用向量检索实体；get_relations 通过 SQL 查询相关边；组合重建子图。

国际化提示词 (i18n.py)

• language 决定使用中/英提示词，包括查询增强、对话模板、前端文案。

快速试用（假设已配置 conf.ini 且 Doris 就绪）

# 1) 安装依赖（示例）
pip install "doris-vector-search>=0.0.5" langchain langchain-community langchain-text-splitters langchain-openai fastapi uvicorn pydantic pandas

# 2) 构建向量索引
python index_md_to_doris.py

# 3) 启动服务
uvicorn rag_service:app --host 0.0.0.0 --port 8000

访问 http://localhost:8000 体验 Web 端，或运行 python rag_cli.py 在终端测试。

实践建议

• 在 Doris 侧预建数据库，并为向量表配置足够的内存与并发资源。
• 根据模型维度调整 IndexOptions、dim 及 metric；必要时尝试混合检索（向量 + keyword 过滤）。
• 对大语料可增加批量写入与分片；对 LLM 查询可加缓存或 streaming。
• 知识图谱模式目前演示性质，可按需完善删除、补全节点信息等逻辑。