9. 构建你的第一个 RAG 系统

from langchain_openai import ChatOpenAI
 
llm = ChatOpenAI(model="gpt-5-mini")
 
# 询问 LLM 从未见过的私有文档
response = llm.invoke("Acme Corp 的退款政策是什么?")
print(response.content)

我没有关于 Acme Corp 退款政策的具体信息。我建议查看他们的官方网站或直接联系他们的客户支持团队以获取最准确和最新的信息。

from langchain_openai import ChatOpenAI
 
llm = ChatOpenAI(model="gpt-5-mini")
 
# 实际上这会长得多,但让我们假设以下是整个文档
document_text = """
退款政策(生效日期:2026年1月):
- 购买后30天内凭原始收据可全额退款。
- 30天后,仅提供商店积分。
- 数字产品下载后不可退款。
- 有缺陷的商品可随时退货并获得全额退款。
"""
 
response = llm.invoke(
    f"""请根据以下文档回答:
{document_text}
 
问题:数字产品的退款政策是什么?"""
)
print(response.content)

数字产品下载后不可退款。
...

from langchain_openai import OpenAIEmbeddings
 
embeddings_model = OpenAIEmbeddings(model="text-embedding-3-small")
 
# 嵌入单个句子
vector = embeddings_model.embed_query("我如何拿回我的钱?")
 
print(f"向量维度: {len(vector)}")
print(f"前5个值: {vector[:5]}")

向量维度: 1536
前5个值: [0.0123, -0.0456, 0.0789, -0.0234, 0.0567]

from langchain_openai import OpenAIEmbeddings
import numpy as np
 
embeddings_model = OpenAIEmbeddings(model="text-embedding-3-small")
 
# 嵌入查询和两个候选文档
query_vec = embeddings_model.embed_query("退款期限是多久?")
doc1_vec = embeddings_model.embed_query("购买后30天内可全额退款。")  # 相关
doc2_vec = embeddings_model.embed_query("我们的办公室位于西雅图市中心。")  # 不相关
 
def cosine_similarity(a, b):
    """计算两个向量之间的余弦相似度。"""
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
 
sim1 = cosine_similarity(query_vec, doc1_vec)
sim2 = cosine_similarity(query_vec, doc2_vec)
 
print(f"查询 vs '退款' 文档:    {sim1:.4f}")
print(f"查询 vs '办公室' 文档:   {sim2:.4f}")

查询 vs '退款' 文档:    0.6415
查询 vs '办公室' 文档:   0.1706

mkdir -p data/docs

# 退款政策
 
**生效日期**:2026年1月1日
 
## 标准退货
 
所有实体产品可在购买后30天内退货并获得全额退款。
需要原始收据或订单确认邮件。商品必须处于原始包装和未使用状态。
 
30天后,接受退货但仅提供商店积分。商店积分不会过期。
 
## 数字产品
 
数字产品(软件许可证、电子书、在线课程)一旦激活下载或访问链接即不可退款。
如果您遇到阻止访问的技术问题,请在7天内联系支持以获得更换或退款。
 
## 有缺陷的商品
 
有缺陷的商品可随时退货并获得全额退款或更换。
请附上缺陷描述。有缺陷退货的运费由公司承担。
 
## 订阅服务
 
月度订阅可随时取消。退款根据计费周期中的剩余天数按比例计算。
年度订阅可在前14天内全额退款。14天后,不提供退款,但访问权限持续到计费期结束。

# 配送信息
 
## 国内配送
 
标准配送(5-7个工作日):订单满$50免费,否则$5.99。
快速配送(2-3个工作日):$12.99。
隔夜配送(下一个工作日):$24.99。
 
## 国际配送
 
国际订单通过跟踪航空邮件配送。配送时间因目的地而异,
通常为10-21个工作日。国际配送费用在结账时根据重量和目的地计算。
 
关税和进口税由买方承担,不包含在配送费用中。
 
## 订单跟踪
 
所有订单都包含跟踪号码,在发货后24小时内通过电子邮件发送。
通过电子邮件中的跟踪链接或通过承运商网站跟踪您的订单。
 
## 丢失或损坏的包裹
 
如果您的包裹丢失或到达时损坏,请在48小时内联系支持。
我们将免费发送替换品。对于损坏的商品,请提供损坏和包装的照片。

from pathlib import Path
from langchain_community.document_loaders import TextLoader
 
# 从 data/docs 目录加载所有 .md 文件
docs_dir = Path("data/docs")
 
for md_file in docs_dir.glob("*.md"):
    loader = TextLoader(str(md_file), encoding="utf-8")
    docs = loader.load()
 
    if docs:  # 检查文件不为空
        doc = docs[0]  # 单个文件 = 单个 Document
        print(f"文件: {doc.metadata['source']}")
        print(f"长度: {len(doc.page_content)} 字符")
        print(f"预览: {doc.page_content[:80]}...")
        print()

文件: data/docs/refund_policy.md
长度: 1166 字符
预览: # 退款政策
...
 
文件: data/docs/shipping_info.md
长度: 972 字符
预览: # 配送信息
...

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
 
# 加载文档
loader = TextLoader("data/docs/refund_policy.md", encoding="utf-8")
docs = loader.load()
 
# 配置分割器
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=400,
    chunk_overlap=80,
    separators=["\n## ", "\n\n", "\n", " "],
)
 
chunks = text_splitter.split_documents(docs)
 
print(f"分割成 {len(chunks)} 个块\n")
for i, chunk in enumerate(chunks):
    print(f"--- 块 {i} (来源: {chunk.metadata['source']}) ---")
    print(f"长度: {len(chunk.page_content)} 字符")
    print(chunk.page_content[:120])
    print()

分割成 4 个块
 
--- 块 0 (来源: data/docs/refund_policy.md) ---
长度: 374 字符
# 退款政策
...
 
--- 块 1 (来源: data/docs/refund_policy.md) ---
长度: 267 字符
## 数字产品
...
 
--- 块 2 (来源: data/docs/refund_policy.md) ---
长度: 204 字符
## 有缺陷的商品
...
 
--- 块 3 (来源: data/docs/refund_policy.md) ---
长度: 315 字符
## 订阅服务
...

pip install chromadb langchain-chroma

from langchain_openai import OpenAIEmbeddings
 
embedding_model = OpenAIEmbeddings(model="text-embedding-3-small")
 
# 验证它是否工作
test_vector = embedding_model.embed_query("测试")
print(f"嵌入维度: {len(test_vector)}")

嵌入维度: 1536

# ingest.py - 完整的摄入流水线
from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
 
# 步骤1:加载文档
# DirectoryLoader:扫描目录并加载匹配的文件。
# 实际加载委托给 loader_cls 中指定的加载器。
loader = DirectoryLoader(
    "data/docs/", glob="**/*.md",
    loader_cls=TextLoader, loader_kwargs={"encoding": "utf-8"},
)
documents = loader.load()
print(f"加载了 {len(documents)} 个文档")
 
# 步骤2:分割成块
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=400,
    chunk_overlap=80,
    separators=["\n## ", "\n\n", "\n", " ", ""],
)
chunks = text_splitter.split_documents(documents)
print(f"创建了 {len(chunks)} 个块")
 
# 步骤3:创建嵌入模型
embedding_model = OpenAIEmbeddings(model="text-embedding-3-small")
 
# 步骤4:创建向量存储并摄入块
vector_store = Chroma.from_documents(
    documents=chunks,
    embedding=embedding_model,
    persist_directory="data/chroma_db",
    collection_name="company_docs",
)
 
print(f"在 data/chroma_db/ 中存储了 {len(chunks)} 个块")

加载了 2 个文档
创建了 8 个块
在 data/chroma_db/ 中存储了 8 个块

from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
 
# 加载持久化的向量存储——无需重新嵌入
embedding_model = OpenAIEmbeddings(model="text-embedding-3-small")
 
vector_store = Chroma(
    persist_directory="data/chroma_db",
    collection_name="company_docs",
    embedding_function=embedding_model,
)
 
print(f"加载了包含 {len(vector_store.get()['ids'])} 个块的向量存储")

加载了包含 8 个块的向量存储

from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
 
embedding_model = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = Chroma(
    persist_directory="data/chroma_db",
    collection_name="company_docs",
    embedding_function=embedding_model,
)
 
# 搜索与问题相关的块
query = "我可以退回数字产品吗?"
results = vector_store.similarity_search(query, k=2)
 
print(f"查询: {query}")
print(f"找到 {len(results)} 个结果\n")
 
for i, doc in enumerate(results):
    print(f"--- 结果 {i + 1} (来源: {doc.metadata['source']}) ---")
    print(doc.page_content[:200])
    print()

查询: 我可以退回数字产品吗?
找到 2 个结果
 
--- 结果 1 (来源: data/docs/refund_policy.md) ---
## 数字产品
 
...
 
--- 结果 2 (来源: data/docs/refund_policy.md) ---
# 退款政策
 
**生效日期**:2026年1月1日
 
## 标准退货
 
...

results_with_scores = vector_store.similarity_search_with_score(query, k=2)
 
for doc, score in results_with_scores:
    # ChromaDB 返回距离(越低越相似)
    print(f"分数: {score:.4f} | 来源: {doc.metadata['source']}")
    print(f"  {doc.page_content[:200]}...")
    print()

分数: 0.5942 | 来源: data/docs/refund_policy.md
  ## 数字产品
 
...
 
分数: 0.9577 | 来源: data/docs/refund_policy.md
  # 退款政策
 
...

from langchain_core.prompts import ChatPromptTemplate
 
rag_prompt = ChatPromptTemplate.from_messages([
    ("system",
     "你是一名客户服务代表。"
     "仅使用提供的上下文回答用户的问题。"
     "如果上下文不包含足够的信息来回答,"
     "请说\"我没有足够的信息来回答这个问题。\"\n\n"
     "上下文:\n{context}"),
    ("human", "{question}"),
])

# rag_chain.py - 完整的 RAG 流水线
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
 
def format_docs(docs):
    """将检索到的文档连接成单个上下文字符串。"""
    return "\n\n---\n\n".join(doc.page_content for doc in docs)
 
 
def build_rag_chain():
    """构建并返回完整的 RAG 链。"""
    # 加载向量存储
    embedding_model = OpenAIEmbeddings(model="text-embedding-3-small")
    vector_store = Chroma(
        persist_directory="data/chroma_db",
        collection_name="company_docs",
        embedding_function=embedding_model,
    )
 
    # 创建检索器(k=3 表示返回前3个块)
    retriever = vector_store.as_retriever(search_kwargs={"k": 3})
 
    # 定义提示
    rag_prompt = ChatPromptTemplate.from_messages([
        ("system",
         "你是一名客户服务代表。"
         "仅使用提供的上下文回答用户的问题。"
         "如果上下文不包含足够的信息来回答,"
         "请说\"我没有足够的信息来回答这个问题。\"\n\n"
         "上下文:\n{context}"),
        ("human", "{question}"),
    ])
 
    # 初始化 LLM
    llm = ChatOpenAI(model="gpt-5-mini")
 
    # 使用 LCEL 组合链
    rag_chain = (
        {"context": retriever | format_docs, "question": lambda x: x}
        | rag_prompt
        | llm
        | StrOutputParser()
    )
 
    return rag_chain
 
 
if __name__ == "__main__":
    chain = build_rag_chain()
    answer = chain.invoke("我可以退回数字产品吗?")
    print(answer)

数字产品一旦激活下载或访问链接即不可退款。
如果您遇到阻止访问的技术问题,请在7天内联系支持以获得更换或退款。

rag_chain = (
    {"context": retriever | format_docs, "question": lambda x: x}
    | rag_prompt
    | llm
    | StrOutputParser()
)

# test_rag.py - 使用各种问题测试 RAG 链
from rag_chain import build_rag_chain
 
chain = build_rag_chain()
 
test_questions = [
    # 可回答——信息在文档中
    "实体产品的退款政策是什么?",
    "快速配送费用是多少?",
    "我可以在6个月后退回有缺陷的商品吗?",
    # 不可回答——信息不在文档中
    "员工休假政策是什么?",
    "使用了哪些编程语言?",
]
 
for question in test_questions:
    print(f"问: {question}")
    answer = chain.invoke(question)
    print(f"答: {answer}\n")
    print("-" * 60)

问: 实体产品的退款政策是什么?
答: 所有实体产品可在购买后30天内退货并获得全额退款。...
 
------------------------------------------------------------
问: 快速配送费用是多少?
答: 快速配送(2-3个工作日)费用为$12.99。
 
------------------------------------------------------------
问: 我可以在6个月后退回有缺陷的商品吗?
答: 可以。有缺陷的商品可随时退货并获得全额退款或更换。...
 
------------------------------------------------------------
问: 员工休假政策是什么?
答: 我没有足够的信息来回答这个问题。
 
------------------------------------------------------------
问: 使用了哪些编程语言?
答: 我没有足够的信息来回答这个问题。
 
------------------------------------------------------------