RAG 应用设计

RAG（Retrieval-Augmented Generation，检索增强生成）是一种将信息检索与文本生成相结合的技术架构。它通过从外部知识库中检索相关文档，为 LLM 提供上下文信息，从而生成更准确、更可靠的回答。

RAG 的核心流程

RAG 的完整工作流包括 6 个步骤：

加载 → 分割 → 嵌入 → 存储 → 检索 → 生成
Load → Split → Embed → Store → Retrieve → Generate

                   ┌─────────────┐
                   │  外部知识库  │
                   └──────┬──────┘
                          │
                ┌─────────▼─────────┐
                │   文档加载器       │
                │ (Document Loader)  │
                └─────────┬─────────┘
                          │
                ┌─────────▼─────────┐
                │   文本分割器       │
                │ (Text Splitter)    │
                └─────────┬─────────┘
                          │
                ┌─────────▼─────────┐
                │   嵌入模型         │
                │ (Embeddings)       │
                └─────────┬─────────┘
                          │
                ┌─────────▼─────────┐
                │   向量存储         │
                │ (Vector Store)     │
                └─────────┬─────────┘
                          │
                ┌─────────▼─────────┐
                │   检索器           │
                │ (Retriever)        │
                └─────────┬─────────┘
                          │
                ┌─────────▼─────────┐
                │   LLM 生成        │
                │ (Generator)        │
                └─────────┬─────────┘
                          │
                   ┌──────▼──────┐
                   │   最终回答   │
                   └─────────────┘

环境准备

pip install langchain langchain-openai langchain-community
pip install chromadb faiss-cpu

实现方式一：使用 create_agent（推荐）

LangChain 的 create_agent API 可以轻松构建 RAG 代理：

定义工具

from langchain.tools import tool
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 加载文档并构建检索索引
loader = TextLoader("knowledge.txt")
docs = loader.load()

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500, chunk_overlap=50
)
chunks = text_splitter.split_documents(docs)

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = FAISS.from_documents(chunks, embeddings)
retriever = vector_store.as_retriever(search_kwargs={"k": 3})

# 定义检索工具
@tool
def search_knowledge(query: str) -> str:
    """搜索知识库，获取与查询相关的内容"""
    docs = retriever.invoke(query)
    return "\n\n".join([doc.page_content for doc in docs])

tools = [search_knowledge]

构建 Agent

from langgraph.prebuilt import create_agent

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

agent = create_agent(
    model=llm,
    tools=tools,
    system_prompt="你是一个知识助手。使用 search_knowledge 工具搜索知识库，"
                  "基于检索到的内容回答用户问题。请注明信息来源。"
)

# 执行查询
result = agent.invoke({"messages": [("human", "什么是 RAG？")]})
print(result["messages"][-1].content)

实现方式二：手动构建 RAG 链

定义完整的 RAG 流程

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 1. 加载文档
loader = DirectoryLoader(
    "./knowledge/",
    glob="**/*.txt",
    loader_cls=TextLoader,
    show_progress=True
)
documents = loader.load()
print(f"加载了 {len(documents)} 个文档")

# 2. 分割文档
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
chunks = text_splitter.split_documents(documents)
print(f"分割为 {len(chunks)} 个块")

# 3. 嵌入并存储
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = FAISS.from_documents(chunks, embeddings)

# 4. 创建检索器
retriever = vector_store.as_retriever(search_kwargs={"k": 4})

# 5. 构建提示模板
template = """你是知识库助手。请基于以下上下文回答用户问题。

上下文：
{context}

问题：{question}

回答："""

prompt = ChatPromptTemplate.from_template(template)

# 6. 构建生成链
def format_docs(docs):
    return "\n\n".join([
        f"[来源 {i+1}] {doc.page_content}" 
        for i, doc in enumerate(docs)
    ])

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | ChatOpenAI(model="gpt-4o-mini", temperature=0)
    | StrOutputParser()
)

# 7. 执行查询
response = rag_chain.invoke("RAG 的核心流程是什么？")
print(response)

实现方式三：带对话历史的 RAG

结合对话历史实现上下文感知的 RAG：

from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.checkpoint import InMemorySaver
from langgraph.prebuilt import create_agent
from langchain_openai import ChatOpenAI

@tool
def search_knowledge(query: str) -> str:
    """搜索知识库"""
    docs = retriever.invoke(query)
    return "\n\n".join([doc.page_content for doc in docs])

tools = [search_knowledge]

# 创建带 checkpointer 的 agent
agent = create_agent(
    model=ChatOpenAI(model="gpt-4o-mini", temperature=0),
    tools=tools,
    checkpointer=InMemorySaver(),
    system_prompt="你是一个知识助手，基于知识库回答问题。"
)

# 第一次对话
result_1 = agent.invoke(
    {"messages": [("human", "什么是 RAG？")]},
    config={"configurable": {"thread_id": "1"}}
)
print(result_1["messages"][-1].content)

# 第二次对话（有上下文）
result_2 = agent.invoke(
    {"messages": [("human", "它的优势是什么？")]},
    config={"configurable": {"thread_id": "1"}}
)
print(result_2["messages"][-1].content)

高级 RAG 技术

带来源引用的 RAG

@tool
def search_with_sources(query: str) -> str:
    """搜索知识库并返回带来源的信息"""
    docs = retriever.invoke(query)
    results = []
    for i, doc in enumerate(docs):
        source = doc.metadata.get("source", f"来源-{i+1}")
        results.append(f"[{source}]: {doc.page_content}")
    return "\n\n".join(results)

多查询增强检索

from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_openai import ChatOpenAI

# 使用 LLM 生成多个查询变体
multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=vector_store.as_retriever(search_kwargs={"k": 3}),
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0)
)

@tool
def enhanced_search(query: str) -> str:
    """增强检索，生成多个查询进行检索"""
    docs = multi_query_retriever.invoke(query)
    return "\n\n".join([doc.page_content for doc in docs])

混合检索（向量 + 关键词）

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

# BM25 关键词检索
bm25_retriever = BM25Retriever.from_texts(
    [chunk.page_content for chunk in chunks]
)
bm25_retriever.k = 3

# 向量检索
vector_retriever = vector_store.as_retriever(search_kwargs={"k": 3})

# 集成检索
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.3, 0.7]
)

@tool
def hybrid_search(query: str) -> str:
    """混合检索（向量 + 关键词）"""
    docs = ensemble_retriever.invoke(query)
    return "\n\n".join([doc.page_content for doc in docs])

完整示例：企业知识库问答系统

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langgraph.checkpoint import InMemorySaver
from langgraph.prebuilt import create_agent
from langchain.tools import tool

# ---- 1. 索引阶段 ----
print("正在构建知识索引...")

loader = DirectoryLoader(
    "./knowledge_base/",
    glob="**/*.txt",
    loader_cls=TextLoader,
    show_progress=True
)
documents = loader.load()

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
chunks = text_splitter.split_documents(documents)

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = Chroma.from_documents(
    chunks,
    embeddings,
    persist_directory="./chroma_rag_db"
)
retriever = vector_store.as_retriever(search_kwargs={"k": 4})
print(f"索引完成：{len(chunks)} 个文档块")

# ---- 2. 定义工具 ----
@tool
def search_database(query: str) -> str:
    """从知识库搜索相关信息"""
    docs = retriever.invoke(query)
    results = []
    for doc in docs:
        source = doc.metadata.get("source", "unknown")
        results.append(f"[来源: {source}]\n{doc.page_content}")
    return "\n\n---\n\n".join(results)

tools = [search_database]

# ---- 3. 创建 RAG Agent ----
agent = create_agent(
    model=ChatOpenAI(model="gpt-4o-mini", temperature=0),
    tools=tools,
    checkpointer=InMemorySaver(),
    system_prompt="""你是一个企业知识库助手。
操作规则：
1. 始终使用 search_database 搜索相关信息
2. 基于检索结果回答问题
3. 如果检索结果不足，请告知用户
4. 引用信息来源
5. 使用中文回答"""
)

# ---- 4. 对话示例 ----
def ask_question(question: str, thread_id: str = "default"):
    result = agent.invoke(
        {"messages": [("human", question)]},
        config={"configurable": {"thread_id": thread_id}}
    )
    return result["messages"][-1].content

# 测试
print("=" * 50)
print("用户：公司的假期政策是什么？")
response = ask_question("公司的假期政策是什么？")
print(f"助手：{response}")

print("\n" + "=" * 50)
print("用户：年假天数是多少？")
response = ask_question("年假天数是多少？")
print(f"助手：{response}")

RAG 优化策略

优化方向	方法	效果
文档加载	支持更多格式（PDF、HTML、Markdown）	扩大知识来源
文本分割	调整 chunk_size 和 chunk_overlap	检索精度提升
嵌入模型	使用中文优化模型（BGE 等）	中文语义理解更好
检索增强	MultiQuery、Hybrid、MMR 等	召回率提升
提示工程	优化 prompt 模板	回答质量提升
后处理	对检索结果重排序	相关性提升

最佳实践

1. 索引与检索分离：索引完成后保存向量存储，检索时加载
2. 合理的块大小：通常 500-1000 字符，根据内容调整
3. 元数据管理：保存来源、时间等元数据便于追踪
4. 召回率与准确率平衡：调整 k 值和检索策略
5. 监控与评估：使用 LangSmith 追踪 RAG 调用
6. 缓存与优化：频繁查询结果可缓存

下一步

• 检索器：深入学习不同的检索策略
• 向量存储：了解向量存储的更多配置
• Memory 记忆：为 RAG 系统添加记忆功能
• Streaming 流式输出：实现流式输出
• 测试与评估：评估 RAG 系统性能