从文档切分到 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字，带元数据）核心代码
fromlangchain_community.document_loadersimportPyPDFLoader,TextLoaderfromlangchain_text_splittersimportRecursiveCharacterTextSplitter# 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）关键点：语义相近的文字，向量距离就近。用户问"没发货能退吗"，虽然跟文档里写的"订单未发货时，用户可以直接申请退款"用词不同，但向量距离很近——这就是语义检索的本质。
核心代码
fromlangchain_community.embeddingsimportDashScopeEmbeddingsfromlangchain_community.vectorstoresimportMilvusimportos# 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)fordocinresults: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 根据上下文生成答案 │ ▼ "根据员工手册，迟到按每分钟扣罚..."核心代码
fromlangchain.agentsimportcreate_agentfromlangchain_core.promptsimportChatPromptTemplate# 1. 从向量库获取 Retrieverretriever=vector_store.as_retriever(search_kwargs={"k":3})# 2. 检索相关文档docs=retriever.invoke("迟到扣多少钱？")# 3. 整理上下文context="\n\n".join(doc.page_contentfordocindocs)# 4. 构造 Promptprompt=ChatPromptTemplate.from_messages([("system","根据以下资料回答问题。资料不足时说明无法确定。\n\n资料：\n{context}"),("human","{question}"),])# 5. 调用模型fromutils.model_factoryimportget_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 组织语言回复用户核心代码
fromlangchain_core.toolsimporttoolfromlangchain.agentsimportcreate_agentfromutils.model_factoryimportget_deepseek_model# 1. 定义工具@tooldefget_order_status(order_id:str)->str:"""根据订单号查询订单状态。"""fake_orders={"A1001":"已付款，等待发货","A1002":"已发货，快递单号 SF123456","A1003":"已签收",}returnfake_orders.get(order_id,"未找到该订单")@tooldefcalculate_discount_price(price:float,discount:float)->float:"""根据原价和折扣比例计算折后价格。"""returnprice*discount# 2. 创建 Agentmodel=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 ──→ 收尾/日志上报/持久化 ───────────┘ │ ▼ 返回结果核心代码：两个最常用的预置中间件
fromlangchain.agentsimportcreate_agentfromlangchain.agents.middlewareimportSummarizationMiddleware,HumanInTheLoopMiddlewarefromlanggraph.checkpoint.memoryimportInMemorySaver# 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 多服务端连接 |
