Langchain 1.+

---

目录

1. 第 01 章　LangChain 1.2 概述

2. 第 02 章　模型的创建与调用

3. 第 03 章　LangSmith 基本使用

4. 第 04 章　消息与提示词模板

5. 第 05 章　Tools（工具）

6. 第 06 章　结构化输出

7. 第 07 章　智能体（Agent）

8. 第 08 章　中间件（Middleware）

9. 第 09 章　上下文与记忆

10. 第 10 章　RAG（检索增强生成）

---

全局约定

• 代码示例中 <YOUR_API_KEY>、<KEY> 等为占位符，使用时请替换为自己的密钥；密钥建议放在 .env 文件并通过 load_dotenv() 加载，不要硬编码或外泄。

• 各章模型初始化大多使用 init_chat_model() 统一接口，示例中常以 gpt-5.4-mini、deepseek-v4-flash 等作为模型名，按所用平台替换。

• 命令行示例默认在已激活的 conda 虚拟环境（如 langchain1.2）中执行。

---

第01章 LangChain 1.2 概述

1、为什么需要 LangChain？

1.1 从传统应用到智能体时代

传统应用基于确定性逻辑，而智能体时代要求应用具备推理、规划与行动能力。单一的大语言模型无法独立完成真实业务，必须与外部工具、数据源、记忆机制结合，这正是 LangChain 框架的设计理念来源。

1.2 单一的大语言模型的局限性

• 知识滞后：训练数据有截止日期，无法获取实时信息。

• 幻觉问题：对未知信息容易编造答案。

• 无法执行外部动作（如订票、查库、写 SQL）。

• 无持久记忆，多轮对话上下文易丢失。

因此要构建真正实用的 AI 应用，必须将大语言模型与外部工具、数据源和记忆机制有机结合，从而催生了 LangChain 框架的设计理念。

1.3 LangChain 框架的定位

LangChain 作为大模型与应用间的中间层，可统一调用各类大模型、管理提示词与上下文，还能集成外部工具和数据源，快速搭建具备推理、行动能力的智能体。它是当前构建生产级 AI 智能体系统的首选。

核心定位三点：

1. 打通大模型与外部资源：统一接口对接数据库、检索引擎、API、文件系统等。

2. 封装底层复杂逻辑：抽象工具调用、记忆等能力，降低智能体开发难度。

3. 支撑多智能体协作：依托 LangGraph 等生态，从单智能体拓展至多智能体协作，构建工业级智能体。

1.4 LangChain 的应用场景

应用场景	痛点	功能
检索增强生成 (RAG)	大模型知识滞后（无实时数据）和幻觉	检索外部知识库（文档/数据库）并向量化，基于最新、最相关资料回答
Agent 智能体构建	LLM 无法直接执行复杂任务	模型作为"推理引擎"，自主规划并动态调用外部工具（订票、查报告、写 SQL）
对话系统与聊天机器人	多轮对话"记忆"流失	集成记忆管理，记住偏好与历史，结合私有数据提供专业服务
多模态应用开发	单一文本交互的限制	融合图像识别、语音转文字等，处理音视频和图片的综合推理
自动化写作与格式化生成	生成内容格式不规范、质量不稳定	配合提示词模板与输出解析器，自动产出规范报告、合同、邮件
数据连接与结构化处理	非结构化数据难以直接利用	从 PDF/Excel 提取关键信息，实现自然语言与 SQL 的自动转换

1.5 大模型相关岗位介绍

应用开发是大模型最值得关注的方向：应用为王。学习 LangChain 框架可高效开发大模型应用。

---

2、LangChain 是什么？

2.1 LangChain 的发展时间线

阶段	时间	要点
第1阶段：诞生	2022年10月	哈里森·蔡斯（Harrison Chase）创建开源框架。名称来自 "Language" + "Chain"，体现"链接大语言模型与其他计算资源和数据"的理念
第2阶段：探索期	2022年Q4—2023年Q1	初版发布，聚焦 PromptTemplate、LLMChain 等基础模块，GitHub Star 快速破万
第3阶段：体系化阶段	2023年Q2—Q4	引入 Tool、Agent、Retrieval，形成"大模型+工具调用+记忆"核心架构；推出 LangChain Hub 与 LangSmith，初步构建生态闭环
第4阶段：平台化阶段	2024年—2025上半年	LangGraph（有向图工作流管理）与 LangServe（服务化部署）发布，完成从框架到智能体平台的跃升
第5阶段：深层智能体阶段	2025年下半年至今	推出 Deep Agent（Agent Harness，智能体执行框架），在 LangGraph 和 LangChain 之上运行，构建多智能体复杂系统

2.2 LangChain 的两个重要版本

1、LangChain v0.3：既爱又恨

• 因 API 变动频繁被戏称为"版本碎钞机"。

• 2024 年架构大变革：推出 LangGraph 作为底层编排框架，原有链和智能体被标记弃用。

• GPT-4 普及后，Function Calling、结构化输出、系统提示词成为模型基础功能，再用 LangChain 封装显得多此一举。

• 这个阶段 LangChain 开发者大规模流失。

2、LangChain v1.x：AI 开发新范式

• 经历阵痛后进行彻底的架构重构与瘦身。

• 2025年10月20日正式发布 LangChain v1.0.0 与 LangGraph v1.0.0，是 Agent 开发领域的里程碑，标志框架成熟与标准化。

• 官方首次明确 API 稳定保证：承诺在 2.0 版本前无破坏性变更（对企业级应用至关重要）。

• 同期完成 1.2 亿美元融资，估值超 12 亿美元。

对比 LangChain v0.3 与 v1.2（了解）

维度	LangChain v0.3	LangChain v1.2
核心架构与设计理念	过渡性版本，以链（Chain）为核心	生产级稳定版本，从"链式调用"到"智能体框架"的范式转变
Agent 构建方式	依赖 initialize_agent 等旧版 API，基于 AgentExecutor 硬编码	官方推荐 create_agent 作为标准构建入口，底层基于 LangGraph
工具定义	@tool 装饰器和 Tool 类，类型安全性与参数验证较弱	支持 Pydantic Schema 定义工具，类型安全，参数清晰
结构化输出	依赖 JSON Parser 和正则，稳定性差	Structured Output 成为一等公民，直接绑定 Pydantic 类，模型底层保证格式稳定
输出解析	纯文本，需正则匹配，繁琐易错	引入标准化 content_blocks，统一推理/文本/工具调用，无需手动解析
扩展机制	缺乏系统性扩展，需改源码或提示词	引入 Middleware（中间件）系统，可在模型调用、工具执行等各生命周期拦截扩展
多模态支持	支持不完善，无法无缝集成图像、音频	完善多模态适配，轻松实现多模态对话、多模态 RAG
异步执行性能	性能一般	性能优化，响应速度提升 30% 以上
包结构与依赖	包结构混乱，模块耦合度高；langchain 为直接依赖；内部从 Pydantic v1 升级到 v2	包结构清晰，主 langchain 包轻量，旧功能（如 Chains）迁移至 langchain-classic；@langchain/core 为对等依赖
Python 版本要求	Python >= 3.9（停止支持 3.8）	Python >= 3.10
推荐用途	维护老旧项目，不推荐新项目	所有新项目和学习首选；官方承诺 1.x 无破坏性变更

2.3 LangChain v1.2 的主要模块

• langchain-core：官方推荐的核心 API，如 Runnable、BaseMessage 等。

• langchain-classic：冗余或不再推荐使用的经典 API（0.x 常用而 1.x 移除的）。

• langchain-community：第三方集成，如合作伙伴包 langchain-openai、langchain-anthropic 等，按需安装、避免臃肿。

• langgraph：深度整合 LangGraph 1.0，协调多个 Chain、Agent、Tools 完成更复杂任务，支持循环调用，是 langchain 图形化的增强版。

不要试图学完 LangChain 的所有 API（那不可能）。搞懂核心逻辑与核心模块，其它的用到再查。把它当成工具箱，而不是教科书。

2.4 API 文档

• 官网：LangChain: Observe, Evaluate, and Deploy Reliable AI Agents

• GitHub：https://github.com/langchain-ai

• 中文文档：LangChain 概述 - LangChain 文档 - LangChain 教程

• 英文文档：LangChain overview - Docs by LangChain

• API 查询：langchain | LangChain Reference

---

3、LangChain 家族四大支柱

截至 2025 年 11 月，LangChain 已成长为覆盖智能体系统全生命周期的技术生态，由四大核心支柱构成，分别对应基础能力层、运行时编排层、智能体抽象层、监控与评估层。

3.1 LangChain：智能体开发的基石

整个生态的核心与起点，提供模型调用、工具与中间件集成、智能体构建等基础能力。

核心价值：

• 统一的模型抽象层：屏蔽不同模型提供商（OpenAI、Anthropic、Ollama 等）的接口差异。

• 高度模块化设计：用 Message、Tool、Agent、Middleware 等组件实现灵活组合扩展。

• 丰富的集成生态：预置数据源、API、中间件，构成 AI 能力枢纽。

结论：若只需构建简单智能体应用、无复杂编排需求，选择 LangChain 即可。

3.2 LangGraph：复杂工作流的编排引擎

当任务从单一指令扩展为多步骤、有状态的复杂工作流时使用。核心思想是将智能体内部抽象为一张有向图。

• 节点（Node）：独立的功能单元或决策点。

• 边（Edge）：节点之间的流转条件与路径。

• 状态（State）：在节点间传递并持久化存储的共享上下文。

通俗理解：

• LangChain = 能力抽象层（LLM / Tool / Message 标准化），负责"有什么能力"。

• LangGraph = 执行与编排层（状态机 / 工作流 / 多 Agent 系统），负责"怎么跑"。

3.3 Deep Agent：智能体的执行框架

新推出的组件，定位为 Agent Harness（智能体执行框架）。构建于 LangChain 与 LangGraph 之上，增加规划能力、文件系统、子 Agent 等高级功能。

核心能力：

• 显式规划：自主生成、执行并动态调整多步任务计划。

• 虚拟文件系统：结构化存储中间结果与知识。

• 子智能体：任务在多个智能体间分解与协作。

• 长期记忆：结合 LangGraph 状态存储，实现跨对话经验积累。

• 可扩展中间件：嵌入安全审计、性能监控或自定义业务逻辑。

3.4 三者的关系

三个框架并非互斥，复杂项目可同时用到三层。完整的 LangChain 生态玩法：从 LangChain 快速搭建 → 用 LangGraph 打磨生产稳定性 → 再用 Deep Agent 赋予 Agent 更强自主能力。

3.5 LangSmith：可视化监控与测试平台

LangChain 官方推出的可视化监控与测试平台，用于跟踪、记录和分析智能体运行过程中的完整调用链路，让内部运行透明、可评估。

核心目标：全链路追踪、调试与优化、评测与质量控制、团队协作。

官网：LangSmith: AI Agent & LLM Observability Platform

---

4、开发前的准备工作

4.1 前置知识

1. Python 基础语法：变量、流程控制、函数与参数机制、类与对象、装饰器；常用容器（列表/元组/集合/字典）、JSON 处理、异常处理；模块导入、包管理（pip 或 conda）、线程与协程。

   • LangChain 生态支持 Python 和 JavaScript，Python 版本功能最完整、更新最及时、社区最活跃。

2. 大语言模型基础：了解 LLM、Token、Prompt、Embedding；OpenAI API 或其他提供商（Anthropic、阿里云百炼、DeepSeek 等）；通过浏览器/app 使用过大模型（豆包、千问、DeepSeek 等）。

4.2 相关环境安装

4.2.1 代码管理方案

虚拟环境相比全局环境拥有独立的 Python 解释器、pip 和第三方依赖包，互不干扰。

工具	管理 Python 解释器	管理 Python 包	管理非 Python 依赖	适合场景
conda	✅ 可以	✅ 可以	✅ 可以	AI、深度学习、科学计算、复杂底层依赖（C/C++、CUDA）
uv	✅ 可以	✅ 可以	❌ 不可以	纯 Python 项目、Web、Agent、RAG 应用层
venv	不支持原生安装，基于已有解释器	✅ 可以	❌ 不可以	简单项目、教学演示、轻量隔离

• 方案1 conda：适合"Python + 非 Python 依赖"的复杂环境，数据科学/深度学习/AI 工程优先推荐。conda 中可用 pip，但建议先 conda 装底层依赖、再 pip 补 Python 包，不随意交替。本课程选择 conda。

• 方案2 uv：现代包管理工具，仅管理 Python 生态依赖，适合纯 Python 项目（FastAPI、LangChain、脚本工具、Web 后端、AI Agent、RAG 应用层）。

• 方案3 venv：Python 自带，轻量；不安装新解释器，基于当前已安装的解释器创建环境。

4.2.2 安装虚拟环境与 Python 解释器

LangChain 1.2 要求 Python >= 3.10，本课程使用 Python 3.13.12。

# 创建一个名为 langchain1.2 的环境，指定 Python 版本是 3.13.12
conda create --name langchain1.2 python=3.13.12
# 查看 anaconda 已安装的 python 环境
conda env list
# 初始化虚拟环境（执行完后重启命令行窗口）
conda init
# 在命令行切换到某 python 环境
conda activate langchain1.2
# 验证 python 版本（输出如：Python 3.13.12）
python -V
​
# 退出当前 python 环境
conda deactivate
# 删除一个已有的 anaconda 管理的 python 环境
conda remove --name langchain1.2 --all

4.2.3 下载 langchain 安装包

注意：包的安装必须显式指明版本，否则可能出现不兼容。

方式1 — 使用 conda（推荐）：

# 安装指定版本（如 1.2.12）
conda install langchain==1.2.12
# 指定频道 conda-forge（更新更快、包更全；-c 是 --channel 的缩写）
conda install -c conda-forge langchain==1.2.12
# 更新 / 卸载 / 查看已安装包
conda update langchain
conda uninstall langchain
conda list

方式2 — 使用 pip：

# 安装指定版本
pip install langchain==1.2.12
# 国内镜像加速（-i 指定镜像源）
pip install langchain==1.2.12 -i https://pypi.tuna.tsinghua.edu.cn/simple
# 从旧版本升级
pip install --upgrade langchain
# 卸载 / 查看已安装包
pip uninstall langchain
pip list

建议：优先 conda install，conda 没有再用 pip install。 区别：conda 依赖检查严格、管环境+依赖+稳定性、支持 Python+非 Python 包；pip 相对宽松、只管 Python 包。

4.2.4 PyCharm 开发环境

PyCharm 是专业的 Python IDE，具有代码编辑、调试和版本控制功能。新建工程并设置 Python 解释器（选择 Anaconda 环境）。

验证安装与版本：

import langchain
print(langchain.__version__)

---

5、大模型应用场景介绍

大模型应用技术特点：门槛低，天花板高。

5.1 RAG 开发

1）背景

• 知识冻结：LLM 训练成本与周期随规模增加，无法实时学习最新信息，难以应对时间敏感问题。

• 幻觉：对训练中未学过的信息，无法准确答复，转而臆想编造。

2）何为 RAG

RAG = Retrieval-Augmented Generation（检索增强生成）。典型流程（检索-增强-生成）：

1. 本地文件（结构化二维表 / 非结构化 PDF、Word、TXT）

2. 非结构化数据加载器（Unstructured Loader） → Text

3. 文本切分（Text Splitter） → 多个 Text Chunk

4. 嵌入模型（Embedding Model） 向量化 → Vector Embeddings

5. 向量数据库（Vector Database） 存储 & 索引

6. User Query 经嵌入模型向量化，相似度搜索（Similarity Search） 召回最相似向量，作为上下文 Context

7. 提示词模板 组合：Prompt = Context + User Query

8. 发送给 LLM 生成 Answer

检索对应召回步骤，增强对应"提示词包含检索到的数据"，生成对应 LLM 输出。

难点与 Reranker（重排器）

四大难点：① 文件解析（PDF 内含图片/表格/图文）② 文件切割（无固定格式）③ 知识检索 ④ 知识重排序。

随着文档数量增加，召回准确率会下降。引入 reranker 对初步召回的较多 chunk（如 top 20/50）精排，提高准确率、防止 LLM 处理无关信息、降低成本（比仅靠 LLM 生成便宜，但比纯矢量搜索贵）。

• 适合：追求高精度高相关性的场景（专业知识库、客服系统）。

• 不适合：会增检索延迟，对响应时间要求高的服务不合适。

5.2 Agent 开发

充分利用 LLM 的推理决策能力，增加规划、记忆和工具调用，构造能独立思考、逐步完成目标的 Agent。

公式表达：

Agent = LLM + Planning + Tools + Memory + Action

智能体核心要素（5 个模块）：

1. 大模型（LLM）作为"大脑"：提供推理、规划和知识理解，是决策中枢，能呈现推理和规划过程，应对未知任务。

2. 规划决策（Planning）：通过任务分解、反思与自省框架处理复杂任务，如思维链（CoT）拆解子任务并通过反馈优化策略。

3. 工具使用（Tool Use）：调用外部工具（API、数据库）扩展能力边界。

4. 记忆（Memory）：

   • 短期记忆：存储单次对话周期的上下文，受限于模型上下文窗口。如 ChatGPT 约 8k、GPT-4 约 32k；新模型如 GPT-5.5/5.4 Pro 100 万 token、Claude Opus 4.7 200 万 token、DeepSeek-V4-Pro 100 万 token。

   • 长期记忆：跨会话/时间周期存储并调用核心知识（用户偏好、历史指令）。可通过模型微调、知识图谱、向量数据库实现。

5. 行动（Action）：实际执行决策的模块，涵盖软件接口操作（自动订票）和物理交互（机器人搬运），如检索、推理、编程。

5.3 大模型应用开发的 4 个场景

场景	通俗类比	特点
纯 Prompt	你说一句，ta 回一句	Prompt 是操作大模型的唯一接口
Agent + Function Calling	你问出差要不要带伞，ta 让你先看天气，你看了告诉 ta，ta 再回答	Agent：AI 主动提要求；Function Calling：对接外部系统时 AI 要求执行某函数
RAG	考试答题时到书上找相关内容，再结合题目作答（智能客服最广）	需要补充领域知识时使用；Embeddings→向量数据库→向量搜索
Fine-tuning（精调/微调）	努力学习考试内容，长期记住，活学活用	成本最高；前面方式解决不了时再用

如何选择技术方案：面对一个需求，常用思路是从简单到复杂递进——优先 Prompt，需要外部系统用 Agent + Function Calling，需要领域知识用 RAG，都不行再考虑 Fine-tuning。

第03章 LangSmith 基本使用

1、LangSmith 概述

1.1 什么是 LangSmith？

LangSmith 是 LangChain 生态系统中专门用于 LLM 应用调试、监控、评估和管理 的平台。

• 🔍 追踪 (tracing)：记录每次 LLM 调用的详细信息

• 📊 监控 (monitoring)：实时查看应用性能

• 🐛 调试 (debug)：排查问题和优化性能

• 📈 评估 (evaluate)：系统化测试 LLM 应用

1.2 具体功能

功能1：核心应用与开发

1. Tracing（追踪）：最核心的功能，完整记录应用每一次调用链路（Trace）。当 Agent/RAG 变慢或报错时，进入项目可看到每步的 Prompt、模型返回、Token 消耗、各节点耗时，方便排查 Bug 和优化性能。

2. Monitoring（监控）：提供生产环境的高级数据可视化看板，宏观监控 Token 消耗趋势、QPS、错误率、平均延迟（Latency）及成本预估。

3. Datasets & Experiments（数据集与实验）：管理测试数据集并运行对比实验。可把用户真实输入、边界情况（Edge Cases）存为数据集，修改 Prompt 或更换底层模型后，运行自动化对比测试。

4. Evaluators（评估器）：配置和自动化评估任务。支持基于规则（关键词匹配）或基于模型（LLM-as-a-judge）的评估指标（答案相关性、是否幻觉等），自动打分。

5. Annotation Queues（标注队列）：人工反馈与数据清洗工具。把 Traces 发送到标注队列，让团队成员手动打分、纠正、贴标签，用于微调模型或充当测试集。

功能2：提示词与调试工具

1. Prompts（提示词管理）：类似"提示词版的 GitHub"。把 Prompt 从代码中解耦，云端统一管理、版本控制（v1、v2），代码中通过 API 动态拉取，支持团队协作与分享。

2. Playground（演练场）：网页端模型交互界面。无需写代码即可选择不同模型（OpenAI、Anthropic、本地模型），快速微调测试 Prompt，可一键保存到 Prompts 仓库。

3. Studio（工作室）：与 LangGraph 深度集成的可视化图形界面。可视化查看状态机（State）在节点间流转，支持在节点"暂停"、手动修改数据后继续执行，是调试复杂 Agent 的利器。

4. Context Hub（上下文中心）：管理全局上下文或通用组件配置，存放可跨项目/Prompt 复用的公共上下文模板、全局变量、系统预设提示。

功能3：部署与沙盒

1. Deployments（部署）：一键将 LangChain 应用或 LangGraph Agent 部署为线上 API 服务（依托 LangGraph Cloud），提供开箱即用的生产端点，处理高并发、队列管理和状态持久化。

2. Sandboxes（沙盒）：轻量级在线运行和测试环境，不污染生产环境，安全试运行新 Agent 或执行自动化脚本。

建议：现阶段重点关注 Tracing（观察调用细节）和 Playground（快速调优提示词）。当应用走向复杂（复杂 RAG 检索、多 Agent 协同）时，再引入 Datasets 量化评估、用 Studio 可视化调试。

---

2、准备账号

2.1 注册或登录

1. 访问官网：LangSmith

2. 自由选择注册或登录方式

3. 登录成功

2.2 获取 API_KEY

1. 打开设置

2. 创建 API_KEY

3. 保存 KEY（点击 copy 复制到剪贴板）

注意：API_KEY 只在创建弹窗出现一次，关闭弹窗后回到设置页面就无法再在官网查看其内容，务必妥善保存。 步骤4（可选）：按需删除 KEY，点击右侧图标即可。

2.3 新增环境变量

在 .env 配置文件中添加四个环境变量：

# 是否启用 Langsmith 监控功能
LANGSMITH_TRACING=true
# Langsmith 监控 WebUI 地址
LANGSMITH_ENDPOINT=https://api.smith.langchain.com
# 创建的 API_KEY
LANGSMITH_API_KEY=<YOUR_API_KEY>
# 自定义项目名称，可在 Langsmith WebUI 监控页面按名称查看对应运行记录
LANGSMITH_PROJECT="pr-clear-harmony-32"

---

3、查看监控指标

添加环境变量后，程序中通过 load_dotenv() 加载，运行 LangChain 代码时 LangSmith 会自动记录运行指标并同步到后台服务，可在官网查看运行记录。

步骤1：运行任意 LangChain 程序

举例1 — 直接使用厂商专用 Chat 类（ChatDeepSeek）：

import os
from dotenv import load_dotenv
from langchain_deepseek import ChatDeepSeek

# 将 env 文件中的变量加载为环境变量；override=True 表示 .env 优先
load_dotenv(override=True)
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
DEEPSEEK_BASE_URL = os.getenv("DEEPSEEK_BASE_URL")

model = ChatDeepSeek(
    api_key=DEEPSEEK_API_KEY,
    api_base=DEEPSEEK_BASE_URL,
    model_name="deepseek-v4-flash"
)
print(model.invoke("你好"))

举例2 — 使用统一的 init_chat_model（指定 model_provider="openai"）：

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

load_dotenv(override=True)
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")

model = init_chat_model(
    model="deepseek-v4-flash",
    model_provider="openai",
    api_key=CLOSEAI_API_KEY,
    base_url=CLOSEAI_BASE_URL
)
print(model.invoke("你好，用一句话回答"))

举例3 — 通过 config 自定义运行元信息（run_name / tags / metadata）与可调参数（configurable）：

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os
from rich import print as rprint

# 从 .env 文件中加载环境变量
load_dotenv(override=True)
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
DEEPSEEK_BASE_URL = os.getenv("DEEPSEEK_BASE_URL")

# 1. 初始化模型
model = init_chat_model(
    model="deepseek-v4-flash",
    model_provider="deepseek",
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL,
    temperature=0.2,
    max_tokens=500,
    # 指定可调整参数
    configurable_fields=("model", "model_provider", "temperature", "max_tokens"),
)

# 2. 准备 config 字典
config = {
    "run_name": "joke_generation",            # 在 LangSmith 中这次运行显示为 joke_generation
    "tags": ["my_tag1", "my_tag2"],           # 打标签便于分类查找
    "metadata": {
        "user_id": "shkstart",                # 记录用户 ID
        "session_id": "sess_123"              # 记录会话 ID
    },
    "configurable": {
        "model": "deepseek-v4-pro",           # 配置模型参数
        "model_provider": "openai",           # 配置模型提供商参数
        "temperature": 0.7,                   # 配置温度参数
        "max_tokens": 1000                    # 配置最大令牌数
    }
}

# 3. 调用模型并传入 config
response = model.invoke(
    "1 + 2 = ？",
    config=config
)
rprint(response)

config 关键字段说明：

字段	作用
run_name	该次运行在 LangSmith 中显示的名称
tags	标签，便于在 WebUI 中分类查找
metadata	元信息（如 user_id、session_id）
configurable	运行时可调参数（配合 configurable_fields 使用）

步骤2：打开监控界面

在 LangSmith 官方 WebUI 的 Tracing 界面下，可以看到按 LANGSMITH_PROJECT 命名的项目。

步骤3：查看运行指标

点击条目任意位置进入详情页，列出详细的运行指标；点击某次运行记录可查看更详细信息。

步骤4：查看运行报表

报表页提供大量指标，点击标签或下滑页面可切换指标。

第02章 模型的创建与调用

本章对应的官方文档：英文 Models - Docs by LangChain ，中文 https://docs.langchain.org.cn/oss/python/langchain/models

1 模型调用的准备工作

1.1 一张老图看大模型的调用

在 LangChain v0.3 中，Model I/O 包含三个环节，分别对应三类核心组件：

环节	作用	对应组件
Format	格式化输入提示	Prompt Template
Predict	调用模型	Model
Parse	解析输出	Output Parser

对话模型与补全模型的历史演进：

• GPT-3 时代：大模型以补全模型为主，只能以"成语接龙"方式补全文本，效果不稳定。此时 LangChain 借助高层封装 API 让模型完成对话、调用工具、结构化输出。

• GPT-3.5 时代：对话模型正式登上舞台并成为主流。得益于更强的指令跟随能力，很多原本需要 LangChain 才能完成的工作，已成为对话模型的原生功能。

• 因此，本章只讲解对话模型的创建与调用。

1.2 模型初始化的分类方式

一句话概括：用谁家的 API，以什么方式，创建存放在哪个位置的大模型。 可从三个角度划分：

• 角度1（用谁家的 API）：

  • 使用模型提供商的专用库

  • 使用 LangChain 统一方式（推荐）

• 角度2（关键参数 BASE_URL、API_KEY 的书写位置）：

  • 使用配置文件（推荐）

  • 硬编码：写在代码中

• 角度3（模型所在位置）：

  • 在线部署的大模型

  • 本地部署的大模型

LangChain 本身只是一个"工具"，不提供任何 LLMs，而是依赖第三方集成各类大模型。

1.3 线上大模型服务平台

使用流程：注册 → 充值 → 创建 API-Key → 用 API-Key 和 URL 调用平台模型。每个平台配置时都需三个要素：模型名、api-key、base-url。

平台	网址	备注
OpenRouter	OpenRouter	全球主流，含国外模型
CloseAI	CloseAI - 亚洲规模最大的企业级OpenAI API中转平台	亚洲最大，含国外模型
阿里云百炼	大模型服务平台百炼控制台	企业端友好
硅基流动	硅基流动 SiliconFlow - 致力于成为全球领先的 AI 能力提供商	性价比高，适合个人
百度千帆	百度智能云千帆大模型平台	主打百度生态
火山引擎	账号登录-火山引擎	主打字节多模态生态

要点：

• 想用国外模型选前两个；只用国内模型选后四个。

• 阿里云百炼：新用户可得超 5000 万 Tokens 免费额度及 4500 张图片生成额度，适合 toB。

• 硅基流动：号称 9B 以下模型永久免费，开源模型价格低，适合个人学习。

• OpenRouter：第三方镜像站，转发厂商 API，支持国内直连，支持支付宝/微信充值（最低 $5，税费 $0.8）；若模型禁止国内使用（如 ChatGPT），会提示 This model is not available in your region，需要魔法才能调用。

1.4 提前安装所有依赖

课程将所有依赖统一写入 requirements.txt，并固定主要版本以避免兼容问题。

# 在项目根目录执行
pip install -r .\requirements.txt

各章节单独列出的依赖仅为说明，已统一包含在 requirements.txt 中，无需重复安装。

---

2 模型初始化角度1：使用模型提供商库

LangChain 初始化模型主要有两种方式：①直接使用特定的 Model Class；②使用统一的 init_chat_model() 函数。

方式1 最直接：导入供应商提供的专门 Model 类（如 ChatOpenAI、ChatAnthropic、ChatDeepSeek、ChatOllama、ChatHunyuan、ChatTongyi、ChatZhipuAI）并实例化。

参考：chat-models | langchain_community | LangChain Reference

2.1 通过专用 API 调用

注意：不同模型传入的参数名称可能不同，可参考对应源码。

2.1.1 DeepSeek 大模型

官网：DeepSeek | 深度求索

依赖： langchain-deepseek 是使用 DeepSeek 的必要依赖，它又依赖 langchain-openai（pip 会自动拉取）。同时推荐 python-dotenv 用于环境变量管理。

conda activate langchain1.2
pip install langchain-openai
pip install langchain-deepseek
pip install python-dotenv

.env 配置： 将占位符替换为自己的 API Key。

DEEPSEEK_API_KEY=<Your API Key>
DEEPSEEK_BASE_URL=https://api.deepseek.com

方式1：手动读取环境变量初始化

from langchain_deepseek import ChatDeepSeek
import os
from dotenv import load_dotenv

# override=True：无论是否已存在同名环境变量，都用 .env 的值强行覆盖
load_dotenv(override=True)

DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
DEEPSEEK_BASE_URL = os.getenv("DEEPSEEK_BASE_URL")

# 创建 DeepSeek LLM
deepseek_llm = ChatDeepSeek(
    api_key=DEEPSEEK_API_KEY,
    api_base=DEEPSEEK_BASE_URL,  # 注意：这里是 api_base，不是 base_url
    model_name="deepseek-v4-flash",
)
print(deepseek_llm.invoke("请介绍一下你自己"))

方式2：依靠默认行为读取 .env 环境变量（推荐简化写法）

ChatDeepSeek 要求系统存在名为 DEEPSEEK_API_KEY 的环境变量；URL 有默认值，无需手动传入：

from langchain_deepseek import ChatDeepSeek
from dotenv import load_dotenv

load_dotenv(override=True)

deepseek_llm = ChatDeepSeek(
    model="deepseek-v4-flash",
)
print(deepseek_llm.invoke("请介绍一下你自己"))

源码默认值参考：

api_key: SecretStr | None = Field(
    default_factory=secret_from_env("DEEPSEEK_API_KEY", default=None),
)
api_base: str = Field(
    default_factory=from_env("DEEPSEEK_API_BASE", default=DEFAULT_API_BASE),
)
DEFAULT_API_BASE = "https://api.deepseek.com/v1"

方式3：硬编码（不推荐）

from langchain_deepseek import ChatDeepSeek

deepseek_llm = ChatDeepSeek(
    api_key="sk-2nkIWkv6M...U1Ra4P0NGa",  # 明文暴露密钥
    api_base="https://api.deepseek.com",
    model="deepseek-v4-flash",
)
print(deepseek_llm.invoke("请介绍一下你自己"))

硬编码仅适用于临时测试，存在密钥泄露风险。生产环境推荐使用 .env 配置文件，并将其加入 .gitignore。

2.1.2 智谱大模型

官网：智谱AI开放平台

# 社区依赖包，包含 ChatHunyuan、ChatTongyi、ChatZhipuAI
pip install langchain-community
# 智谱 AI 认证相关依赖
pip install pyjwt

.env：

ZHIPUAI_API_KEY=<Your API Key>
ZHIPUAI_BASE_URL=https://open.bigmodel.cn/api/paas/v4/

from langchain_community.chat_models import ChatZhipuAI
from dotenv import load_dotenv
import os

load_dotenv(override=True)

zhipu_llm = ChatZhipuAI(
    model="glm-5.1",
    api_base=os.getenv("ZHIPUAI_BASE_URL"),  # 可选
    api_key=os.getenv("ZHIPUAI_API_KEY"),    # 可选
)
print(zhipu_llm.invoke("请介绍一下你自己"))

2.1.3 千问大模型（阿里云百炼）

官网：大模型服务平台百炼控制台

pip install dashscope   # ChatTongyi 依赖包

.env：

DASHSCOPE_API_KEY=<Your API Key>

注意：一般不要添加 DASHSCOPE_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1。该 URL 是百炼 OpenAI 兼容接口专用，而 ChatTongyi 底层基于专用 SDK，指定该 URL 会导致 ConnectionError（远程主机强迫关闭连接）。

import os
from langchain_community.chat_models import ChatTongyi
from dotenv import load_dotenv

load_dotenv(override=True)

tongyi_llm = ChatTongyi(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    model="qwen-plus",
)
print(tongyi_llm.invoke("请介绍一下你自己"))

2.2 兼容用法

许多平台 LangChain 未提供专用接口，且专用接口配置五花八门（如腾讯混元需要 APP_ID + SecretId + SecretKey，繁琐不友好）。由于大多数平台都支持 OpenAI API 接口规范，因此基本都可以通过 ChatOpenAI 集成。

from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os

load_dotenv(override=True)

# 通过 ChatOpenAI 连接 DeepSeek 模型
deepseek_llm2 = ChatOpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
    model="deepseek-v4-flash",
)
print(deepseek_llm2.invoke("1 + 1 = ？"))

同样可用 ChatOpenAI 连接智谱、千问等，只需替换 api_key、base_url、model：

zhipu_llm2 = ChatOpenAI(
    api_key=os.getenv("ZHIPUAI_API_KEY"),
    base_url=os.getenv("ZHIPUAI_BASE_URL"),
    model="glm-5.1",
)
tongyi_llm2 = ChatOpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_BASE_URL"),
    model="qwen-plus",
)

2.3 中转平台

受政策影响，国内无法直接调用国外顶尖闭源模型，可通过中转平台"曲线救国"。

2.3.1 OpenRouter

官网：OpenRouter

OpenRouter 是一个多模型 API 聚合平台，提供统一的 OpenAI 兼容接口，一个 API Key 即可调用 OpenAI、Claude、Gemini、DeepSeek、Qwen 等模型，适合模型对比、路由、Agent 开发和课程实验。知名度最高，但使用时需要魔法。

pip install langchain-openrouter

.env：

OPENROUTER_API_KEY=<YOUR_API_KEY>
OPENROUTER_API_BASE=https://openrouter.ai/api/v1

LangChain 为 OpenRouter 提供了专用集成 ChatOpenRouter，也可用 ChatOpenAI 调用：

from langchain_openrouter import ChatOpenRouter
from dotenv import load_dotenv
import os
​
load_dotenv(override=True)
​
model = ChatOpenRouter(
    model="deepseek/deepseek-v4-flash",
    api_key=os.getenv("OPENROUTER_API_KEY"),
    # base_url=os.getenv("OPENROUTER_API_BASE"),
)
print(model.invoke("一句话介绍下你自己"))

2.3.2 CloseAI

官网：CloseAI - 亚洲规模最大的企业级AI中转平台

CloseAI 是面向国内用户的中转平台，代理 OpenAI、Claude、Gemini 等模型，适合解决国内网络访问、支付和接口统一管理问题。LangChain 未提供专用集成，需通过 ChatOpenAI 兼容接口调用。

.env：

CLOSEAI_API_KEY=<YOUR_API_KEY>
CLOSEAI_BASE_URL=https://api.openai-proxy.org/v1

from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os
​
load_dotenv(override=True)
​
model = ChatOpenAI(
    # model="gpt-5-mini",
    model="deepseek-v4-flash",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL"),
)
print(model.invoke("欧盟都有哪些国家"))

---

3 模型初始化角度1：init_chat_model()

init_chat_model 是 LangChain 1.x 推出的统一初始化接口，只要是 LangChain 支持的模型都能处理，它会根据模型名称自动选择对应的模型类。

基本语法：

from langchain.chat_models import init_chat_model

model = init_chat_model(
    "provider:model_name",   # 提供商:模型名称
    api_key="your-api-key",  # API 密钥（可选，可从环境变量读取）
    temperature=0.7,         # 温度参数（可选）
    max_tokens=1000,         # 最大 token 数（可选）
    **kwargs                 # 其他模型特定参数
)

3.0 与直接使用 ChatXxx 的区别

维度	说明
统一接口	无需记忆每个提供商的不同初始化方式
易于切换	简化智能体系统中模型切换（只需改模型字符串）
简洁明了	减少样板代码
自动适配	内部根据模型标识自动选择驱动类（ChatOpenAI、ChatDeepSeek 等）

3.1 使用举例

举例1：调用 DeepSeek 官网模型（模型名 deepseek-v4-flash 会自动调用 ChatDeepSeek）

import os
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv

load_dotenv(override=True)

model = init_chat_model(
    model="deepseek:deepseek-v4-flash",   # provider:model 形式
    # model_provider="deepseek",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
)
print(model.invoke("你好，用一句话回答"))

举例2：调用阿里百炼模型（dashscope 未被官方纳入注册体系，需指定 model_provider="openai"，且要求服务是 OpenAI 兼容的）

.env：

DASHSCOPE_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
DASHSCOPE_API_KEY=<YOUR_API_KEY>

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

load_dotenv(override=True)

model = init_chat_model(
    model="qwen-plus",
    model_provider="openai",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_BASE_URL"),
)
print(model.invoke("你好，用一句话回答"))

举例3：调用 CloseAI 中转平台模型

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

load_dotenv(override=True)

model = init_chat_model(
    model="deepseek-v4-flash",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL"),
)
print(model.invoke("你好，用一句话回答"))

3.2 两个常见问题

问题1：model_provider 支持哪些 provider？

支持的 providers 包括：anthropic、anthropic_bedrock、azure_ai、azure_openai、bedrock、bedrock_converse、cohere、deepseek、fireworks、google_anthropic_vertex、google_genai、google_vertexai、groq、huggingface、ibm、mistralai、nvidia、ollama、openai、openrouter、perplexity、together、upstage、xai 等。

• model_provider="openai" → 自动加载 langchain-openai，底层调用 ChatOpenAI。

• model_provider="deepseek" → 自动加载 langchain-deepseek，底层调用 ChatDeepSeek。

• 阿里 dashscope 尚未被官方纳入统一注册体系，可设为 openai（要求模型服务是 OpenAI Compatible）。

问题2：未在 model 中指明提供商时，是否必须在 model_provider 中指明？

可在 model 参数中用前缀指定供应商（与模型名以冒号分隔），等价于用 model_provider 指定。若两处都未指明，LangChain 会按内置规则自动推断。但并非所有模型都支持自动推断，例如 qwen-plus 不支持自动推断，未指明供应商会报错。

3.3 小结：模型的创建（按平台与方式）

平台	可用的创建方式
DeepSeek 官网	ChatDeepSeek()、ChatOpenAI()、init_chat_model()
阿里云百炼	ChatTongyi()、ChatOpenAI()、init_chat_model()
OpenRouter	ChatOpenRouter()、ChatOpenAI()、init_chat_model()
CloseAI	ChatOpenAI()、init_chat_model()

---

4 模型初始化参数（常用版）

4.1 常用参数表

Model Class 与 init_chat_model 共同的常用参数（API 文档：https://docs.langchain.org.cn/oss/python/langchain/models#parameters ）：

参数	类型	说明	默认值
model	str	特定提供商的模型名称（必需），如 openai:gpt-4o、groq:gemma2-9b-it	无
model_provider	str	模型提供商名称	无
api_key	str	API 密钥；不提供则从环境变量读取（如 DEEPSEEK_API_KEY）	None
base_url	str	大模型供应商 API 请求地址	None
temperature	float	控制输出随机性，范围 0.0–2.0，越高越随机	0.7
max_tokens	int	限制模型输出的最大 token 数	None
timeout	float	超时时间（秒），超时则请求被取消	None
max_retries	int	请求失败时的最大重试次数	6

temperature 取值参考：

• 0.0–0.3：需一致性、准确性（数学计算、数据提取、分类、代码生成）

• 0.5–0.7：平衡创造性与一致性（聊天、问答）

• 0.8–1.5：创造性任务（写作、头脑风暴）

• 1.5–2.0：高度创造性（诗歌、故事创作）

4.2 temperature 实践对比

场景1：创意文案（temperature=1.5，多次结果差异大）

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

load_dotenv(override=True)

model = init_chat_model(
    model="deepseek-v3.2",
    model_provider="openai",
    temperature=1.5,
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL"),
)
for i in range(3):
    response = model.invoke("帮我写一首描述春天的七言绝句诗")
    print(response.content)

场景2：严格结构化数据提取（temperature=0，确保字段准确）

model = init_chat_model(
    model="deepseek-v3.2",
    model_provider="openai",
    temperature=0,
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL"),
)
response = model.invoke(
    "张三，男，30岁，拥有8年编程开发经验，目前在某互联网大厂担任技术专家。"
    "帮我从上文中提取数据，返回JSON格式"
)
print(response.content)

返回示例：

{
  "name": "张三",
  "gender": "男",
  "age": 30,
  "programming_experience_years": 8,
  "current_position": "技术专家",
  "current_company_type": "互联网大厂"
}

4.3 max_tokens 实践

设置 max_tokens=15 会被截断，finish_reason 变为 length：

model = init_chat_model(
    model="gpt-5.4-mini",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL"),
    max_tokens=15,
)
response = model.invoke("请用中文详细介绍什么是AI")
print(response)

4.4 Token 是什么

• 基本单位：大模型通过分词器（Tokenizer）将文本拆分后的最小语义单元，相当于自然语言中的词或字。不同模型采用不同分词算法（如 BPE、WordPiece），同一段文本在不同模型中的 Token 数可能不同。

• 收费依据：大模型通常以 Token 数量为计量（收费）依据。

• 经验值：1 个中文 Token ≈ 1–1.8 个汉字；1 个英文 Token ≈ 3–4 个字符。

• 可视化工具：OpenAI https://platform.openai.com/tokenizer ；百度智能云 服务与支持

---

5 模型初始化角度3：本地模型（Ollama）

LangChain 也支持 Ollama、vLLM 等框架启动的本地大模型，本节以 Ollama 为例。

5.1 Ollama 介绍与安装

Ollama 是 GitHub 上的开源项目，定位为本地运行大模型的集成框架，可自动化完成 Qwen、DeepSeek 等主流大模型的下载、启动与运行推理。官网：https://ollama.com

支持 Mac、Linux、Windows 跨平台。Linux 安装：

curl -fsSL https://ollama.com/install.sh | sh

Windows 安装时可指定目录（在 cmd 中执行安装包）：

OllamaSetup.exe /DIR=F:\common_tools\Ollama

5.2 模型下载与常用命令

下载/运行模型：

ollama run deepseek-r1:1.5b

Ollama 常用命令：

命令	说明
ollama pull llama3	下载指定模型
ollama run llama3	启动并进入交互对话
ollama list	列出本机已下载的所有模型
ollama rm llama3	删除模型以节省磁盘
ollama cp llama3 my-llama3	本地复制/重命名模型
ollama show llama3	查看模型详细信息（参数、大小等）
ollama create my-model -f Modelfile	用自定义 Modelfile 构建新模型
ollama serve	启动后台服务，供 API 调用
ollama ps	查看当前正在运行的模型进程
ollama stop llama3	停止正在运行的模型
ollama --version	查看安装的 ollama 版本

5.3 LangChain 调用本地模型

pip install -qU langchain-ollama
pip install -U ollama

方式1：使用 ChatOllama

from langchain_ollama import ChatOllama

ollama_llm = ChatOllama(
    model="deepseek-r1:1.5b",
    base_url="http://192.168.1.106:11434"
    # 若 Ollama 在本地默认端口运行，可省略，或用 http://localhost:11434
)
question = "你好，请你介绍一下你自己。"
result = ollama_llm.invoke(question)
print(result)

方式2：使用 init_chat_model

from langchain.chat_models import init_chat_model

ollama_llm = init_chat_model(
    model="deepseek-r1:1.5b",
    model_provider="ollama",
    # base_url="http://192.168.1.106:11434",
)
question = "你好，请你介绍一下你自己。"
result = ollama_llm.invoke(question)
print(result)

---

6 模型的调用

在 LangChain 中，调用（Invocation） 是通过特定方法触发大模型生成输出的过程。核心方法及其异步版本：

方法	模式	适用场景
invoke()	阻塞式，一次性返回	问答、批处理、无需实时反馈
ainvoke()	非阻塞，提高吞吐量	高并发 Web、IO 密集型
stream()	流式输出，实时返回每个 token	聊天机器人、长文本生成、交互应用
astream()	非阻塞流式	高并发、IO 密集型
batch()	批量处理多个输入	高并发、大量请求
abatch()	非阻塞批量	高并发、IO 密集型

6.1 invoke()

invoke() 是最核心的方法，阻塞式工作：程序等待模型完全生成整个响应后，一次性返回。

基本语法：

response = model.invoke(input, config=None)

参数	类型	说明	必需
input	str / list[dict] / list[Message] 等	发送给模型的内容	必需
config	dict	高级配置（回调、元数据、标签等）	可选（None）

6.1.1 输入参数详解

invoke 支持三种输入形式：

(1) 文本输入（最简单）

适用快速测试；缺点是无法设置 system prompt，无法传递对话历史。

prompt = "翻译成英文：你好世界"
response = model.invoke(prompt)
print(response)

(2) 字典列表（推荐，最灵活）

一条消息含 role（角色）、content（内容）。可设置系统提示、表达多轮对话历史、JSON 兼容、易序列化，生产环境推荐。

messages = [
    {"role": "system", "content": "系统提示"},
    {"role": "user", "content": "用户消息"},
    {"role": "assistant", "content": "AI回复"},  # 可选，用于对话历史
    {"role": "user", "content": "继续提问"},
]

角色说明：

角色	英文	作用	示例
system	System	设定 AI 的行为、角色、规则	"你是一个专业的 Python 导师"
user	Human/User	用户的输入/问题	"什么是装饰器？"
assistant	AI/Assistant	AI 的历史回复（对话上下文）	"装饰器是一种设计模式..."

"user" 与 "human" 有时可互换，但遵循所选提供商（如 OpenAI）的惯例使用 "user" 最稳妥。

多轮对话（带历史）： 不传递历史 AI 会"失忆"，传递历史才能记住上下文。

messages = [
    {"role": "system", "content": "你是一个专业的数学老师。"},
    {"role": "user", "content": "2 + 3 * 2 = ？"},
    {"role": "assistant", "content": "8"},
    {"role": "user", "content": "我刚才问了什么问题？"},
]
response = model.invoke(messages)
print(f"AI的回复：{response.content}")

对话记忆的通用做法——把上一轮的 AI 回复追加进消息列表：

conversation = [
    {"role": "system", "content": "你是一个非常友好的AI助手"},
    {"role": "user", "content": "你好，我叫小明"},
]
response1 = model.invoke(conversation)
print(f"AI的回复1：{response1.content}")

# 添加记忆
conversation.append({"role": "assistant", "content": response1.content})
conversation.append({"role": "user", "content": "我叫什么名字？"})
response2 = model.invoke(conversation)
print(f"AI的回复2：{response2.content}")

(3) 消息对象列表

使用内置消息类（适合需要类型检查、IDE 自动补全的大型项目；缺点是代码较长、不如字典简洁、难以序列化）。

消息类	对应字典格式	作用
SystemMessage	{"role": "system", ...}	系统提示
HumanMessage	{"role": "user", ...}	用户输入
AIMessage	{"role": "assistant", ...}	AI 回复

from langchain_core.messages import SystemMessage, AIMessage, HumanMessage
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

load_dotenv(override=True)
model = init_chat_model(
    model="openai:gpt-5.4-mini",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL"),
)

messages = [
    SystemMessage("你是一个专业的数学老师。"),
    HumanMessage("2 + 3 * 2 = ？"),
    AIMessage("8"),
    HumanMessage("我刚才问什么问题了？"),
]
response = model.invoke(messages)
print(f"AI的回复：{response.content}")

6.1.2 返回值详解

invoke 返回一个 AIMessage 对象。

response = model.invoke([HumanMessage("2 + 3 * 2 = ？")])
print(type(response))  # <class 'langchain_core.messages.ai.AIMessage'>

AIMessage 包含的核心信息：

字段	说明
content	模型生成的文本回答（核心输出）
id	LangChain 内部的唯一运行标识（Run ID）
additional_kwargs	供应商特定额外参数；refusal 表示拒绝回答原因（None 为正常）
response_metadata	API 返回的原始详细信息
tool_calls / invalid_tool_calls	工具调用列表 / 格式错误的工具调用

响应元数据关键字段：

• Token 消耗（决定费用）：prompt_tokens/input_tokens（输入）、completion_tokens/output_tokens（输出）、total_tokens（总）；reasoning_tokens（推理模型思考消耗）；cached_tokens（缓存命中，费用更低）。

• 模型信息：model_name（具体版本）、model_provider（供应商）、system_fingerprint（后端配置指纹）。

• finish_reason：stop（正常结束）、length（达最大 Token 被截断）。

• 延迟性能（Latency Checkpoint，单位 ms）：total_duration_ms（总耗时）、user_visible_ttft_ms（首字到达，体感快慢关键）、engine_ttft_ms（引擎首字）、service_tbt_ms（Token 间间隔）、pre_inference_ms（推理前处理）。

• 统一消耗元数据（usage_metadata）：LangChain 标准化后的 input_tokens/output_tokens/total_tokens 等。

访问这些信息的示例：

response = model.invoke("用一句话解释什么是 AI")

print("AI 回复:", response.content)
metadata = response.response_metadata
print(f"使用的模型: {metadata['model_name']}")
print(f"结束原因: {metadata['finish_reason']}")

usage = metadata.get('token_usage', {})
print(f"输入 tokens: {usage.get('prompt_tokens')}")
print(f"输出 tokens: {usage.get('completion_tokens')}")
print(f"总计 tokens: {usage.get('total_tokens')}")

6.2 流式调用 stream()

invoke 在输出完成后一次性返回，长文本用户体验差；stream 实时返回响应片段，返回一个迭代器，循环即可逐块处理。

注意：流式输出依赖模型供应商对流式的支持。

for chunk in model.stream("写一首七言律诗，总结大模型的发展"):
    print(chunk.content, end="", flush=True)  # 逐 token 输出

6.3 批量调用 batch()

batch() 一次性发送一组请求，模型在后台并行处理，返回结果列表，相比逐个 invoke 能大幅减少网络往返和等待时间，显著降低成本。适用场景：文档摘要、批量问答、数据预处理、多样本分类。

messages = [
    "你好，你是谁？",
    "2 + 3 * 5 = ?",
    "中国首都在哪里？",
]
responses = model.batch(messages)
for response in responses:
    print(response)

batch_as_completed()： 每完成一个请求即 yield，结果可能乱序；返回元组 (index, AIMessage)，可按 index 重新排序。

性能对比（4 条翻译任务）： batch() 约 1.91 秒，循环 invoke() 约 3.87 秒，批量调用节省约 50.7%。

import time

inputs = ["翻译成英文：春天来了", "翻译成英文：夏天很热",
          "翻译成英文：秋天落叶", "翻译成英文：冬天下雪"]

start = time.time()
responses = model.batch(inputs)
batch_time = time.time() - start

start = time.time()
loop_responses = [model.invoke(inp) for inp in inputs]
loop_time = time.time() - start

print(f"批量调用节省: {((loop_time - batch_time) / loop_time * 100):.1f}%")

6.4 异步调用（ainvoke / astream / abatch）

• 同步（sync）：发起任务后需等待完成，当前执行流被"阻塞"。

• 异步（async）：发起任务后不必等待，可继续执行其他任务，执行流不被"阻塞"。

异步方法的优势：避免阻塞主线程、优化资源利用（减少空闲等待）。

举例1：ainvoke()（在 .py 文件中执行，非 Jupyter）

import asyncio
import os
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import time

load_dotenv(override=True)
model = init_chat_model(
    model="openai:gpt-5.4-mini",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL"),
)

async def demo_async_invoke():
    print("=== 演示：ainvoke 的异步（非阻塞）效果 ===")
    start_time = time.perf_counter()
    # 1. 创建后台任务
    async_task = asyncio.create_task(model.ainvoke("用一句话解释人工智能。"))
    # 2. 并行执行其他任务（异步等待，释放控制权）
    for i in range(3):
        await asyncio.sleep(1)
        print(f">>> 正在执行第{i + 1}个任务... (已耗时 {time.perf_counter() - start_time:.2f}s)")
    # 3. 获取模型结果
    response = await async_task
    print(f">>> 模型返回: {response.content}")

async def main():
    await demo_async_invoke()

if __name__ == "__main__":
    asyncio.run(main())

要点：用 asyncio.create_task() 让协程立即在后台执行，用 asyncio.sleep 而非 time.sleep 释放事件循环去处理网络 IO。astream()、abatch() 用法类似。

6.5 如何处理 API 调用失败

使用 try-except 捕获异常：

try:
    response = model.invoke("Hello")
    print(response.content)
except ValueError as e:
    print(f"配置错误: {e}")
except ConnectionError as e:
    print(f"网络错误: {e}")
except Exception as e:
    print(f"未知错误: {e}")

---

7 拓展内容

7.1 美化模型输出响应

方法1：pretty_print()

response = model.invoke(conversation)
response.pretty_print()

输出形如：

================================== Ai Message ==================================
我是小王，你是老王。

方法2：rich 库（终端色彩鲜明、排版优雅）

from rich import print as rprint
response = model.invoke(conversation)
rprint(response)

7.2 模型配置信息 profile

LangChain 1.1+ 可通过 profile 属性查看模型的能力画像。是否存在取决于 LangChain 在集成厂商时是否声明了画像。

• DeepSeek 官方模型、CloseAI 平台 gpt 模型：model.profile 为 {}（未声明）。

• OpenRouter 平台模型：已声明画像，例如 gpt-4o-mini：

from langchain_openrouter import ChatOpenRouter
from dotenv import load_dotenv
from rich import print as rprint

load_dotenv(override=True)
model = ChatOpenRouter(
    model="openai/gpt-4o-mini",
    temperature=0.7,
    timeout=30,
    max_tokens=1000,
    max_retries=6,
)
rprint(model.profile)

输出（更换 model ID 即会替换为对应画像）：

{
 'max_input_tokens': 128000,
 'max_output_tokens': 16384,
 'text_inputs': True, 'image_inputs': True,
 'tool_calling': True, 'structured_output': True
 ...
}

注意：查看 profile 只需 OpenRouter 的 API_KEY，不会真正发送请求，不必充值。

7.3 模型初始化参数（完整版）

官方文档/源码注释未给出完整参数列表。可通过类属性 model_fields 获取（参数由类自身定义或从父类 BaseChatModel 继承）。

from langchain_deepseek import ChatDeepSeek
print(ChatDeepSeek.model_fields.keys())

参数构成（以 ChatDeepSeek 为例）

(1) 客户端与连接参数（Networking） —— 决定"怎么连服务端"：api_key/openai_api_key、api_base/openai_api_base、request_timeout、max_retries、http_client/http_async_client、openai_proxy、default_headers/default_query。

(2) 模型推理参数（Model Inference） —— 直接传给模型 API，决定生成质量与风格：model_name、temperature、top_p、max_tokens、stop、streaming、n、reasoning/reasoning_effort（DeepSeek R1 特色，控制思考链深度）、presence_penalty/frequency_penalty、store、logit_bias。

(3) LangChain 框架通用参数（由 BaseChatModel 定义，所有 ChatXxx 都具备）：name、verbose、callbacks、tags/metadata、cache、rate_limiter。

(4) 高级与特定扩展参数：

• 透传参数 model_kwargs：存放 OpenAI Compatible API 支持但 LangChain 未直接列出的字段。

• 透传参数 extra_body：存放模型厂商基于 OpenAI 协议扩展的字段（如 DeepSeek 的 thinking 控制思考模式）。

model_kwargs 示例（透传 tools 字段）：

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
from rich import print as rprint
load_dotenv(override=True)

model = init_chat_model(
    model="deepseek:deepseek-v4-flash",
    model_kwargs={
        "tools": [{
            "type": "function",
            "function": {
                "name": "get_weather",
                "description": "Get weather of a location, the user should supply a location first.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "location": {"type": "string", "description": "The city and state, e.g. San Francisco, CA"}
                    },
                    "required": ["location"],
                },
            },
        }]
    },
)
response = model.invoke("你好，今天北京的天气如何")
rprint(response)  # 输出包含 tool_calls，说明工具被正确识别

extra_body 示例（开启 DeepSeek 思考模式）：

model = init_chat_model(
    model="deepseek:deepseek-v4-flash",
    extra_body={"thinking": {"type": "enabled"}},  # "disabled" 则关闭
)
response = model.invoke("你好，一句话回答")
rprint(response)  # 包含 additional_kwargs.reasoning_content，说明启用思考模式

经验：记住常见参数及用法即可；需要精细控制输出时，查阅 OpenAI 与特定供应商文档，通过 model_kwargs 或 extra_body 传递。

7.4 模型调用中的 config 参数

调用 invoke()、ainvoke()、stream()、batch() 等方法时可传入 config 参数，用于在运行时动态配置和控制模型行为，无需在初始化时固定所有参数。

config 支持的参数：

配置项	类型	描述
run_name	str	为当前运行设置可读名称（便于在 LangSmith 中定位）
tags	List[str]	标签，用于分类和过滤
callbacks	List[BaseCallbackHandler]	回调处理器，可与 LangSmith 集成
metadata	Dict[str, Any]	附加任意键值对元数据（如 {"user_id": "123"}）
max_concurrency	int	限制最大并发运行数，实现简单速率限制
recursion_limit	int	限制递归调用最大深度（防止 Agent 无限递归）
configurable	Dict[str, Any]	万能字典，传递其他可配置参数

要点：

• run_name、tags、callbacks 主要用于 LangSmith 追踪/筛选/调试。

• configurable 与初始化参数的区别：初始化参数是模型实例的默认设置；运行时 config 是单次调用的特定设置，优先级更高。

若要用 configurable 覆盖默认参数，必须在 init_chat_model 初始化时通过 configurable_fields 声明哪些参数运行时可替换。

举例1：通过 config 覆盖模型参数（configurable）

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os
from rich import print as rprint
load_dotenv(override=True)

model = init_chat_model(
    model="deepseek-v4-flash",
    model_provider="deepseek",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
    temperature=0.2,
    max_tokens=500,
    # 声明运行时可替换的参数
    configurable_fields=("model", "model_provider", "temperature", "max_tokens"),
)

config = {
    "run_name": "joke_generation",
    "tags": ["tag1", "tag2"],
    "metadata": {"user_id": "123"},
    "configurable": {
        "model": "deepseek-v4-pro",
        "model_provider": "openai",
        "temperature": 0.7,
        "max_tokens": 1000,
    },
}

response = model.invoke("1 + 2 = ？", config=config)
rprint(response)  # response_metadata 中 model_name 变为 deepseek-v4-pro，证明覆盖生效

举例2：批量调用时限制并发

model.batch(
    large_list_of_inputs,
    config={"max_concurrency": 5}  # 限制最大并发数为 5
)

第04章 消息与提示词模板

1、消息(Message)

1.1 认识消息

大模型没有记忆,输出只与输入模型的上下文有关。多数大模型 API 服务端也不维护会话历史,是“无状态”的。因此若要应用“记住”对话历史,需在程序中自行维护消息列表。

在 LangChain 中,Message(消息)是模型交互的最基本单元,既代表模型的输入(Input),也代表输出(Output)。每一轮对话由一条或多条 Message 构成,每个 Message 除文字内容外,还携带描述上下文状态的元信息(metadata),用于保持对话一致性与可追踪性(理解“谁在说话”“说了什么”“属于哪一轮对话”)。

LangChain 1.0 提供了跨模型统一的 Message 标准,无论 OpenAI、Anthropic、Gemini 还是本地模型都保持一致行为:

• 兼容性强:不同模型的消息格式自动对齐

• 可扩展性高:方便添加多模态内容或自定义字段

• 可追踪性好:为 LangSmith 等调试工具提供一致的上下文数据结构

1.2 消息的内部结构

Message 对象包含三种字段:

• Role:消息所属的角色/类型,如 system、user、assistant

• Content:消息内容

• Metadata(可选):存储额外信息,如消息 ID、响应时间、token 消耗量、消息标签等

1.3 消息的类型

LangChain 通过 role 区分多种消息类型,常用四种:

1. 系统消息(SystemMessage):即系统提示词,用于在对话开始时为模型设定角色、行为准则和上下文背景,相当于给 AI 一份“工作说明书”。

2. 用户消息(HumanMessage):即用户提示词,表示用户的一次输入,可包含文本或复杂多模态内容(图片、音频、文档)。

3. 助手消息(AIMessage):代表模型回复,包括生成文本、工具调用、元数据等。

4. 工具调用消息(ToolMessage):匹配工具调用结果,返回给模型让其基于结果继续生成。

为什么使用不同的消息类型?

• 明确角色:清晰区分系统提示、用户输入和 AI 回复

• 控制行为:通过 SystemMessage 精确控制 AI 行为

• 对话历史:构建完整的多轮对话上下文

• 调试友好:更容易追踪和调试对话流程

四种类型对应的 JSON 结构示例:

{"role": "system", "content": "你是个精通编程的软件架构师"}
{"role": "user", "content": "你好啊~"}
{"role": "assistant", "content": "我也很高兴认识你"}

{
    "role": "assistant",
    "content": "",
    "tool_calls": [{
        "name": "get_weather",
        "args": {"location": "北京"},
        "id": "call_00_nUD2NC9QRN5Cg1GaoIkBJQ4s"
    }]
}

{"role": "tool", "content": "今天天气很好", "tool_call_id": "call_00_nUD2NC9QRN5Cg1GaoIkBJQ4s"}

1.4 消息格式

LangChain 支持两种消息格式:JSON 格式(字典)与对象格式(XxxMessage 类实例)。

角色	字典格式	对象格式	用途	示例
System	{"role": "system", ...}	SystemMessage(...)	设定 AI 的行为、角色、规则	"你是一个专业的数学老师"
User	{"role": "user", ...}	HumanMessage(...)	用户输入	"什么是微积分？"
Assistant	{"role": "assistant", ...}	AIMessage(...)	AI 的回复	"微积分是研究变化率的数学分支..."
Tool	{"role": "tool", ...}	ToolMessage(...)	工具执行的结果	"今天北京天气晴朗,万里无云"

消息对象统一导入:

from langchain_core.messages import (
    HumanMessage,    # 用户消息
    AIMessage,       # AI 消息
    SystemMessage,   # 系统消息
    ToolMessage      # 工具返回消息
)

举例1:JSON 格式

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

load_dotenv(override=True)
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")

model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=CLOSEAI_API_KEY,
    base_url=CLOSEAI_BASE_URL
)

# 通过 JSON 初始化
messages = [
    {"role": "system", "content": "你是一个善于给出通俗易懂解释的AI助手"},
    {"role": "user", "content": "你好"},
    {"role": "assistant", "content": "你好！我能帮你什么？"},
    {"role": "user", "content": "什么是机器学习"}
]
response = model.invoke(messages)
print(response.content)

举例2:对象格式

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

load_dotenv(override=True)
model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)

# 通过对象初始化
messages = [
    SystemMessage("你是一个善于给出通俗易懂解释的AI助手"),
    HumanMessage("你好"),
    AIMessage("你好！我能帮你什么？"),
    HumanMessage("什么是机器学习"),
]
response = model.invoke(messages)
print(response.content)

1.5 消息对象字段说明

仅说明常用字段,完整列表需查阅官方手册或源码。

1.5.1 SystemMessage 参数

content 为消息内容,字段名可省略,等价写法:

SystemMessage("你是个善解人意的助手")
SystemMessage(content="你是个善解人意的助手")

1.5.2 HumanMessage 参数

content 可省略字段名;metadata 为自定义元数据。name 与 id 属于元数据字段,用于在消息类型相同时区分消息。但不是所有模型都支持,取决于模型供应商:

• OpenAI 的 API 手册说明 HumanMessage 支持 name 字段

• DeepSeek 官方文档明确支持 name,但实测模型无法识别

HumanMessage(
    content="Hello!",
    name="alice",   # 可选,用户名
    id="msg_123",   # 可选,message 的 ID
)

name 字段多人对话场景示例(通过 name 区分不同发言者):

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

load_dotenv(override=True)
model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)

messages = [
    SystemMessage("你是一个信息抽取器。你会收到多条来自不同发言者的 user 消息。每条消息可能带有 name 字段。你的任务是：严格根据每条消息的 name 提取发言者及其观点，并输出 JSON。禁止使用“第一个人/第二个人”这种相对称呼。若某条消息没有 name，则输出 unknown。输出格式：{\"speakers\":[{\"name\":\"...\",\"claim\":\"...\"}]}"),
    HumanMessage(content="我认为 1+1=2", name="Bob"),
    HumanMessage(content="我认为 1+1>2", name="Tom"),
    HumanMessage(content="请列出谁说了什么，不要判断对错。", name="audience"),
]
response = model.invoke(messages)
print(response.content)

输出(模型正确加载了 name 传递的信息):

{"speakers":[{"name":"Bob","claim":"我认为 1+1=2"},{"name":"Tom","claim":"我认为 1+1>2"},{"name":"audience","claim":"请列出谁说了什么，不要判断对错。"}]}

拓展:使用 ChatOpenRouter 调用时未能正确传递 name,输出均为 unknown:

from langchain_openrouter import ChatOpenRouter
from dotenv import load_dotenv
import os

load_dotenv(override=True)
model = ChatOpenRouter(
    model="openai/gpt-4o-mini",
    api_key=os.getenv("OPENROUTER_API_KEY"),
    base_url=os.getenv("OPENROUTER_API_BASE"),
)
# 其余 messages 构造同上

1.5.3 AIMessage 参数

• content:模型输出原始内容,字段名可省略(AIMessage("你好~") 等价 AIMessage(content="你好~"))

• response_metadata:AIMessage 特有属性,LLM 响应中附加元数据,如 token 使用量

• tool_calls:AIMessage 特有属性,工具调用信息,无调用则为空。结构为 ToolCall 列表(每个为字典)

• usage_metadata:用量信息

tool_calls 结构示例:

tool_calls=[{
    'name': 'get_weather',
    'args': {'city': '杭州'},
    'id': 'call_00_gIXYOD1Q1OkEXmdDBqXR1578',
    'type': 'tool_call'
}]

AIMessage 直接给出答案:

AIMessage(content="北京今天晴天，温度 15°C")

AIMessage 调用工具:

AIMessage(
    content="",
    tool_calls=[{
        'name': 'get_weather',
        'args': {'city': '北京'},
        'id': 'call_xxx'
    }]
)

实际调用并打印 response 可观察 content、response_metadata(含 token_usage、延迟信息等)、tool_calls、usage_metadata 等字段;response.usage_metadata 单独取用量:

from rich import print as rprint
rprint(response.usage_metadata)

{
 'input_tokens': 34,
 'output_tokens': 118,
 'total_tokens': 152,
 'input_token_details': {'audio': 0, 'cache_read': 0},
 'output_token_details': {'audio': 0, 'reasoning': 0}
}

1.5.4 ToolMessage 参数(拓展)

• content:内容

• name:工具名称

• tool_call_id:工具调用唯一 ID,必须与对应 AIMessage 的 tool_calls 中的 id 一致

ToolMessage(
    content="<工具输出>",
    name="get_weather",
    tool_call_id="call_00_nUD2NC9QRN5Cg1GaoIkBJQ4s"
)

工具调用完整示例(JSON 格式):

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

load_dotenv(override=True)
model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)

def get_weather(city: str) -> str:
    return "不错哦~"

# 模拟模型绑定工具
model_with_tools = model.bind_tools([get_weather])

ai_message = {
    "role": "assistant",
    "content": "",
    "tool_calls": [{
        "name": "get_weather",
        "args": {"location": "北京"},
        "id": "call_00_nUD2NC9QRN5Cg1GaoIkBJQ4s"
    }]
}
tool_message = {
    "role": "tool",
    "content": "今天北京天气晴朗，万里无云~",
    "tool_call_id": "call_00_nUD2NC9QRN5Cg1GaoIkBJQ4s"
}
messages = [
    {"role": "user", "content": "北京天气如何"},
    ai_message,
    tool_message
]
response = model.invoke(messages)
print(response)

对象格式等价写法(用 AIMessage/ToolMessage/HumanMessage 替换字典):

from langchain_core.messages import AIMessage, ToolMessage, HumanMessage

ai_message = AIMessage(
    content=[],
    tool_calls=[{
        "name": "get_weather",
        "args": {"location": "北京"},
        "id": "call_00_nUD2NC9QRN5Cg1GaoIkBJQ4s"
    }]
)
tool_message = ToolMessage(
    content="今天北京天气晴朗，万里无云~",
    tool_call_id="call_00_nUD2NC9QRN5Cg1GaoIkBJQ4s"
)
messages = [
    HumanMessage(content="北京天气如何"),
    ai_message,
    tool_message
]
response = model.invoke(messages)
print(response)

1.6 实战

1.6.1 对话历史管理

关键规则:每次调用必须传递完整的对话历史! 即每次对话都要在原有消息列表中追加新消息,不可重新创建新列表。

各轮的消息列表累积过程:

第 1 轮:[system, user] → AI回复 → 保存回复
第 2 轮:[system, user, assistant, user] → AI回复 → 保存回复
第 3 轮:[system, user, assistant, user, assistant, user] → AI回复

错误示例(丢失历史):

# ❌ 不传历史,模型无记忆
response1 = model.invoke("我叫张三")
response2 = model.invoke("我叫什么？")   # AI 不记得!

# ❌ 重新创建列表,丢失历史
conversation = [{"role": "user", "content": "问题1"}]
response1 = model.invoke(conversation)
conversation = [{"role": "user", "content": "问题2"}]
response2 = model.invoke(conversation)

# ❌ 忘记保存上一轮 AI 回复
conversation = []
conversation.append({"role": "user", "content": "问题1"})
response1 = model.invoke(conversation)
# 忘记保存 response1.content!
conversation.append({"role": "user", "content": "问题2"})
response2 = model.invoke(conversation)   # AI 不知道之前的回答

正确做法:

conversation = []
# 第一次
conversation.append({"role": "user", "content": "我叫张三"})
response1 = model.invoke(conversation)
# 关键:保存 AI 回复
conversation.append({"role": "assistant", "content": response1.content})
# 第二次(传递完整历史)
conversation.append({"role": "user", "content": "我叫什么？"})
response2 = model.invoke(conversation)   # AI 记得!

1.6.2 对话历史优化

问题:对话历史越来越长,消耗大量 tokens 和成本。

解决方案:只保留最近 N 轮对话。

• 总是保留 system 消息(定义角色)

• 只保留最近 N 轮对话,丢弃更早的历史

def keep_recent_messages(messages, max_pairs=3):
    """
    保留最近的 N 轮对话
    max_pairs: 保留的对话轮数（每轮 = user + assistant）
    """
    # 分离 system 和对话
    system_msgs = [m for m in messages if m.get("role") == "system"]
    conversation_msgs = [m for m in messages if m.get("role") != "system"]
    # 只保留最近的
    recent_msgs = conversation_msgs[-(max_pairs * 2):]
    # 返回：system + 最近对话
    return system_msgs + recent_msgs

测试:

long_conversation = [
    {"role": "system", "content": "你是 Python 导师"}
]
# 第 1~3 轮依次追加 user 与 assistant
long_conversation.append({"role": "user", "content": "什么是列表？用一句解释"})
r1 = model.invoke(long_conversation)
long_conversation.append({"role": "assistant", "content": r1.content})
long_conversation.append({"role": "user", "content": "列表和元组有什么区别？用一句解释"})
r2 = model.invoke(long_conversation)
long_conversation.append({"role": "assistant", "content": r2.content})
long_conversation.append({"role": "user", "content": "什么是字典呢？用一句解释"})
r3 = model.invoke(long_conversation)
long_conversation.append({"role": "assistant", "content": r3.content})

print(f"原始消息数: {len(long_conversation)}")
# 优化:只保留最近 2 轮
optimized = keep_recent_messages(long_conversation, max_pairs=2)
print(f"优化后消息数: {len(optimized)}")
# 添加新的用户问题
optimized.append({"role": "user", "content": "我第一个问题问的是什么？"})
response = model.invoke(optimized)
print(f"\nAI 回复: {response.content}")

输出:原始消息数 7,优化后消息数 5(system + 最近 2 轮)。

1.6.3 多轮对话聊天机器人

基于模型初始化、流式响应与消息列表拼接创建多轮聊天机器人:

from langchain.chat_models import init_chat_model
import os
from dotenv import load_dotenv

load_dotenv(override=True)

# 1. 基础配置
MODEL_NAME = "gpt-5.4-mini"
MAX_PAIRS_HISTORY = 10
EXIT_WORD = "quit"

# 2. 初始化模型
model = init_chat_model(
    model=MODEL_NAME,
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)

# 3. 初始化消息列表
messages = [
    {
        "role": "system",
        "content": "你是小谷姐姐，尚硅谷教育的数字员工，也是一名耐心、友好的智能助手。我会用自然、清晰的方式回答用户问题。"
    }
]

# 4. 启动提示
print(f"✨ 请输入问题，输入 {EXIT_WORD} 结束对话\n")

# 5. 多轮对话主循环
i = 1
while True:
    print("\n", "=" * 10, f'-> 第 {i} 轮对话开始 <-', "=" * 10, "\n")
    user_input = input("🙋 请输入：")
    if user_input.lower() == EXIT_WORD:
        print("🌙 对话已结束，欢迎下次再来！")
        break
    # 追加用户消息
    messages.append({"role": "user", "content": user_input})
    # 流式输出模型回复
    print("🧚 小谷姐姐：", end="", flush=True)
    reply_content = ""
    # 优化历史记忆
    memory_messages = keep_recent_messages(messages, max_pairs=MAX_PAIRS_HISTORY)
    # 控制发送给模型的消息长度
    for chunk in model.stream(memory_messages):
        if chunk.content:
            print(chunk.content, end="", flush=True)
            reply_content += chunk.content
    print("\n", "=" * 10, f'-> 第 {i} 轮对话结束 <-', "=" * 10, "\n")
    i += 1
    # 追加 AI 回复
    messages.append({"role": "assistant", "content": reply_content})

其中 keep_recent_messages() 定义见 1.6.2。

1.7 拓展:消息属性 content 与 content_blocks

1.7.1 content

content 是弱类型的,支持字符串和列表(列表元素通常为字典)。

• 存储字符串:纯文本直接传字符串,可省略参数名。

• 存储字典列表:多模态内容需字典列表形式,字典遵循供应商 API 规范(如 OpenAI)。

from langchain_core.messages import HumanMessage

msg1 = HumanMessage(content="你好啊")
msg2 = HumanMessage("你好啊")   # 等价

图片理解示例(将本地图片转 Base64 Data URI):

import base64
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from dotenv import load_dotenv
import os

load_dotenv(override=True)
model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)

def encode_image(img_path, img_type='jpeg'):
    """将本地图片转换成 Base64 编码的 Data URI 字符串"""
    with open(img_path, "rb") as img_file:
        return f"data:image/{img_type};base64,{base64.b64encode(img_file.read()).decode('utf-8')}"

base64_image = encode_image("image_test.png")
response = model.invoke([
    HumanMessage(content=[
        {'type': 'text', 'text': '这张图里有什么？'},
        {'type': 'image_url', "image_url": base64_image},
    ])
])
print(response.content)

1.7.2 content_blocks

content_blocks 是 LangChain 1.x 中 BaseMessage 的重大升级,核心目标是提供跨模型供应商、标准化的多模态数据结构。过去处理图片/音频/思维链时各供应商格式各异,需大量适配代码;content_blocks 统一了这种混乱。

LangChain 1.2 中 content 仍保留(向前兼容),但新增 content_blocks 可将 content 解析为标准、类型安全的表示。

• 数据结构:list[TypedDict]

• 统一格式:每个 block 都有 type 字段区分内容类型

• 支持类型:text(文本)、image(图片)、audio(音频)、video(视频)、tool_call(工具调用)、reasoning(推理/思维链)

① 输入格式化:对复杂对话(带图片或工具结果),建议用 content_blocks 列表构建 HumanMessage/AIMessage,实现一套代码无缝切换不同厂商模型。

OpenAI 模型示例:

from langchain.messages import HumanMessage
import os
from dotenv import load_dotenv
import base64

load_dotenv(override=True)
model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)

def encode_image(img_path):
    with open(img_path, "rb") as img_file:
        return base64.b64encode(img_file.read()).decode("utf-8")

base64_image = encode_image("image_test.png")
response = model.invoke([
    HumanMessage(
        content_blocks=[
            {'type': 'text', 'text': '这张图里有什么？'},
            {
                'type': 'image',
                'base64': base64_image,
                'mime_type': 'image/png',
            }
        ]
    )
])
print(response.content)

切换到 Anthropic 模型(同样写法,仅改 model):

model = init_chat_model(
    model="claude-haiku-4-5",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)

② 输出格式化:content_blocks 也可用于输出格式化。例如 DeepSeek 的输出包含思考内容,位于 additional_kwargs 的 reasoning_content 字段下,不同模型格式不同,切换模型都可能需改代码。content_blocks 提供了统一输出格式。

注意:content_blocks 是懒加载的,调用时才解析。当需要获取“思维链”或“引用(Citations)”信息时,应优先检查 response.content_blocks 而不是 response.content。

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv

load_dotenv(override=True)
model = init_chat_model(
    model="deepseek:deepseek-v4-flash",
    extra_body={"thinking": {"type": "enabled"}},
)
response = model.invoke("你好，一句话回答")
print('=' * 20, '-> response <-', '=' * 20)
print(response)
print('=' * 20, '-> response.content <-', '=' * 20)
print(response.content)
print('=' * 20, '-> response.content_blocks <-', '=' * 20)
print(response.content_blocks)

response.content_blocks 输出统一结构(reasoning 与 text 分离):

[{'type': 'reasoning', 'reasoning': '好的，用户说“一句话回答”……（思维链内容）'},
 {'type': 'text', 'text': '你好，请说出您的问题，我会用一句话回答。'}]

---

2、提示词模板(Prompt Templates)

2.1 为什么推荐提示词模板?

构造提示词既可直接用 Python 字符串拼接(f-string、format()、+),也可用 LangChain 的 PromptTemplate 或 ChatPromptTemplate。

字符串拼接方式:

topic = "Python"
difficulty = "初学者"
prompt_str = f"你是一个{difficulty}级别的编程导师。请用简单易懂的语言解释{topic}。"
response = model.invoke(prompt_str)

• 优点:简单直接、上手快、适合临时 demo、无额外学习成本

• 缺点:可读性差(变量多时混乱)、不易维护、无变量校验、难支持复杂场景(多轮对话/RAG/Few-shot)

提示词模板方式:

from langchain.prompts import PromptTemplate

template = PromptTemplate.from_template(
    "你是一个{difficulty}级别的编程导师。请用简单易懂的语言解释{topic}。"
)
prompt = template.format(difficulty="初学者", topic="Python")
response = model.invoke(prompt)

• 优点:结构清晰(变量占位)、易维护可复用、自动变量校验(更安全)、支持复杂场景、可与 LangChain 生态无缝集成、便于调试与日志追踪

• 缺点:有一定学习成本、初期写法略复杂、对极简单场景略“重”

开发建议:小项目/临时用 → 字符串拼接;正式开发/AI 应用 → 提示词模板(必选)。

2.2 提示词机制演进

LangChain 1.0 架构变革的核心之一在 Prompt 机制:结构化、富含元数据的消息列表取代单一字符串,成为与模型交互的标准数据格式。

1、旧时代:LLM + PromptTemplate(输入与输出均为字符串)

• 模型接口:对应 LLM 类,面向早期文本补全模型

• 工作方式:接受单一字符串,预测生成后续文本(文本补全)

• Prompt 工具:PromptTemplate,接收变量并渲染输出完整字符串

from langchain.prompts import PromptTemplate

prompt_template = PromptTemplate.from_template("请给我一个关于{topic}的{type}解释。")
prompt = prompt_template.format(type="详细", topic="量子力学")
print(prompt)

• 局限性:模拟多轮聊天时需手动拼接伪造对话角色(如 "Human：你好\nAI：你好！...\nHuman：..."),结构混乱、难维护、易让模型混淆边界与上下文。

2、新时代:ChatModel + ChatPromptTemplate(输入与输出均为消息列表)

• 模型接口:ChatModel(LangChain 1.0 主流接口)

• 工作方式:现代聊天模型 API 原生支持角色概念,接受结构化消息列表

• Prompt 工具:ChatPromptTemplate,接收变量并输出 List[BaseMessage],直接传给聊天模型

特性	PromptTemplate	ChatPromptTemplate
输出格式	纯文本字符串	消息列表
角色支持	无	system/user/assistant
对话历史	不支持	支持
适用场景	简单提示	聊天、对话、多轮交互

角色字符串	含义	用途
"system"	系统消息	设定 AI 的行为、角色、规则
"user" / "human"	用户消息	用户的输入/问题
"assistant" / "ai"	AI 消息	AI 的回复(用于对话历史)

ChatPromptTemplate 因此成为 LangChain 1.0 中最核心的 Prompt 工具。

2.3 ChatPromptTemplate 的使用

ChatPromptTemplate 是创建聊天消息列表的提示模板,比普通 PromptTemplate 更适合多角色、多轮次对话,支持 System/Human/AI 等角色消息模板。

2.3.1 两种实例化方式

方式1(推荐):from_messages() —— 传入由元组(role, content)构成的列表。

from langchain_core.prompts import ChatPromptTemplate

chat_template = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的AI机器人，你的名字是{name}。"),
    ("human", "你好，最近怎么样？"),
    ("ai", "我很好，谢谢！"),
    ("human", "{user_input}"),
])
prompt = chat_template.invoke({"name": "小明", "user_input": "你叫什么名字？"})
print(prompt)

方式2:实例初始化方法(from_messages() 底层也是调用 __init__):

from langchain_core.prompts import ChatPromptTemplate

prompt_template = ChatPromptTemplate([
    ("system", "你是一个AI开发工程师. 你的名字是 {name}."),
    ("human", "你能开发哪些AI应用?"),
    ("ai", "我能开发很多AI应用, 比如聊天机器人, 图像识别, 自然语言处理等."),
    ("human", "{user_input}")
])
prompt = prompt_template.invoke({"name": "小谷AI", "user_input": "你能帮我做什么?"})
print(prompt)

2.3.2 模板调用的 3 种方式

方式	返回类型
invoke(**kwargs)	ChatPromptValue(消息列表封装)
format(**kwargs)	字符串(str)
format_messages(**kwargs)	消息列表(list)

# 方式1：invoke() 返回 ChatPromptValue
prompt = prompt_template.invoke({"name": "小谷AI", "user_input": "你能帮我做什么?"})
print(type(prompt))       # ChatPromptValue
print(prompt)
print(len(prompt.messages))

# 方式2：format() 返回字符串
prompt = prompt_template.format(name="小谷AI", user_input="你能帮我做什么?")
print(type(prompt))       # <class 'str'>
print(prompt)

# 方式3：format_messages() 返回消息列表
prompt = prompt_template.format_messages(name="小谷AI", user_input="你能帮我做什么?")
print(type(prompt))       # <class 'list'>
print(prompt)

2.3.3 结合 LLM 调用

from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
import os
from langchain.chat_models import init_chat_model

# 1、提供大模型
load_dotenv(override=True)
model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)

# 2、提供提示词
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个数学家，你可以计算任何算式"),
    ("human", "{text}"),
])
prompt_value = chat_prompt.invoke({
    "text": "我今年18岁，我的舅舅今年38岁，我的爷爷今年72岁，我和舅舅一共多少岁了？"
})

# 3、结合提示词，调用大模型
output = model.invoke(prompt_value)
print(output.content)

2.3.4 更丰富的初始化参数类型

源码 __init__ 签名表明,messages 参数是列表,元素类型多样:str、dict、字符串元组、消息类型、提示词模板类型、消息提示词模板类型等。

类型1:str 列表(不推荐,默认角色为 human):

from langchain_core.prompts import ChatPromptTemplate

chat_template = ChatPromptTemplate.from_messages([
    "Hello, {name}!"   # 等价于 ("human", "Hello, {name}!")
])
messages = chat_template.invoke({"name": "小谷AI"})
print(messages)
# messages=[HumanMessage(content='Hello, 小谷AI!', ...)]

类型2:tuple 列表(最常用):

prompt = ChatPromptTemplate.from_messages([
    ("system", "你的名字是{role}."),
    ("human", "很高兴认识你"),
])
print(prompt.invoke({"role": "小智"}))

类型3:dict 列表:

prompt = ChatPromptTemplate.from_messages([
    {"role": "system", "content": "你的名字是{role}."},
    {"role": "human", "content": "很高兴认识你"},
])
print(prompt.invoke({"role": "小智"}))

类型4:Message 列表 —— 注意:XxxMessage 中不能含占位符(占位符不会被解析):

from langchain_core.messages import SystemMessage, HumanMessage

chat_prompt_template = ChatPromptTemplate.from_messages([
    SystemMessage(content="我是一个贴心的智能助手"),
    HumanMessage(content="我的问题是:{word}英文怎么说？")   # 占位符不会被替换
])
messages = chat_prompt_template.invoke({"word": "人工智能"})
print(messages)
# content 仍为 '我的问题是:{word}英文怎么说？'

类型5:MessagePromptTemplate 列表

LangChain 提供不同 MessagePromptTemplate,最常用为 SystemMessagePromptTemplate、HumanMessagePromptTemplate、AIMessagePromptTemplate。HumanMessagePromptTemplate 专用于生成用户消息模板:

• 模板化:支持变量占位符,运行时填充

• 格式化:模板与输入变量结合生成最终聊天消息

• 输出类型:生成 HumanMessage(content + role="human")

from langchain_core.prompts import (
    ChatPromptTemplate, HumanMessagePromptTemplate, SystemMessagePromptTemplate
)

system_message_prompt = SystemMessagePromptTemplate.from_template("你是一个{role}")
human_message_prompt = HumanMessagePromptTemplate.from_template("给我解释{concept}，用浅显易懂的语言")

chat_prompt = ChatPromptTemplate.from_messages([system_message_prompt, human_message_prompt])
formatted_messages = chat_prompt.invoke({"role": "物理学家", "concept": "相对论"})
print(formatted_messages)

类型6:BaseChatPromptTemplate 列表(嵌套 ChatPromptTemplate):

from langchain_core.prompts import ChatPromptTemplate

nested_prompt_template1 = ChatPromptTemplate.from_messages([
    ("system", "我是一个人工智能助手，我的名字叫{name}")
])
nested_prompt_template2 = ChatPromptTemplate.from_messages([
    ("human", "很高兴认识你,我的问题是{question}")
])
prompt_template = ChatPromptTemplate.from_messages([
    nested_prompt_template1, nested_prompt_template2
])
prompt_template.invoke({"name": "小智", "question": "你为什么这么帅？"})

综合使用各类元素:

from langchain_core.prompts import (
    ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate
)
from langchain_core.messages import SystemMessage, HumanMessage

system_msg = SystemMessage(content="你是一个AI工程师。")          # BaseMessage
human_msg = HumanMessage(content="你好！")
system_prompt = SystemMessagePromptTemplate.from_template("你是一个{role}.")  # 模板
human_prompt = HumanMessagePromptTemplate.from_template("{user_input}")
nested_prompt = ChatPromptTemplate.from_messages([("system", "嵌套提示词")])   # 嵌套

prompt = ChatPromptTemplate.from_messages([
    system_msg, human_msg, system_prompt, human_prompt, nested_prompt
])
prompt.invoke({"role": "人工智能专家", "user_input": "介绍一下大模型的应用场景"})

2.4 高级特性

2.4.1 部分变量预填充:partial()

预填充某些固定不变的变量,创建模板的变体。使用场景:某些变量所有调用都相同;为不同用户/场景创建定制模板。

from langchain_core.prompts import ChatPromptTemplate

template = ChatPromptTemplate.from_messages([
    ("system", "你是{role}，目标用户是{audience}"),
    ("user", "{task}")
])
# 部分填充
customer_support_template = template.partial(
    role="客服专员",
    audience="普通用户"
)
# 现在只需提供 task
messages = customer_support_template.invoke({"task": "解释退款政策"})
print(messages)

为不同部门创建专用模板:

base_template = ChatPromptTemplate.from_messages([
    ("system", "你是{department}的{role}"),
    ("user", "{task}")
])
it_template = base_template.partial(department="IT 部门", role="技术支持")
sales_template = base_template.partial(department="销售部门", role="销售顾问")
sales_template.invoke({"task": "为什么每年年底汽车会促销"})

2.4.2 消息占位符

当不确定消息模板使用什么角色,或希望在格式化过程中插入消息列表时,使用消息占位符负责在特定位置添加消息列表。使用场景:多轮对话系统存储历史消息、Agent 中间步骤处理。

方式1:JSON 形式("placeholder"):

from langchain_core.prompts import ChatPromptTemplate

template = ChatPromptTemplate.from_messages([
    ("system", "你是一个有用的AI助手"),
    ("placeholder", "{conversation}"),
])
prompt_value = template.invoke({
    "conversation": [
        ("human", "你好!"),
        ("ai", "今天我能帮你做什么？"),
        ("human", "你能给我做一个冰激凌吗？"),
        ("ai", "抱歉，我没有这样的能力"),
    ]
})
print(prompt_value)

方式2:MessagesPlaceholder 实例:

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage

prompt_template = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant"),
    MessagesPlaceholder("msgs")
])
prompt_template.invoke({"msgs": [HumanMessage(content="hi!")]})

存储对话历史内容:

prompt_template = ChatPromptTemplate.from_messages([
    ("system", "你是一个非常友好的AI助手"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{question}")
])
prompt_template.invoke({
    "history": [
        ("human", "5 + 2 = ?"),
        ("ai", "5 + 2 = 7")
    ],
    "question": "结果再乘以4呢？"
})

2.4.3 可复用模板库

实际项目中建议创建模板库(templates.py):

from langchain_core.prompts import ChatPromptTemplate

class PromptLibrary:
    """可复用的提示词模板库"""
    TRANSLATOR = ChatPromptTemplate.from_messages([
        ("system", "你是专业翻译，精通{source_lang}和{target_lang}"),
        ("user", "翻译以下文本：\n{text}")
    ])
    CODE_REVIEWER = ChatPromptTemplate.from_messages([
        ("system", "你是{language}代码审查专家，重点关注{focus}"),
        ("user", "审查代码：\n```{language}\n{code}\n```")
    ])
    SUMMARIZER = ChatPromptTemplate.from_messages([
        ("system", "你是内容摘要专家"),
        ("user", "将以下内容总结为{num}个要点：\n{content}")
    ])
    TUTOR = ChatPromptTemplate.from_messages([
        ("system", "你是{subject}导师，学生水平：{level}"),
        ("user", "{question}")
    ])

其他文件中使用:

from templates import PromptLibrary

messages = PromptLibrary.TRANSLATOR.format_messages(
    source_lang="英语",
    target_lang="中文",
    text="Hello World"
)

也可按模块组织(templates/ 下 common.py、translation.py、coding.py)。

2.4.4 模板组合

将多个模板片段组合成复杂提示词。

方法1:字符串组合:

role_part = "你是一个{domain}专家。"
style_part = "回答风格：{style}。"
constraint_part = "限制：{constraint}。"
full_system = role_part + style_part + constraint_part
template = ChatPromptTemplate.from_messages([
    ("system", full_system),
    ("user", "{question}")
])

方法2:使用 + 运算符(LangChain 1.0 支持):

template1 = ChatPromptTemplate.from_messages([("system", "你是助手")])
template2 = ChatPromptTemplate.from_messages([("user", "{input}")])
combined = template1 + template2

---

第06章 结构化输出(Structured Output)

1、结构化输出概述

1.1 什么是结构化输出

LangChain 的结构化输出(Structured Output) 指要求模型最终返回一个符合预定义结构的数据对象(固定字段的 JSON、Pydantic 模型、TypedDict),而不再是无格式的自然语言文本。

核心目标:把“自然语言回答”变成“程序可以稳定消费的数据”。

例如,不是输出:

盗梦空间在2010年上映,导演是克里斯托弗·诺兰,评分9.3。

而是输出结构:

{
    "title": "盗梦空间",
    "year": 2010,
    "director": "克里斯托弗·诺兰",
    "rating": 9.3
}

价值有三点:

• 更容易被代码处理:下游系统直接读字段,无需从自然语言解析

• 结果更稳定:减少“说法变了但意思差不多”导致的解析失败

• 更适合工程化:适用于表单抽取、分类、路由、工具参数生成、工作流状态传递等场景

1.2 传统方式 vs 结构化输出

1、传统方式(繁琐、不推荐):

# 1. 提示词要求 JSON
prompt = "以JSON格式返回：{name, age, occupation}"
response = model.invoke(prompt)

# 2. 手动解析
import json
data = json.loads(response.content)

# 3. 手动验证类型
if not isinstance(data['age'], int):
    raise ValueError("age must be int")

# 4. 手动创建对象
person = Person(**data)

2、结构化输出(简洁):

# 一步到位
structured_llm = model.with_structured_output(Person)
person = structured_llm.invoke("张三是一名 30 岁的软件工程师")
# 自动解析、验证、创建对象

为什么结构化输出受欢迎? 在没有 Pydantic 等方案之前,开发者要写大量 Prompt 求模型“请返回 JSON,不要带解释”,再自己写繁琐的 json.loads() 和 try...except。结合 Pydantic 与 .with_structured_output() 后:

• Prompt 变干净:字段的 description 直接充当 Prompt 的一部分

• 类型安全:编辑器自动补全,运行前可做类型检查

• 极其稳定:依托模型厂商底层 JSON 模式,输出错误率降到极低

1.3 结构化输出模式

LangChain 1.x 支持多种 Schema 与结构化输出方式:

• Pydantic:字段校验、描述、嵌套结构,功能最丰富

• TypedDict:轻量类型约束

• JSON Schema:与前后端/跨语言接口最通用

• dataclass

模型对象可调用 with_structured_output() 绑定输出模式(schema)。关键差异:只有 Pydantic 返回 Schema 类实例,其余三种返回字典;也只有 Pydantic 在类型不匹配时抛出异常。

是否所有模型都支持? 大部分现代模型支持(通过函数调用):OpenAI(gpt-4、gpt-3.5-turbo)、Anthropic(claude-3)、Groq(llama-3)等;某些旧模型不支持。不支持时 LangChain 会回退到提示词 + JSON 解析。

2、四种模式的使用

2.1 模式1:Pydantic

Pydantic 通过运行时强制执行类型提示确保数据正确性与一致性,是生产场景首选。

2.1.1 基本使用

要素:

• 所有结构化输出的数据模型都必须继承 BaseModel

• 使用类型提示(str、int、float、List[xxx]、Optional[xxx] 等)

• 使用 Field() 添加字段默认值和描述,帮助 LLM 理解字段含义(没有描述,LLM 可能格式错误)

模型初始化:

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

load_dotenv(override=True)
model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)

定义 Pydantic 模型:

from pydantic import BaseModel, Field

class Person(BaseModel):
    """人物信息"""
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")
    occupation: str = Field(description="职业")

使用 with_structured_output 引导结构化输出:

structured_llm = model.with_structured_output(Person)
result = structured_llm.invoke("张三是一名 30 岁的软件工程师")
print(result)
print(type(result))
# result 是 Person 实例,可直接点属性访问
print(result.name)        # "张三"
print(result.age)         # 30
print(result.occupation)  # "软件工程师"

输出:

name='张三' age=30 occupation='软件工程师'
<class '__main__.Person'>

电影信息抽取与情感分析示例:

from pydantic import BaseModel, Field

class MovieModel(BaseModel):
    """电影的详细信息"""
    title: str = Field(description="电影标题")
    year: int = Field(description="电影上映年份")
    director: str = Field(description="导演")
    rating: float = Field(description="电影评分，满分十分")

model_with_structure = model.with_structured_output(MovieModel)
response = model_with_structure.invoke("给出盗梦空间的信息")
# title='盗梦空间' year=2010 director='克里斯托弗·诺兰' rating=9.3

class SentimentAnalysis(BaseModel):
    """情感分析结果"""
    sentiment: str = Field(description="情感倾向：positive/negative/neutral")
    confidence: float = Field(description="置信度，0-1之间")
    keywords: list[str] = Field(description="关键词列表")

structured_model = model.with_structured_output(SentimentAnalysis)
text = "这个课程内容很实用，学到了很多知识，强烈推荐！"
result = structured_model.invoke(f"分析以下文本的情感：\n{text}")
print(f"类型: {type(result)}")
print(f"情感: {result.sentiment}")       # positive
print(f"置信度: {result.confidence}")    # 0.99
print(f"关键词: {result.keywords}")      # ['实用', '学到了很多知识', '强烈推荐']

2.1.2 高级特性

情况1:可选字段 —— LLM 未填充字段时,用 Optional 指定字段可选。

from typing import Optional
from pydantic import BaseModel, Field

class Person(BaseModel):
    """人物信息"""
    name: str = Field(description="姓名")
    age: int = Field(description="年龄")
    occupation: str = Field(description="职业")

structured_llm = model.with_structured_output(Person)
# 不带年龄信息时：Person(name='张三', age=0, occupation='医生')

对比使用 Optional[int]:

class Person(BaseModel):
    name: str = Field(description="姓名")
    age: Optional[int] = Field(description="年龄")
    occupation: str = Field(description="职业")

structured_llm = model.with_structured_output(Person)
# Person(name='张三', age=None, occupation='医生')

情况2:默认值 —— LLM 未提供的信息使用默认值 Field(default="默认值", description="描述")。注意:不同模型提供商对 default 字段支持不同。

class Person(BaseModel):
    name: str = Field(description="姓名")
    age: int = Field(1, description="年龄")   # 位置参数即默认值
    occupation: str = Field(description="职业")

OpenRouter 平台返回 Person(name='张三', age=1, occupation='医生')(正确使用默认值);CloseAI 平台此例返回 age=0(支持有差异)。

class Config(BaseModel):
    timeout: Optional[int] = Field(30, description="超时时间(单位秒)")
    retry: bool = Field(False, description="是否支持重试")
    max_attempts: int = Field(6, description="最大重试次数")

structured_llm = model.with_structured_output(Config)
# 输入"支持重试,最多重试5次" → Config(timeout=None, retry=True, max_attempts=5)

class Product(BaseModel):
    """产品信息"""
    name: str = Field(description="产品名称")
    price: float = Field(description="价格")
    description: Optional[str] = Field(description="产品描述")
    stock: int = Field(default=100, description="库存")

structured_llm = model.with_structured_output(Product)
# 完整信息：name='iPhone 15' price=5999.0 description='最新款智能手机' stock=50
# 缺少描述和库存：name='MacBook Pro' price=12999.0 description=None stock=100

情况3:枚举类型 —— 限制字段可选值。

from enum import Enum

class Priority(str, Enum):
    LOW = "低"
    MEDIUM = "中"
    HIGH = "高"

class Task(BaseModel):
    title: str
    priority: Priority   # 只能是 LOW/MEDIUM/HIGH

嫌单独定义 Enum 麻烦,可用 typing.Literal 直接限定:

from typing import Optional, Literal

class CustomerInfo(BaseModel):
    """客户信息"""
    name: str = Field(description="客户姓名")
    phone: str = Field(description="电话号码")
    email: Optional[str] = Field("未提供", description="邮箱")
    issue: str = Field(description="问题描述")
    urgency: Literal["低", "中", "高"] = Field(description="紧急程度")

structured_llm = model.with_structured_output(CustomerInfo)
conversation = """
客服: 您好，请问有什么可以帮助您？
客户: 我是王小明，电话 138-1234-5678，我的订单一直没发货，很着急！
客服: 好的，我帮您查一下
"""
result = structured_llm.invoke(f"从以下客服对话中提取客户信息：\n{conversation}")
print(f"客户: {result.name}, 紧急程度: {result.urgency}")
# urgency='高'

应用场景:自动填充 CRM 系统、工单自动分类、客服辅助。

情况4:列表提取

from typing import List

class Person(BaseModel):
    name: str
    age: int

class PersonList(BaseModel):
    people: List[Person]   # 多个 Person 对象

structured_llm = model.with_structured_output(PersonList)
result = structured_llm.invoke("张三 30岁，李四 25岁")
# people=[Person(name='张三', age=30), Person(name='李四', age=25)]

产品评论分析:

class Review(BaseModel):
    """产品评论"""
    product: str
    rating: int = Field(description="评分 1-5")
    pros: List[str] = Field(description="优点列表")
    cons: List[str] = Field(description="缺点列表")

structured_llm = model.with_structured_output(Review)
review = structured_llm.invoke("""
iPhone 17 很棒！摄像头强大，手感好。但是价格贵，没有充电器。4分。
""")
# product='iPhone 17' rating=4 pros=['摄像头强大', '手感好'] cons=['价格贵', '没有充电器']

文档信息提取(发票):

class Invoice(BaseModel):
    """发票信息"""
    invoice_number: str = Field(description="发票号")
    date: str = Field(description="日期")
    total_amount: float = Field(description="总金额")
    items: List[str] = Field(description="商品")

structured_llm = model.with_structured_output(Invoice)
invoice_text = """
发票号: INV-2024-001
日期: 2024-01-15
总金额: 1299.00
商品: MacBook Pro, AppleCare+
"""
invoice = structured_llm.invoke(f"提取发票信息：{invoice_text}")
# invoice_number='INV-2024-001' date='2024-01-15' total_amount=1299.0 items=['MacBook Pro', 'AppleCare+']

应用场景:批量处理评论、自动生成分析报告、OCR 后结构化、数据录入。

情况5:嵌套结构

from pydantic import BaseModel

class Address(BaseModel):
    city: str
    district: str

class Company(BaseModel):
    name: str
    address: Address   # 嵌套模型

structured_llm = model.with_structured_output(Company)
result = structured_llm.invoke("阿里巴巴在杭州滨江区")
# name='阿里巴巴' address=Address(city='杭州', district='滨江区')

更复杂的电影信息(嵌套列表):

from pydantic import BaseModel, Field
from typing import List

class Actor(BaseModel):
    """演员信息"""
    name: str = Field(description="演员姓名")
    role: str = Field(description="饰演的角色")

class Movie(BaseModel):
    """电影信息"""
    title: str = Field(description="电影标题")
    year: int = Field(description="上映年份")
    director: str = Field(description="导演")
    cast: List[Actor] = Field(description="演员列表")
    rating: float = Field(description="评分")

structured_model = model.with_structured_output(Movie)
response = structured_model.invoke("请介绍电影《盗梦空间》")
print(f"电影名: {response.title}")
print(f"导演: {response.director}")
print(f"演员列表: {response.cast}")

说明:LLM 能力有限,复杂嵌套结构可能出错,建议:

• 嵌套层级 ≤ 3 层

• 使用清晰的 description

• 必要时拆分成多个调用

4 层嵌套容易出错:

class Bad(BaseModel):
    user: User
        company: Company
            address: Address
                country: Country   # 4 层嵌套,容易出错

多维度产品评论分析(嵌套 + 列表):

class Aspect(BaseModel):
    """评论维度"""
    name: str = Field(description="维度名称，如：质量、价格、服务")
    score: int = Field(description="评分，1-5")
    comment: str = Field(description="具体评价")

class ProductReview(BaseModel):
    """产品评论分析"""
    overall_sentiment: str = Field(description="整体情感：positive/negative/neutral")
    overall_score: int = Field(description="综合评分，1-5")
    aspects: List[Aspect] = Field(description="各维度评价")
    summary: str = Field(description="一句话总结")

structured_model = model.with_structured_output(ProductReview)
review_text = """
这款笔记本电脑性能非常强大，运行大型软件毫无压力。
屏幕色彩鲜艳，看视频很舒服。
不过价格有点贵，而且风扇噪音较大。
客服态度很好，物流也快。
"""
result = structured_model.invoke(f"分析以下产品评论：\n{review_text}")
print(f"整体情感: {result.overall_sentiment}")
for aspect in result.aspects:
    print(f"  - {aspect.name}: {aspect.score}/5 - {aspect.comment}")

情况6:限制条件(字段约束)

from pydantic import BaseModel, Field, ValidationError

class User(BaseModel):
    name: str = Field(min_length=2, max_length=20)
    age: int = Field(ge=0, le=150)
    email: str

# 有效数据
try:
    user = User(name="张三", age=30, email="zhang@example.com")
    print(f"[OK] {user.name}, {user.age}, {user.email}")
except ValidationError as e:
    print(f"[FAIL] {e}")

# 无效数据(年龄超出范围)
try:
    user = User(name="李四", age=200, email="li@example.com")
except ValidationError as e:
    print(f"[FAIL] 验证失败（符合预期）: {e.errors()[0]['msg']}")
# Input should be less than or equal to 150

带约束的产品信息(gt=0、ge=0):

class Product(BaseModel):
    """产品信息（严格验证）"""
    name: str = Field(description="产品名称（字符串类型）", min_length=2)
    price: float = Field(description="价格，数字类型", gt=0)
    stock: int = Field(description="库存，整数类型", ge=0)

structured_llm = model_with_closeai.with_structured_output(Product)
# 输入非法价格 -7999, 库存 -100
response = structured_llm.invoke("华为mate 80 promax 价格是-7999，当前库存-100")
# CloseAI: name='华为mate 80 promax' price=7999.0 stock=100（自动修正为合法值）
# OpenRouter: name='华为mate 80 promax' price=1.0 stock=0

2.1.3 工作流程图解

结构化输出分四步:

• 第1步:定义结构 —— 用 Pydantic 定义模型(如 BookInfo,含 title、author、tags)。

• 第2步:协议转换 —— LangChain 内部调用 Pydantic 底层方法(如 model_json_schema())将 Python 代码自动转成标准 JSON Schema(描述字段、类型、描述)。

• 第3步:模型交互与强约束 —— LangChain 将 JSON Schema 包装进请求。现代模型(OpenAI/Anthropic/Gemini)普遍支持函数/工具调用或 JSON Mode,LangChain 把 JSON Schema 作为 Tools 传入;OpenAI 的 strict=True 启动语法采样约束(Grammar-based sampling),模型严格按 JSON Schema 语法树解码,底层保证输出格式不走样。

• 第4步:自动解析与验证 —— 大模型返回符合 JSON 规范的字符串后,PydanticStructuredOutputParser 接管:

  1. 解析(Parsing):字符串解析为 Python 字典

  2. 验证(Validation):字典喂给 Pydantic 模型,检查类型;漏字段或类型错误直接抛验证错误(或触发重试)

  3. 返回(Return):通过验证后返回可直接点属性的 Pydantic 对象(如 result.title)

from pydantic import BaseModel, Field

class BookInfo(BaseModel):
    title: str = Field(description="书名")
    author: str = Field(description="作者名字")
    tags: list[str] = Field(description="书籍的标签或分类")

2.2 模式2:TypedDict

2.2.1 什么是 TypedDict

TypedDict 是 Python 3.8+ 引入的类型提示工具,即带有类型声明的字典结构,适合需快速定义字典结构且无需 Pydantic 重量级功能的场景。

普通 dict 没有类型信息;TypedDict 可说明字典有哪些字段、每个字段类型。TypedDict 主要是类型声明,不是运行时强校验器(字段名不一致时 IDE 静态检查会标记,但不会运行时异常):

from typing_extensions import TypedDict

class MovieDict(TypedDict):
    title: str
    year: int
    director: str
    rating: float

movie: MovieDict = {
    "title1": "盗梦空间",   # 字段名不一致
    "year": 2010,
    "director": "克里斯托弗·诺兰",
    "rating": 8.8,
}
print(movie)
# 运行时正常输出,不报错

Annotated 的使用:Annotated 用来在“类型”之外附加额外信息(元数据),类似 Pydantic 的 Field:

Annotated[类型, 附加信息1, 附加信息2, ...]

2.2.2 基本使用

举例1:返回简单结构

from typing_extensions import TypedDict, Annotated

class MovieTypedDict(TypedDict):
    """电影的详细信息"""
    title: Annotated[str, "电影的正式名称，例如《盗梦空间》"]
    year: Annotated[int, "电影的公映年份，使用四位数字表示"]
    director: Annotated[str, "电影导演的全名"]
    rating: Annotated[float, "电影在10分制下的评分，可包含一位小数"]

structured_llm = model_with_closeai.with_structured_output(MovieTypedDict)
response = structured_llm.invoke("给我介绍下电影《星际穿越》")
print(type(response))   # <class 'dict'>
print(response)
# {'title': '星际穿越', 'year': 2014, 'director': '克里斯托弗·诺兰', 'rating': 8.6}

举例2:返回嵌套结构

from typing import TypedDict, List, Annotated

class Actor(TypedDict):
    """演员情况"""
    name: Annotated[str, "演员姓名"]
    role: Annotated[str, "饰演的角色"]

class Movie(TypedDict):
    """电影情况"""
    title: Annotated[str, "电影标题"]
    year: Annotated[int, "上映年份"]
    director: Annotated[str, "导演"]
    cast: Annotated[List[Actor], "演员列表"]
    rating: Annotated[float, "评分"]

structured_llm = model_with_closeai.with_structured_output(Movie)
resp = structured_llm.invoke("给我介绍下电影《盗梦空间》")
print(f"电影名: {resp['title']}")
print(f"演员列表:{resp['cast']}")   # 列表元素为 dict

举例3:... 的使用 —— ... 是 Python 字面量,等价 Ellipsis,可理解为占位符。LangChain 中 Annotated 的 ... 表示当前字段必须存在,不可省略。

from typing_extensions import TypedDict, Annotated

class MovieDict(TypedDict):
    """电影的详细信息"""
    title: Annotated[str, ..., "电影标题"]
    year: Annotated[int, ..., "电影上映年份"]
    director: Annotated[str, ..., "导演"]
    rating: Annotated[float, "电影评分，满分十分"]

model_with_structure = model_with_closeai.with_structured_output(MovieDict)
response = model_with_structure.invoke(
    "根据这段话抽取盗梦空间的信息，不包含的信息可以留空：盗梦空间在2010年上映，导演是克里斯托弗·诺兰。"
)
# CloseAI: {'title': '盗梦空间', 'year': 2010, 'director': '克里斯托弗·诺兰'}
#          (rating 未标记必填,被省略)
# OpenRouter: {'title': '盗梦空间', 'year': 2010, 'director': '克里斯托弗·诺兰', 'rating': 0}
#          (rating 未省略,填 0)

对比 Ellipsis 写法与省略:

class MovieDict(TypedDict):
    title: Annotated[str, ..., "电影标题"]
    year: Annotated[int, ..., "电影上映年份"]
    director: Annotated[str, Ellipsis, "导演"]   # 必填
    rating: Annotated[float, "电影评分，满分十分"]  # 未标记必填

# director 缺失 → 输出包含该字段但值为空字符串 ''
# rating 缺失 → 字段被省略
# {'title': '盗梦空间', 'year': 2010, 'director': ''}

2.3 模式3:JSON Schema

需按 JSON Schema 规范拼接 JSON 字符串,比较繁琐且缺少校验机制,不推荐。

模型初始化同前。举例1:返回简单结构

json_schema = {
    "title": "Movie",
    "description": "A movie with details",
    "type": "object",
    "properties": {
        "title": {"type": "string", "description": "The title of the movie"},
        "year": {"type": "integer", "description": "The year the movie was released"},
        "director": {"type": "string", "description": "The director of the movie"},
        "rating": {"type": "number", "description": "The movie's rating out of 10"}
    },
    "required": ["title", "year", "director", "rating"]
}
structured_model = model.with_structured_output(
    json_schema,
    method="json_schema"
)
response = structured_model.invoke("给出盗梦空间的信息")
print(response)
# {'title': '盗梦空间', 'year': 2010, 'director': '克里斯托弗·诺兰', 'rating': 8.8}

关键字说明:

• method:结构化输出方式,是否可用依赖模型供应商及适配器实现(如 DeepSeek 不支持 json_schema 模式)

• json_schema:使用模型供应商提供的专用结构化输出功能

JSON Schema 标准关键字(固定写法):

关键字	说明
title	整个 Schema 或属性的人类可读标题,不能是中文,用于提高可读性
description	详细文字描述,说明用途,帮助理解
type	当前数据节点的类型,如 string/number/integer/boolean/object/array/null(object 即 JSON 对象)
properties	定义 object 中可包含哪些属性及每个属性的类型、说明
required	当 type 为 object 时使用,数组形式列出必须存在的属性名

举例2:返回嵌套结构(用 array + items 定义嵌套数组):

project_schema = {
    "title": "MovieInfo",
    "description": "包含电影标题、上映年份、导演、演员和评分的电影对象",
    "type": "object",
    "properties": {
        "title": {"type": "string", "description": "电影标题"},
        "year": {"type": "integer", "description": "上映年份"},
        "director": {"type": "string", "description": "导演"},
        "cast": {
            "type": "array",
            "description": "演员列表",
            "items": {
                "type": "object",
                "properties": {
                    "name": {"type": "string", "description": "演员姓名"},
                    "role": {"type": "string", "description": "演员角色"}
                },
                "required": ["name", "role"]
            }
        },
        "rating": {"type": "number", "description": "评分（10分制）"}
    },
    "required": ["title", "year", "director", "cast", "rating"]
}
structured_model = model.with_structured_output(project_schema)
response = structured_model.invoke("生成一个关于《星际穿越》的电影信息，包含导演、演员、评分")

2.4 模式4:@dataclass

@dataclass 是 Python 标准库 dataclasses 提供的类装饰器,用于简化“以字段为核心”的数据类定义。加 @dataclass 后,Python 自动生成常用方法:__init__、__repr__、__eq__,近似于手写这些方法的普通类。主要价值:让数据结构定义更简洁清晰。

注意:@dataclass 修饰的类仍是普通 python 类,但被标记为数据类并携带 dataclass 字段元信息。手写 __init__ 等方法的普通类不能替代 @dataclass 类 —— 前者可作为 LangChain 的 Schema,后者不行。

from dataclasses import dataclass

@dataclass
class Movie:
    title: str
    year: int
    director: str
    rating: float

带 Field 描述的写法:

from dataclasses import dataclass
from pydantic import Field

@dataclass
class Movie:
    """电影的详细信息"""
    title: str = Field(description="电影标题")
    year: int = Field(description="电影上映年份")
    director: str = Field(description="导演")
    rating: float = Field(description="电影评分，满分十分")

structured_model = model.with_structured_output(Movie)
response = structured_model.invoke("给出盗梦空间的信息")
print(response)        # {'title': '盗梦空间', ...}
print(type(response))  # <class 'dict'>

@dataclass 作为 schema,返回的是未经校验的字典。

3、关于类型校验

构造一个虚拟的 DeepSeek 服务端(Fake Server),客户端收到的响应是人为构造的,目的是观察不同模式下 LangChain 从响应的哪些字段抽取信息。服务端固定返回前两个字段名不匹配(title1、year2)的 JSON:

import json
import time
from http.server import BaseHTTPRequestHandler, HTTPServer

class FakeDeepSeekHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        content_length = int(self.headers.get("Content-Length", 0))
        raw_body = self.rfile.read(content_length).decode("utf-8")
        json_body = None
        try:
            json_body = json.loads(raw_body)
        except Exception:
            pass

        # 固定返回一个普通 JSON(经 tool_calls 携带,字段名故意写错)
        response = {
            "id": "chatcmpl-test",
            "object": "chat.completion",
            "created": int(time.time()),
            "model": "any",
            "choices": [{
                "index": 0,
                "message": {
                    "role": "assistant",
                    "content": "",
                    "tool_calls": [{
                        "id": "call_1",
                        "type": "function",
                        "function": {
                            "name": json_body["tools"][0]["function"]["name"],
                            "arguments": json.dumps(
                                {'title1': '盗梦空间', 'year2': 2010,
                                 'director': '克里斯托弗·诺兰', 'rating': 9.3},
                                ensure_ascii=False
                            )
                        }
                    }]
                },
                "finish_reason": "stop"
            }],
            "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}
        }
        body = json.dumps(response, ensure_ascii=False).encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, format, *args):
        pass

def main():
    server = HTTPServer(("127.0.0.1", 8889), FakeDeepSeekHandler)
    print("Fake DeepSeek server running at http://127.0.0.1:8889")
    server.serve_forever()

if __name__ == "__main__":
    main()

四种模式的校验结果对比

模式	字段不匹配时表现
Pydantic	抛出 ValidationError(校验失败)
TypedDict	按字典原样输出,不报错
JSON Schema	按字典原样输出,不报错
@dataclass	按字典原样输出,不报错

Pydantic(字段 title、year 缺失抛异常):

from pydantic import BaseModel, Field, SecretStr
from langchain_deepseek import ChatDeepSeek

model = ChatDeepSeek(
    model="deepseek-v4-flash",
    api_base="http://localhost:8889",
    api_key=SecretStr("<KEY>")
)
class MovieModel(BaseModel):
    title: str = Field(description="电影标题")
    year: int = Field(description="电影上映年份")
    director: str = Field(description="导演")
    rating: float = Field(description="电影评分，满分十分")

model_with_structure = model.with_structured_output(MovieModel)
response = model_with_structure.invoke("给出盗梦空间的信息")
# ValidationError: 2 validation errors for MovieModel (title / year Field required)

TypedDict(原样输出 title1、year2,不报错):

from typing_extensions import TypedDict, Annotated

class MovieDict(TypedDict):
    title: Annotated[str, "电影标题"]
    year: Annotated[int, "电影上映年份"]
    director: Annotated[str, "导演"]
    rating: Annotated[float, "电影评分，满分十分"]

structured_model = model.with_structured_output(MovieDict)
response = structured_model.invoke("给出盗梦空间的信息")
# {'title1': '盗梦空间', 'year2': 2010, 'director': '克里斯托弗·诺兰', 'rating': 9.3}

小结:用 Pydantic 定义 schema,接收响应后会校验,字段不匹配抛异常;其余三种方式不校验。

4、获取结构化结果方式

除 with_structured_output 外,还可通过输出解析器获取结构化输出。

4.1 使用 with_structured_output

最新、最简洁的 API,直接让模型“理解”数据结构并返回解析好的对象。可在方法中传入 include_raw=True,表示返回解析前的原始 AIMessage,从而访问令牌用量等元数据。

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os
from pydantic import BaseModel, Field
from rich import print as rprint

load_dotenv(override=True)
model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)

class Movie(BaseModel):
    """电影信息"""
    title: str = Field(description="电影标题")
    year: int = Field(description="上映年份")
    director: str = Field(description="导演")
    rating: float = Field(description="评分（10分制）")

# include_raw=True:返回含原始 AIMessage 的字典
model_with_structure = model.with_structured_output(Movie, include_raw=True)
resp = model_with_structure.invoke("给我介绍下电影《星际穿越》")
print(type(resp))   # <class 'dict'>
rprint(resp)

返回字典含三个字段:

• raw:返回的原始 AIMessage(含 token 用量等元数据)

• parsed:解析后的输出(Movie 实例)

• parsing_error:解析错误;Pydantic 校验,格式不符 schema 会报错,其余三种方式不符不会报错

{
    'raw': AIMessage(content='', ...tool_calls=[{'name': 'Movie', 'args': {...}, 'type': 'tool_call'}],
                     usage_metadata={'input_tokens': 373, 'output_tokens': 93, 'total_tokens': 466}),
    'parsed': Movie(title='星际穿越', year=2014, director='克里斯托弗·诺兰', rating=9.3),
    'parsing_error': None
}

4.2 使用输出解析器(不推荐)

更传统的方法,依赖在提示词中明确指示模型输出特定格式文本,再用解析器转换。流程:提示词指导 → 模型生成文本 → 解析器转换。

from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

# 1. 创建提示词模板
prompt_template = ChatPromptTemplate.from_messages([
    ("system", "回答用户问题,必须始终输出一个包含title(电影标题)和year(上映年份)的 JSON 对象"),
    ("human", "问题：{question}")
])

# 2. 模型初始化
load_dotenv(override=True)
model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)

# 3. 定义结构
class Movie(BaseModel):
    """电影信息"""
    title: str = Field(description="电影标题")
    year: int = Field(description="上映年份")

# 4. 创建输出解析器
parser = JsonOutputParser(pydantic_object=Movie)

# 5. 创建链(LCEL)
chain = prompt_template | model | parser

# 6. 调用（返回字典）
response = chain.invoke({"question": "介绍电影《盗梦空间》"})
print(response)   # {'title': '盗梦空间', 'year': 2010}

---

笔记已输出完毕。两章合计约 7000 字,核心 API 代码(消息四类、ChatPromptTemplate 三种调用、partial/MessagesPlaceholder/模板组合、Pydantic 六种情况、TypedDict、JSON Schema、dataclass、四种校验对比、include_raw、输出解析器链)均已保留代表性版本并清理了错位行号。

第05章 Tools

Tools 概述

工具的重要性

要构建强大的 AI 工程应用，仅有"生成文本"这样的"纸上谈兵"能力是不够的。工具（Tools）是赋予大语言模型与外部世界交互能力的关键组件，让智能体能够执行搜索、计算、数据库查询、邮件发送、调用第三方 API 等操作。借助工具，大模型才能从"认识世界"走向"改变世界"。

工具是构建智能体的核心要素之一。

工具调用的两种方式

在 LangChain 中，工具（Tools）实际上是指明确定义了输入和输出的可调用函数。因此工具调用（Tool Calling）也被称为函数调用（Function Calling）。具体有两种调用方式：

调用方式	说明	适用场景
直接调用	通过 .invoke() 手动调用工具	测试时使用
绑定到模型（主流）	通过 model.bind_tools([...]) 让 AI 决定是否调用	开发中使用

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """
    获取指定城市的天气信息
    参数:
        city: 城市名称，如"北京"、"上海"
    返回:
        天气信息字符串
    """
    return city + "晴天，温度 15°C"

直接调用（测试用）：

result = get_weather.invoke({"city": "北京"})
print(result)  # 北京晴天，温度 15°C

模型初始化（贯穿全章）：

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

load_dotenv(override=True)
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")

model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=CLOSEAI_API_KEY,
    base_url=CLOSEAI_BASE_URL
)

工具调用的整体流程

大模型能根据对话上下文决定何时调用工具以及传递哪些参数。完整流程如下：

from langchain_core.tools import tool

# 1. 定义工具
@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    return "晴天，温度 15°C"

# 2. 绑定工具
model_with_tools = model.bind_tools([get_weather])

# 3. AI 可以决定是否调用工具
response = model_with_tools.invoke("北京天气如何？")

# 4. 检查 AI 是否要调用工具
if response.tool_calls:
    print("AI 想调用工具：", response.tool_calls)
else:
    print("AI 直接回答：", response.content)

输出示例：

AI 想调用工具： [{'name': 'get_weather', 'args': {'city': '北京'},
'id': 'call_fR3LE8Wjqh9lnDosQ61Y892E', 'type': 'tool_call'}]

从 Message 流转看工具的调用

完整流程的核心是 消息列表（messages）的流转：HumanMessage → AIMessage（含 tool_calls） → ToolMessage → AIMessage（最终回复）。

参考 1：不使用 @tool 修饰（需手动拼接 ToolMessage）

from langchain.messages import HumanMessage, ToolMessage

def get_weather(city: str):
    """获取天气的工具"""
    return f"{city}天气晴朗"

model_with_tools = model.bind_tools([get_weather])
messages = [HumanMessage("今天北京天气如何")]

response = model_with_tools.invoke(messages)
messages.append(response)

for tool_call in response.tool_calls:
    if tool_call["name"] == "get_weather":
        # 手动拼接 ToolMessage
        tool_response = ToolMessage(
            content=get_weather(**tool_call["args"]),
            tool_call_id=tool_call["id"],
            name=tool_call["name"]
        )
        messages.append(tool_response)

final_response = model_with_tools.invoke(messages)

参考 2：使用 @tool 修饰（无需手动拼接，invoke 直接返回 ToolMessage）

from langchain.messages import HumanMessage, ToolMessage
from langchain_core.tools import tool

@tool
def get_weather(city: str):
    """获取天气的工具"""
    return f"{city}天气晴朗~"

model_with_tools = model.bind_tools([get_weather])
messages = [HumanMessage("今天北京天气如何")]

response = model_with_tools.invoke(messages)
messages.append(response)

for tool_call in response.tool_calls:
    if tool_call["name"] == "get_weather":
        # 被修饰的工具可直接 invoke，返回 ToolMessage 实例
        tool_response = get_weather.invoke(tool_call)
        messages.append(tool_response)

final_response = model_with_tools.invoke(messages)

工具调用流程四步总结

1. 步骤 1：模型绑定工具 — 通过 model.bind_tools([...]) 绑定一个或多个工具。

2. 步骤 2：模型生成工具调用请求 — 用户输入问题并调用模型，若需要调用工具，模型返回包含工具名称和参数的 AIMessage。

3. 步骤 3：开发者手动执行工具 — 从响应中提取工具调用信息并手动调用对应工具（如 工具.invoke()）。

4. 步骤 4：将 ToolMessage 传回模型生成最终结果 — 把用户提问 + 工具结果 ToolMessage 一起返回给模型，模型生成最终回复。

特别注意：大模型调用工具是单次推理、直接响应，需要开发者手动执行工具并管理循环，适合简单、确定的任务。

---

工具定义方式 1：不使用 @tool

模型绑定工具并发送请求

from rich import print as rprint

def get_weather(city: str):
    return f"{city}天气晴朗"

model_with_tools = model.bind_tools([get_weather])
response = model_with_tools.invoke("今天北京天气如何")
rprint(response)

返回的 AIMessage 关键字段（节选）：

AIMessage(
    content='',
    finish_reason='tool_calls',
    tool_calls=[{
        'name': 'get_weather',
        'args': {'city': '北京'},
        'id': 'call_ECvZNV7RLTWpKQSjhvdGzKBd',
        'type': 'tool_call'
    }],
    usage_metadata={'input_tokens': 121, 'output_tokens': 18, 'total_tokens': 139}
)

工具描述的各部分详解

了解：convert_to_openai_tool

执行 model.bind_tools([get_weather]) 时，底层最终会调用 convert_to_openai_tool 生成工具描述：

from langchain_core.utils.function_calling import convert_to_openai_tool
from rich import print as rprint

def get_weather(city: str):
    return f"{city}天气晴朗"

rprint(convert_to_openai_tool(get_weather))

{
  'type': 'function',
  'function': {
      'name': 'get_weather',
      'description': '',
      'parameters': {
          'properties': {'city': {'type': 'string'}},
          'required': ['city'],
          'type': 'object'
      }
  }
}

为什么不加 @tool 的函数也能被当作工具？ 加 @tool 的函数走 BaseTool 分支；普通函数走 callable 分支，会基于函数定义和 docstring 生成 pydantic 模式描述，再转换为规范的 tool_schema。

description 说明

convert_to_openai_tool 会从 docstring（文档字符串） 加载工具描述，docstring 为空则抽取的 description 也为空。docstring 用三引号表示。

参数说明（Args）

参数说明从 docstring 加载，必须遵循 Google 风格。使用 Args: / Returns: / Raises: 等关键字。Agent 通过这些注释理解工具用途与调用时机，因此清晰准确的 docstring 是工具被正确调用的前提。

注意：如果 docstring 中包含参数说明，则对应参数必须有类型注解，否则报错：ValueError: Arg city in docstring not found in function signature.

参数类型说明

参数类型来源于函数的类型注解。若删除类型注解，则工具描述中不包含参数类型。

参数默认值说明

• 参数无默认值 → 出现在 required 列表中。

• 参数有默认值 → 描述信息中包含 default 字段，且不在 required 列表中。

---

工具定义方式 2：使用 @tool 装饰器（推荐）

使用 @tool 装饰器可以自动将普通 Python 函数转化为智能体可调用的工具。此方式最直接、代码量极少，非常适合快速验证想法或创建参数简单的工具。

自定义工具描述：description

情况 1：仅提供 docstring 信息。@tool 同样要求遵循 Google docstring 规范，没有 docstring 则报错：ValueError: Function must have a docstring if description not provided.

情况 2：通过 description 参数显式指定，优先级高于 docstring：

from langchain.tools import tool

@tool(description="根据城市名称查询当日天气的工具")
def get_weather(city: str):
    """天气查询工具"""
    return f"{city}天气晴朗"

情况 3：parse_docstring=True 解析 docstring。默认情况下 @tool 会将 docstring 整体视为 description（包含 Args 段落也被塞进 description）。将 parse_docstring 设为 True，docstring 才会被解析并填充到各字段描述中：

from langchain.tools import tool

@tool(parse_docstring=True)
def get_weather(city: str, units: str = "celsius", include_forecast: bool = False) -> str:
    """
    获取当日天气，可选择是否同时查询未来五日天气预报
    Args:
        city: 城市
        units: 气温单位，可选：celsius-摄氏度，fahrenheit-华氏度
        include_forecast: 是否包含未来五日的天气预报
    """
    ...

注意：不使用 @tool 时，docstring 不合法会被视为普通文本作为 description；但使用 @tool 时 docstring 不合法会抛出异常：ValueError: Found invalid Google-Style docstring.

更改工具名称：name_or_callable

默认使用函数名作为工具名，可通过 @tool 传参 name_or_callable 更改（参数名可省略）：

@tool(name_or_callable="getWeather")  # 等价于 @tool("getWeather")
def get_weather(city: str):
    """天气查询工具"""
    return f"{city}天气晴朗"

注意：不要使用 config 或 runtime 作为参数名，这些是 LangChain 内部保留的。不推荐自定义工具名称。

自定义 args_schema

当工具参数变复杂、需要枚举值、范围限制或复杂业务验证时，Pydantic 模型是理想选择，能精确控制参数格式与验证规则。

方式 1：使用 Pydantic 模型定义

• ① BaseModel 基类：通过继承 BaseModel 声明字段结构、类型约束、默认值及校验规则。BaseModel 子类初始化时不接收位置参数，字段值必须以关键字参数传入。

• ② Field：定制字段的函数，用于设置默认值、描述等。每个字段的 description 至关重要，直接影响大模型理解参数含义的能力。

• ③ Literal：限定参数为固定选项（几个字面量之一）。非法值会触发 ValidationError。

from pydantic import BaseModel, Field
from typing import Literal

class WeatherInput(BaseModel):
    city: str = Field(default="北京", description="城市")
    unit: Literal["celsius", "fahrenheit"] = Field(default="celsius", description="气温单位")
    include_forecast: bool = Field(default=False, description="是否包含未来五日天气预报")

使用 Pydantic 定义 args_schema

通过 @tool(args_schema=PydanticModelCls) 将 Pydantic 模型与工具函数关联。Literal 类型会自动转换为 JSON Schema 的 enum 字段：

from pydantic import BaseModel, Field
from langchain.tools import tool
from typing import Literal

class WeatherInput(BaseModel):
    city: str = Field(default="北京", description="城市")
    unit: Literal["celsius", "fahrenheit"] = Field(default="celsius", description="气温单位")
    include_forecast: bool = Field(default=False, description="是否包含未来五日天气预报")

@tool(args_schema=WeatherInput)
def get_weather(city: str, unit: str = "celsius", include_forecast: bool = False) -> str:
    """获取当日天气，可选未来五日天气预报"""
    ...

方式 2：使用 JSON Schema 定义

也可直接用 JSON Schema 字典定义参数模式，因可基于数据库配置或用户输入在运行时动态生成，特别适合参数结构需动态生成的场景。

from langchain.tools import tool

weather_schema = {
    "type": "object",
    "properties": {
        "location": {"type": "string"},
        "units": {"type": "string"},
        "include_forecast": {"type": "boolean"}
    },
    "required": ["location", "units", "include_forecast"]
}

@tool(args_schema=weather_schema)
def get_weather(city: str, unit: str = "celsius", include_forecast: bool = False) -> str:
    """获取当日天气，可选未来五日天气预报"""
    ...

---

工具的应用案例

案例 1：使用 args_schema

通过 args_schema 给出明确的参数信息（包含 name、description、args_schema 三种自定义）：

from pydantic import BaseModel, Field
from langchain.tools import tool
from langchain.messages import HumanMessage

class WeatherSchema(BaseModel):
    city: str = Field(default="北京", description="城市名称")
    if_forecast: bool = Field(default=False, description="是否包含明日天气预报")

@tool("get_weather_and_forecast",
      description="查询当日天气，可以包含明日天气预报",
      args_schema=WeatherSchema)
def get_weather(city: str, if_forecast: bool):
    res = f"{city} 今天天气不错"
    if if_forecast:
        res += "\n明天也不错"
    return res

model_with_tools = model.bind_tools([get_weather])
messages = [HumanMessage("今天杭州天气如何？明天呢？")]
response = model_with_tools.invoke(messages)
messages.append(response)

for tool_call in response.tool_calls:
    if tool_call["name"] == "get_weather_and_forecast":
        tool_msg = get_weather.invoke(tool_call)
        messages.append(tool_msg)

final_response = model_with_tools.invoke(messages)
for msg in messages:
    msg.pretty_print()

案例 2：撰写 docstring

可在 docstring 中撰写参数描述，此时参数默认值和类型通过函数签名传递。必须将 parse_docstring 设为 True 才能正确解析：

from langchain.tools import tool
from langchain.messages import HumanMessage

@tool("get_weather_and_forecast", parse_docstring=True)
def get_weather(city: str = "北京", if_forecast: bool = False):
    """
    查询当日天气，可以包含明日天气预报
    Args:
        city: 城市名称
        if_forecast: 是否包含明日天气预报
    """
    res = f"{city} 今天天气不错"
    if if_forecast:
        res += "\n明天要下雨"
    return res

model_with_tools = model.bind_tools([get_weather])
messages = [HumanMessage("今天杭州天气如何？明天呢？")]
response = model_with_tools.invoke(messages)
messages.append(response)

for tool_call in response.tool_calls:
    if tool_call["name"] == "get_weather_and_forecast":
        tool_msg = get_weather.invoke(tool_call)
        messages.append(tool_msg)

final_response = model_with_tools.invoke(messages)
for msg in messages:
    msg.pretty_print()

案例 3：多工具调用（while 循环管理）

大模型调用工具是单次推理，每次运行调用一个工具；调用多个工具时需要开发者自行管理多次调用循环。下面用 while True 循环，直到模型不再发起工具调用：

from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage

@tool(parse_docstring=True)
def get_stock_price(company: str, timeframe: str = "today") -> str:
    """获取指定公司的股票价格信息
    Args:
        company: 公司名称（如：苹果公司, 微软公司, 谷歌公司）
        timeframe: 时间范围（today-今日, week-本周, month-本月）
    """
    mock_data = {
        "苹果公司": {"today": 185.20, "week": 183.50, "month": 180.75},
        "微软公司": {"today": 415.86, "week": 412.30, "month": 405.42},
    }
    if company in mock_data:
        price = mock_data[company].get(timeframe, "未知时间范围")
        return f"{company} {timeframe}价格: {price}美元"
    return f"未找到股票代码 {company} 的数据"

@tool(parse_docstring=True)
def search_news(company: str) -> str:
    """搜索指定公司的财经新闻
    Args:
        company: 公司名称
    """
    return f"{company} 的财经新闻..."

tools = [get_stock_price, search_news]
model_with_tools = model.bind_tools(tools)
message_list = [HumanMessage(content="苹果公司今天的股价是多少？最近有什么新闻？")]

while True:
    response = model_with_tools.invoke(message_list)
    message_list.append(response)
    if not response.tool_calls:
        print("没有工具调用，直接返回答案")
        break
    for tool_call in response.tool_calls:
        if tool_call["name"] == "get_stock_price":
            message_list.append(get_stock_price.invoke(tool_call))
        if tool_call["name"] == "search_news":
            message_list.append(search_news.invoke(tool_call))

for msg in message_list:
    msg.pretty_print()

案例 4：一次返回多个 tool_calls

模型可在单次响应中返回多个工具调用（response.tool_calls 是一个列表），遍历挨个调用即可。

---

拓展：强制使用工具（tool_choice）

bind_tools 可传参 tool_choice，控制是否强制使用工具。OpenAI 与 DeepSeek 官方 API 取值规定一致。

取值	含义
none	模型不会调用任何工具
auto	默认值，模型自主决定不调用或调用任意数量的工具
required	模型必须调用工具，数量不限
any	等价于 required
指定工具名（字符串）	强制调用特定的工具

# 不调用任何工具
model.bind_tools([get_weather], tool_choice="none")
# 模型自主决定（默认）
model.bind_tools([get_weather], tool_choice="auto")
# 必须调用工具
model.bind_tools([get_weather], tool_choice="required")
# 强制调用特定工具
model.bind_tools([get_weather1, get_weather2], tool_choice="get_weather2")

---

实践经验总结

1. 清晰的描述：docstring 要清晰明确，让 Agent 能理解工具用途与调用时机。

2. 功能单一：每个工具只做一件事，避免一个工具包揽多种逻辑。

3. 如何处理工具失败——三层防护：

   • 第 1 层：工具内部处理（try/except 捕获异常，返回友好错误信息）

   • 第 2 层：Agent 级重试（通过 prompt 引导）

   • 第 3 层：调用级重试（使用 @retry）

# 第 1 层：工具内部处理
@tool
def divide(a: float, b: float) -> str:
    """除法计算
    Args:
        a: 被除数
        b: 除数"""
    try:
        if b == 0:
            return "错误：除数不能为零"
        return f"{a} / {b} = {a / b}"
    except Exception as e:
        return f"计算错误：{e}"

# 第 3 层：调用级重试（tenacity）
from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(3))  # 最多尝试 3 次
def call_agent(question):
    return agent.invoke({"messages": [{"role": "user", "content": question}]})

1. 返回字符串（str）：LangChain 工具强烈建议返回字符串。大模型（LLM）本质只"吃"文本；返回含中文的 dict 时，强制转字符串默认可能用 Unicode 编码，易干扰模型。通过 json.dumps(..., ensure_ascii=False) 确保喂给模型的是干净直观的纯文本。

import json

# ✅ 好：返回 JSON 字符串
@tool
def get_user_info(user_id: str) -> str:
    """获取用户信息"""
    user = {"id": user_id, "name": "张三"}
    return json.dumps(user, ensure_ascii=False)

1. 选择同步 vs 异步：

类型	适用场景
同步工具	简单场景、CPU 密集型任务
异步工具	IO 密集型（API 调用、数据库、文件操作）

第07章 智能体

1. 理解 Agents

通用人工智能（AGI）是 AI 的终极形态，而构建智能体（Agent）则是当下 AI 工程应用的"终极形态"——Agent 是大模型应用开发的核心。

1.1 什么是 Agent

在大模型应用开发中，智能体通常指一种以大语言模型为推理与决策核心，结合记忆、工具调用与环境交互能力，能够进行规划决策并执行复杂任务以达成目标的软件系统。

Agent 的关键能力：理解用户问题、如何拆解任务、判断是否需要工具、需要调用哪些工具、如何利用好工具结果生成回答并推进任务。

1.2 Agent 的核心组件

实际开发中几个要素并不需要同时出现：

组件	必要性
行动（Action）	必须
工具（Tool）	几乎总是存在
规划决策（Planning）	有条件存在
记忆（Memory）	最容易被省略

1.3 Agent 的创建与调用

历史上的调用（LangChain 0.x）：当时设计理念是"针对场景设计特定 Agent"，存在碎片化问题——ReAct 用 create_react_agent、结构化输出用 create_structured_chat_agent、工具调用用 create_tool_calling_agent。这种方式带来三个问题：心智负担高、可组合性差、生态碎片化。

❌ v0.x 复杂方式（多步骤）：

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.prompts import PromptTemplate

model = ChatOpenAI(model="gpt-4o-mini")
prompt = PromptTemplate.from_template("""
You are a helpful assistant.
Tools: {tools}
Tool Names: {tool_names}
{agent_scratchpad}
""")
agent = create_react_agent(llm=model, tools=tools, prompt=prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
result = executor.invoke({"input": "问题"})

全新的调用（LangChain 1.0+）：团队彻底重构，将所有 Agent 创建方式统一为一个入口 create_agent()，取代了 create_react_agent、create_json_agent、create_tool_calling_agent 等分支函数。底层通过"中间件机制（Middleware）"和"标准模型接口（invoke / stream）"实现全局统一。

✅ v1.x 简洁方式：

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent

model = init_chat_model("gpt-4o-mini", model_provider="openai")
agent = create_agent(
    model=model,
    tools=[tool1, tool2],
    system_prompt="Agent 的行为指令"  # 可选
)
result = agent.invoke({"messages": [{"role": "user", "content": "问题"}]})

---

2. Agent 基本用法 1：模型的传入方式

在 LangChain 1.2 中，create_agent 是构建智能体的核心方式，底层基于 LangGraph 实现。

create_agent 完整参数签名：

from langchain.agents import create_agent

create_agent(
    model: str | BaseChatModel,             # 必需：聊天模型
    tools: List[BaseTool],                  # 必需：工具列表
    *,
    system_prompt: str = "",                # 系统提示词
    middleware: Sequence[AgentMiddleware] = (),  # 中间件
    interrupt_before: List[str] = None,     # 在某些工具前暂停（人机协作）
    interrupt_after: List[str] = None,      # 在某些工具后暂停
    debug: bool = False,                    # 调试模式
    name: str | None = None,                # 设置模型名称
)

2.1 传入模型字符串

Agent 根据传入的模型字符串，自主创建模型对象。

from langchain.agents import create_agent
from dotenv import load_dotenv

load_dotenv(override=True)
agent = create_agent("deepseek-v4-flash")
print(type(agent))  # <class 'langgraph.graph.state.CompiledStateGraph'>

由上可知，agent 本质上是 LangGraph 的 CompiledStateGraph 实例，底层是一个图结构。

2.2 传入模型对象

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from dotenv import load_dotenv
import os

load_dotenv(override=True)
model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)
agent = create_agent(model)
print(type(agent))  # CompiledStateGraph

---

3. Agent 基本用法 2：如何调用 Agent

agent.invoke() 是最基本的同步调用方法，会阻塞程序执行直到返回最终结果。

• 输入：字典类型，通过 messages 字段传递消息列表：{"messages": [{"role": "...", "content": "..."}]}。

• 输出：底层可能经历多轮交互，返回完整的消息列表，封装在字典的 messages 字段中。

response = agent.invoke({"messages": [...]})
# response 是字典类型
{
    "messages": [
        HumanMessage(...),       # 用户问题
        AIMessage(...),          # AI 工具调用
        ToolMessage(...),        # 工具返回结果
        AIMessage(...)           # 最终回答 ← 通常取这个
    ]
}
# 获取最终回答
final_answer = response['messages'][-1].content

举例 2：在 messages 开头加入 system 角色消息定义 Agent 行为

from rich import print as rprint

agent = create_agent(model)
resp = agent.invoke({
    "messages": [
        {"role": "system", "content": "你是一个小学数学老师，耐心，幽默，讲解深入浅出"},
        {"role": "user", "content": "100加上50等于多少？"}
    ]
})
rprint(resp)

invoke 调用的核心是输入一系列消息，每条消息包含 role（user / assistant / system / tool）和 content。

---

4. Agent 基本用法 3：绑定工具

只有接入了工具，create_agent 创建的 Agent 才算完整。工具可以是 LangChain 内置的，也可以是自定义的。

LangChain 内置工具列表：https://docs.langchain.com/oss/python/integrations/tools（典型如 TavilySearch 网络搜索等）。Agent 支持静态和动态绑定工具（动态绑定需要中间件）。

4.1 基本用法

举例 2：接入内置工具 TavilySearch

先在 Tavily 官网注册获得 API-KEY，写入 .env 的 TAVILY_API_KEY 变量。

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_tavily import TavilySearch
from dotenv import load_dotenv
import os

load_dotenv(override=True)
model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)
web_search = TavilySearch(max_results=2)
agent = create_agent(model=model, tools=[web_search])
result = agent.invoke(
    {"messages": [{"role": "user", "content": "请帮我查询2024年诺贝尔物理学奖得主是谁？"}]}
)
print(result['messages'][-1].content)

输出：

2024年诺贝尔物理学奖得主是：
- John J. Hopfield
- Geoffrey E. Hinton
授奖理由是："为实现机器学习的人工神经网络奠定基础性的发现和发明"。

这是一次典型的 Function calling 执行流程：用户首次发起消息后共产生 4 条消息——HumanMessage（用户）、AIMessage（function call）、ToolMessage（function response）、AIMessage（final response）。

注意：只给 Agent 需要的工具，工具太多会混淆。一般 2–5 个工具最佳。

4.2 工具调用流程分析

LangChain 的 Agent 由一个基于 LangGraph 的图结构编排执行流程，本质上就是经典的 ReAct 结构：一个具备"思考（Thought）— 行动（Action）— 观察（Observation）"不断循环的自主工作者。

完整流程：用户问题 → AI 思考 → 调用工具 → 观察结果 → 继续思考 → ... → 最终答案。

4.3 重试机制

Agent 可以在工具调用结果不满足要求时自主重试（通过 system_prompt 指导）。

from langchain.agents import create_agent
from langchain.tools import tool
from langchain.messages import SystemMessage, HumanMessage

flag = 0

@tool
def get_weather(city: str):
    """
    天气查询工具
    Args:
        city: 城市名称
    """
    global flag
    flag += 1
    if flag < 3:
        return "TEMP_UNAVAILABLE: 天气服务暂时不可用，请稍后重试"
    return f"{city}今天天气挺好"

messages = [
    SystemMessage("""
    你是一个天气助手。
    当工具返回以 'TEMP_UNAVAILABLE:' 开头的结果时，
    说明是临时故障，不要立即放弃；
    你应再次调用同一个工具，最多重试 3 次。
    如果 3 次后仍失败，再向用户说明服务暂时不可用。
    """),
    HumanMessage("你好，杭州今天的天气如何？")
]
agent = create_agent(model, tools=[get_weather])
response = agent.invoke({"messages": messages})

模型三次调用 get_weather，最终获得满意结果。

4.4 常见问题

问题 1：Agent 如何选择工具？依据工具的 docstring：AI 根据问题内容、每个工具的描述，自动选择最匹配的工具。

问题 2：Agent 为什么没有调用工具？原因：工具 docstring 不清晰、问题表述不明确、模型认为不需要工具。

问题 3：Agent 选错工具？原因：多个工具功能描述相似、工具太多导致混淆。解决：只给必要的工具、工具描述要有明确区分、在 system_prompt 中说明工具使用场景。

# ❌ 不好：太模糊
@tool
def tool1(x: str) -> str:
    """做一些事情"""

# ✅ 好：明确区分
@tool
def get_weather(city: str) -> str:
    """
    获取指定城市的实时天气信息
    Args:
        city: 城市名称，如"北京"、"上海"
    """

问题 4：如何知道 Agent 何时完成？当 AIMessage 不包含 tool_calls 时表示完成。

问题 5：Agent 可以调用多少次工具？默认没有限制，直到得到最终答案，但可能超时、达到 token 限制或模型决定停止。

问题 6：如何限制工具调用次数？可通过 recursion_limit 限制步数：

config = {"recursion_limit": 5}  # 最多 5 步
response = agent.invoke(input, config=config)

---

5. Agent 高级用法 1：设置 Agent 名称

创建 Agent 时，LangChain 允许指定其 name，输出的 AIMessage 会携带 Name 信息。

agent = create_agent(model=model, name="chat_assistant")
response = agent.invoke({"messages": ["你好"]})

name 的经典使用场景：流式输出归因、消息身份标记、调试与 trace 可读性、组件化封装、前端展示与运行态可观测性、稳定的运行时身份标识（尤其在 Multi-Agent 场景）。

---

6. Agent 高级用法 2：系统提示词

通过 system_prompt 设置 SystemMessage，定义 Agent 行为，类型可以是 str 或 SystemMessage。

使用建议：明确说明 Agent 角色、定义输出格式、说明何时使用工具。

from langchain_tavily import TavilySearch

web_search = TavilySearch(max_results=2)
agent = create_agent(
    model=model,
    tools=[web_search],
    system_prompt="你是一名多才多艺的智能助手，可以调用工具帮助用户解决问题。"
)
result = agent.invoke(
    {"messages": [{"role": "user", "content": "请帮我查询2026年足球世界杯是哪个国家举办的？"}]}
)
print(result['messages'][-1].content)

提示词设置有两种方式：静态设置和动态设置（动态设置需要借助中间件）。

---

7. Agent 高级用法 3：结构化输出

结构化输出是 Agent 的核心功能之一，允许 Agent 以特定、可预测的格式返回数据（Pydantic 模型、JSON 对象或数据类），无需复杂解析。

通过 response_format 参数设置输出模式（Schema），模型生成结构化数据时，系统会自动捕获、验证并将结果存入 Agent 状态的 structured_response 键。

7.1 模型 vs Agent 的结构化输出对比

维度	模型结构化输出	Agent 结构化输出
操作对象	作用于大模型对象	作用于 Agent
解析时机	每次模型调用生成 AIMessage 时解析	仅在 Agent 决定"任务结束"并输出最终答案时解析
数据流转	模型 → 结构化对象	模型 → 工具 → 反思 → ... → 结构化对象
绑定方式	使用 with_structured_output	使用 response_format 参数
适用场景	单次、确定性任务（提取字段、翻译、分类）	多步、复杂推理任务（查文档后汇总报表）

7.2 结构化输出的 4 种策略

create_agent 的 response_format 参数支持四种策略：

① ProviderStrategy：使用模型提供商的原生结构化输出能力，在模型生成阶段就强制保证输出格式，适用于支持原生结构化输出的模型（OpenAI、Anthropic Claude、xAI Grok）。

from pydantic import BaseModel, Field
from langchain.agents import create_agent
from langchain.agents.structured_output import ProviderStrategy
from langchain.messages import HumanMessage

class ContactInfo(BaseModel):
    """用户的联系方式"""
    name: str = Field(description="用户姓名")
    email: str = Field(description="用户邮箱地址")
    phone: str = Field(description="用户的手机号")

agent = create_agent(
    model=model,
    response_format=ProviderStrategy(ContactInfo)
)
response = agent.invoke({
    "messages": [HumanMessage("从这段话中抽取结构化信息：小明的邮箱地址为：shkstart@atguigu.com，手机号：12345678912")]
})

② ToolStrategy：对于不支持原生结构化输出的模型，采用工具调用方式实现。兼容绝大多数支持工具调用的现代模型，核心原理是动态创建一个"虚拟工具"，引导模型调用它从而间接产生结构化数据。

from langchain.agents.structured_output import ToolStrategy

agent = create_agent(
    model=model,
    tools=[search_tool],
    response_format=ToolStrategy(ContactInfo)
)
result = agent.invoke({
    "messages": [{"role": "user", "content": "联系人信息: John Doe, john@atguigu.com, (010) 56253825"}]
})
print(result["structured_response"])
# name='John Doe' email='john@atguigu.com' phone='(010) 56253825'

③ type / AutoStrategy：直接传入一个定义类型时，LangChain 自动包装为 AutoStrategy，触发自动选择——模型支持原生结构化输出则用 ProviderStrategy，否则用 ToolStrategy。

④ None：默认配置，不以结构化输出，以自然语言响应。

总结：在实际 Agent 开发中，若使用结构化输出，推荐使用 ToolStrategy，因为它兼容性最好。

7.3 ToolStrategy 使用详解

ToolStrategy 通过工具调用实现结构化输出，LangChain 会在消息列表末尾追加一条 ToolMessage 让链路完整（实际没有工具执行，是伪消息）。它包含三个参数：

参数	说明
schema（必需）	支持 Pydantic、TypedDict、JSON Schema、@dataclass，还支持 Union[类型1, 类型2]
tool_message_content（可选）	自定义生成结构化输出时会话历史中记录的提示信息，默认使用标准响应语句
handle_errors（可选）	指定数据校验失败时的重试策略，默认 True

7.3.1 schema 参数（4 种 Schema）

模式 1：Pydantic 类型（支持数据验证，优先推荐）：

from pydantic import BaseModel, Field
from langchain.agents.structured_output import ToolStrategy
from langchain.agents import create_agent
from langchain.messages import HumanMessage

class ContactInfo(BaseModel):
    """用户的联系方式"""
    name: str = Field(description="用户姓名")
    email: str = Field(description="用户邮箱地址")
    phone: str = Field(description="用户的手机号")

agent = create_agent(model=model, response_format=ToolStrategy(ContactInfo))
response = agent.invoke({
    "messages": [HumanMessage("从这段话中抽取结构化信息：小明的邮箱地址为：songhk@atguigu.com，手机号：12345678912")]
})

模式 2：TypedDict 类型（带类型提示的字典结构，不支持运行时验证）。字段定义采用 Annotated[类型, 默认值, "描述"] 格式，可选字段使用 Optional 包装。

模式 3：JSON Schema 类型（标准 JSON Schema 字典，适合跨语言交互或复杂数据约束）。title、description、type、properties、required 是遵循 JSON Schema 规范的固定关键字。

from langchain.agents import create_agent

json_schema = {
    "title": "ContactInfo",
    "description": "用户的联系方式",
    "type": "object",
    "properties": {
        "name": {"description": "用户姓名", "type": "string"},
        "email": {"description": "用户邮箱地址", "type": "string"},
        "phone": {"description": "用户的手机号", "type": "string"}
    },
    "required": ["name", "email", "phone"]
}
agent = create_agent(model=model, response_format=ToolStrategy(json_schema))

模式 4：@dataclass 类型（Python 3.7+ 装饰器，简化数据存储类定义）。

多 schema 联合模式：Union[类型1, 类型2]，LLM 根据输入智能选择最合适的一个模型生成结构化输出，但最终只会输出一种类型。

结构化输出与 system_prompt 的注意事项：

• 在系统提示词中应最后提示结构化输出结果，若先提示结构化输出，可能导致部分工具不再被调用。

• 系统提示词中应加入"未找到用户"时的处理提示，避免程序一直调用工具尝试查找。

7.3.2 自定义工具消息：tool_message_content

通过 tool_message_content 可定制追加的 ToolMessage 内容，好处：① 用更自然的消息替代原始数据；② 用简短确认信息替代长数据块，减少 token 消耗。

agent = create_agent(
    model=model,
    response_format=ToolStrategy(ContactInfo, tool_message_content="已成功抽取信息")
)

无论 tool_message_content 如何设置，成功提取的结构化数据都会正确存入 result["structured_response"]，自定义消息仅影响对话历史中的一条记录。

7.3.3 错误处理：handle_errors

取值	说明
True（默认）	捕获所有异常，用内置模板提示模型重试
False	关闭自动重试，任何异常直接抛出，中断程序
"自定义字符串"	捕获所有异常，用固定字符串作为错误消息
ExceptionType	仅捕获指定类型（或元组）异常重试，其他直接抛出
callable	灵活性最高，自定义函数处理异常，可返回差异化提示

默认情况下 LangChain 处理两类异常：MultipleStructuredOutputsError（返回的工具调用请求数量大于 1）、StructuredOutputValidationError（输出格式不符合结构化要求）。

from langchain.agents.structured_output import (
    ToolStrategy, StructuredOutputValidationError, MultipleStructuredOutputsError
)

def custom_error_handler(error: Exception) -> str:
    if isinstance(error, StructuredOutputValidationError):
        return "数据格式有误，请检查字段是否符合要求。"
    elif isinstance(error, MultipleStructuredOutputsError):
        return "检测到多个响应，请选择最相关的一个进行返回。"
    else:
        return f"Error: {error}"

agent = create_agent(
    model=model,
    response_format=ToolStrategy(
        Union[ContactInfo, EventDetails],
        tool_message_content="提取完成！",
        handle_errors=custom_error_handler
    )
)

---

8. Agent 高级用法 4：流式输出及模式

8.1 流式输出的说明

通过 invoke 调用 Agent 时，内部可能经历多次调用，长时间看不到执行情况。流式调用（渐进式显示）可实时显示 Agent 运行过程中的更新，在处理 LLM 延迟时尤其有效。

设置方式：agent.stream(stream_mode=指定模式)，模式有：values、updates（默认）、messages、custom、checkpoints、tasks、debug。

8.2 具体的输出模式

• values 模式：每个步骤执行后输出完整状态信息。

• updates 模式（默认）：每个步骤执行后只增量更新发生变化的内容。

• messages 模式：输出流式返回的 Token 及元数据（来自哪个节点），用于类似 ChatGPT 的打字机效果。

• tasks 模式：输出当前 task 任务的开始/结束时间、结果和错误信息。

• debug 模式：类似 tasks，额外输出任务步骤、时间戳、task 类型。

• checkpoints 模式：每当 checkpoint 被创建时触发输出，需启用 checkpointer。

• custom 模式：开发者通过 get_stream_writer 在工具或节点内自定义发送数据。

for chunk in customer_service_agent.stream(
    {"messages": [{"role": "user", "content": "查询客户ID为 CUST123456 的完整信息和可用优惠"}]},
    stream_mode="updates"
):
    print(chunk)

custom 模式（在工具内用 get_stream_writer 发送业务进度）：

from langchain.agents import create_agent
from langgraph.config import get_stream_writer
from langchain.tools import tool
import time

@tool
def generate_sales_report() -> str:
    """生成销售报告"""
    writer = get_stream_writer()
    writer({"type": "生成销售报告", "message": "开始生成销售报告"})
    for i in range(1, 4):
        time.sleep(0.5)
        writer({"type": "生成销售报告", "message": f"生成销售报告进度百分比：{i * 25}%"})
    writer({"type": "生成销售报告", "message": "报告生成完成"})
    return "销售报告：总收入150万元，同比增长12%"

reporting_agent = create_agent(model=model, tools=[generate_sales_report, generate_inventory_report])
for chunk in reporting_agent.stream(
    {"messages": [{"role": "user", "content": "生成销售报告和库存报告"}]},
    stream_mode="custom"
):
    print(chunk)

8.3 流式输出模式总结

模式	输出内容	使用场景
values	每个步骤执行后输出完整的状态信息	每一步都要获取完整状态、状态持久化场景
updates（默认）	每个步骤执行后只增量更新发生变化的内容	监控 Agent 执行进度（观察调用工具、工具结果）
messages	流式返回的 Token 及元数据（来自哪个节点 model/tool）	实现 ChatGPT 的打字机效果，最佳实时体验
tasks	当前 task 任务开始/结束时间，包含结果和错误信息	监控任务的生命周期
debug	类似 tasks，多输出任务步骤、时间戳、task 类型	调试、监控 task 任务生命周期
checkpoints	当 checkpoint 被创建时触发输出，含检查点状态	状态持久化、工作流恢复、分布式执行跟踪
custom	通过 get_stream_writer 在工具或节点内自定义发送的数据	输出业务进度（如"已处理10/100条"）、自定义日志或指标

选择建议：实时对话交互 → messages；观察思考与执行步骤 → updates；查看每一步状态 → values / tasks / debug；工具执行时输出自定义业务日志 → custom。

多模式组合：模式可组合使用，如 stream_mode=["tasks", "updates"]，此时遍历返回 dict，key 是模式名，value 是该模式输出结果。

---

9. 实战：多功能智能助手

需求：开发一个支持天气查询、数学计算、时间查询、货币转换、信息搜索的多功能智能助手。

9.1 工具定义

from langchain_core.tools import tool
import math
from datetime import datetime, timedelta

@tool
def get_weather(city: str) -> str:
    """获取指定城市的实时天气信息
    Args:
        city: 城市名称，如"北京"、"上海"
    Returns:
        包含温度、天气状况、空气质量的详细信息
    """
    weather_db = {
        "北京": "多云，15-22℃，空气质量良，湿度 45%",
        "上海": "晴天，18-25℃，空气质量优，湿度 60%",
    }
    result = weather_db.get(city)
    return f"{city}：{result}" if result else f"暂不支持查询{city}的天气信息"

@tool
def calculator(expression: str) -> str:
    """执行数学计算
    Args:
        expression: 数学表达式，如 2 + 3 * 4、sqrt(16)、2 ** 10
    """
    try:
        safe_functions = {"sqrt": math.sqrt, "pow": pow, "abs": abs, "pi": math.pi, "e": math.e}
        result = eval(expression, {"__builtins__": {}}, safe_functions)
        return f"{expression} = {result}"
    except Exception as e:
        return f"计算出错：{str(e)}"

@tool
def get_time_info(query_type: str = "current") -> str:
    """获取时间相关信息
    Args:
        query_type: 查询类型（current/date/tomorrow/yesterday/weekday）
    """
    now = datetime.now()
    if query_type == "current":
        return now.strftime("当前时间：%Y年%m月%d日 %H:%M:%S")
    elif query_type == "tomorrow":
        return (now + timedelta(days=1)).strftime("明天是：%Y年%m月%d日")
    return f"不支持的查询类型：{query_type}"

@tool
def convert_currency(amount: float, from_curr: str, to_curr: str) -> str:
    """货币转换工具
    Args:
        amount: 金额数值
        from_curr: 源货币代码（CNY/USD/EUR/GBP/JPY/HKD）
        to_curr: 目标货币代码
    """
    exchange_rates = {"CNY": 1.0, "USD": 0.14, "EUR": 0.13, "GBP": 0.11, "JPY": 20.8, "HKD": 1.09}
    from_curr, to_curr = from_curr.upper(), to_curr.upper()
    cny_amount = amount / exchange_rates[from_curr]
    result_amount = cny_amount * exchange_rates[to_curr]
    return f"{amount} {from_curr} = {result_amount:.2f} {to_curr}"

9.2 Agent 创建（封装为类）

from langchain.agents import create_agent

class SmartAssistant:
    """多功能智能助手"""
    def __init__(self):
        self.model = model
        self.tools = [get_weather, calculator, get_time_info, convert_currency, search_info]
        system_prompt = """你是一个多功能智能助手，可以帮助用户：
          🌤 查询天气：使用 get_weather 工具
          🔢 数学计算：使用 calculator 工具
          ⏰ 时间查询：使用 get_time_info 工具
          💱 货币转换：使用 convert_currency 工具
          重要提示：仔细阅读用户问题，确定需要使用哪个工具。请始终使用中文回答。"""
        self.agent = create_agent(
            model=self.model,
            tools=self.tools,
            system_prompt=system_prompt
        )
        self.messages = []

    def chat(self, user_input: str) -> str:
        """对话接口"""
        self.messages.append({"role": "user", "content": user_input})
        result = self.agent.invoke({"messages": self.messages})
        self.messages = result["messages"]
        for msg in reversed(self.messages):
            if msg.type == "ai" and msg.content:
                return msg.content
        return "抱歉，我无法处理这个请求。"

    def reset(self):
        """重置对话历史"""
        self.messages = []

9.3 主程序

def main():
    assistant = SmartAssistant()
    print("🤖 多功能智能助手（LangChain 1.2）")
    demos = ["北京今天天气怎么样？", "帮我算一下 (25 + 17) * 3", "现在几点了？", "100 美元等于多少人民币？"]
    for demo in demos:
        print(f"👤 {demo}")
        response = assistant.chat(demo)
        print(f"🤖 {response}\n")
    # 交互模式
    while True:
        user_input = input("\n👤 你: ")
        if user_input.lower() == 'quit':
            break
        if user_input.lower() == 'reset':
            assistant.reset()
            continue
        if not user_input.strip():
            continue
        response = assistant.chat(user_input)
        print(f"🤖 助手: {response}")

if __name__ == "__main__":
    main()

---

本章核心要点回顾：

• LangChain 1.x 用统一的 create_agent(model, tools, system_prompt, ...) 取代了 0.x 碎片化的多种创建函数，底层是 LangGraph 的 CompiledStateGraph。

• Agent 调用通过 invoke({"messages": [...]})，输入是消息列表，输出是完整消息列表（result['messages'][-1].content 为最终回答）。

• 工具通过 @tool 装饰器自定义或使用内置工具（如 TavilySearch），工具描述（docstring）决定 Agent 选工具的准确性，2–5 个工具最佳。

• 执行流程本质是 ReAct"思考—行动—观察"循环，直到 AIMessage 不含 tool_calls 结束；可通过 recursion_limit 限制步数。

• 结构化输出推荐用 ToolStrategy(Schema)，支持 Pydantic / TypedDict / JSON Schema / @dataclass 及 Union 多 schema；结果存入 result["structured_response"]。

• 流式输出通过 agent.stream(stream_mode=...) 实现实时反馈，默认 updates，交互场景选 messages，可多模式组合。

第08章 中间件

1. 中间件概述

1.1 什么是中间件

在 create_agent() 的底层运行机制中，有几个重要的核心组件：

组件	比喻	职责
模型(Model)	大脑	理解任务与决策推理
工具(Tools)	手脚	执行模型自身做不到的外部操作
系统提示词(System Prompt)	角色	告诉模型该怎么想、参考什么上下文
中间件(Middleware)	中枢	在执行流程的关键节点拦截、控制、增强

Middleware 本质上是 Agent 执行过程中的钩子函数(hook)，是 LangChain 1.x 的“王牌”工程化能力。

钩子(Hook) 是框架或系统在某些关键执行点暴露的扩展接口。开发者可以“挂上”自己的逻辑，在那些点上插入、修改或替换行为，而无需改变主流程代码——就像在流水线上设置“检查点”或“插入器”。

借助中间件，开发者可以高度定制和控制 Agent 运行的每一个环节，是处理 Agent 生命周期的标准方式。

无中间件时，Agent 架构简单直接；添加中间件后，Agent 架构变为在执行循环的关键节点（如“模型调用前”、“模型调用后”、“工具调用前后”）设置钩子，在不改主体逻辑的情况下实现策略与治理。

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

agent = create_agent(
    model="gpt-5.4-mini",
    tools=[...],
    middleware=[
        SummarizationMiddleware(...),
        HumanInTheLoopMiddleware(...)
    ],
)

1.2 为什么需要中间件

无中间件时 Agent 流程较为直接：

用户输入 → 拼接提示词/消息 → 调用模型 → 如有需要调用工具 → 返回结果

这种方式对简单场景够用，但真实项目常遇到额外需求：

• 根据问题复杂度动态切换模型

• 限制某些用户只能调用部分工具

• 工具报错时自动重试或返回兜底结果

• 模型调用前插入额外的系统提示

• 记录每一步执行日志便于排查

• 敏感信息出现时阻断执行

• 正式执行工具前增加人工审批

这些需求的共同特点：不是核心业务逻辑，但会影响 Agent 的执行过程。直接写进主流程会带来四大问题：

1. 主流程迅速变乱：日志、鉴权、重试、风控、审计全塞进去，主逻辑臃肿。

2. 横切需求难以复用：日志、重试、权限控制是多个 Agent 都需要的，写死会导致大量重复代码。

3. 流程控制粒度不够细：没有统一拦截点，只能手动改主流程，麻烦易错。

4. 后期维护成本高：新增规则往往要修改多处代码。

中间件的价值就在于把这些与业务无关、但与执行过程强相关的横切逻辑从 Agent 主流程中分离出来，实现“拦截流程、修改流程、增强流程”。它能实现：

• 日志与分析 —— 追踪行为、调试、性能监控

• 转换 —— 修改提示词、工具选择、输出格式

• 容错 —— 重试、降级、早期终止

• 安全 —— 限流、守护规则、PII 检测

1.3 中间件的分类

按是否已由 LangChain 定义分类：

• 自定义中间件：开发者自定义，实现更灵活的 Agent 行为管理

• 内置中间件：LangChain 实现并提供的

• 模型供应商定制的中间件：依赖特定模型服务（非本课重点）

• 和模型供应商无关的中间件

LangChain 提供的与供应商无关的内置中间件文档：Overview - Docs by LangChain

1.4 和模型供应商无关的内置中间件分类

LangChain 提供的与供应商无关的内置中间件分为六大类别：

类型 1：成本与资源控制类 —— 控成本、控配额、避免无限调用

中间件	作用
Model call limit	限制模型调用次数，防止反复请求 LLM 导致费用失控
Tool call limit	限制工具调用次数，避免 Agent 无限试错、死循环
Summarization	上下文快满时自动总结历史，减少 token 消耗
Context editing	裁剪上下文、清理工具调用痕迹，节省上下文成本

适用：生产环境的成本治理、配额治理、长会话优化、SaaS 产品控费。

类型 2：稳定性与容错保障类 —— 保证服务不中断、失败后尽量自动恢复

中间件	作用
Model fallback	主模型失败时切换备用模型
Model retry	模型调用失败后自动重试
Tool retry	工具调用失败后自动重试

适用：线上生产系统，尤其是多模型、多工具依赖的 Agent（高可用、容灾、鲁棒性建设）。

类型 3：安全与合规风控类 —— 让 Agent 可控、可审、合规

中间件	作用
Human-in-the-loop	在关键工具调用前暂停，等人工审批
PII detection	检测和处理个人敏感信息
Model/Tool call limit	也可归入风控，防止异常滥用

适用：企业内部系统、客服系统、审批流、数据查询类 Agent（发邮件、调数据库、调财务/人事系统、导出敏感信息等）。

类型 4：决策增强与智能编排类 —— 提升 Agent 的决策质量和任务拆解能力

中间件	作用
To-do list	给 Agent 增加任务规划、分步骤执行和状态跟踪能力
LLM tool selector	工具太多时，用子模型筛选最相关的几个工具交给主模型
Subagent	允许生成子 Agent，把复杂任务拆给不同角色处理

适用：研究型 Agent、多步骤分析、报告生成、多角色协作、长链路任务编排（增强“脑子”与“组织能力”）。

类型 5：执行能力扩展类 —— 给 Agent 更多“手脚”

中间件	作用
Shell tool	给 Agent 持久 shell，可执行命令
File search	给 Agent 文件搜索能力，能做 Glob/Grep
Filesystem	给 Agent 文件系统读写与长期存储能力

适用：工程 Agent、代码 Agent、本地自动化 Agent、运维 Agent（把 Agent 从“纯推理”扩展成“能操作环境的执行体”）。

类型 6：开发调试与测试辅助类 —— 方便开发、测试、验证 Agent 行为

中间件	作用
LLM tool emulator	用 LLM 模拟工具执行，便于测试（最典型）
Summarization	可辅助调试长会话表现
Context editing	可用于测试上下文裁剪效果
Human-in-the-loop	常用于调试高风险步骤

适用：开发阶段快速验证流程、做 mock、减少真实工具依赖。

---

2. 常用内置中间件的使用

LangChain 1.0 提供了 16 个预置中间件，开箱即用。本节详细讲解常用内置中间件。

注：各中间件参数说明不保证包含完整参数列表，不常用或被标记为过时的参数被省略。

2.1 SummarizationMiddleware 中间件

• 作用：对历史消息列表进行摘要总结，达到压缩上下文的效果。

• 原理：在达到触发条件时，调用大模型对历史消息进行摘要，将摘要结果作为 HumanMessage 放到消息列表最开始的位置。

参数说明

参数	说明
model	用于摘要的模型。可以是模型名称或模型对象；传名称时底层调用 init_chat_model 初始化
trigger	摘要触发条件，是一个列表，每个元素对应一个条件，任一条件满足即触发。包括：tokens（历史 token 累计达到该值）、messages（历史消息条数达到该值）、fraction（达到 max_input_tokens*fraction 触发）。若用 fraction，要求模型 profile 含 max_input_tokens，DeepSeek 模型 profile 为空时需手动添加（DeepSeek-V3.2 上下文 128K）
keep	摘要时保留的原始消息。支持 tokens/messages/fraction 三种条件，同一时间只接收一种
token_counter	统计 token 数量的函数，默认使用 count_tokens_approximately，一般不改
summary_prompt	摘要时的自定义提示词，需含 {messages} 占位符；不指定则用内置提示词
trim_token_to_summarize	摘要时历史消息的最大 token 数，超出则裁剪，默认 "4000"

count_tokens_approximately 对纯文本消息的大致思路：先统计字符数 len(字符串)，再除以每个 token 大致的字符数转为粗略 token 数，再加额外开销作为估算。

举例 1：测试 trigger、keep 参数

为支持 fraction，需自定义 profile 指定 max_input_tokens：

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

load_dotenv(override=True)

custom_profile = {
    "max_input_tokens": 128_000
}
model = init_chat_model(
    model="deepseek-v4-flash",
    model_provider="deepseek",
    profile=custom_profile,
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL")
)

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langchain.messages import SystemMessage, HumanMessage, AIMessage

messages = [
    SystemMessage("你是个非常友好的AI助手"),
    HumanMessage("你好啊，我是老王，你是谁？"),
    AIMessage("你好老王，我是小王"),
    HumanMessage("好的小王，很高兴认识你"),
    AIMessage("你高兴得太早了"),
    HumanMessage("呵呵，你什么意思")
]
agent = create_agent(
    model="deepseek-v4-flash",
    middleware=[
        SummarizationMiddleware(
            model=model,
            trigger=[
                ("tokens", 100),
                ("messages", 6),
                ("fraction", 0.001)
            ],
            keep=("messages", 2)
        )
    ]
)
response = agent.invoke({"messages": messages})
for msg in response["messages"]:
    msg.pretty_print()

分析：

1. 通过自定义 profile 指定 max_input_tokens，才能用 fraction 作为度量，否则报错。

2. 三个触发条件至少满足一个，即触发摘要。

3. 摘要结果作为 HumanMessage，传入消息列表头部。

4. keep 要求保留两条消息，则最新两条消息原样保留。

举例 2：测试 summary_prompt 参数

agent = create_agent(
    model="deepseek-v4-flash",
    middleware=[
        SummarizationMiddleware(
            model=model,
            trigger=[
                ("tokens", 100),
                ("messages", 6),
                ("fraction", 0.0001)
            ],
            keep=("messages", 2),
            summary_prompt="对历史消息摘要，消息列表如下\n{messages}"
        )
    ]
)

分析：摘要结果和案例一明显不同，提示词生效；摘要包含了历史对话记录，{messages} 被替换为历史消息列表。

2.2 HumanInTheLoopMiddleware 中间件

HumanInTheLoopMiddleware（人在环中间件、人工审核中间件）在工具调用前中断 Agent 运行，等待用户对工具调用请求决策。可选决策：

• approve：同意执行

• edit：编辑调用配置后执行

• reject：拒绝执行

参数说明

参数	说明
interrupt_on	工具名和中断策略的映射。策略可为 True/False/InterruptOnConfig 对象
description_prefix	自定义中断描述，默认 "Tool execution requires approval"

interrupt_on 策略含义：

• True：所有决策(approve/edit/reject)都可选择，相当于：

  "get_weather": {"allowed_decisions": ["approve", "edit", "reject"]}

• False：不中断，无需审批即可执行。

• InterruptOnConfig：TypedDict 子类，可用字典直接赋值。支持的 Key：

  • allowed_decisions：精细控制中断后允许的决策

  • description：特定工具的中断描述信息，优先级高于 description_prefix

完整示例：

interrupt_on={
    "get_weather": True,
    "read_email_tool": False,
    "send_email_tool": {
        "allowed_decisions": ["approve", "reject"],
    },
}

举例过程 1：调用前中断

本例需要从中断的位置让 Agent 继续运行，需用到短期记忆（checkpointer）。

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langchain.messages import HumanMessage
from langchain.tools import tool
from langgraph.types import Command
from rich import print as rprint

@tool
def get_weather(city: str, is_forcast: bool = False) -> str:
    """查询指定城市天气"""
    res = f"{city}今天天气不错"
    if is_forcast:
        res += "\n明天下雨"
    return res

@tool
def get_news() -> str:
    """查询当日新闻"""
    return "中方三艘油轮通过霍尔木兹海峡"

@tool
def read_email_tool(email_id: str) -> str:
    """通过邮件ID读取内容的伪函数"""
    return f"邮件ID：{email_id}\n是空的"

@tool
def send_email_tool(recipient: str, subject: str, body: str) -> str:
    """发送邮件伪函数"""
    print(">>> 真的执行发送邮件工具了")
    return f"发送给 {recipient} 的邮件标题是：{subject}，内容：{body}"

agent = create_agent(
    model=model,
    tools=[get_weather, get_news, read_email_tool, send_email_tool],
    checkpointer=InMemorySaver(),
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "get_weather": True,
                "get_news": True,
                "read_email_tool": False,
                "send_email_tool": {
                    "allowed_decisions": ["approve", "reject"],
                    "description": "发送邮件中断啦"
                },
            },
            description_prefix="中断啦"
        ),
    ],
)
config = {"configurable": {"thread_id": "1"}}
# 第一次调用：会暂停在发送邮件前
response = agent.invoke(
    {
        "messages": [
            HumanMessage(content="请帮我查询今天北京的天气"
                                 "查询今日新闻"
                                 "查看ID为 'sk2131421' 的邮件内容，"
                                 "向15641685664@qq.com发送邮件，标题是'哈哈哈'，内容是：'你好啊'"
                                 "同时做这四件事")
        ]
    },
    config=config,
)

关键：查看中断信息——response["__interrupt__"] 是一个 Interrupt 列表，其中 interrupt.value 包含 action_requests（各工具调用请求，含 name/args/description）和 review_configs（每个工具的 allowed_decisions）。

interrupts = response.get("__interrupt__", [])
action_requests = interrupts[0].value["action_requests"]
for action_request in action_requests:
    rprint(action_request)

举例过程 2：指明工具调用请求决策

通过 Command(resume=decisions) 让 Agent 从中断处继续，decisions 的顺序必须和返回的中断请求顺序一致，且必须用同一个 thread_id：

weather_decision = {
    "type": "edit",
    "edited_action": {
        "name": "get_weather",
        "args": {"city": "中国北京市", "is_forcast": True}
    }
}
news_decision = {"type": "approve"}
send_email_decision = {"type": "approve"}
decisions = {"decisions": []}

# 决策的顺序必须和返回的中断请求顺序一致
for action_request in action_requests:
    if action_request["name"] == "get_weather":
        decisions["decisions"].append(weather_decision)
    if action_request["name"] == "get_news":
        decisions["decisions"].append(news_decision)
    if action_request["name"] == "send_email_tool":
        decisions["decisions"].append(send_email_decision)

if interrupts:
    # 审批通过
    resumed_response = agent.invoke(
        Command(resume=decisions),
        config=config,  # 必须是同一个 thread_id
    )

效果：get_weather 被 edit（参数改为“中国北京市”+含预报），get_news/send_email_tool 被 approve，工具真实执行（“真的执行发送邮件工具了”）。

2.3 PIIMiddleware 中间件

PII 中间件用于检测和处理对话中的个人身份信息（Personally Identifiable Information，PII），支持自定义处理策略。

参数说明

参数	说明
pii_type	检测的 PII 数据类型。内置类型有 email/credit_card/url/mac_address/ip，也可自定义
strategy	处理策略，支持四种：redact/mask/hash/block
detector	自定义 PII 检测函数或正则表达式，未提供则使用内置检测函数
apply_to_input	是否在调用模型前检测，默认 True
apply_to_output	是否在模型调用后检测，默认 False
apply_to_tool_results	是否在工具调用后检测其输出，默认 False

四种处理策略对比：

策略	处理方式	示例	适用场景
redact	用 [REDACTED_[PII_TYPE]] 替换	[REDACTED_EMAIL]、[REDACTED_CREDIT_CARD]	日志清洗、合规、公开输出隐藏敏感内容
mask	用 *** 遮蔽前面部分，保留部分可辨识性	****-****-****-1234	用户服务界面、前端显示
hash	用哈希值替代原值	<email_hash:a1b2c3d4>	analytics、调试、统计分析、匿名追踪
block	检测到 PII 直接抛异常	抛出异常	隐私要求极高、绝不允许泄露的场景

通常只在模型调用前检测（apply_to_input=True），因为 PII 检测的主要目的是避免将敏感信息发送给模型服务导致信息泄露。

LangChain 内置检测函数：

BUILTIN_DETECTORS: dict[str, Detector] = {
    "email": detect_email,
    "credit_card": detect_credit_card,
    "ip": detect_ip,
    "mac_address": detect_mac_address,
    "url": detect_url,
}

举例 1：使用内置检测器

from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware
from langchain.messages import HumanMessage

agent = create_agent(
    model=model,
    tools=[],
    middleware=[
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
        PIIMiddleware("url", strategy="hash", apply_to_input=True),
        PIIMiddleware("mac_address", strategy="mask", apply_to_input=True),
        PIIMiddleware("ip", strategy="block", apply_to_input=True),
    ]
)

response = agent.invoke({
    "messages": [HumanMessage("""
    帮我向 156168188@qq.com 发送一封邮件
    同时查看银行卡号： 5105-1051-0510-5100 的余额
    访问 https://localhost:12345
    确认这是不是 MAC地址： 11-11-11-11-11-11
    """)]
})

try:
    response1 = agent.invoke({
        "messages": [HumanMessage("看看这个 IP 能不能 ping 通：192.168.10.1")]
    })
except Exception as e:
    print('=' * 30, '-> 抛异常 <-', '=' * 30)
    print(f"检测到IP，抛出异常：{e}")

输出效果：邮箱→[REDACTED_EMAIL]，卡号→****-****-****-5100，url→<url_hash:dd5fc2a9>，MAC→**-**-**-**-**-11，IP→抛出异常 Detected 1 instance(s) of ip in text content。

举例 2：自定义检测器/函数

自定义检测函数（用 re.finditer 返回匹配对象迭代器）：

import re

def detect_phone_number(content: str):
    return [
        {
            "text": m.group(0),   # 提取匹配到的 11 位数字文本
            "start": m.start(),   # 在原文本中的起始索引
            "end": m.end()        # 在原文本中的结束索引
        } for m in re.finditer(r"[0-9]{11}", content)
    ]

text = "尚硅谷的电话是13812345678，康师傅的电话是13987654321。"
result = detect_phone_number(text)
print(result)
# [{'text': '13812345678', 'start': 7, 'end': 18}, {'text': '13987654321', 'start': 26, 'end': 37}]

re.finditer(pattern, content) 在 content 中从左到右扫描，每找到一个符合条件的匹配就生成一个匹配对象，返回迭代器（Iterator）。

在 PIIMiddleware 中使用自定义 detector（支持正则字符串或检测函数）：

agent = create_agent(
    model=model,
    tools=[],
    middleware=[
        PIIMiddleware("api_key", strategy="hash", apply_to_input=True,
                      detector=r"sk-[a-zA-Z0-9]+"),
        PIIMiddleware("phone_number", strategy="mask", apply_to_input=True,
                      detector=detect_phone_number)
    ]
)

2.4 TodoListMiddleware 中间件

TodoListMiddleware 中间件赋予 Agent 任务规划和追踪进度的能力，可应对复杂的多步任务。当大任务需拆解为 3 个以上子任务且前后步骤有依赖时，若不列 Todo 列表，大模型执行到第 3 步容易忘记最初目标，或在工具返回大量报错后“应激”跳过验证。TodoListMiddleware 强制把计划挂在全局状态里，时刻提醒“下一步该干什么”。

把普通的 Agent 比作“想到哪写到哪”的实习生，引入 TodoListMiddleware 的 Agent 就是“先写方案、再列 CheckList、最后按部就班执行”的资深工程师。

To-do list 的创建和维护是通过调用 write_todos 工具实现的。典型场景：任务链路长、步骤多且有严格先后依赖关系；需要在前端 UI 实时展示 Agent 的思考与执行进度。

是否需要 TodoListMiddleware 的决策树：

你的任务是否需要拆解？
  ├── 否 (如问答、翻译、单次函数调用) ──> ❌ 绝不需要，浪费算力
  └── 是 (如写多文件工程)
      └── 步骤是否多变且需要应对失败？
          ├── 否 (步骤完全固定如 A->B->C) ──> ❌ 传统 LangGraph 线性节点即可
          └── 是 (AI 需要边做边调计划) ──> ✅ 引入 TodoListMiddleware

参数说明

参数	说明
system_prompt	自定义指导 todo 列表使用的提示词，不提供则用内置提示词
tool_description	自定义 write_todos 工具的描述信息，不提供则用内置描述

案例设计

任务目标：扫描工作目录，测试并修复工作区下的 my_add.py 文件。工具列表：

• list_files：扫描工作目录，列出所有文件

• read_file：扫描指定文件，返回文件内容

• write_file：向指定文件写入内容

• run_tests：运行测试，底层基于 pytest 实现

待修复文件 my_add.py（bug：把加法写成了减法）：

def add(a: int, b: int) -> int:
    """返回两个整数的和"""
    return a - b

测试文件 test_my_add.py：

from my_add import add

def test_add():
    """测试加法功能"""
    assert add(2, 3) == 5
    assert add(-1, 1) == 0
    assert add(0, 0) == 0
    assert add(10, -5) == 5

pytest 用法：会扫描目录下所有以 test_ 开头或 _test 结尾的文件视为测试文件，执行其中以 test 开头的函数，出错打印到控制台。运行：pytest -q。

业务代码

模型初始化（略），提供工具列表：

from langchain.tools import tool
from pathlib import Path
import subprocess

WORKSPACE = Path("../todo_workspace")

@tool
def list_files(path: str = ".") -> str:
    """列出工作区指定目录下的文件和子目录。path 只能是相对路径。"""
    target = (WORKSPACE / path).resolve()
    workspace_root = WORKSPACE.resolve()
    if not str(target).startswith(str(workspace_root)):
        return "错误：只允许访问工作区内的目录。"
    if not target.exists():
        return f"错误：目录不存在: {path}"
    if not target.is_dir():
        return f"错误：不是目录: {path}"
    items = sorted(target.iterdir(), key=lambda p: (p.is_file(), p.name.lower()))
    if not items:
        return f"目录为空: {path}"
    lines = []
    for item in items:
        rel = item.relative_to(workspace_root)
        kind = "[DIR]" if item.is_dir() else "[FILE]"
        lines.append(f"{kind} {rel.as_posix()}")
    return "\n".join(lines)

@tool
def read_file(path: str) -> str:
    """读取工作区中的文本文件内容。path 只能是相对路径。"""
    file_path = (WORKSPACE / path).resolve()
    if not str(file_path).startswith(str(WORKSPACE.resolve())):
        return "错误：只允许读取工作区内的文件。"
    if not file_path.exists():
        return f"错误：文件不存在: {path}"
    return file_path.read_text(encoding="utf-8")

@tool
def write_file(path: str, content: str) -> str:
    """写入工作区中的文本文件。path 只能是相对路径。"""
    file_path = (WORKSPACE / path).resolve()
    if not str(file_path).startswith(str(WORKSPACE.resolve())):
        return "错误：只允许写入工作区内的文件。"
    file_path.write_text(content, encoding="utf-8")
    return f"已写入文件: {path}"

@tool
def run_tests() -> str:
    """在工作区运行 pytest -q，并返回输出。"""
    try:
        result = subprocess.run(
            ["pytest", "-q"],
            cwd=str(WORKSPACE),
            capture_output=True,
            text=True,
            timeout=20,
        )
        return (
            f"returncode={result.returncode}\n\n"
            f"STDOUT:\n{result.stdout}\n\n"
            f"STDERR:\n{result.stderr}"
        )
    except Exception as e:
        return f"运行测试失败: {e}"

创建 Agent 并执行：

from langchain.agents import create_agent
from langchain.agents.middleware import TodoListMiddleware
from langchain.messages import HumanMessage
from rich import print as rprint

agent = create_agent(
    model=model,
    tools=[list_files, read_file, write_file, run_tests],
    middleware=[TodoListMiddleware()],
    system_prompt=(
        "你是一个代码修复助手。遇到多步骤任务时，先使用 write_todos 制定待办事项；"
        "然后读取文件、修复代码并运行测试。工作全部在工作区下进行。"
    ),
)

print("正在执行 Agent 任务...")
final_state = agent.invoke(
    {
        "messages": [
            HumanMessage(content="请测试并修复工作区下 my_add.py 文件中的代码")
        ]
    }
)
rprint(final_state)

输出与分析

Agent 会先调用 write_todos 制定待办列表，TodoListMiddleware 自动将规划好的步骤注入到 state 的 "todos" 字段中，每完成一步更新状态。执行流程：

[用户请求] -> "修复 my_add.py"
     │
     ▼
[Agent 思考] -> 意识到是多步骤复杂任务
     │
     ▼
[触发工具] -> 调用 write_todos(todos=[...])
     │
  ┌─┴────────────────────────┐
  │ TodoListMiddleware 拦截 │ -> 自动解析工具参数，更新 State 中的 {"todos": [...]}
  └─┬────────────────────────┘
     │
     ▼
[继续执行] -> 读取文件、修改、测试...
     │
     ▼
[最终返回] -> final_state 携带了被中间件更新过的最新 "todos" 列表

为了让 TodoListMiddleware 生效，Agent、工具和中间件三者之间必须满足协同契约：

1. todos 列表的维护是通过工具调用（write_todos）实现的。

2. todos 列表信息分为两部分：status（状态）和 content（内容）。状态共三种取值：

   • in_progress：正在进行

   • completed：已完成

   • pending：待执行

3. 每进行一个步骤，Agent 会更新 To-do lists。

---

3. 其它内置中间件

本节为大部分中间件提供测试代码和输出。

3.1 ModelCallLimitMiddleware 中间件

限制模型调用次数，避免无限循环，控制调用成本。

参数	说明
thread_limit	每个线程最多调用模型次数
run_limit	每次运行最多调用模型次数
exit_behavior	达到限制后的退出行为：end（优雅退出）/error（抛异常）

使用 thread_limit 必须配置 checkpointer（如 InMemorySaver()）。

举例 1：整个会话限制-优雅退出

from langchain.agents import create_agent
from langchain.agents.middleware import ModelCallLimitMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langchain.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage
from typing import List

agent = create_agent(
    model=model,
    checkpointer=InMemorySaver(),  # Required for thread limiting
    tools=[],
    middleware=[
        ModelCallLimitMiddleware(
            thread_limit=2,  # 每个线程最多2次模型调用
            # run_limit=5,   # 每次运行最多5次
            exit_behavior="end",  # 达到限制后退出
        ),
    ],
)

def pretty_iterate_msg(messages: List[SystemMessage | HumanMessage | AIMessage | ToolMessage]):
    for msg in messages:
        msg.pretty_print()

config = {"configurable": {"thread_id": "1"}}
response_first = agent.invoke({"messages": [HumanMessage("你好")]}, config=config)
print("=" * 30, "> first <", "=" * 30)
pretty_iterate_msg(response_first["messages"])

response_second = agent.invoke({"messages": [HumanMessage("你是谁？")]}, config=config)
print("=" * 30, "> second <", "=" * 30)
pretty_iterate_msg(response_second["messages"])

response_third = agent.invoke({"messages": [HumanMessage("你能帮我做什么？")]}, config=config)
print("=" * 30, "> third <", "=" * 30)
pretty_iterate_msg(response_third["messages"])

分析：thread_limit=2 表示每个线程最多 2 次模型调用。前两次正常返回，第三次超出限制：

• exit_behavior="end"：返回消息内容为 Model call limits exceeded: thread limit (2/2)。

• exit_behavior="error"：抛出 ModelCallLimitExceededError: Model call limits exceeded: thread limit (2/2)，错误发生在 'ModelCallLimitMiddleware.before_model' 任务中。

举例 2/3/4：单次调用限制（run_limit）需配合 fake-server 重复触发工具调用。fake server 以 80% 概率输出非法响应，重复触发模型调用，从而能观察到 run_limit 超限效果。

from langchain_deepseek import ChatDeepSeek
from pydantic import BaseModel, Field, SecretStr
from typing import List, Union
from dotenv import load_dotenv
load_dotenv()

model = ChatDeepSeek(
    model="any",
    api_base="http://localhost:8889",
    api_key=SecretStr("<KEY>")
)

class ContactInfo(BaseModel):
    """用户的联系方式"""
    name: str = Field(description="用户姓名")
    email: str = Field(description="用户邮箱地址")
    phone: str = Field(description="用户的手机号")

class EventInfo(BaseModel):
    event_name: str = Field(description="事件名称")
    date: str = Field(description="事件发生日期")

agent = create_agent(
    model=model,
    checkpointer=InMemorySaver(),
    tools=[],
    middleware=[
        ModelCallLimitMiddleware(
            run_limit=3,
            exit_behavior="end",  # 或 "error"
        ),
    ],
    response_format=Union[ContactInfo, EventInfo]
)
config = {"configurable": {"thread_id": "1"}}
response = agent.invoke({"messages": [HumanMessage("你好")]}, config=config)
for msg in response["messages"]:
    msg.pretty_print()

fake server 关键逻辑：80% 概率返回非法响应（同时返回 ContactInfo 和 EventInfo 两个结构化响应），触发模型反复重试，从而测试 run_limit 超限行为。

3.2 ToolCallLimitMiddleware 中间件

限制工具调用次数，可以限制所有工具调用总次数，也可限制特定工具调用次数。

作用：避免过多调用昂贵外部 API；限制网络爬虫或数据库查询请求数量；避免 Agent 陷入无限循环。

退出行为有三种模式：

模式	说明
error	直接抛异常
end	结束整个会话
continue	继续运行 Agent（默认行为），将工具调用超出限制的信息传递给模型，后者自主决定后续行为。模型能力不足可能导致死循环，为避免死循环 fake server 以 20% 概率输出正确响应来终止循环

from langchain.agents.middleware import ToolCallLimitMiddleware

agent = create_agent(
    model=model,
    checkpointer=InMemorySaver(),
    tools=[],
    middleware=[
        ToolCallLimitMiddleware(
            # thread_limit=2,  # 每个线程最多2次工具调用
            run_limit=2,       # 每次运行最多2次
            exit_behavior="end",  # 或 "error" / "continue"
        ),
    ],
    response_format=Union[ContactInfo, EventInfo]
)

三种模式输出对比：

• exit_behavior="end"：Agent 返回 Tool call limit reached: run limit exceeded (4/2 calls).。

• exit_behavior="error"：抛出 ToolCallLimitExceededError: Tool call limit reached: run limit exceeded (4/2 calls).，发生在 'ToolCallLimitMiddleware.after_model' 任务中。

• exit_behavior="continue"：返回 ToolMessage Tool call limit exceeded. Do not make additional tool calls.，Agent 继续运行让模型自主决策。

3.3 ModelFallbackMiddleware 中间件

用于故障转移，当主模型无法访问时启用备用模型。

from langchain.agents.middleware import ModelFallbackMiddleware
from langchain.chat_models import init_chat_model

# 方式1：传模型对象列表
primary_model = init_chat_model("openai:gpt-5.4-mini")
fallback = ModelFallbackMiddleware(
    fallback_models=[
        init_chat_model("openai:gpt-4o-mini"),
        init_chat_model("anthropic:claude-3-haiku")
    ]
)
agent = create_agent(model=primary_model, tools=[], middleware=[fallback])

# 方式2：传模型名称（可变参数）
agent = create_agent(
    model="deepseek:fake_model",
    tools=[],
    middleware=[
        ModelFallbackMiddleware(
            "deepseek-v4-flash",
            "deepseek-v4-pro",
        ),
    ],
)
response = agent.invoke({"messages": [HumanMessage("你是谁？")]})
last_msg = response["messages"][-1]
print(last_msg.response_metadata.get("model_name"))  # deepseek-v4-flash（实际服务的模型）

主模型 deepseek:fake_model 不存在，自动 fallback 到 deepseek-v4-flash，最终响应的 model_name 为 deepseek-v4-flash。

3.4 LLMToolSelectorMiddleware 中间件

智能工具筛选。当工具太多时，用子模型筛选最相关的几个工具。

参数	说明
model	用于工具筛选的子模型
max_tools	限定可以调用的工具总数
always_include	指定的工具不被计数（始终保留）

from langchain.agents.middleware import LLMToolSelectorMiddleware

# 主模型 model_out、筛选子模型 model_in
agent = create_agent(
    model=model_out,
    tools=[get_weather, get_news, calculate, search_stock],
    middleware=[
        LLMToolSelectorMiddleware(
            model=model_in,
            max_tools=0,                          # 非 always_include 工具最多选 0 个
            always_include=["get_weather"]        # 始终保留 get_weather
        ),
    ],
)

举例对比（输入“北京今天天气如何？今日新闻概要”）：

配置	结果
max_tools=0, always_include=["get_weather"]	只能调 get_weather，无法获取新闻
max_tools=0, always_include=["get_news"]	只能调 get_news，无法获取天气
max_tools=0, always_include=["get_weather","get_news"]	两个工具都被保留，两个都能调用
max_tools=1, always_include=["get_weather"]	get_weather 必选 + 子模型再选 1 个（如 get_news）
max_tools=1, always_include=["get_news"]	get_news 必选 + 子模型再选 1 个（如 get_weather）

always_include 指定的工具不计入 max_tools 名额，是强制保留的。

3.5 ToolRetryMiddleware 中间件

基于指数退避（Exponential Backoff）算法，设置工具调用失败时的重试策略。

指数退避核心思想：当操作失败时，不立刻重试也不固定等待，而是让每次重试的延迟时间按指数级增长。避免所有失败客户端同时重试对瘫痪服务器造成 DDoS。

Jitter（抖动）：为避免大量工具重试请求集中在固定时间点，引入随机性。假设按策略两次调用间隔应为 10 秒，加入抖动后可能为 8.9 秒或 10.2 秒。

参数	说明
max_retries	最大重试次数（不含初始那次调用，最多调 1 + max_retries 次）
backoff_factor	指数退避因子（每次重试等待时间乘以该因子）
initial_delay	第一次重试前的初始等待时间（秒）
max_delay	最大等待延迟上限（防止指数增长无限大）
jitter	是否开启抖动（True/False）
retry_on / retry_on_exceptions	仅捕获指定异常时才触发重试，如 (TimeoutError,)
on_failure	达到最大重试仍失败时的行为：continue（包装错误塞回对话历史）/error

from langchain.agents import create_agent
from langchain.agents.middleware import ToolRetryMiddleware
from langchain.messages import HumanMessage
import datetime

def write_times(s):
    """将每次工具调用的时间戳和间隔写入本地文件，方便观察退避策略"""
    with open("call_times_with_jitter.txt", "a", encoding="utf-8") as f:
        f.write(s + "\n")

count = 1
start_time = None

@tool
def get_weather(city: str):
    """查询指定城市天气"""
    global count
    global start_time
    interval = 0
    current_time = datetime.datetime.now()
    if not start_time:
        interval = 0
    else:
        # 计算当前调用与上一次调用之间的时间差（秒）
        interval = (current_time - start_time).total_seconds()
    start_time = current_time
    res_str = f"第 {count} 次调用，当前时间： {start_time}, 和上次调用间隔 {interval} 秒"
    count += 1
    write_times(res_str)
    # 故意抛出 TimeoutError，以此触发中间件的重试机制
    raise TimeoutError("Not Implemented")

agent = create_agent(
    model=model,
    tools=[get_weather],
    middleware=[
        ToolRetryMiddleware(
            max_retries=6,          # 一共最多调 1 + 6 = 7 次
            backoff_factor=2.0,     # 每次重试等待时间乘以 2
            initial_delay=1.0,      # 第一次重试前等待 1 秒
            max_delay=10.0,         # 等待上限 10 秒
            jitter=True,            # 开启抖动
            retry_on=(TimeoutError,),
            on_failure="continue"   # 失败信息包装后塞回对话历史
        ),
    ],
)
response = agent.invoke({"messages": [HumanMessage("今天北京天气如何？")]})
for msg in response["messages"]:
    msg.pretty_print()

理论等待时间（initial_delay=1.0，backoff_factor=2.0，max_delay=10.0）：

重试轮次	理想基础延迟	jitter=False 实际等待	jitter=True 实际等待
第 1 次	1.0 秒	严格 1.0 秒	在 ~1 秒附近随机
第 2 次	2.0 秒	严格 2.0 秒	在 ~2 秒附近随机
第 3 次	4.0 秒	严格 4.0 秒	在 ~4 秒附近随机
第 4 次	8.0 秒	严格 8.0 秒	在 ~8 秒附近随机
第 5 次	10.0 秒	严格 10.0 秒（受 max_delay 限制）	在 ~10 秒附近随机
第 6 次	10.0 秒	严格 10.0 秒（受 max_delay 限制）	在 ~10 秒附近随机

公式：等待时间 ≈ initial_delay * (backoff_factor ** retry_number)，受 max_delay 上限约束。若 backoff_factor=0，则不使用指数增长，重试间始终用固定 initial_delay。

Jitter 开关对比：

• jitter=False：重试间隔死板、精准、可预测，适合本地调试、测试重试逻辑是否生效。

• jitter=True：重试间隔随机、错开、更安全，适合线上生产环境。

惊群效应（Thundering Herd Problem）：高并发生产环境下，若关闭 jitter，假设某 API 突然宕机 1 秒，1000 个请求同时失败后都严格等待 1/2/4 秒，会在固定时间点整齐地再次轰炸服务器，刚复活的服务器瞬间又被压垮，形成恶性循环。引入 jitter 后请求在区间内均匀错开（削峰填谷），流量平摊到时间轴上。

简写用法：

from langchain.agents.middleware import ToolRetryMiddleware
retry = ToolRetryMiddleware(
    max_retries=3,
    backoff_factor=2.0,
    retry_on_exceptions=[ConnectionError, TimeoutError]
)
agent = create_agent(model="openai:gpt-4o", tools=[web_search, api_call], middleware=[retry])

3.6 ModelRetryMiddleware 中间件

模型调用失败时重试，策略和工具重试一样，都基于指数退避算法。本节重点测试不同退出模式。

from langchain.agents import create_agent
from langchain.agents.middleware import ModelRetryMiddleware
from langchain.messages import HumanMessage
from dotenv import load_dotenv
load_dotenv(override=True)

agent = create_agent(
    model="deepseek-cat",  # 不存在的模型，触发 BadRequestError
    middleware=[
        ModelRetryMiddleware(
            max_retries=6,
            backoff_factor=2.0,
            initial_delay=1.0,
            max_delay=10.0,
            on_failure="continue",  # 或 "error"
            jitter=False,
        ),
    ],
)
response = agent.invoke({"messages": [HumanMessage("你好")]})
for msg in response["messages"]:
    msg.pretty_print()

退出模式对比：

• on_failure="continue"：返回消息 Model call failed after 7 attempts with BadRequestError: ...。

• on_failure="error"：抛出 BadRequestError: Error code: 400 - {'error': {'message': 'Model Not Exist', ...}}，发生在 'model' 任务中。

3.7 LLMToolEmulator 中间件

某些情况下工具尚未开发完成，希望先测试工具调用，可用 LLM tool emulator 模拟工具。

from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolEmulator
from langchain.messages import HumanMessage

@tool
def get_weather(city: str):
    """查询指定城市天气"""
    return f"{city}今天天气晴朗"

agent = create_agent(
    model=model_out,
    tools=[get_weather],
    middleware=[LLMToolEmulator(model=model_in)]
)
response = agent.invoke({"messages": [HumanMessage("今天北京天气如何")]})

模拟器会用子模型 model_in 生成一段逼真的工具返回 JSON（包含 weather/temperature/humidity/wind/aqi 等字段），而非真正执行 get_weather。

3.8 ContextEditingMiddleware 中间件

上下文编辑中间件，提供上下文管理的一种方式，通过更改发送给模型的消息列表来控制成本。注意：不会更改（持久化的）消息列表，因此只能通过 token 用量推测是否对消息列表进行了裁剪。

from langchain.agents import create_agent
from langchain.agents.middleware import ContextEditingMiddleware, ClearToolUsesEdit
from langchain.messages import HumanMessage, AIMessage
from langgraph.checkpoint.memory import InMemorySaver
from dotenv import load_dotenv
load_dotenv()

count = 0
@tool
def get_weather(city: str):
    """查询指定城市天气"""
    global count
    return (f"当前是第 {count} 次调用工具，{city}今天天气晴朗"
            f"天气非常好，北风，非常适合出行，盼望着，盼望着，"
            f"春天来了。我喜欢春天，你喜欢吗，天气真的很不错"
            f"万里无云，天气晴朗，春和景明，哈哈哈哈哈哈，这是凑字数的"
            f"真不错，天气非常好，适合出行，这里token挺多的"
            f"可以出门玩，尅有跑步，钓鱼，爬山，一切都很好哈哈哈")

agent = create_agent(
    model="deepseek-chat",
    tools=[get_weather],
    middleware=[
        ContextEditingMiddleware(
            edits=[
                ClearToolUsesEdit(
                    trigger=50,
                    keep=0,
                ),
            ],
        ),
    ],
    checkpointer=InMemorySaver()
)
config = {"configurable": {"thread_id": "1"}}
for i in range(3):
    count = i + 1
    response = agent.invoke(
        {"messages": [HumanMessage(f"第 {i + 1} 次询问：今天北京天气如何，一句话回答")]},
        config=config
    )
    for msg in response["messages"]:
        if isinstance(msg, AIMessage):
            if not msg.tool_calls:
                print(f"本次token用量：{msg.usage_metadata}")

ContextEditingMiddleware 的价值：多轮对话中，若频繁调用产生大量文本的工具（如代码执行、网页爬取），历史记录会急剧膨胀。该中间件像“上下文抽脂手术”，在影响当前对话的前提下自动删掉之前沉淀的工具调用废话，节省 Token 费用并防止超出最大上下文窗口。

InMemorySaver：在内存中开辟空间，第二轮和第三轮提问时 Agent 能通过 thread_id 自动找回前几轮记忆。

对照实验结论：对照组（不裁剪）第 3 轮 input_tokens 累计到 640；实验组（启用 ClearToolUsesEdit）稳定在 169，明显小于对照组。

3.9 FilesystemFileSearchMiddleware 中间件

基于系统的 Glob 和 Grep 检索工具，为 Agent 赋予本地文件搜索和分析能力。Glob 根据文件路径检索，Grep 根据文件内容检索。

参数	说明
root_path	搜索目录
allowed_extensions	限制搜索的文件后缀，防止读取非代码或无关文件
use_ripgrep	是否启用 ripgrep 搜索引擎（True 性能更好，需系统已安装 ripgrep）
max_file_size_mb	单个文件最大读取限制（MB），防止读取超大文件导致 OOM

from langchain.agents import create_agent
from langchain.agents.middleware import FilesystemFileSearchMiddleware
from langchain.messages import HumanMessage

agent = create_agent(
    model=model,
    tools=[],  # 自动添加 Glob 和 Grep 工具
    middleware=[
        FilesystemFileSearchMiddleware(
            root_path="../todo_workspace",
            # allowed_extensions=[".py", ".ipynb", ".js", ".md"],
            use_ripgrep=True,
            max_file_size_mb=10
        ),
    ],
)
result = agent.invoke({
    "messages": [HumanMessage("找到包含add函数的Python或Jupyter文件")]
})

Agent 会自动调用 glob_search 工具（如 pattern: **/*.py）找到 /my_add.py、/test_my_add.py。

3.10 Shell tool 中间件

为 Agent 提供一个可以执行命令的 Shell 环境。Windows 下无法测试。

3.11 Filesystem 中间件

源自 deepagents（基于 LangChain 的另一个框架）的中间件，内置四个工具，分别用于查看目录、读文件、写文件和改文件。

3.12 Subagent 中间件

来自 deepagents 的中间件，用于便捷地创建子 Agent。

---

4. 多个中间件组合及执行顺序

Middleware 可以叠加使用，那么多个中间件书写顺序重要吗？非常重要！

middleware=[
    TrimmerMiddleware(),        # 1. 先修剪消息
    SummarizationMiddleware(),  # 2. 再摘要
    LoggingMiddleware()         # 3. 最后记录日志
]

典型组合：

agent = create_agent(
    model=model,
    tools=[get_weather, get_news],
    middleware=[
        PIIMiddleware(strategy="redact"),                    # 1. 最先检查 PII
        ModelCallLimitMiddleware(run_limit=10),              # 2. 限制调用次数
        SummarizationMiddleware(max_tokens_before_summary=500),  # 3. 总结历史
        ToolRetryMiddleware(max_retries=3),                  # 4. 重试工具
    ]
)

执行顺序示例（三个自定义中间件）：

from langchain.agents.middleware import AgentMiddleware

class Middleware1(AgentMiddleware):
    def before_model(self, state, runtime):
        print("[中间件1] before_model")
        return None
    def after_model(self, state, runtime):
        print("[中间件1] after_model")
        return None

class Middleware2(AgentMiddleware):
    def before_model(self, state, runtime):
        print("[中间件2] before_model")
        return None
    def after_model(self, state, runtime):
        print("[中间件2] after_model")
        return None

class Middleware3(AgentMiddleware):
    def before_model(self, state, runtime):
        print("[中间件3] before_model")
        return None
    def after_model(self, state, runtime):
        print("[中间件3] after_model")
        return None

agent = create_agent(
    model=model,
    tools=[],
    middleware=[Middleware1(), Middleware2(), Middleware3()]
)
agent.invoke({"messages": [{"role": "user", "content": "测试"}]})

输出：

[中间件1] before_model
[中间件2] before_model
[中间件3] before_model
[中间件3] after_model
[中间件2] after_model
[中间件1] after_model

类似洋葱模型：外层先进后出，执行顺序为 1→2→3→模型→3→2→1。

---

5. 自定义中间件

某些复杂场景下官方内置中间件不能完全满足需求，可通过实现 LangChain 暴露的中间件 hook 函数构建自定义中间件。

说明：尽可能使用内置中间件。

5.1 什么是 hook 函数（钩子函数）

Hook 函数（钩子函数）指的是：在某个既定流程的特定时机，被框架、系统或主程序自动调用的扩展函数。

主流程预留了一些插槽，允许你在这些位置挂上自己的函数，这种被挂进去并在特定时机执行的函数就是 hook 函数。

核心特点：

1. 不是你主动调用，而是当流程运行到某个“钩子点”时系统自动触发。

2. 依附于一个更大的执行流程，如“请求开始前”“模型调用前”“任务结束后”“异常发生时”等。

3. 作用是在不改主流程源码的前提下插入自己的逻辑，如日志、鉴权、修改输入、拦截输出、清理资源等。

LangChain 的中间件作用在 Agent 架构中（后者基于 LangGraph 构建的流程图）。LangChain 暴露了六个 hook 函数。无论是官方内置中间件、自定义中间件，还是便捷装饰器中间件，通常都是通过实现其中的一个或多个 hook 来生效的。

5.2 LangChain 的 hook 函数分类

官方将六个钩子函数按风格分为两类：

类型 1：Node-style hooks（节点风格钩子） —— 在流程特定节点运行，适合顺序逻辑（如记录日志、验证）：

钩子	时机
before_agent	在 Agent 开始运行之前执行
before_model	在模型调用之前执行
after_model	在模型调用之后执行
after_agent	在 Agent 流程全部完成后执行

类型 2：Wrap-style hooks（包装风格钩子） —— 在模型或工具调用前后运行，适合控制流（如重试、回退、缓存）：

钩子	说明
wrap_model_call	包裹模型调用（模型调用前后运行）
wrap_tool_call	包裹工具调用（工具调用前后运行）

5.3 Node-style hooks 函数用法

支持两种用法：装饰器（函数式挂载，把一个 hook 快速挂到 Agent 的某个节点）和类写法（对象化中间件，封装为可配置、可复用、可扩展的组件）。

基本用法（基于装饰器实现）

from langchain.agents.middleware import (
    before_model, after_model, before_agent, after_agent,
    AgentState, AgentMiddleware
)
from langchain.messages import HumanMessage
from langgraph.runtime import Runtime
from langchain.agents import create_agent
from typing import Any

# 1. 定义 before_model 钩子
@before_model
def before_model_middleware(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    state["messages"][-1].content += " -> before_model <- "
    return None

# 2. 定义 after_model 钩子
@after_model
def after_model_middleware(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    state["messages"][-1].content += " -> after_model <- "
    return None

# 3. 定义 before_agent 钩子
@before_agent
def before_agent_middleware(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    state["messages"][-1].content += " -> before_agent <- "
    return None

# 4. 定义 after_agent 钩子
@after_agent
def after_agent_middleware(state: AgentState, runtime: Runtime) -> None:
    state["messages"][-1].content += " -> after_agent <- "
    return None

agent = create_agent(
    model=model,
    middleware=[before_model_middleware, after_model_middleware,
                before_agent_middleware, after_agent_middleware]
)
response = agent.invoke({"messages": [HumanMessage("你好啊")]})
for msg in response["messages"]:
    msg.pretty_print()

输出：

• HumanMessage: 你好啊 -> before_agent <- -> before_model <-

• AIMessage: 你好！... -> after_model <- -> after_agent <-

分析：

1. before_agent 钩子先于 before_model 执行，二者都在调用模型之前执行。

2. after_agent 晚于 after_model 执行，二者都在模型调用后执行。

基本用法（基于类实现）

关键规则：

1. 必须继承 AgentMiddleware（固定）。

2. 方法名固定（before_model/after_model/before_agent/after_agent）。

3. 类名随意。

LangGraph 只看：是否继承 AgentMiddleware？是否有 before_model/after_model 等方法？

from langchain.agents.middleware import AgentMiddleware, AgentState, hook_config
from langchain.messages import HumanMessage
from langgraph.runtime import Runtime
from langchain.agents import create_agent
from typing import Any

class MyMiddleware(AgentMiddleware):
    def __init__(self):
        super().__init__()

    def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        state["messages"][-1].content += " -> before_model <- "
        return None

    def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        state["messages"][-1].content += " -> after_model <- "
        return None

    def before_agent(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        state["messages"][-1].content += " -> before_agent <- "
        return None

    def after_agent(self, state: AgentState, runtime: Runtime) -> None:
        state["messages"][-1].content += " -> after_agent <- "
        return None

my_middleware = MyMiddleware()
agent = create_agent(model=model, middleware=[my_middleware])

before_model 典型场景：消息修剪（trim messages）、PII 脱敏、输入验证、条件路由。 after_model 典型场景：输出验证、格式化响应、统计信息、状态更新。

两种方法的统一

装饰器底层会基于重写的方法构造一个 AgentMiddleware 子类的实例。以 @after_model 装饰器底层实现为例：

def wrapped(
    _self: AgentMiddleware[StateT, ContextT],
    state: StateT,
    runtime: Runtime[ContextT],
) -> dict[str, Any] | Command[Any] | None:
    return func(state, runtime)  # type: ignore[return-value]

return type(
    middleware_name,
    (AgentMiddleware,),
    {
        "state_schema": state_schema or AgentState,
        "tools": tools or [],
        "after_model": wrapped,
    },
)()

上述代码含义：

1. 创建一个 AgentMiddleware 的子类。

2. 类名为 middleware_name（即创建 agent 时传递的中间件名称）。

3. 这个子类有两个属性 state_schema 和 tools。

4. 有一个方法 after_model，逻辑等同于 func(state, runtime)。

5. 最后的括号 () 表示实例化子类，返回一个对象。

所以装饰器最终返回的也是一个 AgentMiddleware 的子类对象，并重写了 after_model 方法，和基于类的自定义方式本质是一样的。

参数说明

Node-style hooks 函数有两个参数：

参数	说明
state	AgentState 实例，维护 Agent 运行过程中的状态，会随运行变化，包括消息列表
runtime	Runtime 实例，维护 Agent 运行过程中的上下文环境，包括上下文、长期记忆等

返回值说明

返回值	含义
None	不修改状态
字典	更新状态
{"jump_to": "..."}	控制流程

jump_to 目标："__end__"（结束 Agent）/"tools"（跳到工具节点）/ 其他自定义节点。

def before_model(self, state, runtime):
    print("日志记录")
    return None  # 不做任何修改，继续流程

def after_model(self, state, runtime):
    count = state.get("count", 0)
    return {"count": count + 1}  # 更新状态中的 count

def before_model(self, state, runtime):
    if state.get("count", 0) > 10:
        return {"jump_to": "__end__"}  # 跳过模型，直接结束
    return None

装饰器参数：can_jump_to

钩子函数可以改变 Agent 正常运行轨迹（如发现上下文窗口溢出直接跳转至结尾）。Node-style 的四个 hook 函数可接收额外参数 can_jump_to，决定了钩子函数可以直接跳转至流程的哪些位置：

取值	跳转目标
end	跳转至 Agent 流程末尾，或第一个 after_agent 钩子，直接终止整个流程
tools	跳转至工具节点
model	跳转至模型节点，或第一个 before_model 钩子

基于装饰器实现 can_jump_to（三个典型场景）：

from typing import Any
from langchain.agents import create_agent
from langchain.agents.middleware import before_model, after_model, AgentState
from langchain.messages import AIMessage, SystemMessage
from langchain.tools import tool
from langgraph.runtime import Runtime

@tool
def get_news() -> str:
    """获取当日新闻"""
    return f"美加墨世界杯今日开幕"

# 场景1：在模型执行前触发，允许跳转到 "tools" 节点（强行拦截并触发工具）
@before_model(can_jump_to=["tools"])
def force_tool_first(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """如果用户输入包含 'direct tool'，则跳过本次大模型思考/生成，
    直接伪造一个 AIMessage 的 tool_calls，强行把控制权移交给工具节点。"""
    text = state["messages"][-1].content
    if isinstance(text, str) and "direct tool" in text.lower():
        print("[MIDDLEWARE] before_model: jump_to='tools'")
        fake_tool_call = AIMessage(
            content="人工构造的消息",
            tool_calls=[
                {
                    "name": "get_news",
                    "args": {},
                    "id": "call_force_weather_001",
                }
            ],
        )
        return {
            "messages": [fake_tool_call],
            "jump_to": "tools",
        }
    return None

# 场景2：在模型执行后触发，允许重新跳转回 "model" 节点（反思/重试机制）
@after_model(can_jump_to=["model"])
def retry_with_extra_instruction(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """如果模型已生成回答，但发现用户最初请求包含 'retry model'，
    则动态追加一条系统提示词，强行让模型重新生成（重试）一次。"""
    user_text = ""
    for msg in reversed(state["messages"]):
        if getattr(msg, "type", "") == "human":
            user_text = getattr(msg, "content", "")
            break
    if isinstance(user_text, str) and "retry model" in user_text.lower():
        # 核心防御：防止无限循环重跳（死循环）
        already_injected = any(
            isinstance(getattr(msg, "content", None), str)
            and "你必须以【二次回答】开头" in msg.content
            for msg in state["messages"]
        )
        if already_injected:
            return None  # 已注入过，直接放行，结束重试流程
        print("[MIDDLEWARE] after_model: jump_to='model' with extra system instruction")
        return {
            "messages": [
                SystemMessage("你必须以【二次回答】开头，并且只用一句话回答。")
            ],
            "jump_to": "model",
        }
    return None

# 场景3：在模型执行前触发，允许直接跳转到 "end" 节点（强行终止/异常拦截）
@before_model(can_jump_to=["end"])
def overflow_context_processor(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """模拟上下文窗口溢出或其他严重阻断情况，直接熔断流程。"""
    if "overflow" in state["messages"][-1].content:
        print("[MIDDLEWARE] before_model: jump_to='end' when contenxt window overflow")
        return {
            "messages": [AIMessage("上下文窗口溢出，终止")],
            "jump_to": "end",
        }

agent = create_agent(
    model=model,
    tools=[get_news],
    middleware=[force_tool_first, retry_with_extra_instruction, overflow_context_processor],
)

def run_once(user_input: str):
    result = agent.invoke(
        {"messages": [{"role": "user", "content": user_input}]}
    )
    for msg in result["messages"]:
        msg.pretty_print()

if __name__ == "__main__":
    # Case 1: 直接跳 tools —— 触发 force_tool_first，绕过 LLM 首轮思考直接调用 get_news
    run_once("请帮我查今日新闻 direct tool")
    # Case 2: 输出后跳回 model —— 触发 retry_with_extra_instruction，注入提示后强行拉回重新生成
    run_once("请随便介绍一下 LangChain retry model")
    # Case 3: 触发 overflow_context_processor，直接打印终止信息并退出，LLM 不会接收请求
    run_once("你好 overflow")
    # Case 4: 正常流程，标准工作流 User -> Model -> Call Tool -> Model -> End
    run_once("今日新闻摘要？")

基于类实现 can_jump_to：关键区别在于需要引入额外的装饰器 @hook_config 为 can_jump_to 传参：

from typing import Any
from langchain.agents import create_agent
from langchain.agents.middleware import hook_config, AgentState, AgentMiddleware
from langchain.messages import AIMessage, SystemMessage
from langchain.tools import tool
from langgraph.runtime import Runtime

@tool
def get_news() -> str:
    """获取当日新闻"""
    return f"美加墨世界杯今日开幕"

class MyMiddleware(AgentMiddleware):
    @hook_config(can_jump_to=["tools", "end"])
    def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        text = state["messages"][-1].content
        if "overflow" in text:
            print("[MIDDLEWARE] before_model: jump_to='end' when contenxt window overflow")
            return {
                "messages": [AIMessage("上下文窗口溢出，终止")],
                "jump_to": "end",
            }
        if isinstance(text, str) and "direct tool" in text.lower():
            print("[MIDDLEWARE] before_model: jump_to='tools'")
            fake_tool_call = AIMessage(
                content="人工构造的消息",
                tool_calls=[
                    {"name": "get_news", "args": {}, "id": "call_force_weather_001"}
                ],
            )
            return {
                "messages": [fake_tool_call],
                "jump_to": "tools",
            }
        return None

    @hook_config(can_jump_to=["model"])
    def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        user_text = ""
        for msg in reversed(state["messages"]):
            if getattr(msg, "type", "") == "human":
                user_text = getattr(msg, "content", "")
                break
        if isinstance(user_text, str) and "retry model" in user_text.lower():
            already_injected = any(
                isinstance(getattr(msg, "content", None), str)
                and "你必须以【二次回答】开头" in msg.content
                for msg in state["messages"]
            )
            if already_injected:
                return None
            print("[MIDDLEWARE] after_model: jump_to='model' with extra system instruction")
            return {
                "messages": [SystemMessage("你必须以【二次回答】开头，并且只用一句话回答。")],
                "jump_to": "model",
            }
        return None

agent = create_agent(
    model=model,
    tools=[get_news],
    middleware=[MyMiddleware()],
)

四个 Case 的预期表现：

1. Case 1：提前判定需要调用工具，直接在 before_model 跳转至工具节点，省去一次模型调用。

2. Case 2：通过约定的 retry model 标记，在 after_model 之后再次跳转到模型节点，触发模型重复调用，最终输出带“【二次回答】”前缀。

3. Case 3：通过约定的 overflow 标记，模拟上下文窗口溢出，在 before_model 直接跳转至结尾，提前终止流程。

4. Case 4：未被干预的正常 Agent 流程，作为对照。

5.4 Wrap-style hooks 函数用法

wrap_model_call

可以同时在模型调用前后做事，命名为 wrap_model_call（wrap 意为包裹）。源码签名：

def wrap_model_call(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
    """
    request: 包含 model, messages, system_message, tools, state
    handler: 执行实际模型调用的函数
    返回：ModelResponse
    """

基于装饰器实现：

from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from langchain.messages import HumanMessage
from langchain.agents import create_agent
from typing import Callable

@wrap_model_call
def wrap_model_call_middleware(
    request: ModelRequest,  # 包含即将发送给大模型的所有请求数据（如消息列表）
    handler: Callable[[ModelRequest], ModelResponse],  # 核心句柄：下一个中间件或最终大模型调用
) -> ModelResponse | None:
    # 调用前：篡改最后一条消息内容（典型：统一追加特殊 Prompt 提示词）
    request.messages[-1].content += " -> wrap_model_call_before <- "
    # 将修改后的请求传递给 handler，真正调用大模型
    response = handler(request)
    # 调用后：篡改返回内容（典型：敏感词过滤、输出格式化、后处理标记）
    response.result[0].content += " -> wrap_model_call_after <- "
    return response

agent = create_agent(model=model, middleware=[wrap_model_call_middleware])
response = agent.invoke({"messages": [HumanMessage("你好啊")]})

模型调用前最后一条是 HumanMessage，调用后最后一条是 AIMessage，前后更改都生效。

基于类实现：

from langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse
from typing import Callable

class WrapModelCallMiddleWare(AgentMiddleware):
    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelResponse | None:
        request.messages[-1].content += " -> wrap_model_call_before <- "
        response = handler(request)
        response.result[0].content += " -> wrap_model_call_after <- "
        return response

agent = create_agent(model=model, middleware=[WrapModelCallMiddleWare()])

典型使用场景（拦截、重试、缓存模型调用）：

场景 1：重试逻辑

from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from typing import Callable
import time

@wrap_model_call
def retry_model(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
    """自动重试失败的模型调用"""
    max_retries = 3
    for attempt in range(max_retries):
        try:
            print(f"🔄 尝试调用模型（第 {attempt + 1}/{max_retries} 次）")
            return handler(request)
        except Exception as e:
            if attempt == max_retries - 1:
                print(f"❌ 所有重试失败：{e}")
                raise
            # 指数退避
            wait_time = 2 ** attempt
            print(f"⚠  调用失败：{e}，{wait_time} 秒后重试")
            time.sleep(wait_time)

场景 2：响应缓存

from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from typing import Callable
import hashlib
import json

class ModelCache:
    """模型响应缓存"""
    def __init__(self):
        self.cache = {}

    def create_hook(self):
        @wrap_model_call
        def cache_model(
            request: ModelRequest,
            handler: Callable[[ModelRequest], ModelResponse]
        ) -> ModelResponse:
            # 生成缓存键
            cache_key = hashlib.md5(
                json.dumps({
                    "messages": [str(m) for m in request.messages],
                    "system": str(request.system_message)
                }).encode()
            ).hexdigest()
            # 检查缓存
            if cache_key in self.cache:
                print("💾 缓存命中！")
                return self.cache[cache_key]
            # 调用模型
            print("🔍 缓存未命中，调用模型")
            response = handler(request)
            # 存入缓存
            self.cache[cache_key] = response
            return response
        return cache_model

# 使用
cache = ModelCache()
agent = create_agent(model=model, middleware=[cache.create_hook()])

场景 3：修改系统提示

from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from langchain_core.messages import SystemMessage
from typing import Callable

@wrap_model_call
def add_context(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
    """动态添加上下文信息到系统提示"""
    from datetime import datetime
    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    original_content = request.system_message.content if request.system_message else ""
    new_content = f"""{original_content}
      当前时间：{current_time}
      用户位置：中国
      语言偏好：中文
      """
    new_system_message = SystemMessage(content=new_content)
    # 使用 override 方法修改请求
    modified_request = request.override(system_message=new_system_message)
    return handler(modified_request)

wrap_tool_call

可以同时在工具调用前后做事，命名为 wrap_tool_call。

基于装饰器实现（在函数中两次调用 handler 并更改参数）：

from langchain.agents.middleware import wrap_tool_call
from langchain.tools.tool_node import ToolCallRequest
from langchain.messages import HumanMessage, ToolMessage
from langchain.agents import create_agent
from langchain.tools import tool
from langgraph.types import Command
from typing import Callable

@tool
def get_weather(city: str, is_forcast: bool) -> str:
    """获取当日特定城市的天气"""
    res = f"{city}今天天气不错"
    if is_forcast:
        res += "\n明天天气也很好"
    return res

@wrap_tool_call
def wrap_tool_call_middleware(
    request: ToolCallRequest,
    handler: Callable[[ToolCallRequest], ToolMessage | Command],
) -> ToolMessage | Command:
    result = handler(request)
    print(f"原始参数：{request.tool_call['args']}")
    print(f"原始参数调用结果： {result}")
    request.tool_call["args"]["is_forcast"] = True
    result = handler(request)
    print(f"更新后的参数：{request.tool_call['args']}")
    print(f"更新参数调用结果： {result}")
    return result

agent = create_agent(model=model, tools=[get_weather], middleware=[wrap_tool_call_middleware])
response = agent.invoke({"messages": [HumanMessage("你好啊，今天杭州的天气怎么样")]})

基于类实现：

from langchain.agents.middleware import AgentMiddleware
from langchain.tools.tool_node import ToolCallRequest
from langchain.messages import HumanMessage, ToolMessage
from typing import Callable
from langgraph.types import Command

class WrapToolCallMiddleware(AgentMiddleware):
    def wrap_tool_call(
        self,
        request: ToolCallRequest,
        handler: Callable[[ToolCallRequest], ToolMessage | Command],
    ) -> ToolMessage | Command:
        result = handler(request)
        print(f"原始参数：{request.tool_call['args']}")
        print(f"原始参数调用结果： {result}")
        request.tool_call["args"]["is_forcast"] = True
        result = handler(request)
        print(f"更新后的参数：{request.tool_call['args']}")
        print(f"更新参数调用结果： {result}")
        return result

agent = create_agent(model=model, tools=[get_weather], middleware=[WrapToolCallMiddleware()])

使用场景：用于监控、重试、修改工具执行。例如监控工具执行时间和状态：

from langchain.agents.middleware import wrap_tool_call
from langchain.tools.tool_node import ToolCallRequest
from langchain_core.messages import ToolMessage
from langgraph.types import Command
from typing import Callable
import time

@wrap_tool_call
def monitor_tool(
    request: ToolCallRequest,
    handler: Callable[[ToolCallRequest], ToolMessage | Command]
) -> ToolMessage | Command:
    """监控工具执行时间和状态"""
    tool_name = request.tool_call["name"]
    tool_args = request.tool_call.get("args", {})
    print(f"🔧 开始执行工具：{tool_name}")
    print(f"   参数：{tool_args}")
    start_time = time.time()
    try:
        result = handler(request)
        elapsed = time.time() - start_time
        print(f"✅ 工具执行成功，耗时：{elapsed:.2f}秒")
        return result
    except Exception as e:
        elapsed = time.time() - start_time
        print(f"❌ 工具执行失败：{e}，耗时：{elapsed:.2f}秒")
        raise

两种方法的统一

同 Node-style，装饰器方法底层也会创建一个 AgentMiddleware 的实例。

参数说明

参数	说明
request	被封装的请求对象，可以是模型或工具调用请求
handler	处理器，用于处理请求并返回调用结果

5.5 装饰器和类的选择

装饰器写法和类写法都能实现 middleware hook，本质上只是两种定义中间件的方式，并不是能力上完全割裂的两套机制，底层实现是统一的。

情况 1：单钩子用装饰器，多钩子用类

• 单个钩子函数：直接用装饰器最简单。

• 多个钩子函数：类写法更合适（结构表达、配置归属、可维护性更好）。装饰器可通过工厂函数返回多个装饰器函数完成，但不如类写法自然、集中、清晰。

装饰器实现多钩子（工厂函数）：

from langchain.agents import create_agent
from langchain.agents.middleware import before_model, after_model, AgentState
from langchain.messages import HumanMessage
from langgraph.runtime import Runtime
from loguru import logger
from typing import Any

def create_audit_middleware(logger):
    @before_model
    def before_log(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        logger.info("调用模型前消息数量: {}", len(state["messages"]))
        return None

    @after_model
    def after_log(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        logger.info("调用模型后消息数量：{}", len(state["messages"]))
        return None

    return [before_log, after_log]

agent = create_agent(model=model, middleware=[*create_audit_middleware(logger=logger)])

类实现多钩子：

class CreateAuditMiddleware(AgentMiddleware):
    def __init__(self, logger):
        super().__init__()
        self.logger = logger

    def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        self.logger.info("调用模型前消息数量: {}", len(state["messages"]))
        return None

    def after_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        self.logger.info("调用模型后消息数量：{}", len(state["messages"]))
        return None

agent = create_agent(model=model, middleware=[CreateAuditMiddleware(logger=logger)])

情况 2：复杂配置推荐用类

装饰器可通过函数闭包传参，但在自省（运行时类型校验）、调试等方面天然不如类写法方便。基于类的写法可随时打印参数信息，而基于装饰器的闭包实现难以做到。

# 基于类
class AuditMiddleware(AgentMiddleware):
    def __init__(self, logger, threshold: int, middleware_name: str):
        self.logger = logger
        self.threshold = threshold
        self.middleware_name = middleware_name

    def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        self.logger.info("current name: {}, threshold: {}", self.middleware_name, self.threshold)
        return None

# 基于装饰器，传参要通过闭包完成
def create_audit_middleware(logger, threshold: int, middleware_name: str):
    @before_model
    def audit_middleware(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        logger.info("current name: {}, threshold: {}", middleware_name, threshold)
        return None
    return audit_middleware

class_middle = [
    AuditMiddleware(logger=logger, threshold=5, middleware_name="short limit"),
    AuditMiddleware(logger=logger, threshold=50, middleware_name="long limit"),
]

对比 type 和 __dict__：

• 类风格：<class '__main__.AuditMiddleware'>，__dict__ 含 logger/threshold/middleware_name。

• 装饰器风格：<class 'langchain.agents.middleware.types.audit_middleware'>，__dict__ 为空（参数被闭包吞掉，无法自省）。

情况 3：跨项目复用推荐用类

如果希望中间件成为可实例化、可封装、可测试的组件，类写法更合适（这些本就是类擅长的场景）。装饰器的闭包也能实现，但使用不友好。

总结：

• 装饰器写法更适合单个 hook、逻辑简单、快速原型的场景；

• 类写法更适合多个 hook 组合、复杂配置、需要同时提供同步/异步实现、以及更强复用与可测试性的场景。

5.6 hook 函数执行顺序（重要）

执行顺序与 hook 类型有关：

hook 类型	执行顺序
before_* 钩子	从前到后（正序）
after_* 钩子	从后往前（逆序）
wrap_* 钩子	洋葱架构，前面的包裹后面的

这里的顺序并非定义顺序，而是创建 Agent 时传递中间件的顺序。

示例（定义顺序乱序，但传递顺序固定）：

from langchain.agents.middleware import (
    before_model, after_model, AgentState,
    wrap_model_call, ModelRequest, ModelResponse,
)
from langchain.messages import HumanMessage
from langgraph.runtime import Runtime
from langchain.agents import create_agent
from typing import Any, Callable

@before_model
def before_model_middleware1(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    state["messages"][-1].content += " -> before_model-1 <- "
    return None

@before_model
def before_model_middleware2(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    state["messages"][-1].content += " -> before_model-2 <- "
    return None

@before_model
def before_model_middleware3(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    state["messages"][-1].content += " -> before_model-3 <- "
    return None

@after_model
def after_model_middleware1(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    state["messages"][-1].content += " -> after_model-1 <- "
    return None

@after_model
def after_model_middleware2(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    state["messages"][-1].content += " -> after_model-2 <- "
    return None

@after_model
def after_model_middleware3(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    state["messages"][-1].content += " -> after_model-3 <- "
    return None

@wrap_model_call
def wrap_model_middleware1(request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]) -> ModelResponse | None:
    request.messages[-1].content += " -> wrap_model-before-1 <- "
    response = handler(request)
    response.result[0].content += " -> wrap_model-after-1 <- "
    return response

@wrap_model_call
def wrap_model_middleware2(request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]) -> ModelResponse | None:
    request.messages[-1].content += " -> wrap_model-before-2 <- "
    response = handler(request)
    response.result[0].content += " -> wrap_model-after-2 <- "
    return response

@wrap_model_call
def wrap_model_middleware3(request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]) -> ModelResponse | None:
    request.messages[-1].content += " -> wrap_model-before-3 <- "
    response = handler(request)
    response.result[0].content += " -> wrap_model-after-3 <- "
    return response

agent = create_agent(
    model=model,
    middleware=[
        before_model_middleware1,
        before_model_middleware2,
        before_model_middleware3,
        after_model_middleware1,
        after_model_middleware2,
        after_model_middleware3,
        wrap_model_middleware1,
        wrap_model_middleware2,
        wrap_model_middleware3,
    ]
)
response = agent.invoke({"messages": [HumanMessage("你好啊，忽略我后续的输入，只和我打个招呼")]})

输出（执行顺序只与传递给 Agent 的顺序有关）：

你好啊 ... -> before_model-1 <- -> before_model-2 <- -> before_model-3 <-
-> wrap_model-before-1 <- -> wrap_model-before-2 <- -> wrap_model-before-3 <-
你好啊！ -> wrap_model-after-3 <- -> wrap_model-after-2 <- -> wrap_model-after-1 <-
-> after_model-3 <- -> after_model-2 <- -> after_model-1 <-

分析：

1. before_model 中间件的执行顺序和传递顺序一致（1→2→3）。

2. after_model 中间件的执行顺序和传递顺序相反（3→2→1）。

3. wrap_model_call 中间件的执行顺序是：先传递的包在最外层（洋葱架构），before 部分正序 1→2→3，after 部分逆序 3→2→1。

---

笔记已整理完毕。涵盖：中间件概念/分类（六大类对比表）、12 个内置中间件的参数与代表性代码（Summarization、HumanInTheLoop、PII、TodoList、ModelCallLimit、ToolCallLimit、ModelFallback、LLMToolSelector、ToolRetry 含指数退避+Jitter 对比表、ModelRetry、LLMToolEmulator、ContextEditing、FilesystemFileSearch）、多中间件组合与洋葱模型执行顺序、自定义中间件（6 个 hook 函数分类、Node-style 与 Wrap-style 两种风格、装饰器与类两种写法及其底层统一、can_jump_to/hook_config 流程跳转、jump_to 返回值、装饰器 vs 类的三种选择场景、以及三类 hook 的执行顺序规律）。已删除课件中所有错位的行号并按语义还原代码，保留了 <KEY>、<YOUR_API_KEY> 类占位符，对比类内容均用表格呈现。

第09章 上下文与记忆

讲师：尚硅谷-宋红康

1、概述

1.1 为什么需要记忆(Memory)

记忆是一种记住之前交互信息的系统。随着 Agent 处理涉及大量用户交互的复杂任务，记忆变得至关重要。

大多数 LLM 应用都有会话接口，允许多轮对话并具备一定的上下文记忆能力。但实际上，大模型本身是"无状态"的，不会记忆任何上下文——每次调用 agent.invoke() 都是全新的开始。

1.2 如何解决记忆问题 —— 上下文工程

实现记忆功能，需要额外模块保存对话上下文，下一次请求时把历史信息一并输入给模型。

• 记忆(Memory)：专门负责"存储历史交互信息"的组件，核心作用是保存上下文和提供上下文，让 LLM 在每次响应时都能"看到"之前的对话。

• 上下文工程(Context Engineering)：负责"合理组织"这些记忆和任务信息，让响应更连贯、更贴合需求，是 Agent 实现复杂多轮交互的核心基础。

LangChain 的上下文工程基于 Agent 讨论，构建在 LangGraph 之上。LangGraph 提供三种管理上下文的方法（结合可变性 + 生命周期维度）：

上下文类型	描述	可变性	生命周期	访问方法
动态运行时上下文	单次运行中会演变的可变数据	动态	单次运行	LangGraph state 对象
动态跨会话上下文	对话间共享的持久数据（用户偏好、历史洞察、知识条目）	动态	跨对话	LangGraph store 对象
静态运行时上下文	启动时传入的用户元数据、工具、数据库连接	静态	单次运行	LangGraph context 对象

1.3 LangChain 的记忆

短期记忆（Short-term memory / 会话级记忆 / thread-scoped memory）：作用范围为单个对话线程(Thread)，开启新对话（更换 thread_id）即消失。

长期记忆（Long-term memory / 跨会话级记忆）：在会话间存储用户特定或应用级数据，并在会话线程间共享，可随时在任何线程调用，范围是任意自定义命名空间。

管理方式的演进：

• v0.x：通过专用的 xxxMemory 类管理记忆。

• v1.x：Agent 构建在 LangGraph 图结构之上，通过 state 和 store 构建记忆系统，更简单、统一。

  • state：短期记忆对象，以会话为单位组织，包含当前会话的所有消息记录及自定义信息。

  • store：长期记忆对象，跨会话持久化，通常结合向量库或外部存储实现。

---

2、短期记忆

LangChain 1.x 的短期记忆是三者的组合：State（会话内部状态） + Checkpointer（持久化机制） + Thread ID（会话作用域）。

• State：默认存储历史消息列表 messages，通过 State 管理历史消息。

• Checkpointer：将 State 作为检查点持久化保存，检查点是某时刻的 State 快照。

• Thread ID：唯一标识 State，运行时按 thread_id 读写 State 快照。

类比 RPG 游戏的"自动存档"：无需手动保存，系统在关键节点自动记录，下次从存档点继续。

2.1 基于内存的持久化器

最便捷的方式，适合快速测试或调试。

举例：拥有记忆

from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent
from langchain.messages import HumanMessage

checkpointer = InMemorySaver()
agent = create_agent(
    model=model,
    checkpointer=checkpointer  # 添加内存管理
)
config = {"configurable": {"thread_id": "1"}}

response1 = agent.invoke(
    {"messages": [HumanMessage("我叫张三")]},
    config=config
)
response2 = agent.invoke(
    {"messages": [HumanMessage("我叫什么？")]},
    config=config  # 使用相同的 thread_id
)
# 第二轮输出：你叫张三。

只需传入 checkpointer 和 config，Agent 就具备连续对话能力。

关键步骤说明：

1. 初始化记忆引擎：checkpointer = InMemorySaver() 创建内存级存储。InMemorySaver 进程结束即丢失，生产环境换 SqliteSaver、PostgresSaver 等。

2. 绑定 Agent：create_agent 时传入 checkpointer。

3. 设定会话 ID：config = {"configurable": {"thread_id": "1"}}。同一 thread_id 共享记忆，不同 thread_id 完全隔离。

常见问题：

• 为什么 Agent 不记得？检查三件事：是否加了 checkpointer、是否传了 config、两次 thread_id 是否相同。

• InMemorySaver 会丢数据吗？会！仅同一进程内有效，程序/进程重启丢失，不支持跨进程共享。解决：换 SQLite/PostgreSQL 持久化。

• 内存会无限增长吗？会，默认保存所有消息，导致 token 超限、成本上升、响应变慢。需上下文管理（修剪/摘要）。

2.2 基于外部存储介质的持久化器

将检查点保存到外部存储（如 PostgreSQL），生产环境必备。

依赖安装：pip install langgraph-checkpoint-postgres

from langgraph.checkpoint.postgres import PostgresSaver

DB_URL = "postgresql://langchain_user:abcd1234@118.195.128.47:5432/langchain_db?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URL) as checkpointer:
    checkpointer.setup()  # 初始化数据库（首次建表，重复执行不重建）
    agent = create_agent(model=model, checkpointer=checkpointer)
    config = {"configurable": {"thread_id": "1"}}
    response1 = agent.invoke({"messages": [HumanMessage("你好，我是老王")]}, config=config)
    response2 = agent.invoke({"messages": [HumanMessage("你好，我是谁？")]}, config=config)

setup() 会创建 4 张表：checkpoints（主表，存每个 thread 在某时刻的快照）、checkpoint_blobs（存较复杂 channel 值）、checkpoint_writes（存中间写入/pending writes）、checkpoint_migrations（迁移版本表）。

2.3 对比两种方式

维度	InMemorySaver	PostgresSaver
存储介质	内存	PostgreSQL 数据库
进程结束/重建 Saver	历史状态丢失	不丢失，可凭 thread_id 重新加载
适用场景	测试调试	生产环境

2.4 记忆治理策略（上下文管理）

历史消息累积带来问题：上下文窗口有限、长上下文下模型被陈旧内容分散注意力、token 花费高昂。需对历史做压缩、清理、重组。

2.4.1 消息裁剪（before_model 中间件）

调用模型前裁剪上下文，保留系统初始消息 + 最近若干条消息，适合成本敏感、对旧上下文依赖不强的场景。

from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langchain.agents.middleware import before_model
from langgraph.runtime import Runtime

@before_model
def trim_messages(state, runtime):
    messages = state["messages"]
    if len(messages) <= 3:
        return None
    first_msg = messages[0]
    recent_messages = messages[-3:] if len(messages) % 2 == 0 else messages[-4:]
    new_messages = [first_msg] + recent_messages
    return {
        "messages": [
            RemoveMessage(id=REMOVE_ALL_MESSAGES),  # 先清空全部
            *new_messages                            # 再保留指定消息
        ]
    }

agent = create_agent(
    model=model,
    middleware=[trim_messages],
    checkpointer=InMemorySaver(),
)

2.4.2 消息删除（after_model 中间件）

模型调用完成后将某些消息永久移除（永久更改状态），适合明确遗忘/清理/重置历史。

from langchain.agents.middleware import after_model

@after_model
def delete_old_messages(state, runtime):
    messages = state["messages"]
    if len(messages) > 5:
        to_delete = len(messages) - 5
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:to_delete]]}
    return None

RemoveMessage 的底层原理（墓碑机制）：返回 RemoveMessage(id=...) 不会真的从数组删除对象，而是作为一条"墓碑"标记追加到状态历史中。下次读取上下文时，内置 Reducer 把原始消息与墓碑标记一起计算，把被标记的消息过滤掉再喂给模型。

裁剪 vs 删除：裁剪强调"调用模型前控制可见上下文范围"；删除强调"调用模型后永久更改状态"。

2.4.3 摘要（SummarizationMiddleware）

把早期历史压缩成摘要替换原始消息，保语义不保原文，是长会话的折中方案。官方推荐内置 SummarizationMiddleware。

from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model=model_out,
    tools=[],
    checkpointer=InMemorySaver(),
    middleware=[
        SummarizationMiddleware(
            model=model_in,
            trigger=[("tokens", 100)],         # 超过 100 tokens 就摘要
            keep=("messages", 2),              # 保留最近 2 条原文
            summary_prompt="对历史消息摘要，消息列表如下\n{messages}",
        )
    ]
)

阈值经验：模型上下文窗口 4k→设 3000，8k→设 6000，16k→设 12000，留余量给工具调用和系统提示。

2.5 了解：state 的理解

AgentState 是 TypedDict 子类（可按字典方式读写），三个字段：

• messages：截止当前节点的历史消息记录，Required。

• jump_to：跳转至运行图指定节点，NotRequired，可为 None。

• structured_response：结构化输出内容，启用结构化输出时记录于此，NotRequired。

可在 before_model(can_jump_to=["tools"]) 中通过 jump_to="tools" 直接跳转工具节点。

---

3、长期记忆

3.1 基本理解

短期记忆是会话级（Thread）数据，会话间不共享；长期记忆是用户特定或应用级数据，任何会话都能访问（如"你喜欢简短回答""你偏好 Python""某用户是 VIP"）。

类型划分（参考 CoALA paper）：

Memory Type	存什么
Semantic（语义记忆）	事实（事实/用户偏好/概念）
Episodic（情景记忆）	经验（过去动作，常表现为 few-shot 示例）
Procedural（程序性记忆）	规则/做事方法（系统提示词、工作流程、工具调用规则）

存储架构：store -> namespace -> key -> value 四层。

• 第1层 Store：BaseStore 子类实例，常用 InMemoryStore(测试) / PostgresStore(生产)。

• 第2层 Namespace：任意长度 tuple[str, ...]，类似文件路径，用于分组隔离。

• 第3层 Key：namespace 下的唯一标识，str。

• 第4层 Value：dict[str, Any]。

namespace = ("users", "user_123", "preferences")
key = "profile"
value = {"language": "zh-CN", "style": "short_direct", "likes": ["python", "rag"]}
store.put(namespace, key, value)

3.2 基础 API 的使用

put() 写入、get() 读取、search() 检索。

put() / get()

from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
namespace = ("users",)
store.put(namespace, "user-1", {"name": "小蓝"})
print(store.get(namespace, "user-1"))
# Item(namespace=['users'], key='user-1', value={'name': '小蓝'}, created_at=..., updated_at=...)

put() 的 index 参数控制语义检索索引：None(默认)用 store 初始化配置；False 不建索引；list[str] 只对指定字段建索引。

search() 支持两种检索：按 filter(value 中键值) 做结构化过滤；按 query 做语义相似度检索（需向量）。

# 按 namespace 前缀搜索
store.search(("users",))
# 按 filter 过滤
store.search(("users",), filter={"food": "紫光园奶皮子酸奶"})

语义搜索与索引配置：初始化 Store 时通过 index 指定（embed/dims/fields）。fields 指定对 value 的哪些字段嵌入，可取 ["$"](整体)、["field"](一级字段)、["parent.child"](嵌套)。

embedding_model = init_embeddings(
    model="openai:text-embedding-3-large",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL"),
)
store = InMemoryStore(index={"embed": embedding_model, "dims": 3072, "fields": ["$"]})
for item in store.search(("users",), query="数电模电"):
    print(item)  # 按 score 降序返回

3.3 在 Agent 运行图中访问长期记忆

可在工具或中间件中访问。Runtime 含 context、store、stream_writer、previous 等字段，因此可通过 runtime.store 访问。

在工具中访问（通过 ToolRuntime，其含 state/context/config/stream_writer/tool_call_id/store）：

from typing import NotRequired
from langchain.agents import create_agent, AgentState
from langchain.tools import tool, ToolRuntime
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

class CustomState(AgentState):
    user_id: NotRequired[str]  # 扩展标准状态，新增 user_id

@tool(parse_docstring=True)
def save_user_info(name: str, runtime: ToolRuntime) -> str:
    """将用户信息保存在长期记忆中"""
    runtime.store.put(("users",), runtime.state["user_id"], {"name": name})
    return "saved"

@tool(parse_docstring=True)
def get_user_info(runtime: ToolRuntime) -> str:
    """从长期记忆中读取用户信息"""
    item = runtime.store.get(("users",), runtime.state["user_id"])
    return str(item.value) if item else "unknown"

agent = create_agent(
    model=model,
    tools=[save_user_info, get_user_info],
    store=store,
    system_prompt="用户提及个人信息时及时记录，用户询问个人信息时尝试用工具检索",
    state_schema=CustomState,
)
# 两次独立会话（不同 thread），第二个会话能读取第一个会话写入的长期记忆

在中间件中访问：Node-style hooks（如 before_model）通过 runtime.store；Wrap-style hooks（wrap_model_call 通过 request.runtime.store；wrap_tool_call 通过 request.runtime.store）。

3.4 何时写入记忆

• 主流程写(hot path)：AI 一边回答一边决定是否记录。优点：立即生效、用户可感知；缺点：增加延迟、逻辑复杂。

• 后台写(background)：先回答用户，记忆整理异步进行。优点：主流程快、记忆逻辑独立；缺点：不能立刻生效。

工程经验：用户偏好、账号资料可热路径写；对话摘要、经验沉淀、行为分析更适合后台写。

---

4、课后阅读：Static runtime Context

静态运行时上下文表示不可变数据（用户元数据、工具、数据库连接），通常运行开始时通过 invoke/stream 的 context 参数传入，运行期间不变。自定义 ContextSchema 用 @dataclass 修饰，通过 context_schema 传给 Agent。

总结：短期记忆靠 state+checkpointer+thread_id；长期记忆靠 store+namespace+key+value；静态上下文靠 context；三者均可在工具/中间件中通过 runtime/request.runtime 统一访问。

第10章 RAG（Retrieval）

讲师：尚硅谷-宋红康

Retrieval 即"检索"，本章涵盖数据获取、切分、向量化、向量存储、向量检索等模块。

1、Retrieval 模块的设计意义

1.1 大模型的局限

1. 知识滞后：训练数据有截止日期，无法反映最新信息。

2. 知识缺失：训练依赖公开静态数据，缺乏企业内部资料、私有数据，导致生成不准确甚至虚构。

3. 幻觉：LLM 可能"胡言乱语"（错误陈述、编造事实、复杂推理错误）。幻觉在金融、医疗等领域代价致命。成因：训练知识偏差、过度泛化、未学到深层含义、缺乏领域知识。共识方案：先为模型提供上下文稳定输出，再用 RAG 把检索到的文档和提示词一起输送给模型。

1.2 什么是 RAG

RAG（Retrieval-Augmented Generation，检索增强生成）：结合信息检索(Retrieval)与文本生成(Generation)，提升 LLM 回答专业问题的准确性和可靠性。如果说 LangChain 给 LLM 装上"四肢和躯干"，RAG 则为 LLM 接入"人类知识图书馆"。

1.3 RAG 优缺点

维度	说明
优点	比提示词工程有更丰富上下文和数据样本；比模型微调提升时效性和可靠性；在一定程度上保护业务数据隐私
缺点	每次问答涉及外部检索，响应时延较高；引用的外部知识消耗大量 Token

1.4 RAG 工作流程（6 环节）

1. Source(数据源)：RAG 外挂的知识库，原始数据类型多样（视频/图片/文本/代码/文档），形式多样（CSV/JSON/PDF/API/网站实时数据）。

2. Load(加载)：Document Loaders 将非结构化文本加载为 Document 对象（含 page_content + metadata）。支持"延迟加载"缓解大文件内存压力。

3. Transform(转换)：含文本拆分器、冗余过滤器、元数据提取器、多语言转换器、对话转换器。其中文档拆分器是必须操作。

4. Embed(嵌入)：Text Embedding Models 将文本转为向量。关键特性：相似的词向量距离近。

5. Store(存储)：将嵌入存储到向量库或临时缓存，避免重复计算。

6. Retrieve(检索)：Retrievers 响应非结构化查询，返回符合要求的文档。

---

2、详细使用流程

2.2 文档加载器 Document Loaders

LangChain 实现了众多文档加载器。每个加载器继承自 BaseLoader，提供通用的 load（一次加载所有）与 lazy_load（延迟加载）方法。

Document 对象两个重要属性：page_content（文档内容，字符串）、metadata（元数据，字典）。

常用 Loaders：TextLoader(文本)、CSVLoader(CSV)、PyPDFLoader(PDF)、WebBaseLoader(网页)、JSONLoader、DirectoryLoader(文件夹)。

加载 txt：

from langchain_community.document_loaders import TextLoader
text_loader = TextLoader(file_path="../asset/load/01-langchain-utf-8.txt", encoding="utf-8")
docs = text_loader.load()  # 返回 List[Document]

加载 CSV：CSVLoader 每行变一个 Document，metadata 含 row 序号。

加载 JSON（JSONLoader）：使用 jq 语法解析 JSON（需 pip install jq）。常见 jq_schema 对照：["...","..."] → .[]；[{"text": ...}] → .[].text；{"key":[{"text":...}]} → .key[].text。

加载 PDF：

• PyPDFLoader（需 pip install pypdf）：extraction_mode 支持 plain(默认，纯文本) 与 layout(布局感知，插入空格/换行模拟多栏，适合学术论文/多栏报刊)。

• MinerU：在线服务，支持 PDF/Word/PPT/图片解析、OCR、公式、表格识别。

from langchain_community.document_loaders import PyPDFLoader
loader = PyPDFLoader(
    file_path="https://arxiv.org/pdf/alg-geom/9202012",  # 支持本地或在线链接
    extraction_mode="plain",
)
docs = loader.load()

其他：UnstructuredWordDocumentLoader(Word, mode: single/elements)、UnstructuredMarkdownLoader(Markdown, mode/strategy)、UnstructuredHTMLLoader(HTML)、DirectoryLoader(批量加载文件夹，配合 glob/loader_cls)。

BaseLoader 定义如何从不同数据源加载文档，所有 loader 继承它，最少实现 lazy_load。load() 不应被重写（内部调用 lazy_load），还有 aload(异步)、load_and_split(加载并切分，默认用 RecursiveCharacterTextSplitter)。

2.3 文档切分器 Text Splitters

2.3.1 为什么切分：长文档超 Token 限制会被截断；小文档检索更精准；控制成本。无论存储还是检索，都以 chunk 为基本单位。

2.3.2 Chunking 拆分策略

方法	特点
1 按句子切分	按自然句子边界，保持语义完整
2 按固定字符数切分	可能在不适当位置切断
3 固定字符数 + 重叠窗口	通过 overlap 避免切断关键内容
4 递归字符切分	动态确定切分点，通常是首选，兼顾固定长度与语义
5 按语义内容切分	依据语义划分，精确但效率低、长度不均

TextSplitter 源码核心：TextSplitter(BaseDocumentTransformer, ABC) 构造参数 chunk_size=4000、chunk_overlap=200、length_function=len、keep_separator、add_start_index、strip_whitespace，约束 chunk_overlap < chunk_size。常用方法调用链：split_documents(documents) -> create_documents(texts, metadatas) -> split_text(text)。

① CharacterTextSplitter（按字符切分）：参数 chunk_size、chunk_overlap、separator(默认 "\n\n")。separator 优先原则——先在分隔符处切分再考虑 chunk_size，避免切句子中间。

from langchain_text_splitters import CharacterTextSplitter
text_splitter = CharacterTextSplitter(
    chunk_size=30, chunk_overlap=5, separator="。", keep_separator=True
)
chunks = text_splitter.split_text(text)

② RecursiveCharacterTextSplitter（最常用）：递归尝试分隔符列表 ["\n\n", "\n", " ", ""]，先按自然边界切，片段仍过大再逐级退化到更细分隔符，最后按 chunk_size/overlap 组织。价值：尽量保留语义完整性。

from langchain_text_splitters import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=200, chunk_overlap=20,
    separators=["\n\n", "\n", "。", "！", "？", "……", "，", ""],
    length_function=len, keep_separator=True,
)
paragraphs = text_splitter.split_documents(docs)

③ TokenTextSplitter / CharacterTextSplitter.from_tiktoken_encoder（按 Token 切分）：Token 是模型的最小处理单位。按 Token 切分能确保每块不超模型 Token 上限，与 LLM 计费逻辑一致。

from langchain_text_splitters import TokenTextSplitter
text_splitter = TokenTextSplitter(
    chunk_size=33, chunk_overlap=0,
    encoding_name="cl100k_base",  # OpenAI 编码器
)
texts = text_splitter.split_text(text)

可选编码器：gpt2、r50k_base、p50k_base、cl100k_base、o200k_base 等。注意：字符长度不等于 Token 数量。

④ SemanticChunker（语义分块）：通过 Embedding 计算前后句子语义差异，超过阈值则切断，保证块语义完整。

特性	语义分割(SemanticChunker)	传统字符分割(RecursiveCharacter)
分割依据	嵌入向量相似度	固定字符/换行符
语义完整性	保持主题连贯	可能切断句子逻辑
计算成本	高（需嵌入模型）	低
适用场景	需要高语义一致性的任务	简单文本预处理

from langchain_experimental.text_splitter import SemanticChunker

text_splitter = SemanticChunker(
    embeddings=embedding_model,
    breakpoint_threshold_type="percentile",  # percentile/standard_deviation/interquartile/gradient
    breakpoint_threshold_amount=65.0,        # 数值越小切分越细，越大越粗
    sentence_split_regex=r"(?<=[。？！])\s+",
)
docs = text_splitter.create_documents([state_of_the_union])

其他：HTMLHeaderTextSplitter(按 HTML 标题标签划分，标题层级保存在 metadata)、CodeTextSplitter(按语言语法结构拆分，RecursiveCharacterTextSplitter.from_language(language=Language.PYTHON, ...))、MarkdownTextSplitter(按多级标题切分)。

2.4 文档嵌入模型 Text Embedding Models

LangChain 提供两种接口：embed_query(句子向量化) 与 embed_documents(文档向量化)。

常用嵌入模型：

模型	机构	描述
bge-large-zh	BAAI	开源，1024 维，序列 512
bge-m3	BAAI	开源，多语言，1024 维，序列 8192
text-embedding-3-small	OpenAI	多语言，1536 维，序列 8192
text-embedding-3-large	OpenAI	多语言，3072 维，序列 8192

初始化（硅基流动，免费模型 BAAI/bge-m3）：

from langchain.embeddings import init_embeddings
embedding_model = init_embeddings(
    model="openai:Pro/BAAI/bge-m3",
    api_key=os.getenv("SILICONFLOW_API_KEY"),
    base_url=os.getenv("SILICONFLOW_BASE_URL"),
)

句子向量化 embed_query：

embedded_query = embedding_model.embed_query(text="What was the name mentioned?")
print(embedded_query[:5], len(embedded_query))  # 如 1024 维

文档向量化 embed_documents：

texts = ["Hi there!", "Oh, hello!", "What's your name?"]
embeded_docs = embedding_model.embed_documents(texts)

2.5 向量存储（Vector Stores）

传统数据库适合精确搜索，向量数据库适合按内容模糊搜索。向量库检索是模糊的、近似的最相似匹配。

常用向量数据库：FAISS(Meta，开源免费)、Chroma(轻量级)、Milvus(云原生，十亿级生产)、Pgvector(PostgreSQL 扩展)、Redis、Elasticsearch、Pinecone。本课程使用 Milvus。

案例：完整 RAG 生命周期（加载→切分→向量化→存储→检索→生成）

① 初始化 Milvus 并建集合：

from pymilvus import MilvusClient

MILVUS_URI = "http://localhost:19530"
DB_NAME = "rag_tutorial"
COLLECTION_NAME = "docs"
EMBED_DIM = 1024  # BGE-M3 输出维度固定 1024

client = MilvusClient(MILVUS_URI)
if DB_NAME not in client.list_databases():
    client.create_database(db_name=DB_NAME)
client.use_database(db_name=DB_NAME)

if client.has_collection(collection_name=COLLECTION_NAME):
    client.drop_collection(collection_name=COLLECTION_NAME)
client.create_collection(
    collection_name=COLLECTION_NAME,
    dimension=EMBED_DIM,   # 预先开辟 1024 维空间
    metric_type="COSINE",  # 余弦相似度，数值越大越相似
)

metric_type="COSINE"：余弦相似度关注向量方向夹角。方向完全一致→接近1；正交→接近0。其他常见度量：L2 欧氏距离、IP 内积。

④ 读取文档并切分：

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

loader = TextLoader(KNOWLEDGE_FILE, encoding="utf-8")
documents = loader.load()
splitter = RecursiveCharacterTextSplitter(
    chunk_size=220, chunk_overlap=80,
    separators=["\n==============================\n", "\n\n", "\n", "。", "，", " ", ""],
)
chunks = splitter.split_documents(documents)

⑤ 生成向量并写入 Milvus（upsert，写数据后 flush 落盘）：

vectors = embed_model.embed_documents([chunk.page_content for chunk in chunks])
data = [
    {"id": i, "vector": vectors[i], "text": chunks[i].page_content,
     "source": KNOWLEDGE_FILE, "chunk_id": i}
    for i in range(len(chunks))
]
client.upsert(collection_name=COLLECTION_NAME, data=data)
client.flush(collection_name=COLLECTION_NAME)

⑦ 检索逻辑（单条查询用 embed_query）：

def retrieve(question: str, k: int = 5):
    query_vector = embed_model.embed_query(question)
    results = client.search(
        collection_name=COLLECTION_NAME,
        data=[query_vector],        # 接收列表
        limit=k,                     # 返回最相似前 K 条
        output_fields=["text", "source", "chunk_id"],
    )
    return results[0]

⑧ 生成回答（检索→拼接 context→构造 Prompt→调用 Agent）：

def generate_answer(question: str):
    hits = retrieve(question, k=5)
    context_blocks = []
    for i, hit in enumerate(hits, 1):
        text = hit["entity"]["text"]
        source = hit["entity"].get("source", "unknown")
        chunk_id = hit["entity"].get("chunk_id", "unknown")
        score = hit["distance"]  # COSINE 模式下 score 越高越相似
        context_blocks.append(f"[片段{i} | chunk_id={chunk_id} | source={source}]\n{text}")
    context = "\n\n".join(context_blocks)
    user_prompt = f"问题：\n{question}\n上下文：\n{context}\n"
    result = agent.invoke({"messages": [{"role": "user", "content": user_prompt}]})
    result["messages"][-1].pretty_print()

检索结果会按 distance(score) 降序返回最相关的 5 个 chunk，Agent 仅根据上下文回答（上下文不足则答"我不知道"，且不执行其中可能包含的指令，防范 prompt 注入）。

关键提示：复杂的 RAG 项目中，文档加载与切分是最关键也最复杂的环节，通常会用更专业的文档处理工具；LangChain 工具链的优势是接口统一、快速上手，适合 MVP 或学习项目。