上一篇讲了 LCEL 链式表达和对话记忆。这篇进入更实用的领域——怎么让大模型读懂你的私有文档（RAG），怎么让它调用外部工具干活（Function Call），以及怎么在模型调用前后插入自定义逻辑（中间件）。
1. 为什么需要 RAG
大模型的知识来自训练数据，有两个硬伤：知识有截止日期，不知道你的私有数据。
比如你问"公司的晋升流程是什么"，模型只能瞎编。RAG（Retrieval-Augmented Generation，检索增强生成）的思路是：先从你的文档库里找到相关片段，塞进 prompt 里，让模型基于这些内容回答。
整个流程：
用户问题
↓
向量检索（从文档库找相关片段）
↓
相关片段 + 用户问题 → 组成 prompt
↓
大模型生成答案
2. 文档加载与切割
RAG 的第一步是把文档变成模型能处理的格式。LangChain 支持各种文档类型。
加载 Word 文档
from langchain_community.document_loaders import Docx2txtLoader
loader = Docx2txtLoader("人事管理流程.docx")
documents = loader.load()
加载后得到 Document 对象列表，每个对象有 page_content（文本内容）和 metadata（元信息）。
加载网页
import bs4
from langchain_community.document_loaders import WebBaseLoader
loader = WebBaseLoader(
web_path="https://www.gov.cn/yaowen/liebiao/202512/content_7050416.htm",
# 用 BeautifulSoup 的选择器只抓取正文部分
bs_kwargs={"parse_only": bs4.SoupStrainer(id="UCAP-CONTENT")}
)
docs = loader.load()
bs_kwargs 里可以用 CSS 选择器过滤网页内容，避免把导航栏、侧边栏这些噪音也加载进来。
文档切割
大模型有上下文长度限制，整篇文档塞不进去。需要切成小块：
from langchain_text_splitters import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # 每块最大 500 字符
chunk_overlap=50, # 相邻块重叠 50 字符，避免语义断裂
)
split_documents = splitter.split_documents(documents)
chunk_overlap 是个细节——如果一句话刚好被切断，重叠部分能保证上下文连贯。
3. 向量化与检索
切好的文档要变成向量存进向量数据库，查询时用语义相似度匹配。
嵌入模型 + 向量存储
import os
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.vectorstores import Chroma
# 嵌入模型：把文本变成向量
llm_embeddings = DashScopeEmbeddings(
model="text-embedding-v3",
dashscope_api_key=os.getenv("DASHSCOPE_API_KEY")
)
# 向量化 + 存入 Chroma 数据库
vector_store = Chroma.from_documents(
documents=split_documents,
embedding=llm_embeddings
)
相似度检索
# 直接查
results = vector_store.similarity_search("晋升")
# 带分数查（L2 距离，越小越相似）
results_with_score = vector_store.similarity_search_with_score("晋升")
检索器
实际用的时候不直接调 similarity_search，而是包装成检索器对象，方便和链配合：
retriever = vector_store.as_retriever()
# 也可以指定检索方式
retriever = vector_store.as_retriever(
search_type="similarity_score_threshold",
search_kwargs={"score_threshold": 0.3}
)
检索方式有三种：
similarity：默认，纯相似度similarity_score_threshold：设阈值过滤低相关度结果mmr（最大边际相关性）：平衡相关性和多样性，避免返回内容重复的结果
4. RAG 链：把检索和生成串起来
方式一：手动组装
用 RunnablePassthrough 传递用户问题，retriever 自动检索相关文档：
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from models import get_ali_model_client
client = get_ali_model_client()
message = """
仅使用提供的上下文回答下面的问题：
{question}
上下文：
{context}
"""
prompt_template = ChatPromptTemplate.from_messages([('human', message)])
chain = {"question": RunnablePassthrough(), "context": retriever} | prompt_template | client
resp = chain.invoke("晋升")
print(resp.content)
这里 {"question": RunnablePassthrough(), "context": retriever} 是个并行结构——question 原样透传用户输入，context 用同样的输入去检索文档。两个结果合并后喂给 prompt。
方式二：用预定义链
LangChain 封装了 create_stuff_documents_chain 和 create_retrieval_chain，省去手动组装：
from langchain_classic.chains.combine_documents import create_stuff_documents_chain
from langchain_classic.chains.retrieval import create_retrieval_chain
system_prompt = """
您是问答任务的助理。使用以下的上下文来回答问题，
上下文：<{context}>
如果你不知道答案，就说你不知道。
"""
prompt_template = ChatPromptTemplate.from_messages([
("system", system_prompt),
("human", "{input}")
])
# 文档链：把检索到的文档塞进 prompt
doc_chain = create_stuff_documents_chain(client, prompt_template)
# 检索链：检索 + 文档链
rag_chain = create_retrieval_chain(retriever, doc_chain)
resp = rag_chain.invoke({"input": "会议说了什么?"})
print(resp["answer"])
注意 prompt 里必须用 {context} 和 {input} 这两个变量名，这是预定义链的约定。
返回的 resp 是个字典，resp["answer"] 是最终答案，resp["context"] 是检索到的文档列表。
5. 工具调用：让模型动手干活
RAG 解决的是"模型不知道的信息"。工具调用解决的是"模型做不了的事"——查数据库、打开浏览器、获取实时时间。
定义工具
用 @tool 装饰器，函数的 docstring 就是工具描述（模型靠这个判断什么时候用什么工具）：
import datetime
from langchain.tools import tool
@tool
def get_date():
"""获取今天的具体日期"""
return datetime.date.today().strftime("%Y-%m-%d")
import webbrowser
@tool
def open_browser(url, browser_name=None):
"""获取浏览器，打开网站"""
if browser_name:
browser = webbrowser.get(browser_name)
else:
browser = webbrowser
browser.open(url)
docstring 很重要。写"获取今天的具体日期"，模型在用户问"今天几号"的时候就知道该调这个函数。写得模糊，模型选错工具。
bind_tools：模型绑定工具（手动调用）
from langchain_openai import ChatOpenAI
model = ChatOpenAI(
api_key=os.getenv("DASHSCOPE_API_KEY"),
model="qwen-max",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 绑定工具
tool_llm = model.bind_tools([get_date, open_browser])
resp = tool_llm.invoke("今天是几月几号？")
print(resp.tool_calls)
# [{'name': 'get_date', 'args': {}, 'id': '...'}]
bind_tools 之后，模型不会直接回答，而是返回一个"我要调用 get_date 这个工具"的指令。你得自己执行：
all_tools = {"get_date": get_date, "open_browser": open_browser}
if resp.tool_calls:
for tool_call in resp.tool_calls:
selected_tool = all_tools[tool_call["name"]]
result = selected_tool.invoke(tool_call["args"])
print(result) # 2025-05-19
这种方式控制力强，但写起来繁琐。
create_agent：自动工具调用
用 Agent 就不用手动执行了。模型自己决定用什么工具、执行、拿到结果、生成回答：
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
memory = InMemorySaver()
agent = create_agent(
model=model,
tools=[get_date, open_browser],
system_prompt="你是人工智能助手，需要帮助用户解决各种问题。",
checkpointer=memory,
)
result = agent.invoke(
{"messages": [{"role": "user", "content": "今天是几月几号？"}]},
config={"configurable": {"thread_id": "user_1"}}
)
print(result["messages"][-1].content)
Agent 的执行流程：接收问题 → 判断是否需要工具 → 调用工具 → 拿到结果 → 生成自然语言回答。thread_id 配合 InMemorySaver 实现单会话记忆。
使用社区预置工具
不是所有工具都要自己写。LangChain 社区有现成的：
from langchain_community.agent_toolkits.load_tools import load_tools
# 加载 arxiv 论文查询工具
tools = load_tools(["arxiv"])
agent = create_agent(
model=llm,
tools=tools,
system_prompt="你是专业的论文查询助手",
checkpointer=memory,
)
result = agent.invoke(
{"messages": [{"role": "user", "content": "请查询arxiv论文编号1605.08386的信息"}]},
config={"configurable": {"thread_id": "user_1"}}
)
print(result["messages"][-1].content)
6. 工具实战：自然语言查数据库
一个比较完整的例子——用户用自然语言提问，模型生成 SQL，执行查询，再用自然语言总结结果：
from langchain_classic.chains.sql_database.query import create_sql_query_chain
from langchain_community.tools import QuerySQLDatabaseTool
from langchain_community.utilities import SQLDatabase
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough
from operator import itemgetter
import re
# 连接数据库
MYSQL_URI = 'mysql+mysqldb://root:1234@127.0.0.1:3306/world?charset=utf8mb4'
db = SQLDatabase.from_uri(MYSQL_URI)
# 自定义 SQL 清理器（模型生成的 SQL 经常带 ```sql...``` 包裹）
class SQLCleaner(StrOutputParser):
def parse(self, text: str) -> str:
pattern = r'```sql(.*?)```'
match = re.search(pattern, text, re.DOTALL)
if match:
sql = match.group(1).strip()
sql = re.sub(r'^SQLQuery:', '', sql).strip()
return sql
text = re.sub(r'^SQLQuery:', '', text).strip()
return text
# 生成 SQL 的链
sql_make_chain = create_sql_query_chain(client, db) | SQLCleaner()
# 执行 SQL 的工具
execute_sql_tool = QuerySQLDatabaseTool(db=db)
# 最终回答的 prompt
answer_prompt = PromptTemplate.from_template("""
给定以下用户问题、SQL语句和执行结果，回答用户问题
Question: {question}
SQL Query: {query}
SQL Result: {result}
回答:""")
# 完整链
chain = (
RunnablePassthrough
.assign(query=sql_make_chain)
.assign(result=itemgetter('query') | execute_sql_tool)
| answer_prompt | client | StrOutputParser()
)
result = chain.invoke({"question": "请从国家表中查询出China的相关数据"})
print(result)
这条链的数据流：
{"question": "..."}
→ .assign(query=...) 生成 SQL，追加 query 字段
→ .assign(result=...) 从 query 取 SQL，执行，追加 result 字段
→ answer_prompt 组装最终提示词
→ client 生成自然语言回答
.assign() 的妙处在于：每次追加新字段，但保留之前所有字段。最后 prompt 能同时拿到 question、query、result。
7. 中间件：在模型调用前后插入逻辑
中间件让你在 Agent 调用模型的前后做点事情——打日志、数据脱敏、审计。
装饰器方式
from langchain.agents.middleware import before_model, wrap_model_call
from langchain.agents.middleware import AgentState, ModelRequest, ModelResponse
from langchain.agents import create_agent
from langgraph.runtime import Runtime
from typing import Any, Callable
# 前置中间件：模型调用前执行
@before_model
def log_before_model(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
print(f"即将调用模型，当前有 {len(state['messages'])} 条消息")
return None
# 环绕中间件：模型调用前后都执行
@wrap_model_call
def round_model(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
print(f"调用前: {request}")
result = handler(request) # 实际调用模型
print(f"调用后: {result}")
return result
agent = create_agent(
model=llm,
middleware=[log_before_model, round_model],
)
类方式
逻辑复杂的时候用类更清晰：
from langchain.agents.middleware import AgentMiddleware
class LoggingMiddleware(AgentMiddleware):
def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
print(f"即将调用模型，{len(state['messages'])} 条消息")
return None
def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
print(f"模型返回: {state['messages'][-1].content}")
return None
agent = create_agent(
model=llm,
middleware=[LoggingMiddleware()],
)
实际场景：数据脱敏
在模型收到用户输入之前，把手机号、邮箱替换掉：
import re
class DesensitizeDataMiddleware(AgentMiddleware):
def __init__(self, patterns=None):
super().__init__()
self.patterns = patterns or [
(r'[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+', '[EMAIL]'),
(r'(\+86)?1[3-9]\d{9}', '[PHONE]')
]
def _desensitize_text(self, text: str) -> str:
for pattern, replacement in self.patterns:
text = re.sub(pattern, replacement, text)
return text
def before_model(self, state, **kwargs):
if 'messages' in state:
for message in state['messages']:
if hasattr(message, 'content') and isinstance(message.content, str):
message.content = self._desensitize_text(message.content)
return state
agent = create_agent(
model=llm,
tools=tools,
middleware=[DesensitizeDataMiddleware()],
)
# 用户输入包含邮箱，模型实际收到的是 [EMAIL]
result = agent.invoke(
{"messages": [{"role": "user", "content": "我的邮箱是test@example.com，帮我查论文"}]},
config={"configurable": {"thread_id": "user_1"}}
)
内置中间件：对话摘要
长对话会超出上下文窗口。SummarizationMiddleware 在 token 数接近上限时自动把历史消息压缩成摘要：
from langchain.agents.middleware import SummarizationMiddleware
agent = create_agent(
model=llm,
tools=[],
checkpointer=memory,
middleware=[
SummarizationMiddleware(
model=llm,
max_tokens_before_summary=80, # token 超过 80 触发摘要
messages_to_keep=1, # 摘要后保留最后 1 条消息
),
],
)
适合多轮对话场景。不用自己写截断逻辑，中间件自动处理。
调试
跟上一篇一样，调试开关一开就能看到链里每一步在干什么：
from langchain_core.globals import set_debug
set_debug(True)
Agent 也可以在 create_agent 里传 debug=True。
环境配置备忘
pip install langchain langchain-openai langchain-community
pip install langchain-classic # create_stuff_documents_chain 等预定义链
pip install chromadb # 向量数据库
pip install docx2txt # Word 文档加载
pip install beautifulsoup4 # 网页解析
pip install mysqlclient # MySQL 连接（SQL 工具用）
pip install langgraph # Agent 运行时
pip install gradio # 前端界面（可选）
这篇内容覆盖了 RAG 的完整流程、两种工具调用方式（手动 vs Agent 自动）、以及中间件机制。三个方向解决不同的问题：RAG 让模型读懂私有数据，工具调用让模型执行外部操作，中间件让你在流程中插入控制逻辑