06. 构建 RAG Agent

学习目标

理解 RAG 的基本数据流，以及它和普通聊天模型调用的区别。
把本地 Markdown 文档切分、向量化，并写入本地 Chroma 向量库。
把检索器包装成 Agent 可调用的工具，让模型在需要事实依据时主动检索。
学会从文档、chunk、embedding、vector store、retriever 和 tool 这几层定位 RAG 问题。

RAG 解决什么问题

模型本身不知道你的私有文档，也不会自动读取项目目录。RAG 的做法不是把文档永久写进模型参数，而是在回答前先检索相关内容，再把这些内容作为上下文交给模型。

在这个教程里，知识库来自 data/knowledge_base/*.md。脚本会读取这些文档，把它们切成较小的 chunks，然后用 .env 中的 OLLAMA_EMBEDDING_MODEL 把每个 chunk 转成向量。Chroma 会保存这些向量和原始文本。用户提问时，retriever 根据问题向量找出最相关的 chunk，再由 Agent 把检索器当作工具使用。

关键概念：

document：原始文档，例如一个 Markdown 文件。
chunk：从文档切出来的小文本块，便于检索和放入上下文。
embedding：文本的向量表示，用来计算语义相似度。
vector store：保存向量、文本和元数据的数据库，这里使用本地 Chroma。
retriever：面向查询的检索接口，负责返回相关文档片段。
tool：暴露给 Agent 的外部能力。这里的检索就是一个工具。

数据流

完整流程分成索引阶段和问答阶段。

索引阶段：

读取 data/knowledge_base/*.md。
把文档切成 chunk_size=500、chunk_overlap=80 的文本块。
使用 OllamaEmbeddings 和 OLLAMA_EMBEDDING_MODEL 为每个 chunk 生成向量。
写入本地 Chroma collection：langchain_agent_tutorial。

问答阶段：

用户在 CLI 中提出问题。
Agent 判断这是事实性教程问题，调用 search_knowledge_base。
工具用 retriever 检索最相关的 3 个 chunk。
工具把来源和内容返回给模型。
模型基于检索结果回答；如果上下文不足，应明确说明资料不足。

这种设计把 retrieval 放进 tool 层，而不是在每次提问前由程序固定检索。好处是 Agent 可以在需要本地知识时主动查资料，在不需要时直接回答或澄清。

建立索引

先确认 Ollama 已启动，并且已经安装 .env 中配置的聊天模型和 embedding 模型。默认配置是：

bash

OLLAMA_MODEL=qwen3
OLLAMA_EMBEDDING_MODEL=embeddinggemma

如果还没有拉取 embedding 模型，先运行：

bash

ollama pull embeddinggemma

运行索引脚本：

bash

uv run python examples/05_rag_ingest.py

成功后会看到写入 chunk 数量和 Chroma 目录。脚本每次运行都会清空并重建 langchain_agent_tutorial collection，所以修改知识库文档后重新运行即可。

如果脚本提示找不到 Markdown 文档，检查 KNOWLEDGE_BASE_DIR 是否指向了包含 .md 文件的目录。默认目录是 data/knowledge_base。

运行 RAG Agent

建立索引后运行：

bash

uv run python examples/06_rag_agent_cli.py

可以尝试提问：

text

LangGraph 适合解决什么问题？

也可以问：

text

RAG 和 Tool 分别是什么？

资料不足时不要编造

这个 CLI 会创建一个 search_knowledge_base 工具。工具返回的内容包含来源和 chunk 文本，便于你确认模型实际拿到了哪些上下文。模型最终回答不应该脱离这些检索结果；如果知识库没有足够资料，它应该说明上下文不足。

调试 RAG

RAG 质量问题通常来自几个位置：

文档内容不够：知识库里根本没有答案，模型只能看到空上下文或弱相关上下文。
chunk 切分不合适：chunk 太大时噪声多，太小时语义被切碎，overlap 太小时上下文容易断裂。
embedding 模型效果有限：不同本地模型的 embedding 质量不同，语义相似度可能不稳定。
retriever 参数不合适：k 太小可能漏掉关键内容，太大可能把无关内容塞给模型。
工具没有被调用：本地模型工具调用能力不稳定时，Agent 可能绕过检索直接回答。
检索结果没有进入最终回答：工具返回了内容，但模型忽略或误读了上下文。

调试时先单独看检索结果，再看 Agent 最终回答。不要直接把失败归因于模型。更稳妥的顺序是：确认文档存在，确认索引成功，确认 retriever 能返回相关 chunk，最后再检查 Agent 是否调用工具并正确使用上下文。

代码阅读重点

阅读 examples/05_rag_ingest.py 时，重点看索引阶段：

load_markdown_documents() 读取 Markdown 文件并保留来源 metadata。
RecursiveCharacterTextSplitter 把长文档切成 chunk。
OllamaEmbeddings 使用 OLLAMA_EMBEDDING_MODEL 生成向量。
Chroma(... persist_directory=...) 把向量和文本保存到本地目录。

阅读 examples/06_rag_agent_cli.py 时，重点看问答阶段：

build_retriever() 从同一个 Chroma collection 构建 retriever。
search_knowledge_base() 把 retriever 包装成工具。
create_agent() 让模型在回答事实性问题前调用检索工具。

索引和问答拆成两个脚本，是为了让你清楚地区分“准备知识库”和“使用知识库”。真实项目里，这两个流程通常也不会混在同一个请求里。

RAG 参数怎么调

初学阶段优先调三个参数：

chunk_size：太小会丢上下文，太大会混入噪声。
chunk_overlap：太小会让跨段信息断裂，太大会增加重复内容。
k：返回 chunk 数量。太小可能漏信息，太大可能让模型被无关内容干扰。

不要一开始就引入复杂重排、混合检索或多向量索引。先用小知识库确认基本链路稳定，再逐步增加复杂度。

实验练习

修改 data/knowledge_base/sample.md，加入一段你自己的项目说明。
重新运行 examples/05_rag_ingest.py，再问 RAG Agent 新增内容。
把 retriever 的 k 从 3 改成 1，观察回答是否更容易漏信息。
故意问一个知识库里没有的问题，观察 Agent 是否承认上下文不足。

自测问题

为什么 RAG 要先索引再问答？
embedding 模型和聊天模型为什么要分开配置？
如果 RAG 答错了，你会先检查模型，还是先检查检索结果？

深入理解：RAG 的质量瓶颈通常不在生成阶段

很多 RAG 问题看起来像“模型答错了”，但根因常常在生成之前：

文档没有被正确读入。
chunk 把关键上下文切断。
embedding 模型不适合当前语言或领域。
retriever 找到的是弱相关片段。
工具返回内容太长，模型忽略了关键句。
prompt 没有要求模型基于检索结果回答。

因此，RAG 调试要把链路拆开。不要一开始就问 Agent，而是先问 retriever：给定这个 query，它到底返回了什么？如果检索结果不对，模型再强也只是基于错误上下文生成更流畅的错误答案。

Chunk 策略的取舍

chunk 不是越小越好，也不是越大越好。

小 chunk 的优点：

检索更精确。
放入上下文的噪声少。
来源定位更容易。

小 chunk 的缺点：

容易丢失跨段语义。
单个 chunk 可能缺少完整答案。
需要 overlap 弥补断裂。

大 chunk 的优点：

上下文更完整。
适合长段落、规范文档、教程类文本。

大 chunk 的缺点：

相似度可能被无关内容稀释。
返回给模型的噪声更多。
更容易占用上下文窗口。

本教程使用 chunk_size=500 和 chunk_overlap=80，不是通用最优值，而是适合小型 Markdown 教程的起点。真实项目应该基于评估集调参。

RAG 反模式

把所有文档一次性塞进 prompt：上下文会爆炸，也很难定位来源。
不保存 source metadata：答错后无法知道错误来自哪份文档。
用聊天模型做 embedding：聊天生成和向量化是不同任务，应使用 embedding 模型。
不检查检索结果：只看最终回答会掩盖 retrieval 问题。
知识库不清洗：重复、过期、冲突文档会让模型难以判断事实。
检索不足时仍要求模型回答：这会鼓励编造。

评估 RAG 的最小方法

准备一个小表格：

问题	期望来源	期望要点	实际检索来源	回答是否正确

每次改 chunk、embedding 模型或 k，都跑同一组问题。不要凭单个例子判断 RAG 变好了。RAG 优化需要看检索命中率和最终答案质量两个指标。

完整示例

建立索引RAG Agent

python

from __future__ import annotations

import sys
from pathlib import Path

sys.path.append(str(Path(__file__).resolve().parents[1]))

from langchain_chroma import Chroma
from langchain_core.documents import Document
from langchain_ollama import OllamaEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

from examples.common.config import load_config, ollama_client_kwargs
from examples.common.ollama_check import assert_model_available, assert_ollama_running


COLLECTION_NAME = "langchain_agent_tutorial"


def _friendly_error(exc: RuntimeError) -> str:
    message = str(exc)
    if message.startswith("Cannot reach Ollama."):
        return (
            "无法连接 Ollama。请先打开 Ollama 应用，或运行 `ollama serve`，"
            f"并确认服务地址可访问。原始信息：{message}"
        )
    if message.startswith("Cannot read Ollama model list"):
        return f"无法读取 Ollama 模型列表。请检查 OLLAMA_BASE_URL 是否正确。原始信息：{message}"
    if message.startswith("Ollama model list"):
        return f"Ollama 返回的模型列表格式异常。请检查 OLLAMA_BASE_URL 是否指向 Ollama。原始信息：{message}"
    if message.startswith("Ollama model `"):
        return (
            "当前配置的 Ollama embedding 模型未安装。"
            f"请先运行 `ollama pull {load_config().ollama_embedding_model}`。"
        )
    return message


def load_markdown_documents() -> list[Document]:
    config = load_config()
    documents: list[Document] = []

    for path in sorted(config.knowledge_base_dir.glob("*.md")):
        documents.append(
            Document(
                page_content=path.read_text(encoding="utf-8"),
                metadata={"source": str(path.relative_to(config.knowledge_base_dir.parent))},
            )
        )

    return documents


def assert_embedding_available(embeddings: OllamaEmbeddings, model_name: str) -> None:
    try:
        embeddings.embed_query("embedding health check")
    except Exception as exc:
        raise RuntimeError(
            f"Ollama embedding 模型 `{model_name}` 无法生成向量。"
            f"请确认已运行 `ollama pull {model_name}`，并且 Ollama 支持 embeddings。"
        ) from exc


def main() -> None:
    config = load_config()
    try:
        assert_ollama_running(config.ollama_base_url)
        assert_model_available(config.ollama_base_url, config.ollama_embedding_model)

        documents = load_markdown_documents()
        if not documents:
            raise RuntimeError(f"没有在 `{config.knowledge_base_dir}` 找到 Markdown 文档。")

        splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=80)
        chunks = splitter.split_documents(documents)

        embeddings = OllamaEmbeddings(
            model=config.ollama_embedding_model,
            base_url=config.ollama_base_url,
            client_kwargs=ollama_client_kwargs(),
        )
        assert_embedding_available(embeddings, config.ollama_embedding_model)

        vector_store = Chroma(
            collection_name=COLLECTION_NAME,
            embedding_function=embeddings,
            persist_directory=str(config.chroma_dir),
        )
        vector_store.reset_collection()
        vector_store.add_documents(chunks)
    except RuntimeError as exc:
        print(f"错误：{_friendly_error(exc)}")
        return
    except Exception as exc:
        print(f"错误：知识库索引构建失败：{exc}")
        return

    print(f"成功：已写入 {len(chunks)} 个文本块到 Chroma 目录 `{config.chroma_dir}`。")


if __name__ == "__main__":
    main()

python

from __future__ import annotations

import sys
from pathlib import Path

sys.path.append(str(Path(__file__).resolve().parents[1]))

from langchain.agents import create_agent
from langchain_chroma import Chroma
from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langchain_ollama import ChatOllama, OllamaEmbeddings

from examples.common.config import load_config, ollama_client_kwargs
from examples.common.ollama_check import assert_model_available, assert_ollama_running


COLLECTION_NAME = "langchain_agent_tutorial"
_retriever = None


def _index_guidance() -> str:
    return "请先运行 `uv run python examples/05_rag_ingest.py` 建立本地知识库索引。"


def _friendly_error(exc: RuntimeError, missing_model_hint: str | None = None) -> str:
    message = str(exc)
    if message.startswith("Cannot reach Ollama."):
        return (
            "无法连接 Ollama。请先打开 Ollama 应用，或运行 `ollama serve`，"
            f"并确认服务地址可访问。原始信息：{message}"
        )
    if message.startswith("Cannot read Ollama model list"):
        return f"无法读取 Ollama 模型列表。请检查 OLLAMA_BASE_URL 是否正确。原始信息：{message}"
    if message.startswith("Ollama model list"):
        return f"Ollama 返回的模型列表格式异常。请检查 OLLAMA_BASE_URL 是否指向 Ollama。原始信息：{message}"
    if message.startswith("Ollama model `"):
        if missing_model_hint is not None:
            return f"当前配置的 Ollama 模型未安装。请先运行 `ollama pull {missing_model_hint}`。"
        return "当前配置的 Ollama 模型未安装。请检查 `.env` 中的模型配置。"
    return message


def build_retriever():
    config = load_config()
    embeddings = OllamaEmbeddings(
        model=config.ollama_embedding_model,
        base_url=config.ollama_base_url,
        client_kwargs=ollama_client_kwargs(),
    )
    try:
        embeddings.embed_query("embedding health check")
    except Exception as exc:
        raise RuntimeError(
            f"Ollama embedding 模型 `{config.ollama_embedding_model}` 无法生成向量。"
            f"请确认已运行 `ollama pull {config.ollama_embedding_model}`，"
            "并且 Ollama 支持 embeddings。"
        ) from exc

    vector_store = Chroma(
        collection_name=COLLECTION_NAME,
        embedding_function=embeddings,
        persist_directory=str(config.chroma_dir),
    )

    try:
        if not vector_store.get(limit=1).get("ids"):
            raise RuntimeError("Chroma collection is empty.")
    except Exception as exc:
        raise RuntimeError(f"没有找到可用的 Chroma 索引。{_index_guidance()}") from exc

    return vector_store.as_retriever(search_kwargs={"k": 3})


@tool
def search_knowledge_base(query: str) -> str:
    """Search the local LangChain tutorial knowledge base for relevant context."""
    if _retriever is None:
        return f"没有可用的本地知识库检索器。{_index_guidance()}"

    try:
        docs = _retriever.invoke(query)
    except Exception as exc:
        return f"知识库检索失败：{exc}\n{_index_guidance()}"

    if not docs:
        return "没有检索到相关本地上下文。请说明知识库内容不足，不能编造答案。"

    return "\n\n".join(
        f"来源：{doc.metadata.get('source', 'unknown')}\n内容：{doc.page_content}"
        for doc in docs
    )


def _final_message_content(agent_response: dict) -> str:
    messages = agent_response.get("messages", [])
    if not messages:
        return ""

    final_message = messages[-1]
    if isinstance(final_message, AIMessage):
        return str(final_message.content)
    return str(getattr(final_message, "content", final_message))


def main() -> None:
    global _retriever

    config = load_config()
    try:
        assert_ollama_running(config.ollama_base_url)
        try:
            assert_model_available(config.ollama_base_url, config.ollama_model)
        except RuntimeError as exc:
            print(f"错误：{_friendly_error(exc, config.ollama_model)}")
            return
        try:
            assert_model_available(config.ollama_base_url, config.ollama_embedding_model)
        except RuntimeError as exc:
            print(f"错误：{_friendly_error(exc, config.ollama_embedding_model)}")
            return
        _retriever = build_retriever()
    except RuntimeError as exc:
        print(f"错误：{_friendly_error(exc)}")
        return

    model = ChatOllama(
        model=config.ollama_model,
        base_url=config.ollama_base_url,
        client_kwargs=ollama_client_kwargs(),
        temperature=0,
    )

    agent = create_agent(
        model=model,
        tools=[search_knowledge_base],
        system_prompt=(
            "你是 LangChain Agent 教程助手。回答事实性教程问题前，必须先调用 "
            "search_knowledge_base 检索本地知识库。回答时优先依据检索结果，并用中文简洁说明。"
            "如果知识库上下文不足，要明确说资料不足，不能编造。"
        ),
    )

    print("LangChain RAG Agent 示例。输入 `exit` 或 `quit` 退出。")
    print(f"索引目录：{config.chroma_dir}")
    while True:
        user_input = input("\n你：").strip()
        if user_input.lower() in {"exit", "quit"}:
            break
        if not user_input:
            continue

        try:
            response = agent.invoke({"messages": [{"role": "user", "content": user_input}]})
        except Exception as exc:
            print(f"错误：RAG Agent 执行失败：{exc}")
            print(_index_guidance())
            continue

        print(f"AI：{_final_message_content(response)}")


if __name__ == "__main__":
    main()

06. 构建 RAG Agent ​

学习目标 ​

RAG 解决什么问题 ​

数据流 ​

建立索引 ​

运行 RAG Agent ​

调试 RAG ​

代码阅读重点 ​

RAG 参数怎么调 ​

实验练习 ​

自测问题 ​

深入理解：RAG 的质量瓶颈通常不在生成阶段 ​

Chunk 策略的取舍 ​

RAG 反模式 ​

评估 RAG 的最小方法 ​

完整示例 ​

06. 构建 RAG Agent

学习目标

RAG 解决什么问题

数据流

建立索引

运行 RAG Agent

调试 RAG

代码阅读重点

RAG 参数怎么调

实验练习

自测问题

深入理解：RAG 的质量瓶颈通常不在生成阶段

Chunk 策略的取舍

RAG 反模式

评估 RAG 的最小方法

完整示例