8年Java开发者AI转型第一周：从零搭建RAG文档问答系统(Day 5-7)

本篇继续记录一位拥有8年Java开发经验的工程师的AI转型之旅，重点聚焦第五至第七天：从零搭建RAG（检索增强生成）文档问答系统。上一期我们已经掌握了LLM API的调用方式、Prompt工程和参数调优，但仅会与AI对话远远不够——核心目标是让AI基于我们自己的私有知识库来准确回答问题。

先来谈谈RAG（检索增强生成）的核心价值。它解决了一个关键问题：如何在不将整个知识库都塞进prompt的前提下，让AI给出精准回答。传统方式是把知识一股脑地堆进prompt，结果导致Token数量爆炸，成本随之飙升。而RAG的思路是将知识存入向量数据库，检索出相关片段，再喂给LLM，最终输出精确答案。

用Java开发者的视角来理解：传统方式就像把整个MySQL库的表结构写进SQL注释里，依赖DBA的记忆；而RAG则像是建立一个Elasticsearch索引，先搜索到相关表再交给DBA处理。这一期，我们就从零搭建一个个人文档问答系统，将之前学习的Embedding、向量数据库、LangChain管道串联成一条完整的RAG链路。

二、Embedding：文本如何变成向量

2.1 什么是Embedding

先搞清楚Embedding是怎么回事。Embedding（嵌入）就是将一段文本映射到一个高维向量（一组浮点数）。举一个直观的例子：

"猫"→ [0.12, -0.45, 0.78, ..., 0.33] (768维)
"小狗"→ [0.15, -0.42, 0.75, ..., 0.31] (768维)
"汽车"→ [0.89, 0.23, -0.56, ..., -0.72] (768维)

“猫”和“小狗”的向量非常接近，但“汽车”的向量离它们很远。这就是语义相似的本质——向量空间中的距离。

2.2 为什么需要Embedding

在RAG系统中，我们需要回答这样的问题：用户问“kkFileView支持哪些文档格式？”，如果文档里写的是“支持doc、docx、xls、xlsx、ppt、pptx、pdf、txt”，关键词匹配（LIKE ‘%格式%’）是搜不到的。但Embedding进行的是语义级别的匹配，即使用户没说“格式”这个词，只要语义相关，向量距离就近，就能命中。

2.3 余弦相似度：向量怎么比

衡量两个向量有多接近，最常用的方法是余弦相似度：

余弦相似度 = (A·B) / (||A|| × ||B||)

值域是[-1, 1]，1代表完全同向（语义相同），0代表正交（无关），-1代表完全反向（相反）。这个公式其实很直观，值越接近1，说明语义越相关。用Java类比，就像HashMap.get(key)用的是精确匹配，而Embedding加余弦相似度做的是模糊匹配——key不完全一样，但语义相近也能命中。

2.4 用代码实现Embedding

这里我们使用智谱的Embedding API来生成向量。下面是一个自定义的Embedding类，实现了LangChain的标准接口：

# BigModelEmbedding.py
import requests
from langchain_core.embeddings import Embeddings
from pydantic import BaseModel, Field
from typing import List

class BigModelEmbeddingFunc(Embeddings, BaseModel):
    """自定义智谱Embedding嵌入类，符合LangChain标准"""
    api_key: str = Field(..., description="API Key")
    base_url: str = Field(default="https://open.bigmodel.cn/api/paas/v4/embeddings", description="向量接口地址")
    model: str = Field(default="embedding-3", description="嵌入模型名称")

    def embed_query(self, text: str) -> List[float]:
        """单条文本向量化：用户提问时调用（检索用）"""
        resp = requests.post(url=self.base_url, headers={"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"}, json={"model": self.model, "input": text})
        resp.raise_for_status()
        return resp.json()["data"][0]["embedding"]

    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        """批量文档向量化：Chroma.from_documents入库时批量调用"""
        resp = requests.post(url=self.base_url, headers={"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"}, json={"model": self.model, "input": texts})
        resp.raise_for_status()
        data = resp.json()
        embeds = sorted(data["data"], key=lambda x: x["index"])
        return [item["embedding"] for item in embeds]

三、ChromaDB：轻量级向量数据库

3.1 为什么需要向量数据库

Embedding生成了向量，但向量本身不会自动完成检索。你需要一个数据库来存储海量向量，并快速搜索语义相似的向量——暴力遍历O(n)实在太慢了。同时还要能持久化，避免重启后数据丢失。ChromaDB正是专为AI应用设计的轻量级向量数据库，本地运行，零配置，非常便捷。

3.2 基础CRUD操作

先看ChromaDB的基本用法，跟Java操作MySQL很像：

# chromadbStudy.py
import chromadb
import os
import json
from chromadb import Documents, EmbeddingFunction, Embeddings

# 自定义Embedding函数（适配ChromaDB原生API）
class DeepSeekEmbeddingFunc(EmbeddingFunction):
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://open.bigmodel.cn/api/paas/v4/embeddings"
        self.headers = {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"}

    def __call__(self, texts: Documents) -> Embeddings:
        resp = requests.post(url=self.base_url, headers=self.headers, json={"model": "embedding-3", "input": texts})
        resp.raise_for_status()
        embeds = [item["embedding"] for item in resp.json()["data"]]
        return embeds

if __name__ == '__main__':
    with open("config_bigmodel.json", "r", encoding="utf-8") as f:
        config = json.load(f)

    # 1. 持久化客户端
    client = chromadb.PersistentClient(path="./my_knowledge_base")
    # 2. 创建/获取集合（类似MySQL的Table）
    ds_embed = DeepSeekEmbeddingFunc(api_key=config["embedding"]["api_key"])
    coll = client.get_or_create_collection(name="article", embedding_function=ds_embed, metadata={"hnsw:space": "cosine"})
    # 3. 写入数据（INSERT）
    coll.add(documents=["Chroma是轻量级向量数据库", "bigmodel API可以生成文本Embedding向量", "RAG架构依靠向量库检索知识库"], metadatas=[{"source": "note1"}, {"source": "note2"}, {"source": "note3"}], ids=["id1", "id2", "id3"])
    # 4. 更新数据（UPDATE）
    coll.update(ids=["id1"], documents=["Chroma是轻量级向量数据库，支持多种存储后端"], metadatas=[{"source": "note1"}])
    # 5. 删除数据（DELETE）
    coll.delete(ids=["id3"])
    # 6. 查询（SELECT COUNT）
    print("集合中文档数量:", coll.count())
    # 7. 语义查询（SELECT WHERE语义相似度）
    query_result = coll.query(query_texts=["向量数据库能用来做什么"], n_results=2)
    print("检索到的原文：")
    print(query_result["documents"])
    print("对应相似度距离：")
    print(query_result["distances"])

3.3 元数据过滤

ChromaDB支持在语义检索的同时加上元数据过滤，类似SQL中的WHERE子句：

# 只查source="note1"的文档中，跟"向量数据库"最相似的
query_result = coll.query(query_texts=["向量数据库能用来做什么"], where={"source": "note1"}, n_results=2)

3.4 ChromaDB vs Java ORM类比

四、LangChain RAG管道全链路

4.1 RAG的六步流水线

LangChain将RAG拆解为一个清晰的管道，每一步各司其职：

文档 → Loader → Splitter → Embedding → VectorStore → Retriever → Chain → 回答

用Java类比，这就像一个工厂流水线：

4.2 第一步：加载文档

from langchain_community.document_loaders import TextLoader
loader = TextLoader("doc/kkfile.md", encoding="utf-8")
docs = loader.load()
print(f"加载文档 {len(docs)} 篇")

TextLoader读取文件后返回Document对象列表，每个Document包含page_content（文本内容）和metadata（元数据，如文件路径）。

4.3 第二步：文本分割

文档不能整段丢给Embedding，需要切成小块（Chunk）。太碎会丢失上下文，太大则检索不精准：

from langchain_text_splitters import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = splitter.split_documents(docs)
print(f"分块数: {len(chunks)}")

4.4 第三步 + 第四步：向量化 + 存储

from BigModelEmbedding import BigModelEmbeddingFunc
from langchain_community.vectorstores import Chroma

embeddings = BigModelEmbeddingFunc(api_key=EMBEDDING_KEY)
if os.path.exists(CHROMA_DB_PATH):
    vectorstore = Chroma(persist_directory=CHROMA_DB_PATH, embedding_function=embeddings)
    print(f"加载已有向量库，共 {vectorstore._collection.count()} 条记录")
else:
    vectorstore = Chroma.from_documents(chunks, embeddings, persist_directory=CHROMA_DB_PATH)
    vectorstore.persist()
    print(f"新建向量库，存入 {len(chunks)} 个分块")

4.5 第五步：构建检索器

retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

k=5表示每次检索返回最相似的5个文档块。

4.6 第六步：构建RAG管道（LangChain 1.x写法）

LangChain 1.x推荐使用Runnable管道，用|操作符将各个组件串联起来：

from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# 自定义提示词
prompt = PromptTemplate.from_template("""请严格依据下面参考资料回答用户问题，给出回答的依据文件，不要编造内容：
参考资料：{context}
用户问题：{input}""")

# 初始化LLM
llm = ChatOpenAI(model="deepseek-v4-pro", temperature=0.7, api_key=DEEPSEEK_KEY, base_url=DEEPSEEK_URL)

# 构建RAG管道
rag_chain = ({"context": retriever, "input": RunnablePassthrough()} | prompt | llm | StrOutputParser())

管道执行流程非常清晰：用户输入后，retriever检索Top-5文档块，RunnablePassthrough透传用户输入，prompt将检索结果和用户问题填入模板，llm发送给DeepSeek API，最后StrOutputParser提取纯文本回答。

五、端到端：个人文档问答系统

将前面所有步骤串联起来，就是一个完整的RAG文档问答系统：

5.1 完整代码

# RAGqa.py - 个人文档问答系统
import os
import json
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from BigModelEmbedding import BigModelEmbeddingFunc
from langchain_community.vectorstores import Chroma
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

if __name__ == '__main__':
    # 配置加载
    try:
        config_path = os.path.join(os.path.dirname(__file__), "config_bigmodel.json")
        with open(config_path, "r", encoding="utf-8") as f:
            config = json.load(f)
    except FileNotFoundError:
        print(f"错误: 配置文件不存在: {config_path}")
        raise
    except json.JSONDecodeError as e:
        print(f"错误: 配置文件格式异常: {e}")
        raise

    # ========== 配置区 ==========
    DEEPSEEK_KEY = config["deepseek"]["api_key"]
    EMBEDDING_KEY = config["embedding"]["api_key"]
    DEEPSEEK_URL = config["deepseek"]["base_url"]
    CHROMA_DB_PATH = config["chroma"]["persist_directory"]
    # ============================

    # 1. 加载文档
    doc_path = os.path.join(os.path.dirname(__file__), "doc", "kkfile.md")
    print("当前文档路径:", os.path.dirname(__file__))
    try:
        loader = TextLoader(doc_path, encoding="utf-8")
        docs = loader.load()
    except Exception as e:
        print(f"错误: 文档加载失败: {e}")
        raise
    if not docs:
        print("警告: 文档为空，请检查文件路径和内容")
        raise SystemExit(1)
    print(f"加载文档 {len(docs)} 篇")

    # 2. 文本分割
    splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
    chunks = splitter.split_documents(docs)
    print(f"分块数: {len(chunks)}")

    # 3. 向量化 + 存储
    try:
        embeddings = BigModelEmbeddingFunc(api_key=EMBEDDING_KEY)
        if os.path.exists(CHROMA_DB_PATH):
            vectorstore = Chroma(persist_directory=CHROMA_DB_PATH, embedding_function=embeddings)
            print(f"加载已有向量库，共 {vectorstore._collection.count()} 条记录")
        else:
            vectorstore = Chroma.from_documents(chunks, embeddings, persist_directory=CHROMA_DB_PATH)
            vectorstore.persist()
            print(f"新建向量库，存入 {len(chunks)} 个分块")
    except Exception as e:
        print(f"错误: 向量化失败: {e}")
        raise

    # 4. 构建问答链
    retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
    prompt = PromptTemplate.from_template("""请严格依据下面参考资料回答用户问题，给出回答的依据文件，不要编造内容：
参考资料：{context}
用户问题：{input}""")
    llm = ChatOpenAI(model="deepseek-v4-pro", temperature=0.7, api_key=DEEPSEEK_KEY, base_url=DEEPSEEK_URL)
    rag_chain = ({"context": retriever, "input": RunnablePassthrough()} | prompt | llm | StrOutputParser())

    # 5. 提问
    print("========== RAG问答系统 ==========")
    try:
        result = rag_chain.invoke("历史更新记录")
        print("回答结果:", result)
        print("========== 第二次问答 ==========")
        result = rag_chain.invoke("kkFileView的官网地址？")
        print("回答结果:", result)
    except Exception as e:
        print(f"错误: LLM调用失败: {e}")
        raise

5.2 运行效果

第一次运行（新建向量库）：

当前文档路径: d:AIaiLearningoneWeekday5~7RAGDemo
加载文档 1 篇
分块数: 42
新建向量库，存入 42 个分块
========== RAG问答系统 ==========
回答结果: kkFileView的历史更新记录包括多个版本的迭代...

第二次运行（加载已有向量库）：

当前文档路径: d:AIaiLearningoneWeekday5~7RAGDemo
加载文档 1 篇
分块数: 42
加载已有向量库，共 42 条记录
========== RAG问答系统 ==========
回答结果: kkFileView的历史更新记录包括多个版本的迭代...

5.3 配置说明

config_bigmodel.json中需要三个配置项：

{
  "embedding": {
    "api_key": "你的智谱Embedding API Key",
    "base_url": "https://open.bigmodel.cn/api/paas/v4/embeddings",
    "model": "embedding-3"
  },
  "deepseek": {
    "api_key": "你的DeepSeek API Key",
    "base_url": "https://api.deepseek.com",
    "model": "deepseek-v4-pro"
  },
  "chroma": {
    "persist_directory": "./rag_knowledge_base"
  }
}

六、总结

6.1 知识地图回顾

经过前三期的学习，你已经掌握了RAG的全套基础技能：

6.2 RAG全流程一图流

┌─────────────┐┌──────────────┐┌─────────────┐
│kkfile.md    │───▶│ TextSplitter │───▶│   Chunks    │
│(原始文档)   │    │(切分成小块)  │    │ (500字/块) │
└─────────────┘└──────────────┘└──────┬──────┘
                                      │
                               ┌─────────▼─────────┐
                               │BigModelEmbedding  │
                               │(文本→向量)        │
                               └─────────┬─────────┘
                                         │
                               ┌─────────▼─────────┐
                               │     ChromaDB      │
                               │(向量持久化存储)   │
                               └─────────┬─────────┘
                                         │
                    ┌─────────────────▼─────────────────┐
                    │          用户提问                  │
                    │ "kkFileView官网地址？"            │
                    └─────────────────┬─────────────────┘
                                      │
                               ┌─────────▼─────────┐
                               │    Retriever      │
                               │(语义检索 Top-5)   │
                               └─────────┬─────────┘
                                         │
                               ┌─────────▼─────────┐
                               │  PromptTemplate   │
                               │(组装上下文+问题)  │
                               └─────────┬─────────┘
                                         │
                               ┌─────────▼─────────┐
                               │   ChatOpenAI      │
                               │(DeepSeek生成)     │
                               └─────────┬─────────┘
                                         │
                               ┌─────────▼─────────┐
                               │    精准回答       │
                               │ "官网地址是..."   │
                               └───────────────────┘

6.3 下一步

下期我们将深入RAG系统的进阶优化：

• Chunking策略实验：chunk_size=500 vs 800 vs 1000，哪个检索效果最好？
• Rerank重排序：召回Top-10后精排Top-2，提升回答精度
• LangChain管道进阶：流式输出、中间结果调试
• 评估RAG质量：学习用RAGAS框架评估答案的忠实度和相关性