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

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

一、背景

上一篇我们搞清楚了 LLM API 的调用方式、Prompt 工程和参数调优。但光会跟 AI 聊天还不够——我们要让 AI 基于我们的私有知识回答问题。

这就是 RAG（Retrieval-Augmented Generation，检索增强生成）的核心价值：

传统方式：把知识塞进 prompt → Token 爆炸，成本爆炸RAG 方式：知识存向量库 → 检索相关片段 → 喂给 LLM → 精准回答

打个 Java 类比：

• 传统方式 = 把整个 MySQL 库的表结构写进 SQL 注释里，靠 DBA 记忆
• RAG = 建个 Elasticsearch 索引，搜到相关表再交给 DBA

这一期，我们从零搭建一个个人文档问答系统，把之前学的 Embedding、向量数据库、LangChain 管道串成一条完整的 RAG 链路。

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

2.1 什么是 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 系统中，我们需要回答这样的问题：

如果用传统的关键词匹配（LIKE '%格式%'），会漏掉很多语义相关的内容。而 Embedding 做的是语义级别的匹配——即使用户没说"格式"这个词，只要语义相关，向量距离就近。

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

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

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

• 值域：[-1, 1]
• 1 = 完全同向（语义相同）
• 0 = 正交（无关）
• -1 = 完全反向（相反）

Java 类比：就像 HashMap.get(key) 用的是精确匹配，Embedding + 余弦相似度做的是模糊匹配——key 不完全一样，但语义相近也能命中。

2.4 用代码实现 Embedding

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

# BigModelEmbedding.pyimport requestsfrom langchain_core.embeddings import Embeddingsfrom pydantic import BaseModel, Fieldfrom typing import Listclass 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 生成了向量，但向量本身不会帮你做检索。你需要一个数据库来：

1. 存储海量向量
2. 快速搜索语义相似的向量（暴力遍历 O(n) 太慢了）
3. 持久化，重启不丢数据

ChromaDB 就是一个专为 AI 应用设计的轻量级向量数据库，本地运行，零配置。

3.2 基础 CRUD 操作

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

# chromadbStudy.pyimport chromadbimport osimport jsonfrom chromadb import Documents, EmbeddingFunction, Embeddings# 自定义 Embedding 函数（适配 ChromaDB 原生 API）class DeepSeekEmbeddingFunc(EmbeddingFunction):def __init__(self, api_key: str):self.api_key = api_keyself.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 embedsif __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 TextLoaderloader = 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 RecursiveCharacterTextSplittersplitter = RecursiveCharacterTextSplitter(chunk_size=500,# 每块最多 500 字符chunk_overlap=50 # 相邻块重叠 50 字符，防止语义断裂)chunks = splitter.split_documents(docs)print(f"分块数: {len(chunks)}")

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

from BigModelEmbedding import BigModelEmbeddingFuncfrom langchain_community.vectorstores import Chromaembeddings = 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 PromptTemplatefrom langchain_core.runnables import RunnablePassthroughfrom langchain_core.output_parsers import StrOutputParserfrom langchain_openai import ChatOpenAI# 自定义提示词prompt = PromptTemplate.from_template("""请严格依据下面参考资料回答用户问题，给出回答的依据文件，不要编造内容：参考资料：{context}用户问题：{input}""")# 初始化LLMllm = 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())

管道执行流程：

用户输入 "kkFileView的官网地址？"↓{"context": retriever, "input": RunnablePassthrough()}→ retriever 检索 Top-5 文档块→ RunnablePassthrough 透传用户输入↓prompt→ 把检索结果和用户输入填入模板↓llm→ 发送给 DeepSeek API↓StrOutputParser()→ 提取纯文本回答

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

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

5.1 完整代码

# RAGqa.py - 个人文档问答系统import osimport jsonfrom langchain_community.document_loaders import TextLoaderfrom langchain_text_splitters import RecursiveCharacterTextSplitterfrom BigModelEmbedding import BigModelEmbeddingFuncfrom langchain_community.vectorstores import Chromafrom langchain_openai import ChatOpenAIfrom langchain_core.prompts import PromptTemplatefrom langchain_core.runnables import RunnablePassthroughfrom langchain_core.output_parsers import StrOutputParserif __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}")raiseexcept 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}")raiseif 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 框架评估答案的忠实度和相关性

本文代码基于 LangChain 1.x + DeepSeek API，完整项目见：ai-learning

系列文章：Day 1-2：Python速学 + 环境搭建 · Day 3-5：LLM API调用 + Prompt工程基础