LangChain 从入门到实战：从 0 构建一个可追溯的本地知识库 RAG 问答系统

LangChain 从入门到实战：从 0 构建一个可追溯的本地知识库 RAG 问答系统

本文面向第一次接触 LangChain 的 Python 开发者。我们不会停留在“调用一次大模型”，而是从核心概念、环境搭建开始，最终完成一个能检索本地资料、基于资料回答、展示引用来源的 RAG 应用。

文章按 LangChain 当前主线 API 编写，更新时间：2026 年 7 月。LangChain 迭代较快，安装时建议使用虚拟环境并锁定主版本。

前言：为什么调用大模型之后，还需要 LangChain？

只调用一次大模型 API 并不困难：准备一段提示词，发送请求，读取回答即可。真正的 AI 应用却通常还需要处理下面这些问题：

• 怎样在不同模型供应商之间切换？
• 怎样组织系统提示词、用户消息和历史消息？
• 怎样让模型查询 PDF、Markdown、数据库或业务 API？
• 怎样保存对话状态，并支持流式输出？
• 怎样记录每一步调用，定位“为什么答错了”？
• 怎样让模型决定何时调用搜索、计算器等工具？

LangChain 的价值不在于“让模型变聪明”，而在于给模型周围的应用逻辑提供统一组件和编排方式。

可以把一个大模型应用粗略理解成：

用户问题
↓
提示词 / 历史消息 / 外部知识
↓
大模型 ←→ 工具、数据库、搜索服务
↓
结构化结果 / 流式回答 / 运行轨迹

LangChain 负责连接这些环节；大模型仍然负责理解与生成，向量数据库负责检索，业务系统负责提供真实数据。它不是模型，也不是数据库。

一、先理清 LangChain、LangGraph 和 LangSmith

初学者最容易被这几个名称绕晕。

当前 LangChain 的 Agent 构建在 LangGraph 之上。简单项目先学 LangChain 即可；当流程包含分支、循环、持久化和人工介入时，再深入 LangGraph。

二、必须掌握的核心组件

1. Model：模型统一接口

LangChain 将不同供应商的聊天模型包装成相近的调用方式。换模型时，业务代码通常不需要整体重写。

response = model.invoke("请用一句话解释 RAG")
print(response.content)

需要注意：统一接口不代表所有模型能力完全相同。工具调用、结构化输出、上下文窗口和多模态能力仍以具体模型为准。

2. Message：消息

聊天模型接收的不是一整段普通字符串，而是一组带角色的消息：

• SystemMessage：规定身份、边界和回答规则；
• HumanMessage：用户输入；
• AIMessage：模型回复，可能包含工具调用信息；
• ToolMessage：工具执行结果。

3. Prompt Template：提示词模板

模板将动态变量安全地注入提示词，使提示词从“拼字符串”升级为可复用组件。

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
("system", "你是一名{role}，回答必须简洁准确。"),
("human", "{question}"),
])

4. Document：文档

LangChain 使用 Document 表示一段文本及其元数据：

from langchain_core.documents import Document

doc = Document(
page_content="员工每年享有 10 天带薪年假。",
metadata={"source": "员工手册", "section": "休假制度"},
)

metadata 非常重要。来源、页码、章节、时间和权限标签都应该保留下来，后续才能引用来源和做过滤检索。

5. Embedding：文本向量

Embedding 模型把文本转换成数值向量。语义相近的文本，在向量空间中的距离通常也更近。

“年假有多少天” ─┐
├─ 向量距离较近
“带薪休假额度” ─┘

Embedding 模型和负责回答问题的聊天模型是两类模型，不要混为一谈。

6. Vector Store 与 Retriever

• Vector Store 保存文本块及其向量，并执行相似度搜索；
• Retriever 接收自然语言问题，返回相关 Document；
• Retriever 不一定来自向量数据库，也可以连接关键词搜索、API 或现有知识系统。

7. Tool 与 Agent

Tool 是模型可以调用的函数；Agent 则让模型在运行时决定调用哪个工具、传什么参数、是否继续执行。

固定流程适合普通 Chain，动态决策适合 Agent。能用确定性流程解决的问题，不必一开始就上 Agent。

三、RAG 到底是什么？

RAG 全称是 Retrieval-Augmented Generation，即检索增强生成。

普通问答：

问题 → 大模型 → 回答

RAG 问答：

问题 → 检索相关资料 → 问题 + 资料 → 大模型 → 有依据的回答

它主要解决两个问题：

典型的离线建库流程是：

加载文档 → 文本切分 → 生成向量 → 写入向量库

在线问答流程是：

用户提问 → 问题向量化 → 召回相关文本块 → 拼接上下文 → 模型回答

RAG 不是“彻底消灭幻觉”。它只是给模型提供依据。检索不到、切分不合理、资料过期或提示词约束不足，都会导致错误。

四、从 0 搭建开发环境

1. 创建项目

mkdir langchain-rag-demo
cd langchain-rag-demo

python -m venv .venv

激活虚拟环境：

# Windows PowerShell
.venv\\Scripts\\Activate.ps1

# macOS / Linux
source .venv/bin/activate

建议使用 Python 3.10 或更高版本。

2. 安装依赖

pip install "langchain>=1,<2" "langchain-core>=1,<2" "langchain-openai>=1,<2" "langchain-text-splitters>=1,<2" "python-dotenv>=1,<2"

为什么要安装多个包？因为 LangChain 采用模块化设计：核心抽象、模型供应商、文本切分器分别发布，项目只安装真正需要的集成。

3. 配置环境变量

新建 .env：

OPENAI_API_KEY=你的_API_Key
OPENAI_CHAT_MODEL=填写你的聊天模型名称
OPENAI_EMBEDDING_MODEL=填写你的向量模型名称

# 使用 OpenAI 兼容服务时再填写，官方服务可以留空
# OPENAI_BASE_URL=https://your-provider.example/v1

不要把真实密钥直接写进 Python 文件，也不要提交到 Git 仓库。记得在 .gitignore 中添加：

.env
.venv/
__pycache__/

聊天模型和向量模型必须由所使用的服务实际支持。某些“OpenAI 兼容接口”只兼容聊天接口，并不提供 Embedding API，使用前需要检查供应商文档。

五、第一步：完成最小模型调用

新建 hello_langchain.py：

import os

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

model = ChatOpenAI(
model=os.environ["OPENAI_CHAT_MODEL"],
api_key=os.environ["OPENAI_API_KEY"],
base_url=os.getenv("OPENAI_BASE_URL") or None,
temperature=0,
)

response = model.invoke("请用不超过 50 个字解释 LangChain。")
print(response.content)

运行：

python hello_langchain.py

这里有两个值得注意的细节：

• invoke() 是 LangChain Runnable 体系中的标准同步调用方法；
• temperature=0 会降低回答的随机性，但不等于保证事实正确。

六、第二步：用提示词模板和 LCEL 组织流程

LangChain Expression Language，简称 LCEL，使用管道符 | 连接 Runnable 组件。

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

prompt = ChatPromptTemplate.from_messages([
("system", "你是 Python 教师，使用通俗中文回答。"),
("human", "请解释：{topic}"),
])

chain = prompt | model | StrOutputParser()

answer = chain.invoke({"topic": "向量数据库为什么能做语义搜索"})
print(answer)

数据依次经过三个组件：

字典参数 → ChatPromptTemplate → ChatModel → StrOutputParser → 字符串

这种写法的优点是组件边界清晰，而且大部分 Runnable 都支持：

• invoke()：单次同步调用；
• ainvoke()：单次异步调用；
• batch()：批量调用；
• stream()：流式调用。

七、完整实战：本地员工手册 RAG 问答系统

我们的目标是构建一个命令行程序：

1. 项目结构

langchain-rag-demo/
├── .env
├── knowledge.md
├── rag_app.py
└── requirements.txt

2. 准备知识库

新建 knowledge.md：

# 星河科技员工手册

## 工作时间

公司实行每日 8 小时、每周 5 天工作制。标准工作时间为 9:00 至 18:00，
午休时间为 12:00 至 13:00。员工每周可以申请两天远程办公，需提前一天
在办公系统中提交申请并获得直属主管批准。

## 年假制度

入职满一年但不满十年的员工，每个自然年享有 10 天带薪年假；入职满十年
的员工享有 15 天带薪年假。年假最小申请单位为半天，原则上应提前 3 个
工作日提交申请。未使用的年假最多可结转 5 天，并须在次年 3 月 31 日前用完。

## 报销制度

报销实行逐级审批。单笔金额不超过 500 元的日常费用，由直属主管审批；超过
500 元且不超过 5000 元的费用，还需要部门负责人审批；超过 5000 元的费用
还需要财务负责人审批。所有报销应在费用发生后 30 天内提交，并附合法有效的发票。

## 设备借用

员工可以在 IT 服务台借用显示器、键盘和测试手机。测试手机最长借用 14 天，
如需续借，应在到期前提交申请。设备损坏或丢失后，应立即联系 IT 服务台。

3. 编写完整程序

新建 rag_app.py：

import os
from pathlib import Path

from dotenv import load_dotenv
from langchain_core.documents import Document
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

load_dotenv()

def require_env(name: str) –> str:
"""读取必需的环境变量，缺失时给出清晰错误。"""
value = os.getenv(name)
if not value:
raise RuntimeError(f"缺少环境变量：{name}")
return value

api_key = require_env("OPENAI_API_KEY")
chat_model_name = require_env("OPENAI_CHAT_MODEL")
embedding_model_name = require_env("OPENAI_EMBEDDING_MODEL")
base_url = os.getenv("OPENAI_BASE_URL") or None

# 1. 初始化聊天模型和向量模型
llm = ChatOpenAI(
model=chat_model_name,
api_key=api_key,
base_url=base_url,
temperature=0,
)

embeddings = OpenAIEmbeddings(
model=embedding_model_name,
api_key=api_key,
base_url=base_url,
)

# 2. 加载文档，并保留来源信息
knowledge_path = Path(__file__).with_name("knowledge.md")
raw_text = knowledge_path.read_text(encoding="utf-8")

documents = [
Document(
page_content=raw_text,
metadata={"source": knowledge_path.name},
)
]

# 3. 将长文档切分为适合检索的小块
splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=100,
add_start_index=True,
separators=["\\n## ", "\\n### ", "\\n\\n", "\\n", "。", "；", "，", " ", ""],
)
chunks = splitter.split_documents(documents)

for index, chunk in enumerate(chunks):
chunk.metadata["chunk_id"] = index

# 4. 建立内存向量库
vector_store = InMemoryVectorStore(embeddings)
vector_store.add_documents(documents=chunks)

# 5. 将向量库包装成检索器
retriever = vector_store.as_retriever(
search_type="similarity",
search_kwargs={"k": 3},
)

# 6. 定义“只能根据资料回答”的提示词
prompt = ChatPromptTemplate.from_messages([
(
"system",
"""你是企业制度问答助手。请严格根据给定资料回答问题。
如果资料中没有答案，直接回答“根据现有资料无法确定”，不要猜测。
回答应简洁、准确；涉及条件、时间或金额时必须保留关键数字。

资料：
{context}""",
),
("human", "问题：{question}"),
])

answer_chain = prompt | llm | StrOutputParser()

def format_documents(docs: list[Document]) –> str:
"""将检索结果整理成适合放入提示词的文本。"""
parts = []
for doc in docs:
source = doc.metadata.get("source", "未知来源")
chunk_id = doc.metadata.get("chunk_id", "?")
parts.append(
f"[来源：{source}，文本块：{chunk_id}]\\n{doc.page_content}"
)
return "\\n\\n".join(parts)

def ask(question: str) –> tuple[str, list[Document]]:
"""先检索，再生成答案，同时返回引用文档。"""
retrieved_docs = retriever.invoke(question)
answer = answer_chain.invoke({
"question": question,
"context": format_documents(retrieved_docs),
})
return answer, retrieved_docs

if __name__ == "__main__":
print(f"知识库加载完成，共生成 {len(chunks)} 个文本块。")
print("输入 exit 退出。\\n")

while True:
question = input("你：").strip()
if question.lower() in {"exit", "quit", "q"}:
break
if not question:
continue

answer, sources = ask(question)
print(f"\\n助手：{answer}")
print("引用来源：")
for doc in sources:
print(
f"- {doc.metadata.get('source')}"
f" / chunk={doc.metadata.get('chunk_id')}"
f" / start={doc.metadata.get('start_index')}"
)
print()

4. 运行项目

python rag_app.py

可以测试以下问题：

你：入职两年的员工有几天年假？
助手：入职满一年但不满十年的员工，每个自然年享有 10 天带薪年假。

你：一笔 6000 元的费用需要谁审批？
助手：需要直属主管、部门负责人以及财务负责人审批。

你：公司每月提供多少餐补？
助手：根据现有资料无法确定。

第三个问题很重要：一个合格的知识库助手不仅要知道“如何回答”，还要知道“资料不足时拒绝猜测”。

八、逐行理解 RAG 的关键设计

1. 为什么要切分文本？

如果直接把整本手册生成一个向量，多个主题会被压缩在同一个表示中，检索精度通常较差；如果文本块太小，又会丢失上下文。

本文使用：

chunk_size=500
chunk_overlap=100

这不是通用最优值。中文制度文档可以从 300～800 字符开始实验；代码、表格和扫描 PDF 应使用更有针对性的切分策略。

chunk_overlap 让相邻文本块保留一部分重复内容，降低关键信息刚好被边界切断的概率，但重叠过大会增加存储量并产生重复召回。

2. 为什么使用 RecursiveCharacterTextSplitter？

它会按照给定分隔符逐级尝试切分，优先保留章节、段落和句子边界。对于一般文本，它通常是比“固定长度硬切”更稳妥的起点。

3. 为什么要保留来源？

模型生成的“引用”不一定真实。可靠做法是由程序读取检索结果的 metadata，并在界面上展示来源，而不是让模型凭空编造页码。

4. 为什么演示使用内存向量库？

InMemoryVectorStore 零配置，适合学习和小型实验。但程序关闭后索引会丢失，而且每次启动都会重新调用 Embedding API。

生产项目应该根据数据规模和运维条件选择 Chroma、FAISS、PGVector、Milvus、Qdrant、Pinecone 等持久化方案，并将“建库”和“查询”拆成两个独立流程。

5. 为什么显式写“检索 → 生成”？

这是一种 2-Step RAG：每个问题都先检索，再生成。它的调用次数和延迟更可预测，特别适合 FAQ、企业制度和产品文档机器人。

如果让 Agent 自己决定何时检索，就变成 Agentic RAG，灵活性更高，但行为、延迟和成本也更难预测。

九、进阶：把知识库封装成 Agent 工具

当应用不只有一个知识库，还需要查询天气、订单、数据库或计算器时，可以把检索能力封装成 Tool。

from langchain.agents import create_agent
from langchain.tools import tool

@tool
def search_employee_handbook(question: str) –> str:
"""查询员工手册中的工作时间、休假、报销和设备制度。"""
docs = retriever.invoke(question)
return format_documents(docs)

agent = create_agent(
model=llm,
tools=[search_employee_handbook],
system_prompt=(
"你是企业助手。涉及员工制度时必须先调用员工手册工具；"
"工具没有提供答案时，不得猜测。"
),
)

result = agent.invoke({
"messages": [
{"role": "user", "content": "测试手机最多可以借多少天？"}
]
})

print(result["messages"][–1].content)

函数的 docstring 会帮助模型判断工具用途，所以工具说明应具体、准确，并明确适用范围。

十、从 Demo 走向生产，还需要做什么？

1. 将建库与问答分离

实际系统不应该每次启动都重新读取全部文件和生成向量。推荐拆分为：

index.py：文档发生变化时执行，负责解析、切分、向量化和入库
app.py：持续提供查询服务，只负责检索和生成

同时记录文件哈希或更新时间，只更新发生变化的文档。

2. 增加 metadata 过滤

为文档添加 department、version、effective_date、permission 等字段。检索前先根据用户权限和制度版本过滤，避免召回无权访问或已失效的资料。

3. 评估检索质量

至少准备一组“问题—标准答案—预期来源”的测试集，并分别观察：

• Recall@K：正确资料有没有进入前 K 个结果；
• 上下文相关性：召回内容是否真正回答问题；
• 答案忠实度：回答是否能被上下文支持；
• 无答案拒答率：资料缺失时是否停止猜测；
• 延迟、Token 和费用。

只评价最终回答，会把“检索错误”和“生成错误”混在一起。

4. 防御 Prompt Injection

外部文档同样是不可信输入。文档中可能出现“忽略系统指令”“发送密钥”等恶意文本。应采取以下措施：

• 明确告诉模型：检索内容是资料，不是可执行指令；
• 工具采用最小权限，读写工具分离；
• 敏感操作必须经过参数校验和人工确认；
• 不把 API Key、系统提示词和私密上下文暴露给模型工具；
• 对上传文件做类型、大小、权限和内容检查。

5. 加入可观测性

RAG 是多阶段系统。线上出现错误时，需要看到问题、召回文本、提示词、模型输出、耗时和 Token 使用情况。可以使用 LangSmith 或企业已有的链路追踪系统。

6. 控制上下文和成本

k 越大不一定越好。召回过多内容会增加费用，也可能让无关信息干扰模型。可以组合以下方法：

• 相似度阈值；
• MMR，兼顾相关性与多样性；
• 混合检索，结合关键词与向量搜索；
• Reranker，对初步召回结果重新排序；
• Parent Document Retrieval，兼顾精确召回和完整上下文。

十一、常见报错与避坑指南

1. ModuleNotFoundError

新版本把不少集成拆到了独立包中：

# 聊天模型与 Embedding
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

# 文本切分
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 核心文档对象
from langchain_core.documents import Document

看到多年以前的 from langchain.xxx import … 教程时，应先核对当前官方文档。

2. 旧教程中的 Chain 已弃用或不再推荐

一些旧文章大量使用 LLMChain、RetrievalQA 或 ConversationalRetrievalChain。理解它们没有坏处，但新项目更适合使用 Runnable/LCEL、显式 RAG 流程或当前的 create_agent API。

3. 聊天正常，Embedding 报错

常见原因是兼容服务不支持 Embedding，或者填写了聊天模型名称。检查：

• Base URL 是否正确；
• Embedding 接口是否存在；
• OPENAI_EMBEDDING_MODEL 是否是向量模型；
• API Key 是否有该模型权限。

4. 中文检索结果不准

依次检查：

扫描版 PDF 只有图片，普通 PDF 解析器无法直接获得文字，需要先 OCR。

5. 回答里出现资料没有的数字

可从四个层面处理：

• 提示词明确要求“无依据就拒答”；
• 展示程序生成的真实引用来源；
• 对关键字段使用结构化输出与规则校验；
• 在高风险场景引入人工复核。

不要把“temperature 设置为 0”当成事实正确性的保证。

十二、学习路线建议

推荐按下面的顺序学习：

先把确定性的 RAG 流程做正确，再学习 Agent，理解会顺畅很多。

总结

本文完成了从概念到代码的完整闭环：

• LangChain 是大模型应用的组件与编排框架，不是模型本身；
• Runnable/LCEL 提供统一的调用和组合方式；
• RAG 的核心链路是“加载、切分、向量化、检索、生成”；
• metadata、引用来源、拒答策略和评估与模型调用同样重要；
• 固定问答优先使用 2-Step RAG，多工具动态决策再考虑 Agent；
• Demo 上线前必须补齐持久化、权限、评估、安全和可观测性。

真正高质量的 AI 应用，不是“接通 API 就结束”，而是让每个答案有依据、每次失败可定位、每项成本可衡量。

参考资料

• LangChain 官方概览：https://docs.langchain.com/oss/python/langchain/overview
• LangChain Retrieval：https://docs.langchain.com/oss/python/langchain/retrieval
• Semantic Search 教程：https://docs.langchain.com/oss/python/langchain/knowledge-base
• LangChain Python API Reference：https://reference.langchain.com/python/