LangChain 实战第13课：如何用Langchain做RAG，开发企业级知识库？

今天分享下 LangChain 如何实现 RAG，并用“公司规章制度问答”做一个完整的例子。

一节课给你讲明白:

1. 1. RAG到底是个什么东西
2. 2. RAG做了什么
3. 3. langchain做了哪些RAG相关的内容。

先解释下，什么是 RAG？

RAG = 检索增强生成（Retrieval-Augmented Generation），分为两部分：

1. 1. 检索（Retrieval）：先从外部知识库（数据库 / 文档 / 向量库）找出相关信息。
2. 2. 生成（Generation）：再把检索结果交给大语言模型，让它基于这些资料生成答案。

好处：

• • 减少幻觉（LLM 不瞎编，回答基于真实文档）；
• • 支持私有知识（比如公司制度、内部文档）；
• • 可追溯（回答能给出处，方便溯源）。

LangChain 是怎么帮我们做 RAG 的？

LangChain 提供了很多 RAG 相关的组件，可以自由组合：

• • 文档处理
• • Document：统一文档格式（包含 page_content 和 metadata）。
• • TextSplitter：把长文档切成小块，提升检索准确率。
• • 向量化 & 存储
• • Embeddings：把文本变成向量（支持 OpenAI / HuggingFace / Cohere 等）。
• • VectorStore：常见的向量数据库封装（Chroma、FAISS、Weaviate 等）。
• • 检索器
• • as_retriever()：把向量数据库转成统一的检索接口，支持 similarity_search / mmr / hybrid search。
• • RAG 封装链
• • RetrievalQA：官方最常用的“RAG 一键链”，帮你自动完成“检索 + 生成”。
• • ConversationalRetrievalChain：带历史对话的 RAG，常用于企业问答机器人。

在这节课，我们选择 手写链条（prompt + LLM + parser），跑完案例就能清晰看到：

LangChain 的底层其实就是把这些拼装在一起。

案例

我们准备了 10 条公司规章制度，存入向量数据库 Chroma，让 HR 助手来回答员工问题。

步骤 1：准备文档

在 LangChain 里，文档用 Document 表示，除了正文（page_content），还可以附加 metadata，方便检索和过滤。

             
             
             
             from langchain.schema import Document

docs = [
    Document(page_content="员工必须每天打卡。", metadata={"category": "attendance", "rule_id": "R001"}),
    Document(page_content="上班期间禁止使用私人手机。", metadata={"category": "discipline", "rule_id": "R002"}),
    Document(page_content="员工请假需提前 3 天申请。", metadata={"category": "leave", "rule_id": "R004", "notice_days": 3}),
    Document(page_content="公司提供每年 10 天带薪年假。", metadata={"category": "leave", "rule_id": "R005", "days": 10}),
    Document(page_content="办公室内禁止吸烟。", metadata={"category": "discipline", "rule_id": "R006"}),
    # ... 共 10 条
]

步骤 2：文档分块

LangChain 内置了 RecursiveCharacterTextSplitter，会优先按段落 / 句子切割，避免“截断语义”。

             
             
             
             from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
split_docs = splitter.split_documents(docs)

步骤 3：向量化 & 存储

• • 我们使用 HuggingFace 的 bge-small-zh-v1.5 中文向量模型；
• • 使用 Chroma 向量数据库，支持持久化到本地目录。

             
             
             
             from langchain_huggingface import HuggingFaceEmbeddings
from langchain_chroma import Chroma

embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-small-zh-v1.5")
vectorstore = Chroma.from_documents(
    documents=split_docs,
    embedding=embeddings,
    collection_name="course_rag",
    persist_directory="./chroma_db",
)

步骤 4：构建 RAG 链

我们这里没有直接用 RetrievalQA，而是手写了一个 Prompt 链：

• • 先用向量数据库 similarity_search 找到相关文档；
• • 再拼接进 Prompt，让 LLM 生成答案；
• • 这样更清晰地看到 RAG 是怎么一步步完成的。

             
             
             
             from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="deepseek-chat", base_url="https://api.deepseek.com/v1", api_key="你的KEY")

prompt = ChatPromptTemplate.from_template(
    "你是一名企业HR助手。请基于【检索上下文】回答问题；如无依据，请说明。\n问题: {question}\n【检索上下文】:\n{context}"
)

rag_chain = prompt | llm | StrOutputParser()

步骤 5：运行查询

我们写了一个小函数 run_query，能展示 ReAct 风格的思考过程，帮助大家理解。

             
             
             
             def run_query(query: str, top_k: int = 4, filter: dict | None = None):
    print(f"Question: {query}")
    print("Thought: 我需要从向量库检索相关片段，并据此回答。")
    print("Action: Retriever")
    print(f"Action Input: query='{query}', k={top_k}, filter={filter}")

    results = vectorstore.similarity_search_with_score(query, k=top_k, filter=filter)

    print("Observation:")
    for i, (doc, score) in enumerate(results, start=1):
        print(f"  [{i}] score={score:.4f} content={doc.page_content} metadata={doc.metadata}")

    context = "\n".join([doc.page_content for doc, _ in results])
    print("Thought: 我已获得相关片段，接下来组织答案。")
    answer = rag_chain.invoke({"question": query, "context": context})
    print("Final Answer:", answer)
    print("出处：", [doc.page_content for doc, _ in results], "\n")

完整代码

             
             
             
             import os
import sys
from dotenv import load_dotenv

from langchain_openai import ChatOpenAI
from langchain_huggingface import HuggingFaceEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain.schema import Document

load_dotenv()
os.environ.setdefault("HF_HUB_DISABLE_SYMLINKS_WARNING", "1")

# 初始化 LLM
llm = ChatOpenAI(
    model="deepseek-chat",
    base_url=os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com/v1"),
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    temperature=0
)

# 准备文档（公司规章制度），附带元数据 metadata 便于教学与过滤
docs = [
    Document(page_content="员工必须每天打卡。", metadata={"source": "company_policy", "category": "attendance", "rule_id": "R001"}),
    Document(page_content="上班期间禁止使用私人手机。", metadata={"source": "company_policy", "category": "discipline", "rule_id": "R002"}),
    Document(page_content="每周一下午有全员例会。", metadata={"source": "company_policy", "category": "meeting", "rule_id": "R003"}),
    Document(page_content="员工请假需提前 3 天申请。", metadata={"source": "company_policy", "category": "leave", "rule_id": "R004", "notice_days": 3}),
    Document(page_content="公司提供每年 10 天带薪年假。", metadata={"source": "company_policy", "category": "leave", "rule_id": "R005", "days": 10}),
    Document(page_content="办公室内禁止吸烟。", metadata={"source": "company_policy", "category": "discipline", "rule_id": "R006"}),
    Document(page_content="员工需遵守信息安全规范。", metadata={"source": "company_policy", "category": "security", "rule_id": "R007"}),
    Document(page_content="加班需经理批准并记录。", metadata={"source": "company_policy", "category": "overtime", "rule_id": "R008"}),
    Document(page_content="公司内部文档禁止外传。", metadata={"source": "company_policy", "category": "security", "rule_id": "R009"}),
    Document(page_content="员工每季度需完成一次培训课程。", metadata={"source": "company_policy", "category": "training", "rule_id": "R010"}),
]

# 启动时打印知识库文档，便于教学查看语料与元数据
def print_corpus(items):
    print("=== 知识库文档（原始） ===")
    for i, d in enumerate(items, start=1):
        print(f"[{i}] content={d.page_content} metadata={d.metadata}")
    print(f"总文档数: {len(items)}\n")

print_corpus(docs)

# 文本分块
# 文本分块（内容较短，为避免过碎，适当增大 chunk）
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
split_docs = splitter.split_documents(docs)
print(f"分块后文档数: {len(split_docs)}\n")

# 向量化 + 向量数据库（用 Chroma 并持久化）
embeddings = HuggingFaceEmbeddings(
    model_name=os.getenv("EMBEDDING_MODEL_NAME", "BAAI/bge-small-zh-v1.5")
)
vectorstore = Chroma.from_documents(
    documents=split_docs,
    embedding=embeddings,
    collection_name="course_rag",
    persist_directory="./chroma_db",
)

# 构建手写 RAG Prompt 链（便于过程透明展示）
prompt = ChatPromptTemplate.from_template(
    "你是一名企业HR助手。请基于【检索上下文】用中文、简洁且准确地回答问题；如无充分依据请明确说明。\n问题: {question}\n【检索上下文】:\n{context}"
)
rag_chain = prompt | llm | StrOutputParser()

def run_query(query: str, top_k: int = 4, filter: dict | None = None):
    print(f"Question: {query}")
    print("Thought: 我需要从向量库检索相关片段，并据此回答。")
    print("Action: Retriever")
    print(f"Action Input: query='{query}', k={top_k}, search_type='similarity', filter={filter}")

    results = vectorstore.similarity_search_with_score(query, k=top_k, filter=filter)

    print("Observation:")
    for i, (doc, score) in enumerate(results, start=1):
        try:
            score_val = float(score)
        except Exception:
            score_val = score
        print(f"  [{i}] score={score_val} content={doc.page_content} metadata={doc.metadata}")

    context = "\n".join([doc.page_content for doc, _ in results])
    print("Thought: 我已获得相关片段，接下来组织答案。")
    answer = rag_chain.invoke({"question": query, "context": context})
    print("Final Answer:", answer)
    print("出处：", [doc.page_content for doc, _ in results], "\n")

print("=== 企业HR助手 RAG Demo（请输入你的问题） ===")
while True:
    query = input("你：")
    if query.lower() in ["exit", "quit", "退出"]:
        break
    run_query(query)
    try:
        vectorstore.persist()
    except Exception:
        pass

运行效果如下：

你看这效果，是不是AI能够根据我们的问题，从“知识库”中找到相关答案，并且告诉我们准确的结果？

其实，跑通RAG知识库，就是这么简单。