从文档切分到Agent穿甲——LangChain 1.0五步造一个会干活的生产级AI
·
从文档切分到 Agent 穿甲——LangChain 1.0 五步造一个会干活的生产级 AI
你是不是也卡在这五步上？
想做一个企业级 AI Agent，搜了一堆教程，学了文档切分、Embedding、RAG、Tool、中间件——但每个都是孤立的知识点，串不起来？
- 知道要切分文档，但不知道切完之后怎么跟向量库对接？
- 知道 Embedding 是把文字变成数字，但不知道它在整条链路里到底扮演什么角色？
- 知道 RAG 是检索增强生成，但不知道它跟 Tool 有什么区别？
- 知道中间件可以做日志/重试/人工审批，但不知道它卡在 Agent 生命周期的哪个环节？
今天这篇，就用一条完整的链路把五步全部串起来。
┌─────────────────────────────────────────────────────────────────────┐
│ 生产级 AI Agent 开发全链路 │
│ │
│ 第1步 第2步 第3步 第4步 第5步 │
│ 文档切分 ──→ Embedding ──→ RAG检索 ──→ Tool调用 ──→ 中间件 │
│ │
│ 把200页 把文字变 用问题找 让AI能 给Agent穿 │
│ 手册切成 成1024维 相关文档， 调外部工具 上日志/重试│
│ 小文档块 向量 喂给大模型 干活 /审批铁甲 │
│ │
│ Recursive DashScope Retriever @tool Middleware │
│ TextSplitter Embeddings + LCEL Chain + create_agent │
│ + PyPDFLoader + Chroma + Prompt + Agent循环 │
│ + Milvus │
└─────────────────────────────────────────────────────────────────────┘
下面逐步拆解。
第1步：文档加载与切分——把 200 页手册变成一口能吃下的小块
这一步在链路中的角色
原料预处理。你有一堆 PDF、Markdown、TXT——模型一次吃不下整份文档，必须先切块。
200页员工手册.pdf
│
▼
PyPDFLoader（每页→1个Document）
│
▼
RecursiveCharacterTextSplitter（按段落递归切分）
│
▼
16个小文档块（每块约300字，带元数据）
核心代码
from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
# 1. 加载 PDF（每页变成一个 Document）
loader = PyPDFLoader("data/employee_handbook.pdf")
documents = loader.load()
# 2. 递归切分（优先按段落切，段落太长再按句号切）
splitter = RecursiveCharacterTextSplitter(
chunk_size=300, # 每块最大 300 字符
chunk_overlap=50, # 相邻块重叠 50 字符，防止句子被切断
separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
)
chunks = splitter.split_documents(documents)
print(f"切分完成：{len(documents)} 页 → {len(chunks)} 个文档块")
两个关键参数
| 参数 | 作用 | 建议值 |
|---|---|---|
chunk_size |
每个文档块最大长度 | 300-500（普通文本） |
chunk_overlap |
相邻块重叠长度 | chunk_size 的 10%-20% |
比喻：切分就像把一整头牛切成牛排——不能乱剁，要顺着纹理（段落）切，每块大小适中，切面要有重叠（ overlap），不然肉汁（上下文）就流失了。
深入阅读：本文只讲核心概念。完整教程（Document 对象、元数据保留、Path 遍历、企业级预处理脚本）请看系列第 1 篇。
第2步：Embedding 与向量数据库——给文字装上「语义 GPS」
这一步在链路中的角色
把文字变成数字，让计算机能算「语义相似度」。切分后的文档块是文字，但文字没法直接做数学比较——Embedding 把每个文档块变成一个 1024 维的浮点数向量，存进向量数据库。
文档块："未发货可以退款吗？"
│
▼
Embedding 模型（text-embedding-v4）
│
▼
向量：[0.012, -0.035, 0.078, ..., -0.008] ← 1024 个数字
│
▼
存入向量数据库（Milvus / Chroma）
关键点：语义相近的文字，向量距离就近。用户问"没发货能退吗"，虽然跟文档里写的"订单未发货时，用户可以直接申请退款"用词不同，但向量距离很近——这就是语义检索的本质。
核心代码
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.vectorstores import Milvus
import os
# 1. 初始化 Embedding 模型
embeddings = DashScopeEmbeddings(
model="text-embedding-v4",
dashscope_api_key=os.getenv("DASHSCOPE_API_KEY"),
)
# 2. 把文档块写入 Milvus 向量数据库
vector_store = Milvus(
embedding_function=embeddings,
collection_name="employee_handbook",
connection_args={"uri": "http://localhost:19530"},
index_params={
"index_type": "HNSW",
"metric_type": "COSINE",
"params": {"M": 16, "efConstruction": 128}
},
auto_id=True,
)
# 3. 写入数据（第1步切分好的 chunks）
ids = vector_store.add_documents(chunks)
print(f"已写入 {len(ids)} 条向量")
# 4. 语义检索测试
results = vector_store.similarity_search("迟到扣多少钱？", k=3)
for doc in results:
print(doc.page_content[:100])
print(f" 来源：{doc.metadata.get('source', '未知')}")
向量数据库选型速查
| 数据库 | 类型 | 适用场景 | 推荐指数 |
|---|---|---|---|
| Milvus | 开源 | 大规模企业 RAG，百万级以上 | 首选 |
| Chroma | 开源 | 本地开发，中小规模 | 入门首选 |
| Pinecone | 商用 | 无需运维，开箱即用 | 预算充足选 |
| pgvector | 扩展 | 已有 PostgreSQL 的项目 | 复用现有设施 |
比喻：Embedding 就是给每段文字装一个「语义 GPS 坐标」。向量数据库就是一座「坐标仓库」——用户提问时，先把问题也变成坐标，然后在仓库里找最近的几个坐标。
深入阅读：向量维度原理、余弦相似度计算、HNSW 索引参数调优等完整内容，请看系列第 2 篇。
第3步：RAG 检索增强生成——让 AI 不再编答案
这一步在链路中的角色
把检索到的文档喂给大模型。前两步建好了索引，现在用户提问时：先从向量库检索相关文档块 → 把文档块拼进 Prompt → 让模型根据文档回答。
用户："公司迟到扣多少钱？"
│
▼
Retriever 从向量库检索 Top 3 文档块
│
▼
拼接上下文：
"根据以下资料回答问题：
[文档块1] 迟到早退按分钟扣罚...
[文档块2] 超过三次需部门审批...
问题：公司迟到扣多少钱？"
│
▼
DeepSeek 根据上下文生成答案
│
▼
"根据员工手册，迟到按每分钟扣罚..."
核心代码
from langchain.agents import create_agent
from langchain_core.prompts import ChatPromptTemplate
# 1. 从向量库获取 Retriever
retriever = vector_store.as_retriever(search_kwargs={"k": 3})
# 2. 检索相关文档
docs = retriever.invoke("迟到扣多少钱？")
# 3. 整理上下文
context = "\n\n".join(doc.page_content for doc in docs)
# 4. 构造 Prompt
prompt = ChatPromptTemplate.from_messages([
("system", "根据以下资料回答问题。资料不足时说明无法确定。\n\n资料：\n{context}"),
("human", "{question}"),
])
# 5. 调用模型
from utils.model_factory import get_deepseek_model
model = get_deepseek_model()
chain = prompt | model
response = chain.invoke({"context": context, "question": "迟到扣多少钱？"})
print(response.content)
RAG vs 直接问模型
| 对比维度 | 直接问模型 | RAG |
|---|---|---|
| 知识来源 | 模型训练数据 | 企业文档 |
| 时效性 | 训练截止日期 | 实时更新 |
| 准确性 | 可能编造 | 基于文档回答 |
| 来源追溯 | 无法追溯 | 可展示出处 |
| 适用场景 | 通用知识 | 企业内部知识 |
比喻：直接问模型就像闭卷考试——它只能凭记忆答，记不清就编。RAG 就是开卷考试——先翻到相关页码（检索），再照着内容答（生成），答案有据可查。
深入阅读：Retriever 高级配置、LCEL 链式组合、上下文压缩、流式输出等完整内容，请看系列第 3 篇。
第4步：Tool 工具调用——给 AI 装上真干活的机械臂
这一步在链路中的角色
让 Agent 能调外部系统。RAG 解决了「知识」问题，但 Agent 还需要「行动」——查订单、算价格、调 API。Tool 就是把外部能力包装成模型可调用的函数。
用户："帮我查一下订单 A1002 发了没"
│
▼
Agent 推理：需要查订单 → 调用 get_order_status 工具
│
▼
Tool 执行：get_order_status(order_id="A1002")
│
▼
返回结果："已发货，快递单号 SF123456"
│
▼
Agent 组织语言回复用户
核心代码
from langchain_core.tools import tool
from langchain.agents import create_agent
from utils.model_factory import get_deepseek_model
# 1. 定义工具
@tool
def get_order_status(order_id: str) -> str:
"""根据订单号查询订单状态。"""
fake_orders = {
"A1001": "已付款，等待发货",
"A1002": "已发货，快递单号 SF123456",
"A1003": "已签收",
}
return fake_orders.get(order_id, "未找到该订单")
@tool
def calculate_discount_price(price: float, discount: float) -> float:
"""根据原价和折扣比例计算折后价格。"""
return price * discount
# 2. 创建 Agent
model = get_deepseek_model()
agent = create_agent(
model=model,
tools=[get_order_status, calculate_discount_price],
system_prompt="你是一个电商客服助手。根据用户问题选择合适的工具。",
)
# 3. 运行
result = agent.invoke({
"messages": [{"role": "user", "content": "帮我查一下订单 A1002 发了没"}]
})
print(result["messages"][-1].content)
# 输出：订单 A1002 已发货，快递单号 SF123456。
Tool vs RAG：什么时候用哪个？
| 对比维度 | RAG | Tool |
|---|---|---|
| 解决什么 | 知识检索 | 业务执行 |
| 数据来源 | 文档库 | 实时系统/数据库/API |
| 典型场景 | “公司迟到扣多少？” | “查我的订单状态” |
| 返回内容 | 文本片段 | 结构化数据 |
| 触发方式 | 每次问答都检索 | Agent 按需调用 |
比喻：RAG 是给 Agent 配了一本「参考书」——翻到相关页码照着答。Tool 是给 Agent 配了一部「电话」——遇到答不了的问题，打个电话问业务系统。
深入阅读：工具描述六要素、Agent 执行循环原理、多工具电商客服完整案例等，请看系列第 4 篇。
第5步：中间件——给 Agent 穿上生产级铁甲
这一步在链路中的角色
生产级保障。前四步搭好了 Agent 的核心能力，但上生产还缺：日志、重试、限流、对话摘要、人工审批。中间件就是在 Agent 执行流程的特定阶段自动触发的钩子。
Agent 执行流程 + 中间件拦截点：
agent.invoke()
│
▼
┌─ @before_agent ──→ 初始化/权限校验 ──────────────┐
│ │
│ ┌─ @before_model ──→ 输入校验/日志 ──────────┐ │
│ │ │ │
│ │ ┌─ 模型实际调用 ─────────────────────┐ │ │
│ │ │ │ │ │
│ │ └──────────────────────────────────────┘ │ │
│ │ │ │
│ └─ @after_model ──→ 结果分析/风险检测 ────────┘ │
│ │
└─ @after_agent ──→ 收尾/日志上报/持久化 ───────────┘
│
▼
返回结果
核心代码：两个最常用的预置中间件
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware, HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
# 1. 对话摘要中间件（防止 Token 超限）
summarization = SummarizationMiddleware(
model=model,
trigger=('tokens', 1000), # 达到 1000 token 触发摘要
keep=('messages', 2), # 保留最近 2 条原始消息
)
# 2. 人工审批中间件（敏感操作需人工确认）
hitl = HumanInTheLoopMiddleware(
interrupt_on={
"dangerous_write": {
"allowed_decisions": ["approve", "reject", "edit"]
}
}
)
agent = create_agent(
model=model,
tools=[get_order_status, dangerous_write],
middleware=[summarization, hitl],
checkpointer=InMemorySaver(), # HITL 必须配置
system_prompt="你是一个企业运维助手。",
)
中间件分类速查
| 类型 | 触发时机 | 典型场景 |
|---|---|---|
@before_model |
模型调用前 | 输入校验、日志记录 |
@after_model |
模型响应后 | 结果分析、风险检测 |
@wrap_model_call |
包裹模型调用 | 计时、重试、限流 |
@dynamic_prompt |
模型调用前 | 千人千面 Prompt |
SummarizationMiddleware |
Token 超阈值 | 长对话自动压缩 |
HumanInTheLoopMiddleware |
敏感工具调用前 | 人工审批/编辑/拒绝 |
比喻：前四步造了一个能干活的机器人，但它还是裸奔的——没有安全帽（限流）、没有保险绳（重试）、没有审批流程（HITL）。中间件就是给机器人穿上铁甲，让它能安全上岗。
深入阅读：MCP 协议、异步编程、装饰器/类中间件完整实战、洋葱模型等，请看系列第 5 篇。
全链路总结：五步如何串成一条完整流水线
把五步放在一起，就是一个完整的生产级 AI Agent 系统：
┌──────────────────────────────────────────────────────────────────────┐
│ 生产级 AI Agent 全链路 │
│ │
│ ┌─────────┐ ┌──────────┐ ┌─────────┐ ┌────────┐ ┌──────┐ │
│ │ 文档切分 │───→│ Embedding│───→│ RAG检索 │───→│ Tool │───→│中间件│ │
│ │ │ │ + 向量库 │ │ + 上下文 │ │ + Agent│ │ │ │
│ └─────────┘ └──────────┘ └─────────┘ └────────┘ └──────┘ │
│ │ │ │ │ │ │
│ 原料预处理 语义索引 知识检索 行动执行 生产保障 │
│ │ │ │ │ │ │
│ Recursive Milvus Retriever create_agent 中间件 │
│ TextSplitter Chroma + LCEL + @tool 链 │
│ + PyPDFLoader + HNSW + Prompt @before│
│ 索引 @after │
│ @wrap │
└──────────────────────────────────────────────────────────────────────┘
一句话串起五步：
切分文档成小块（第1步）→ 把小块变成向量存入数据库（第2步）→ 用户提问时检索相关文档喂给模型（第3步）→ 模型根据问题选择工具执行业务操作（第4步）→ 全程有中间件做日志、重试、摘要、人工审批保障（第5步）。
各步核心 API 速查表
| 步骤 | 核心 API | 作用 |
|---|---|---|
| 第1步 | TextLoader / PyPDFLoader |
加载 TXT/MD/PDF |
| 第1步 | RecursiveCharacterTextSplitter |
递归切分文档 |
| 第1步 | split_documents() |
切分并保留元数据 |
| 第2步 | DashScopeEmbeddings |
中文 Embedding 模型 |
| 第2步 | embed_documents() / embed_query() |
文档/查询向量化 |
| 第2步 | Milvus / Chroma |
向量数据库 |
| 第2步 | similarity_search() |
语义检索 |
| 第3步 | as_retriever() |
VectorStore → Retriever |
| 第3步 | ChatPromptTemplate |
构造上下文 Prompt |
| 第3步 | prompt | model |
LCEL 链式调用 |
| 第4步 | @tool |
定义工具 |
| 第4步 | create_agent() |
创建 Agent |
| 第4步 | agent.invoke() |
运行 Agent |
| 第5步 | SummarizationMiddleware |
对话摘要 |
| 第5步 | HumanInTheLoopMiddleware |
人工审批 |
| 第5步 | @before_model / @after_model |
装饰器钩子 |
| 第5步 | @wrap_model_call |
包裹式钩子 |
| 第5步 | AgentMiddleware |
类中间件基类 |
| 第5步 | MultiServerMCPClient |
MCP 多服务端连接 |
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
8
0- 0
已为社区贡献2条内容
所有评论(0)