LangChain：从“链”到“智能体”，一篇带你跑通全套AI应用

🚀 前言

💬 “模型再强，落地才是王道。”

受够了只跑 demo 却落不了地的尴尬？😫

当模型得自己选工具、容错、防泄露、记住偏好时，代码瞬间乱成一锅粥。LangChain 就是来终结这场混乱的！🤖

本文用最轻松的方式，从一行 create_agent 开始，带你搭出能上线、可管控、有记忆的生产级 AI。

新手友好，老手也能抄作业～ 坐稳，出发啦！🚀✨

🌐 第 1 章 LangChain 生态全景与技术定位

1.1 核心价值与四层架构体系

一句话先说清 LangChain 是什么：

🎯 LangChain 是一个「把大语言模型接到真实世界」的工程框架——把模型、提示、工具、数据源、记忆、编排逻辑，拼装成一个可上线的应用。

用「四层」的视角看它的架构：

逐层拆解：

这套设计的精髓是 「低门槛入口 + 高上限定制」：

• 🚀 新手：用 create_agent 三个参数就能跑起来。
• 🔧 老手：用中间件钩子把脱敏、审计、限流、动态换模型一层层叠上去。

💡 关键在于：叠加这些能力，完全不需要重写主流程。

1.2 2026 年重大更新概览

1.0 这一代（截至 2026 年）最值得你关注的变化：

• ✨ create_agent 成为「唯一推荐」的建法：取代了 AgentExecutor、initialize_agent 等碎片化写法，底层统一跑在 LangGraph 上，自带 ReAct（推理 + 行动）循环。
• 🔗 中间件成为一等公民：通过 before_model / after_model / wrap_model_call / wrap_tool_call / before_agent / after_agent 钩子，在任意环节插入逻辑；还内置了脱敏、摘要、人工审批三款中间件。
• 📦 标准内容块（Content Blocks）：新增 content_blocks 属性，跨厂商一致地读文本、推理过程、工具调用、引用来源。
• 🎯 结构化输出进主循环：ToolStrategy 让结构化输出在主循环内完成，省掉「额外再调一次 LLM 做格式化」的开销。
• 📦 包结构瘦身：主 langchain 只留核心命名空间（agents、chat_models、tools、messages、embeddings）；旧链、旧检索器、hub、索引 API 迁到 langchain-classic。
• 🤖 Deep Agents 成为独立范式：deepagents 包提供 create_deep_agent，内建「规划 + 虚拟文件系统 + 子智能体」。

📌 升级一句话：pip install -U langchain。老代码若用到旧链或 langchain.retrievers，记得 pip install langchain-classic，并把 from langchain… 改成 from langchain_classic…。

1.3 与主流 AI 框架对比

从「场景适配」角度横向对比：

选型口诀 🎵：

🔹 只想调一次模型做翻译/问答 → 直接用模型 SDK，别上框架。 🔹 要让模型自己用工具、带记忆、能上线 → LangChain（create_agent）。 🔹 流程极复杂（多分支、循环、人工审批、定时） → 下探 LangGraph 手写图。 🔹 想快速验证想法 → 先用 LangChain，后期可平滑迁移到 LangGraph（前者本就跑在后者之上）。

⚡ 第 2 章 快速上手：5 分钟构建第一个 AI 应用

2.1 环境搭建与依赖安装

推荐用 uv（更快的包管理器），当然 pip 也完全 OK。

# 方式一：pip
pip install -U langchain

# 方式二：uv（推荐，速度更快）
uv add langchain

# 按需安装你要用的模型供应商集成包（任选其一或多个）
pip install langchain-openai # OpenAI
pip install langchain-anthropic # Anthropic Claude
pip install langchain-deepseek # DeepSeek
pip install langchain-google-genai # Google Gemini

# 本文后续章节会用到的额外依赖（用到时再装即可）
pip install langchain-mcp-adapters # 第 4 章 连接 MCP 服务器
pip install fastmcp # 第 4 章 编写本地 MCP 服务器
pip install langchain-chroma chromadb # 第 5/6 章 向量库
pip install langchain-community rank_bm25 # 第 6 章 BM25 混合检索
pip install langchain-classic # 第 6 章 EnsembleRetriever
pip install deepagents mypy # 第 7 章 Deep Agents（mypy 供类型检查示例）

设置 API Key（以环境变量方式，避免把密钥写进代码）：

# Windows PowerShell 示例
$env:DEEPSEEK_API_KEY = "你的key"
# 或 OpenAI
$env:OPENAI_API_KEY = "你的key"

💡 提示：本文统一用 init_chat_model 初始化模型，切换厂商只改一个字符串，业务代码一行都不用动。

2.2 核心概念速通

写第一个应用前，先把四个最基础的概念讲明白：

• 🧠 模型（Model）：真正「动脑子」的部分。你给它消息，它返回消息。用 init_chat_model("厂商:模型名") 统一拿到聊天模型对象。
• 💬 提示（Prompt）：你怎么「跟模型说话」。含系统提示（设角色）和用户输入，用 ChatPromptTemplate 把变量插槽化、复用化。
• ⛓️ 链（Chain）：把「提示 → 模型 → 后处理」用管道符 | 串成一条流水线。这就是 LCEL（LangChain 表达式语言），第 3 章深挖。
• 📤 解析器（Output Parser）：把模型返回的原始消息「翻译」成程序好用的格式，比如纯字符串、JSON、Pydantic 对象。

💡 一句话连起来：提示塑形输入 → 模型生成结果 → 解析器落地 —— prompt | model | parser。

2.3 第一个实用应用：商品评论情感分析器

🎯 场景：电商运营每天要看成千上万条评论。我们做个小工具：输入一条评论，自动判断情感倾向、星级预测、关键吐槽点，并输出结构化结果，方便写入数据库。

就用「提示模板 + 模型 + 结构化解析器」这条最经典的链来实现。

"""商品评论情感分析器 —— LangChain LCEL 入门示例。"""
from typing import Literal

from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate

# 1) 定义我们想要的结构化输出（Pydantic 模型即「契约」）
class ReviewInsight(BaseModel):
sentiment: Literal["正面", "负面", "中性"] = Field(description="整体情感倾向")
predicted_star: int = Field(description="预测星级，1~5 的整数", ge=1, le=5)
pain_points: list[str] = Field(description="用户主要的吐槽点，没有则为空列表")
summary: str = Field(description="一句话总结这条评论")

# 2) 初始化模型（换厂商只改这一行字符串）
model = init_chat_model("deepseek:deepseek-chat", temperature=0)

# 3) 构造提示模板
prompt = ChatPromptTemplate.from_messages([
("system", "你是资深电商运营分析师，擅长从评论中提炼客观结论。只依据评论内容判断，不要臆测。"),
("human", "请分析这条商品评论：\\n\\n{review}"),
])

# 4) 让模型直接产出结构化对象（with_structured_output 是 1.0 统一能力）
structured_model = model.with_structured_output(ReviewInsight)

# 5) 用管道符把「提示 → 结构化模型」串成一条链
analyzer = prompt | structured_model

# 6) 运行
review_text = "包装很精致，物流也快，但是用了三天充电口就接触不良了，有点失望，客服倒是回复挺及时的。"
result: ReviewInsight = analyzer.invoke({"review": review_text})

print("情感倾向：", result.sentiment)
print("预测星级：", result.predicted_star)
print("吐槽点：", result.pain_points)
print("一句话总结：", result.summary)

运行后你会得到类似：

情感倾向： 负面
预测星级： 2
吐槽点： ['充电口接触不良', '产品耐用性差']
一句话总结： 包装与物流体验不错，但产品质量问题导致整体满意度低。

✅ 你已经用到了哪些 1.0 特性？

• init_chat_model 统一模型层；
• ChatPromptTemplate 提示模板；
• with_structured_output 直接拿到 Pydantic 对象（告别手写正则解析 JSON）；
• | 管道组合成 LCEL 链。

🔥 短短 30 行，你就有了一个能批量处理的情感分析器。把 analyzer.invoke 换成 analyzer.batch([…]) 即可批量跑。

🔗 第 3 章 LCEL：LangChain 表达式语言深度解析

3.1 LCEL 基础语法与 Runnable 接口

LCEL 的核心思想就一句话：

🧩 万物皆 Runnable，Runnable 之间用 | 拼接。

什么是 Runnable？它是所有可执行组件的统一接口——提示模板、模型、解析器、甚至你自己写的函数，只要实现了 Runnable，就都能用 | 串起来，并自动获得一整套标准方法：

下面用一个「把文案改写成不同语气」的小链演示 Runnable 组合，并引入两个常用工具：RunnableLambda（把普通函数变 Runnable）和 RunnablePassthrough（透传输入）。

"""LCEL 基础：把一段文案改写为「专业版」并附带字数统计。"""
from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda, RunnablePassthrough

model = init_chat_model("deepseek:deepseek-chat", temperature=0.7)

rewrite_prompt = ChatPromptTemplate.from_messages([
("system", "你是品牌文案专家，请把用户的口语化文案改写成专业、克制、可信赖的版本。只输出改写后的文案。"),
("human", "{draft}"),
])

# 一条链：提示 -> 模型 -> 取出纯字符串
rewrite_chain = rewrite_prompt | model | StrOutputParser()

# 用 RunnableLambda 把普通函数接进流水线，做字数统计
def add_stats(text: str) –> dict:
return {"rewritten": text, "char_count": len(text)}

# RunnablePassthrough 让原始草稿也一并保留下来
full_chain = (
{"draft": RunnablePassthrough()} # 输入字符串塞进 draft 键
| rewrite_chain
| RunnableLambda(add_stats)
)

output = full_chain.invoke("这个吸尘器超好用！吸力贼大，我家猫毛全没了！")
print(output["rewritten"])
print("字数：", output["char_count"])

要点 ✨：

• prompt | model | StrOutputParser() 是最高频的三段式；
• RunnableLambda 把任意 text -> dict 的普通函数无缝接入；
• RunnablePassthrough / 字典写法用于「组装并行输入」。

3.2 核心高级特性：流式、异步、错误处理、中间件

🌊（1）流式输出 —— 打字机效果

# stream 会逐 token 吐出，前端就能做「正在输入」的体验
for chunk in rewrite_chain.stream("这耳机音质绝了，戴一天耳朵也不疼"):
print(chunk, end="", flush=True)

⏳（2）异步 —— 扛并发

import asyncio

async def main():
drafts = ["手机拍照真清晰", "这书太催眠了", "空调制冷快但有点吵"]
results = await rewrite_chain.abatch(drafts) # 并发跑三条
for r in results:
print("-", r)

asyncio.run(main())

🛡️（3）错误处理 —— 回退与重试

LCEL 提供 with_fallbacks（主模型失败时自动切备用）和 with_retry（自动重试）：

primary = init_chat_model("deepseek:deepseek-chat")
backup = init_chat_model("openai:gpt-4o-mini")

# 主模型重试 2 次仍失败，就自动回退到备用模型
robust_model = primary.with_retry(stop_after_attempt=2).with_fallbacks([backup])
robust_chain = rewrite_prompt | robust_model | StrOutputParser()

💡 主模型超时/报错时，链会自动重试，再不行就回退备用模型，对调用方完全透明。

⚙️（4）中间件 —— 1.0 的灵魂特性

⚠️ 先分清两个概念：

• 回退/重试是 LCEL 链层面的容错；
• 中间件（Middleware）是 create_agent 智能体 层面的治理钩子——它才是 1.0 标志性的设计。

🎯 演示场景：「订单状态查询智能体」，加两个自定义中间件：

• 🔒 输入审核中间件：调模型前，检查输入是否含疑似提示注入，命中直接拦截。
• 📝 输出格式化中间件：模型返回后，统一在末尾追加客服话术，保证风格一致。

中间件通过继承 AgentMiddleware 并实现钩子来编写：before_model（调模型前）、after_model（调模型后）。

"""订单查询智能体 + 自定义输入审核/输出格式化中间件。"""
from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware
from langchain.chat_models import init_chat_model
from langchain.messages import AIMessage, HumanMessage
from langchain.tools import tool

# — 一个假的订单查询工具（实际可换成查数据库）—
FAKE_ORDERS = {
"20260518A": "已发货，预计明天送达，快递单号 SF1234567890",
"20260517B": "仓库正在打包，今晚发出",
}

@tool
def query_order(order_id: str) –> str:
"""根据订单号查询订单的物流状态。"""
return FAKE_ORDERS.get(order_id, "未找到该订单，请核对订单号。")

# — 中间件 1：输入审核（提示注入检测）—
class InputGuardMiddleware(AgentMiddleware):
BANNED = ["忽略以上指令", "ignore previous", "system prompt"]

def before_model(self, state, runtime):
last = state["messages"][–1]
if isinstance(last, HumanMessage):
text = last.text if hasattr(last, "text") else str(last.content)
# 简单的提示注入检测
if any(b in text.lower() for b in self.BANNED):
# 直接短路：注入一条 AI 消息并结束本轮
return {
"messages": [AIMessage(content="检测到异常指令，本次请求已被安全策略拦截。")],
"jump_to": "end",
}
return None # 返回 None 表示放行，不修改状态

# — 中间件 2：输出统一格式化 —
class OutputStyleMiddleware(AgentMiddleware):
SIGNATURE = "\\n\\n—— 智选优品客服中心，竭诚为您服务 🤝"

def after_model(self, state, runtime):
last = state["messages"][–1]
if isinstance(last, AIMessage) and last.content and not last.tool_calls:
new_text = (last.text if hasattr(last, "text") else str(last.content)) + self.SIGNATURE
return {"messages": [AIMessage(content=new_text, id=last.id)]}
return None

model = init_chat_model("deepseek:deepseek-chat", temperature=0)

order_agent = create_agent(
model=model,
tools=[query_order],
system_prompt="你是电商客服助手，用户给出订单号时调用 query_order 查询并友好回复。",
middleware=[InputGuardMiddleware(), OutputStyleMiddleware()],
)

resp = order_agent.invoke({
"messages": [{"role": "user", "content": "帮我查下订单 20260518A 到哪了？"}]
})
print(resp["messages"][–1].content)

预期输出（末尾自动带上统一签名）：

您的订单 20260518A 已发货，预计明天送达，快递单号 SF1234567890。

—— 智选优品客服中心，竭诚为您服务 🤝

中间件钩子速查表 🪝（1.0 官方提供）：

📦 想要现成能力？1.0 内置了几款中间件，直接传进 middleware=[…] 即可：

from langchain.agents.middleware import (
PIIMiddleware, # 敏感信息脱敏（邮箱/手机号等）🔒
SummarizationMiddleware, # 对话太长时自动摘要压缩 📝
HumanInTheLoopMiddleware, # 敏感操作前要求人工审批 👤
)

agent = create_agent(
model="deepseek:deepseek-chat",
tools=[query_order],
middleware=[
PIIMiddleware("phone_number", strategy="redact", apply_to_input=True),
SummarizationMiddleware(model="deepseek:deepseek-chat", trigger={"tokens": 4000}),
],
)

3.3 实战：构建可组合的复杂工作流

🎯 场景：做一个「智能回复路由器」——先识别用户情绪，再按情绪分支走不同回复风格：

• 😡 愤怒 → 安抚致歉风
• 😄 开心 → 热情共鸣风
• 😐 中性 → 简洁专业风

同时并行生成一条「内部工单摘要」。一次演示 LCEL 的分支（RunnableBranch）与并行（RunnableParallel）。

"""情绪路由 + 并行工单摘要 工作流。"""
from typing import Literal

from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableBranch, RunnableParallel, RunnableLambda

model = init_chat_model("deepseek:deepseek-chat", temperature=0.6)
parser = StrOutputParser()

# 1) 情绪分类（结构化输出，保证只返回三类之一）
class Emotion(BaseModel):
label: Literal["愤怒", "开心", "中性"] = Field(description="用户当前情绪")

classify_prompt = ChatPromptTemplate.from_messages([
("system", "判断用户消息的情绪，只能是 愤怒 / 开心 / 中性 之一。"),
("human", "{message}"),
])
classifier = classify_prompt | model.with_structured_output(Emotion)

# 2) 三种回复风格链
def style_chain(system_prompt: str):
return ChatPromptTemplate.from_messages([
("system", system_prompt),
("human", "{message}"),
]) | model | parser

angry_chain = style_chain("用户在生气。请先真诚致歉、表达理解，再给出解决方案，语气温和耐心。")
happy_chain = style_chain("用户很满意。请热情共鸣、表达感谢，并自然地邀请其分享或复购。")
neutral_chain = style_chain("请用简洁、专业、中立的语气直接回应用户。")

# 3) 用 RunnableBranch 做情绪分支路由
reply_router = RunnableBranch(
(lambda x: x["emotion"] == "愤怒", angry_chain),
(lambda x: x["emotion"] == "开心", happy_chain),
neutral_chain, # 默认分支
)

# 4) 一条内部工单摘要链（并行用）
ticket_prompt = ChatPromptTemplate.from_messages([
("system", "用一句话为客服团队总结该用户问题与情绪，便于建工单。"),
("human", "{message}"),
])
ticket_chain = ticket_prompt | model | parser

# 5) 组装：先分类 -> 把情绪和原文一起往下传 -> 并行产出「给用户的回复」+「给客服的工单」
def assemble(inputs: dict) –> dict:
emotion = classifier.invoke({"message": inputs["message"]}).label
return {"message": inputs["message"], "emotion": emotion}

workflow = (
RunnableLambda(assemble)
| RunnableParallel(
reply_to_user=reply_router,
internal_ticket=ticket_chain,
detected_emotion=RunnableLambda(lambda x: x["emotion"]),
)
)

result = workflow.invoke({"message": "你们这破网络机顶盒又卡死了！第三次了！我要退货！"})
print("识别情绪：", result["detected_emotion"])
print("\\n给用户的回复：\\n", result["reply_to_user"])
print("\\n内部工单：\\n", result["internal_ticket"])

这段代码一次展示了 LCEL 的三种组合范式：

• ➡️ 顺序（|）：分类 → 路由；
• 🔀 分支（RunnableBranch）：按情绪选链；
• ⏩ 并行（RunnableParallel）：同时产出用户回复与内部工单。

🌐 关于 LangGraph：什么时候需要「下探」一层？

到这里，你已经掌握了「链」与「智能体 + 中间件」两种构建方式，足以覆盖大多数应用。

那什么时候需要 LangGraph？当编排需求超出「分支 + 并行」时，比如：

• 🔁 显式循环；
• 🤝 多智能体互相移交（handoff）；
• 👤 带状态机的人工审批流；
• ⏸️ 长时间运行、可中断可恢复的任务。

这时单靠 LCEL 链就会很别扭，LangGraph 才是主场。

📌 这部分自成体系，详见专文：LangGraph：构建复杂有状态智能体的核心框架。

下面回到能力层，看看模型与工具如何统一集成。

🧠 第 4 章 模型与工具集成

4.1 主流大模型统一接口

init_chat_model 提供了跨 20+ 厂商的统一初始化。魔力在于：

🎯 模型名格式是 "厂商前缀:模型名"，切换厂商只改字符串，下游代码（链、智能体、工具绑定）完全不动。

截至 2026 年，常见模型标识举例（以各厂商最新发布为准）：

来个「同一段逻辑、跑遍多家模型」的对比小工具：

"""用统一接口，让多家模型回答同一个问题并对比。"""
from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_messages([
("system", "你是旅行规划师，用不超过 50 字给出一句话建议。"),
("human", "{question}"),
])
parser = StrOutputParser()

# 想测哪些模型，就在这里列出标识字符串
candidates = [
"deepseek:deepseek-chat",
# "openai:gpt-4o-mini",
# "anthropic:claude-sonnet-4-6",
]

question = "六月想去一个凉快又不挤的国内小众目的地，推荐哪里？"

for model_id in candidates:
model = init_chat_model(model_id, temperature=0.7)
chain = prompt | model | parser
print(f"\\n【{model_id}】\\n{chain.invoke({'question': question})}")

💡 只要改 candidates 里的模型标识，就能横评不同模型。

📦 顺便看看「标准内容块」

新增的 content_blocks 让你跨厂商统一读取文本、推理过程、工具调用等结构化内容：

model = init_chat_model("deepseek:deepseek-chat")
resp = model.invoke("用一句话解释什么是时区")

for block in resp.content_blocks:
if block["type"] == "text":
print("回答：", block["text"])
elif block["type"] == "reasoning":
print("推理过程：", block["reasoning"])

4.2 工具系统与 MCP 协议

🔧 自定义工具：@tool 装饰器

让模型「会用工具」的第一步，是把普通 Python 函数标记成工具。

⚠️ 函数的名字、docstring、参数名都会变成模型理解工具用途的依据，所以一定要写清楚。

下面做两个原创工具——航班查询 和 汇率转换，再绑定到一个旅行助手智能体上。

"""旅行助手：航班查询 + 汇率转换 工具。"""
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool

@tool
def search_flights(origin: str, destination: str, date: str) –> str:
"""查询两地之间在指定日期的航班。origin/destination 为城市名，date 格式 YYYY-MM-DD。"""
# 演示用的假数据，真实场景请调用航司/聚合 API
return (
f"{date} {origin}→{destination} 航班：\\n"
f"- CZ3456 08:20 起飞，直飞，¥780\\n"
f"- MU5678 14:05 起飞，经停，¥520"
)

@tool
def convert_currency(amount: float, from_cur: str, to_cur: str) –> str:
"""把一笔金额从一种货币换算到另一种货币。from_cur/to_cur 用 ISO 代码，如 CNY、USD、JPY。"""
rates_to_cny = {"CNY": 1.0, "USD": 7.18, "JPY": 0.048, "EUR": 7.75}
if from_cur not in rates_to_cny or to_cur not in rates_to_cny:
return "暂不支持该货币。"
cny = amount * rates_to_cny[from_cur]
result = cny / rates_to_cny[to_cur]
return f"{amount} {from_cur} ≈ {result:.2f} {to_cur}"

model = init_chat_model("deepseek:deepseek-chat", temperature=0)

travel_agent = create_agent(
model=model,
tools=[search_flights, convert_currency],
system_prompt="你是贴心的旅行助手，需要时调用工具查询航班和换算汇率，再用自然语言汇总给用户。",
)

resp = travel_agent.invoke({
"messages": [{"role": "user", "content": "帮我查 2026-07-01 上海到东京的航班，并把 520 元换算成日元。"}]
})
print(resp["messages"][–1].content)

💡 智能体会自己决定先调 search_flights、再调 convert_currency，最后把结果汇总——这就是 create_agent 内建的 ReAct 循环。

🌊 流式观测智能体的「中间步骤」

第 3 章讲的 chain.stream() 是 LCEL 链层面的 token 流；而智能体往往要经过「思考 → 调工具 → 再思考」多轮，光看最终答案体验很差。

create_agent 返回的智能体提供两种流式手段：

• 🪜 stream_mode：流式吐出中间步骤（每次模型输出、每次工具调用与返回）；
• 🔬 astream_events：更细粒度的事件流（精确到 token 级、节点级），适合做实时前端。

# 方式一：stream_mode —— 按「步」观测智能体推进（同步）
for chunk in travel_agent.stream(
{"messages": [{"role": "user", "content": "查 2026-07-01 上海到东京的航班"}]},
stream_mode="updates", # "updates"=每个节点的增量；"values"=每步后的完整状态
):
print(chunk) # 能看到工具调用、工具返回、模型回复逐步出现

# 方式二：astream_events —— 细粒度事件流（异步），适合实时前端
import asyncio

async def watch():
async for event in travel_agent.astream_events(
{"messages": [{"role": "user", "content": "把 520 元换算成日元"}]},
):
kind = event["event"]
if kind == "on_chat_model_stream": # 模型逐 token 输出
print(event["data"]["chunk"].content, end="", flush=True)
elif kind == "on_tool_start": # 工具开始执行
print(f"\\n[调用工具] {event['name']} 入参={event['data'].get('input')}")
elif kind == "on_tool_end": # 工具返回
print(f"[工具返回] {event['data'].get('output')}")

asyncio.run(watch())

要点 ✨：

• 🪜 stream_mode="updates" 只推增量（这一步新增了什么），"values" 推每步后的完整状态，"messages" 则专门流式消息 token；
• 🔬 astream_events 把执行拆成 on_chat_model_stream / on_tool_start / on_tool_end 等事件，能同时做到「打字机效果 + 工具调用可视化」；
• 🖥️ 生产前端常用 astream_events 把「正在查航班…」「正在换算汇率…」等中间态实时推给用户，体验远胜「转圈等最终结果」。

🔌 MCP：让工具「即插即用」

MCP（Model Context Protocol，模型上下文协议） 是一个开放标准，让「工具/数据源」以统一方式暴露给任意 AI 应用。

🧩 你可以把它理解成「AI 工具界的 USB 接口」。

LangChain 通过 langchain-mcp-adapters 把 MCP 服务器暴露的工具自动转换成 LangChain 工具，直接喂给 create_agent。

第一步，写一个本地 MCP 服务器（用 fastmcp 库，最简单）。这里做一个「景点门票价格查询」服务：

# 文件名：ticket_mcp_server.py
"""一个最小 MCP 服务器：景点门票查询。"""
from fastmcp import FastMCP

mcp = FastMCP("ticket-service")

PRICES = {"故宫": 60, "黄山": 190, "西湖": 0, "迪士尼": 599}

@mcp.tool()
def get_ticket_price(spot: str) –> str:
"""查询某个景点的门票价格（人民币）。"""
price = PRICES.get(spot)
if price is None:
return f"未收录「{spot}」的票价。"
return f"{spot} 门票：¥{price}" if price else f"{spot} 免费开放"

if __name__ == "__main__":
mcp.run(transport="stdio") # 通过标准输入输出通信

第二步，在主程序里用 MultiServerMCPClient 连接它，把 MCP 工具加载进智能体：

"""把本地 MCP 工具接入旅行助手。"""
import asyncio

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model

async def main():
client = MultiServerMCPClient({
"ticket": {
"command": "python",
"args": ["ticket_mcp_server.py"], # 指向上面的服务器文件
"transport": "stdio",
}
})

mcp_tools = await client.get_tools() # 自动转成 LangChain 工具
print("已加载 MCP 工具：", [t.name for t in mcp_tools])

agent = create_agent(
model=init_chat_model("deepseek:deepseek-chat", temperature=0),
tools=mcp_tools,
system_prompt="你是旅行助手，用户问景点票价时调用工具查询。",
)

resp = await agent.ainvoke({
"messages": [{"role": "user", "content": "故宫和黄山的门票分别多少钱？"}]
})
print(resp["messages"][–1].content)

asyncio.run(main())

MultiServerMCPClient 支持同时连接多个服务器（本地 stdio 或远程 http），把它们的工具汇总成一个列表喂给智能体，扩展能力时无需改动智能体主逻辑。🧩

🖼️ 4.3 多模态能力增强：菜品图片 + 文字 → 营养建议

场景：做一个「饮食教练」助手——用户上传一张菜品照片，再补一句文字（比如「我在减脂」），助手结合图片内容和减脂目标，给出营养评估与改进建议。这需要多模态输入：在一条 HumanMessage 里同时塞「文字块」和「图片块」。

1.0 用统一的内容块结构表达多模态输入（type: "text" 与 type: "image"）。注意要选支持视觉的模型。

"""多模态饮食教练：图片 + 文字 -> 营养建议。"""
import base64

from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage, SystemMessage

def encode_image(path: str) –> str:
with open(path, "rb") as f:
return base64.b64encode(f.read()).decode("utf-8")

# 选一个具备视觉能力的模型
model = init_chat_model("openai:gpt-4o", temperature=0.3)

image_b64 = encode_image("my_lunch.jpg")

messages = [
SystemMessage(content="你是注册营养师，依据图片中的食物和用户目标给出客观、可执行的饮食建议。"),
HumanMessage(content=[
{"type": "text", "text": "这是我今天的午餐，我正在减脂，请评估热量大概区间并给出改进建议。"},
{"type": "image", "base64": image_b64, "mime_type": "image/jpeg"},
]),
]

resp = model.invoke(messages)
print(resp.content)

要点 ✨：

• 🧩 多模态输入就是把 HumanMessage.content 写成内容块列表，文字块 + 图片块并存；
• 🖼️ 图片可用 base64（如上）或直接给 URL（{"type": "image", "url": "https://…"}）；
• 🔄 换不同视觉模型时，统一的内容块表示让你几乎不用改代码。

🧠 第 5 章 记忆系统：让 AI 拥有「长期记忆」

5.1 短期记忆与长期记忆

先把两个概念分清：

• 📄 短期记忆：一次会话内的上下文，也就是「这轮对话说过的话」。一旦会话结束或换了 thread_id，它就没了。技术上靠消息历史实现。
• 🗄️ 长期记忆：跨会话、跨天甚至跨设备都还记得的信息，比如「这个用户偏好素食」「学到了第几课」。技术上靠向量数据库 / 持久化存储实现。

场景：做一个「语言学习伙伴」——它要在一次对话里记住你刚说过的话（短期），用 RunnableWithMessageHistory 实现。

"""语言学习伙伴：用 RunnableWithMessageHistory 注入短期记忆。"""
from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

model = init_chat_model("deepseek:deepseek-chat", temperature=0.5)

prompt = ChatPromptTemplate.from_messages([
("system", "你是耐心的英语口语陪练。纠正用户的语法错误，并用简单英语回应，必要时给中文提示。"),
MessagesPlaceholder(variable_name="history"), # 历史消息插槽
("human", "{input}"),
])

chain = prompt | model | StrOutputParser()

# 用一个字典按 session_id 存放每个用户的历史
_store: dict[str, InMemoryChatMessageHistory] = {}

def get_history(session_id: str) –> InMemoryChatMessageHistory:
if session_id not in _store:
_store[session_id] = InMemoryChatMessageHistory()
return _store[session_id]

chat = RunnableWithMessageHistory(
chain,
get_history,
input_messages_key="input",
history_messages_key="history",
)

cfg = {"configurable": {"session_id": "learner_amy"}}

print(chat.invoke({"input": "I very like learning English."}, config=cfg))
# 下一轮它能记得上文，比如知道你刚才在练口语
print(chat.invoke({"input": "What did I say wrong just now?"}, config=cfg))

💬 RunnableWithMessageHistory 自动帮你把历史读进来、把新对话存回去，你只管按 session_id 区分用户即可。

5.2 向量数据库集成：存储用户学习进度

短期记忆会随会话消失。要让伙伴跨天记得「你学到哪了、哪些单词总错」，就把这些「学习进度」写进向量库（这里用 Chroma），下次对话时按语义检索回来注入提示。

📌 这里设计的记忆存储结构不是普通对话，而是结构化的学习画像条目：

"""把用户学习进度存进 Chroma，按需检索注入。"""
from langchain.embeddings import init_embeddings
from langchain_chroma import Chroma
from langchain_core.documents import Document

# 1) 初始化向量库
embeddings = init_embeddings("openai:text-embedding-3-small")
profile_db = Chroma(
collection_name="learner_profile",
embedding_function=embeddings,
persist_directory="./learner_memory", # 落盘，跨进程持久化
)

# 2) 写入「学习进度」记忆（结构化文本 + 元数据）
def remember_progress(user_id: str, note: str, topic: str):
profile_db.add_documents([
Document(page_content=note, metadata={"user_id": user_id, "topic": topic})
])

remember_progress("amy", "Amy 经常混淆 'very like' 和 'really like'，需要强化动词搭配。", "语法")
remember_progress("amy", "Amy 已完成 Unit 3 旅行主题词汇，掌握度约 70%。", "进度")

# 3) 下次对话时，按当前话题检索相关记忆
def recall(user_id: str, query: str, k: int = 2) –> str:
docs = profile_db.similarity_search(
query, k=k, filter={"user_id": user_id}
)
return "\\n".join(f"- {d.page_content}" for d in docs)

memory_context = recall("amy", "我们继续练习旅行相关的口语吧")
print("检索到的长期记忆：\\n", memory_context)

🔄 然后把 memory_context 拼进系统提示，伙伴就「记得」Amy 的薄弱点和进度了。这就是长期记忆 = 向量库存储 + 语义检索 + 注入提示的完整闭环。

5.3 上下文压缩技术：客服长对话历史压缩

⚠️ 问题：客服对话动辄几十上百轮，全部塞进上下文既贵又容易超出窗口。

上下文压缩的思路：当历史超过某个阈值时，用模型把早期对话摘要成一小段「记忆纲要」，只保留最近几轮原文。

📌 下面先手写一个「滚动摘要」函数帮你理解原理（生产中可直接用内置的 SummarizationMiddleware）：

"""客服长对话：超过阈值就把旧消息摘要压缩。"""
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage, AIMessage, SystemMessage

model = init_chat_model("deepseek:deepseek-chat", temperature=0)

def compress_history(messages: list, keep_recent: int = 4) –> list:
"""保留最近 keep_recent 条，其余压成一段摘要。"""
if len(messages) <= keep_recent + 1:
return messages

head, recent = messages[:–keep_recent], messages[–keep_recent:]
transcript = "\\n".join(
f"{'用户' if isinstance(m, HumanMessage) else '客服'}：{m.content}"
for m in head if isinstance(m, (HumanMessage, AIMessage))
)
summary = model.invoke(
f"把下面这段客服对话压缩成要点摘要，保留关键诉求、已承诺事项和未决问题：\\n\\n{transcript}"
).content

return [SystemMessage(content=f"【历史对话摘要】{summary}")] + recent

# 模拟一段长对话
long_history = [
HumanMessage(content="我买的咖啡机漏水"),
AIMessage(content="抱歉给您带来不便，方便提供订单号吗？"),
HumanMessage(content="订单 20260510X"),
AIMessage(content="已为您登记，将安排上门检修"),
HumanMessage(content="检修要等几天？"),
AIMessage(content="预计 2 个工作日内联系您"),
HumanMessage(content="那这期间能先退我一部分钱吗？"),
AIMessage(content="我帮您申请 20 元补偿券"),
]

compressed = compress_history(long_history, keep_recent=2)
for m in compressed:
print(type(m).__name__, ":", m.content[:60])

生产项目里，更推荐直接用内置中间件，一行接入、自动触发：

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
model="deepseek:deepseek-chat",
tools=[],
middleware=[
SummarizationMiddleware(model="deepseek:deepseek-chat", trigger={"tokens": 3000}),
],
)

📋 当对话累计超过约 3000 tokens 时，它会自动把旧历史压缩，对你的业务代码完全透明。

📚 第 6 章 RAG：检索增强生成技术实战

6.1 RAG 基础原理与 2026 年演进

RAG（Retrieval-Augmented Generation，检索增强生成） 解决一个根本问题：

❌ 模型不知道你的私有数据、也记不住最新信息。

与其把所有资料硬塞进提示，不如先「检索」出最相关的几段，再让模型「基于这几段」回答。

💡 经典 RAG 五步：加载 → 分割 → 向量化 → 检索 → 生成。

#mermaid-svg-g4nCMG4XmYx7WXeU{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-g4nCMG4XmYx7WXeU .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-g4nCMG4XmYx7WXeU .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-g4nCMG4XmYx7WXeU .error-icon{fill:#552222;}#mermaid-svg-g4nCMG4XmYx7WXeU .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-g4nCMG4XmYx7WXeU .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-g4nCMG4XmYx7WXeU .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-g4nCMG4XmYx7WXeU .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-g4nCMG4XmYx7WXeU .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-g4nCMG4XmYx7WXeU .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-g4nCMG4XmYx7WXeU .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-g4nCMG4XmYx7WXeU .marker{fill:#333333;stroke:#333333;}#mermaid-svg-g4nCMG4XmYx7WXeU .marker.cross{stroke:#333333;}#mermaid-svg-g4nCMG4XmYx7WXeU svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-g4nCMG4XmYx7WXeU p{margin:0;}#mermaid-svg-g4nCMG4XmYx7WXeU .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-g4nCMG4XmYx7WXeU .cluster-label text{fill:#333;}#mermaid-svg-g4nCMG4XmYx7WXeU .cluster-label span{color:#333;}#mermaid-svg-g4nCMG4XmYx7WXeU .cluster-label span p{background-color:transparent;}#mermaid-svg-g4nCMG4XmYx7WXeU .label text,#mermaid-svg-g4nCMG4XmYx7WXeU span{fill:#333;color:#333;}#mermaid-svg-g4nCMG4XmYx7WXeU .node rect,#mermaid-svg-g4nCMG4XmYx7WXeU .node circle,#mermaid-svg-g4nCMG4XmYx7WXeU .node ellipse,#mermaid-svg-g4nCMG4XmYx7WXeU .node polygon,#mermaid-svg-g4nCMG4XmYx7WXeU .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-g4nCMG4XmYx7WXeU .rough-node .label text,#mermaid-svg-g4nCMG4XmYx7WXeU .node .label text,#mermaid-svg-g4nCMG4XmYx7WXeU .image-shape .label,#mermaid-svg-g4nCMG4XmYx7WXeU .icon-shape .label{text-anchor:middle;}#mermaid-svg-g4nCMG4XmYx7WXeU .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-g4nCMG4XmYx7WXeU .rough-node .label,#mermaid-svg-g4nCMG4XmYx7WXeU .node .label,#mermaid-svg-g4nCMG4XmYx7WXeU .image-shape .label,#mermaid-svg-g4nCMG4XmYx7WXeU .icon-shape .label{text-align:center;}#mermaid-svg-g4nCMG4XmYx7WXeU .node.clickable{cursor:pointer;}#mermaid-svg-g4nCMG4XmYx7WXeU .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-g4nCMG4XmYx7WXeU .arrowheadPath{fill:#333333;}#mermaid-svg-g4nCMG4XmYx7WXeU .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-g4nCMG4XmYx7WXeU .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-g4nCMG4XmYx7WXeU .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-g4nCMG4XmYx7WXeU .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-g4nCMG4XmYx7WXeU .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-g4nCMG4XmYx7WXeU .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-g4nCMG4XmYx7WXeU .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-g4nCMG4XmYx7WXeU .cluster text{fill:#333;}#mermaid-svg-g4nCMG4XmYx7WXeU .cluster span{color:#333;}#mermaid-svg-g4nCMG4XmYx7WXeU div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-g4nCMG4XmYx7WXeU .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-g4nCMG4XmYx7WXeU rect.text{fill:none;stroke-width:0;}#mermaid-svg-g4nCMG4XmYx7WXeU .icon-shape,#mermaid-svg-g4nCMG4XmYx7WXeU .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-g4nCMG4XmYx7WXeU .icon-shape p,#mermaid-svg-g4nCMG4XmYx7WXeU .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-g4nCMG4XmYx7WXeU .icon-shape .label rect,#mermaid-svg-g4nCMG4XmYx7WXeU .image-shape .label rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-g4nCMG4XmYx7WXeU .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-g4nCMG4XmYx7WXeU .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-g4nCMG4XmYx7WXeU :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

📄 原始文档

✂️ 分割成小块 Chunk

🧮 Embedding 向量化

🗄️ 向量数据库

❓ 用户提问

🧮 向量化提问

📊 召回 Top-K 相关块

🧩 拼进提示

🤖 模型生成有据回答

到 2026 年，RAG 已从「朴素 RAG」演进出几条主线：

• 🔀 混合检索：向量语义检索 + 关键词（BM25）检索结合，兼顾「意思相近」和「精确命中」。
• 🥇 重排序（Re-ranking）：先粗召回一批，再用更强的 reranker 模型精排，把最相关的放前面。
• 🤖 Agentic RAG：不再「一问一检索」，而是让智能体自己决定要不要检索、检索几次、怎么改写查询——本质就是把「检索」做成一个工具交给 create_agent。
• 🧩 上下文工程：配合中间件做查询改写、结果裁剪，把「对的信息在对的时机」喂给模型。

6.2 数据处理流水线：公司章程问答

🎯 场景：员工经常问「年假怎么算」「报销额度多少」，我们用公司章程文档搭一个问答系统。下面从零构建完整流水线。

"""公司章程问答 RAG：加载 -> 分割 -> 向量化 -> 检索 -> 生成。"""
from langchain.chat_models import init_chat_model
from langchain.embeddings import init_embeddings
from langchain_chroma import Chroma
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# 1) 加载：这里直接用字符串模拟章程，真实项目可用 PyPDFLoader 等
charter_text = """
第三章 假期管理
第十二条 员工入职满一年享有 5 天带薪年假，满三年享有 10 天，满五年享有 15 天。
第十三条 年假需提前 3 个工作日在系统提交申请，经直属主管批准后生效。
第四章 报销制度
第二十条 市内交通费每月报销上限为 500 元，需提供正规发票。
第二十一条 差旅住宿一线城市每晚上限 600 元，其它城市 400 元。
"""

# 2) 分割：把长文切成带重叠的小块，避免语义被切断
splitter = RecursiveCharacterTextSplitter(chunk_size=120, chunk_overlap=30)
chunks = splitter.create_documents([charter_text])

# 3) 向量化并入库
embeddings = init_embeddings("openai:text-embedding-3-small")
vectordb = Chroma.from_documents(chunks, embedding=embeddings, collection_name="company_charter")
retriever = vectordb.as_retriever(search_kwargs={"k": 3})

# 4) 构造「带上下文约束」的提示
prompt = ChatPromptTemplate.from_messages([
("system",
"你是公司制度问答助手。只能依据下面提供的【章程片段】回答；"
"若片段中没有相关信息，明确说『章程中未规定』，不要编造。\\n\\n【章程片段】\\n{context}"),
("human", "{question}"),
])

model = init_chat_model("deepseek:deepseek-chat", temperature=0)

def format_docs(docs: list[Document]) –> str:
return "\\n—\\n".join(d.page_content for d in docs)

# 5) 组装 RAG 链：检索 + 透传问题 -> 提示 -> 模型
rag_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| model
| StrOutputParser()
)

print(rag_chain.invoke("入职两年能休几天年假？需要怎么申请？"))
print("\\n—\\n")
print(rag_chain.invoke("出差去成都住宿一晚最多能报多少？"))

🎯 注意这条链的关键设计：系统提示里强约束「只依据片段回答、查不到就说没有」，这是降低 RAG 幻觉的最有效手段之一。

6.3 高级 RAG 技术与优化方法

在 6.2 的「朴素 RAG」基础上，加两个常用增强：混合检索（向量 + BM25） 和 重排序。

🔀 混合检索：用 EnsembleRetriever 融合两路召回

"""混合检索：向量检索 + BM25 关键词检索，加权融合。"""
# 需要额外安装：pip install rank_bm25
from langchain_community.retrievers import BM25Retriever
from langchain_classic.retrievers import EnsembleRetriever

# 复用 6.2 的 chunks
bm25 = BM25Retriever.from_documents(chunks)
bm25.k = 3

vector_retriever = vectordb.as_retriever(search_kwargs={"k": 3})

# 0.5 / 0.5 加权融合两路结果
hybrid_retriever = EnsembleRetriever(
retrievers=[bm25, vector_retriever],
weights=[0.5, 0.5],
)

docs = hybrid_retriever.invoke("年假申请要提前几天")
for d in docs:
print("•", d.page_content[:40])

💪 混合检索的好处：像「第十三条」这种带具体数字/术语的查询，BM25 能精确命中关键词；而「请假流程」这种语义化查询，向量检索更擅长。两者融合，召回更稳。

🥇 重排序：粗召回后再精排

先用混合检索召回较多候选（比如 8 条），再用一个轻量「打分」步骤把最相关的留前面。

📌 这里演示用模型做简易 LLM 重排（生产中可换成专用 reranker，如 Cohere Rerank 或本地交叉编码器）：

"""LLM 简易重排序：对候选块按相关性打分，取 Top-N。"""
from langchain.chat_models import init_chat_model

scorer = init_chat_model("deepseek:deepseek-chat", temperature=0)

def rerank(query: str, docs: list, top_n: int = 3) –> list:
scored = []
for d in docs:
prompt = (
f"问题：{query}\\n片段：{d.page_content}\\n"
"请只回答 0-10 的整数，表示该片段对回答问题的相关度。"
)
try:
score = int("".join(filter(str.isdigit, scorer.invoke(prompt).content))[:2] or 0)
except ValueError:
score = 0
scored.append((score, d))
scored.sort(key=lambda x: x[0], reverse=True)
return [d for _, d in scored[:top_n]]

candidates = hybrid_retriever.invoke("差旅住宿能报多少")
top_docs = rerank("差旅住宿能报多少", candidates, top_n=2)
for d in top_docs:
print("✔", d.page_content[:40])

✨ 把 rerank 接到 6.2 的 RAG 链里（检索 → 重排 → 拼提示 → 生成），就得到了一条「混合检索 + 重排序」的高级 RAG 流水线，召回质量明显优于朴素版本。

🤖 第 7 章 Deep Agents：下一代智能体开发范式

7.1 Deep Agents 核心理念与架构

普通 create_agent 走的是「思考 → 调工具 → 观察 → 再思考」的循环，处理几步的任务很顺手。

但面对「写一份完整测试报告」「重构一个模块」这种需要几十步、跨越很长时间的任务，普通智能体容易迷失目标、忘记上下文、把工具结果堆成一团乱麻——这就是所谓的「浅层智能体陷阱」。

Deep Agents（深度智能体） 是 1.0 生态里专门解决长程任务的范式，由 deepagents 包提供。打个比方：

🧩 普通智能体像单兵实习生——埋头硬干、走一步看一步；Deep Agent 则像带着方法论的项目经理：先列计划、把中间产物写进文件当外部记忆、专业子任务派给子智能体，最后汇总验收。

Deep Agent 的四大内建机制：

7.2 核心特性：自主管理与自我验证 —— 自动化测试报告生成代理

🎯 场景：给定一个项目的「测试结果数据」，让 Deep Agent 自主规划 → 逐步执行 → 自我验证，最终产出一份结构化的测试报告。它会体现「计划-执行-验证」循环。

💡 create_deep_agent 的用法和 create_agent 几乎一致，但自带规划与文件系统能力。

"""自动化测试报告生成代理（Deep Agent）。"""
from deepagents import create_deep_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool

# 一个「跑测试」工具：返回结构化的测试结果（演示用假数据）
@tool
def run_test_suite(module: str) –> str:
"""对指定模块运行测试套件，返回通过/失败统计与失败用例。"""
data = {
"payment": "总计 24 通过 21 失败 3；失败用例：test_refund_timeout, test_currency_round, test_duplicate_charge",
"auth": "总计 18 通过 18 失败 0；全部通过",
}
return data.get(module, "该模块无测试记录")

REPORT_PROMPT = """你是资深测试工程师代理。你的任务是生成一份专业的测试报告。

工作流程要求：
1. 先用 write_todos 制定计划（至少包含：收集各模块结果、分析失败原因、撰写报告、自我校验）。
2. 对每个待测模块调用 run_test_suite 收集数据。
3. 把中间结果写入文件系统保存。
4. 撰写报告：包含总体通过率、各模块明细、失败用例分析、修复优先级建议。
5. 自我验证：检查报告中的数字是否与工具返回一致，确认无遗漏后再输出最终报告。
"""

model = init_chat_model("deepseek:deepseek-chat", temperature=0)

report_agent = create_deep_agent(
model=model,
tools=[run_test_suite],
system_prompt=REPORT_PROMPT,
)

result = report_agent.invoke({
"messages": [{"role": "user", "content": "请为 payment 和 auth 两个模块生成本次回归测试报告。"}]
})
print(result["messages"][–1].content)

运行时你会观察到 Deep Agent 先列 todos、再逐模块取数、写文件、最后自检数字一致性——这套「计划-执行-验证」正是它区别于普通智能体的关键。✅

✅ 运行时你会观察到 Deep Agent 先列 todos、再逐模块取数、写文件、最后自检数字一致性——这套「计划-执行-验证」正是它区别于普通智能体的关键。

7.3 实战：构建自主代码助手 —— Python 类型错误自动修复

🎯 场景：做一个「类型错误修复助手」。给它一段有类型问题的 Python 脚本，它要自主地：

我们给 Deep Agent 配三个工具：读文件、写文件、跑类型检查（用 mypy 模拟）。

"""自主类型错误修复助手（Deep Agent）。"""
import subprocess
import tempfile
import os

from deepagents import create_deep_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool

# 用一个临时工作目录承载被修复的脚本
WORKDIR = tempfile.mkdtemp()
SCRIPT = os.path.join(WORKDIR, "buggy.py")

# 准备一段有类型错误的脚本
with open(SCRIPT, "w", encoding="utf-8") as f:
f.write(
"def add(a: int, b: int) -> int:\\n"
" return a + b\\n\\n"
"result: str = add(3, 4)\\n" # 类型错误：int 赋给 str
"print(add('x', 5))\\n" # 类型错误：str 传给 int 参数
)

@tool
def read_code() –> str:
"""读取当前待修复的 Python 脚本内容。"""
with open(SCRIPT, encoding="utf-8") as f:
return f.read()

@tool
def write_code(new_content: str) –> str:
"""用新内容覆盖写入待修复脚本。"""
with open(SCRIPT, "w", encoding="utf-8") as f:
f.write(new_content)
return "写入成功"

@tool
def type_check() –> str:
"""运行 mypy 静态类型检查，返回检查结果。"""
proc = subprocess.run(
["mypy", "–no-color-output", SCRIPT],
capture_output=True, text=True,
)
return proc.stdout or proc.stderr or "检查完成，无输出"

FIX_PROMPT = """你是自主代码修复助手，专门修复 Python 类型错误。

严格按以下循环工作：
1. 用 write_todos 规划修复步骤。
2. read_code 读取代码。
3. type_check 定位所有类型错误。
4. 分析每个错误的根因，write_code 写入修复后的代码（保持原有业务逻辑不变，只修类型问题）。
5. 再次 type_check 验证。若仍有错误，回到第 4 步继续修，直到检查通过。
6. 输出最终修复后的完整代码，并说明每处改动的原因。
"""

model = init_chat_model("deepseek:deepseek-chat", temperature=0)

fixer = create_deep_agent(
model=model,
tools=[read_code, write_code, type_check],
system_prompt=FIX_PROMPT,
)

result = fixer.invoke({
"messages": [{"role": "user", "content": "请修复脚本中的所有类型错误，并验证通过。"}]
})
print(result["messages"][–1].content)

🔄 这个助手的精髓在于「自我验证闭环」：它不会改完就交差，而是反复 type_check 直到真正通过——这正是 Deep Agents「执行后必验证」理念的落地。

⚠️ 运行前请先 pip install mypy。

🏁 结语

到这里，我们从「30 行的情感分析器」一路走到了「会自我验证的代码修复代理」。

回顾这趟旅程里你真正掌握的 LangChain 能力：

• ✅ 统一入口：init_chat_model 一行切换厂商，create_agent 三参数起步；
• ✅ LCEL 组合：| 管道、分支、并行、流式、异步、回退重试；
• ✅ 中间件治理：自定义 before_model / after_model 钩子，内置脱敏 / 摘要 / 人工审批；
• ✅ 工具与 MCP：@tool 装饰器 + langchain-mcp-adapters 即插即用；
• ✅ 多模态：内容块统一表达「文字 + 图片」；
• ✅ 记忆：RunnableWithMessageHistory 管短期、向量库管长期、摘要做压缩；
• ✅ RAG：从朴素五步到混合检索 + 重排序；
• ✅ Deep Agents：规划 + 文件系统 + 子智能体，应对长程任务。

给新手的三条上手建议 🌟：

💡 小提示：LangChain 仍在快速演进，本文示例以 1.x 为准，运行时请以你所安装版本的官方文档为最终准绳。

祝你构建愉快，Happy Building with LangChain！🚀✨