前言：为什么你的AI应用需要RAG
2026年，企业级AI应用的核心矛盾已经不是「模型强不强」，而是「模型能不能回答我的问题」。
通用大模型很厉害，但它的知识有截止日期，而且对企业的私有数据一无所知。你问它「我们公司的年假制度」，它只能瞎编。这不行。
RAG（Retrieval-Augmented Generation，检索增强生成）就是为了解决这个问题而生的。它的思路很直接：先从你的知识库里找到相关内容，再让AI基于这些内容回答问题。
LangChain是构建RAG系统最流行的开发框架。它把大模型应用开发的各个模块抽象出来，让开发者能快速搭建从简单到复杂的AI应用。
这篇文章，我会带你从零构建一个企业知识库问答系统。涉及的知识包括：
- LangChain核心概念和开发模式
- 主流向量数据库的选型与使用
- Embedding模型的选择和优化
- 对话链的完整设计
- 生产环境的部署考量
一、RAG技术原理：从原理到实践
1.1 什么是RAG
RAG的中文叫「检索增强生成」，拆开来看：
- 检索（Retrieval）：从大量文档中找到和问题相关的片段
- 增强（Augmented）：把这些片段作为上下文喂给大模型
- 生成（Generation）：大模型基于上下文生成回答
这比直接让AI「凭记忆回答」靠谱多了。知识库里的内容是准确的，AI只需要理解和整合这些内容。
1.2 RAG的工作流程
一个完整的RAG系统包含以下步骤：
第一步：文档处理
把PDF、Word、网页等格式的文档转成纯文本，然后切分成小块（Chunk）。切分策略很重要——太长了引入太多无关信息，太短了丢失上下文。
第二步：向量化
用Embedding模型把每个文本块转成向量。这些向量是「语义」的表示——意思相近的文本，向量也相近。
第三步：存储
把文本块和对应的向量存到向量数据库里。常用的有Chroma、Pinecone、Milvus、Weaviate等。
第四步：检索
用户提问时，把问题也转成向量，然后在向量数据库里找到最相似的K个文本块。
第五步：生成
把用户的问题和检索到的相关片段组装成Prompt，喂给大模型生成回答。
1.3 为什么选LangChain
LangChain是目前最成熟的LLM应用开发框架，它的优势在于：
- 模块化设计：数据处理、Embedding、向量库、Prompt模板、大模型调用都被抽象成独立模块
- 丰富的集成：支持OpenAI、Claude、国产大模型（通义、文心等）、各种向量数据库
- 活跃的生态：大量的第三方组件和学习资源
二、开发环境准备
2.1 依赖安装
bash
# 创建虚拟环境
python -m venv rag_env
source rag_env/bin/activate # Windows下用 rag_env\Scripts\activate
# 核心依赖
pip install langchain langchain-community
pip install langchain-openai # 或者 langchain-zhipuai
pip install chromadb # 向量数据库
pip install python-dotenv # 环境变量管理
pip install unstructured # 文档解析
pip install tiktoken # Token计数
2.2 环境变量配置
创建一个.env文件：
bash
# OpenAI配置（如果你用OpenAI的模型）
OPENAI_API_KEY=sk-your-api-key
# 国内可以用智谱AI
ZHIPUAI_API_KEY=your-zhipu-api-key
# 向量数据库配置
CHROMA_PERSIST_DIR=./chroma_db
2.3 项目结构
plaintext
rag_project/
├── .env
├── requirements.txt
├── data/ # 存放知识库文档
│ ├── company_policy.pdf
│ └── product_manual.md
├── storage/ # 向量数据库存储
│ └── chroma_db
├── src/
│ ├── __init__.py
│ ├── document_loader.py # 文档加载和切分
│ ├── embeddings.py # Embedding配置
│ ├── vectorstore.py # 向量数据库
│ ├── chain.py # 对话链
│ └── app.py # 应用入口
└── main.py
三、文档处理：从原始文档到可检索数据
3.1 文档加载
LangChain提供了丰富的文档加载器，支持PDF、Word、网页、Notion等各种格式：
python
from langchain_community.document_loaders import (
PyPDFLoader,
UnstructuredMarkdownLoader,
WebBaseLoader
)
def load_documents():
"""加载多个来源的文档"""
docs = []
# PDF文档
pdf_loader = PyPDFLoader("data/company_policy.pdf")
docs.extend(pdf_loader.load())
# Markdown文档
md_loader = UnstructuredMarkdownLoader("data/product_manual.md")
docs.extend(md_loader.load())
# 网页内容
web_loader = WebBaseLoader([
"https://docs.example.com/api-guide",
"https://help.example.com/faq"
])
docs.extend(web_loader.load())
return docs
3.2 文本切分策略
切分是RAG系统中很容易被忽视但很重要的环节。切分策略直接影响检索效果：
python
from langchain.text_splitter import RecursiveCharacterTextSplitter
def split_documents(documents, chunk_size=500, chunk_overlap=50):
"""
递归字符切分
参数说明：
- chunk_size: 每个文本块的目标大小（按字符数）
- chunk_overlap: 块之间的重叠大小，保留上下文连贯性
"""
text_splitter = RecursiveCharacterTextSplitter(
separators=["\n\n", "\n", "。", "，", " ", ""],
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
length_function=len,
)
return text_splitter.split_documents(documents)
切分策略建议：
- 通用场景：500-1000字符，50-100重叠
- 代码检索：按函数/类切分，保留完整代码单元
- 长文档问答：可以尝试更大的块（1500字符），让上下文更完整
3.3 文档元数据
给文档添加元数据，有助于后续的精准检索：
python
def add_metadata(docs, source_type="manual"):
"""给文档添加元数据"""
for doc in docs:
doc.metadata["source_type"] = source_type
doc.metadata["indexed_at"] = datetime.now().isoformat()
return docs
四、向量化与向量数据库
4.1 Embedding模型选择
Embedding模型负责把文本转成向量。选对模型很重要：
海外主流选择：
- OpenAI text-embedding-3-small：效果好，API调用方便，成本适中
- Cohere Embed：多语言支持好，适合中英文混合场景
国产选择：
- 智谱Embedding：支持中文，效果不错
- 通义千问Embedding：阿里出品，和阿里云生态集成好
python
from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import ZhipuAIEmbeddings
# OpenAI方案
def get_openai_embeddings():
return OpenAIEmbeddings(
model="text-embedding-3-small",
openai_api_key=os.getenv("OPENAI_API_KEY")
)
# 智谱方案（国内可用）
def get_zhipu_embeddings():
return ZhipuAIEmbeddings(
model="embedding-2",
zhipuai_api_key=os.getenv("ZHIPUAI_API_KEY")
)
4.2 Chroma向量数据库
Chroma是轻量级、易用的向量数据库，适合本地开发和中小规模应用：
python
import chromadb
from langchain_community.vectorstores import Chroma
def create_vectorstore(documents, embeddings, persist_directory="./storage/chroma_db"):
"""创建并持久化向量数据库"""
vectorstore = Chroma.from_documents(
documents=documents,
embedding=embeddings,
persist_directory=persist_directory
)
return vectorstore
def load_vectorstore(embeddings, persist_directory="./storage/chroma_db"):
"""加载已有的向量数据库"""
return Chroma(
embedding_function=embeddings,
persist_directory=persist_directory
)
4.3 检索测试
创建完向量库后，先测试一下检索效果：
python
def test_retrieval(vectorstore, query, k=4):
"""测试检索效果"""
results = vectorstore.similarity_search(query, k=k)
print(f"查询: {query}\n")
print("检索到的文档片段:")
for i, doc in enumerate(results, 1):
print(f"\n--- 结果 {i} ---")
print(f"来源: {doc.metadata.get('source', '未知')}")
print(f"内容: {doc.page_content[:200]}...")
五、对话链设计
5.1 基础RAG链
用LangChain的LCEL（LangChain Expression Language）构建RAG链：
python
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnablePassthrough
def create_basic_rag_chain(vectorstore, model_name="gpt-4o-mini"):
"""创建基础RAG对话链"""
# 1. 定义Prompt模板
prompt = ChatPromptTemplate.from_messages([
("system", """你是一个专业的企业知识库问答助手。
请基于以下参考信息回答用户的问题。
如果参考信息中没有相关内容，请如实说明，不要编造答案。
参考信息：
{context}"""),
("human", "{question}")
])
# 2. 初始化大模型
llm = ChatOpenAI(model=model_name, temperature=0)
# 3. 定义检索函数
def get_context(question):
docs = vectorstore.similarity_search(question, k=4)
return "\n\n".join([doc.page_content for doc in docs])
# 4. 组装链
chain = (
{"context": get_context, "question": RunnablePassthrough()}
| prompt
| llm
)
return chain
5.2 带历史记录的对话链
单轮问答不够用，需要支持多轮对话：
python
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.runnables import RunnableLambda
def create_conversational_chain(vectorstore, model_name="gpt-4o-mini"):
"""创建支持多轮对话的RAG链"""
# 历史消息存储（实际项目用数据库）
chat_history = []
prompt = ChatPromptTemplate.from_messages([
("system", """你是一个专业的企业知识库问答助手。
请基于以下参考信息和对话历史回答用户的问题。
如果参考信息中没有相关内容，请如实说明，不要编造答案。
参考信息：
{context}"""),
("placeholder", "{chat_history}"),
("human", "{question}")
])
llm = ChatOpenAI(model=model_name, temperature=0)
def get_context_with_history(question, history):
# 把历史对话也加入检索上下文
context_query = question
if history:
context_query = f"{question}\n\n对话历史: {history}"
docs = vectorstore.similarity_search(context_query, k=4)
return "\n\n".join([doc.page_content for doc in docs])
def format_history(history):
return "\n".join([
f"用户: {h.user}\n助手: {h.ai}"
for h in history
])
chain = (
{
"context": lambda x: get_context_with_history(
x["question"],
x.get("chat_history", [])
),
"chat_history": lambda x: format_history(x.get("chat_history", [])),
"question": lambda x: x["question"]
}
| prompt
| llm
)
return chain
5.3 输出优化：Query改写和答案压缩
提升RAG效果的高级技巧：
python
# 查询改写：让检索更精准
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
rewrite_prompt = PromptTemplate.from_template(
"""请将以下用户问题改写得更清晰、更适合检索。
保持原意，但使用更精确的表述。
原问题: {question}
改写后: """
)
def create_query_rewriter():
"""创建查询改写链"""
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
return LLMChain(llm=llm, prompt=rewrite_prompt)
六、生产环境部署
6.1 性能优化
批量索引：文档量大时用批处理
python
from langchain.embeddings import CacheBackedEmbeddings
from langchain.storage import LocalFileStore
# 添加缓存避免重复计算Embedding
store = LocalFileStore("./cache")
cached_embedder = CacheBackedEmbeddings.from_bytes_store(
underlying_embeddings=embeddings,
document_embedding_cache=store,
namespace=embeddings.model
)
异步处理：用asyncio提升吞吐量
python
import asyncio
from langchain.callbacks import AsyncIteratorCallbackHandler
async def astream_rag_response(chain, question):
"""异步流式输出"""
async_handler = AsyncIteratorCallbackHandler()
# 触发异步调用
task = asyncio.create_task(
chain.ainvoke(question, config={"callbacks": [async_handler]})
)
# 流式返回结果
async for token in async_handler.aiter():
yield token
await task
6.2 监控和评估
python
from langchain.evaluation import load_evaluator
def evaluate_rag_system(chain, test_questions):
"""评估RAG系统效果"""
evaluators = [
load_evaluator("qa", llm=llm),
load_evaluator("context_quality", llm=llm),
]
results = []
for q in test_questions:
result = {"question": q}
for evaluator in evaluators:
eval_result = evaluator.evaluate_strings(
prediction=chain.invoke({"question": q}),
reference="", # 参考答案（如果有）
input=q
)
result[evaluator.evaluation_name] = eval_result
results.append(result)
return results
结语
LangChain和RAG是企业级大模型应用开发的基础设施。掌握这套技术栈，你就能构建出真正能在业务场景落地的AI应用。
关键点回顾：
- 文档处理是基础，切分策略直接影响效果
- Embedding模型要选对，中英文场景有差异
- 对话链设计决定用户体验，历史记录和多轮对话很重要
- 生产部署要考虑性能、监控和持续优化
RAG只是起点。当你熟悉了这套架构，可以进一步探索：混合检索、重新排序、多模态RAG、Agent化RAG……大模型应用的世界很大，慢慢探索。
动手试试吧，从一个PDF开始，构建你自己的知识库问答系统。
发表回复