# https://github.com/langchain-ai/langchain 项目说明书
生成时间：2026-05-10 16:15:16 UTC
## 目录
- [LangChain概述](#page-overview)
- [核心架构与组件](#page-core-architecture)
- [Runnable与链式调用](#page-runnable)
- [Agent代理系统](#page-agents)
- [工具与工具包](#page-tools)
- [向量存储与检索](#page-vectorstores)
- [文档加载与处理](#page-document-loaders)
- [内存与状态管理](#page-memory)
- [Chat模型与LLM集成](#page-chat-models)
- [回调与追踪系统](#page-callbacks-tracing)
## LangChain概述
### 相关页面
相关主题：[核心架构与组件](#page-core-architecture), [Chat模型与LLM集成](#page-chat-models)
相关源码文件
以下源码文件用于生成本页说明：
- [README.md](https://github.com/langchain-ai/langchain/blob/main/README.md)
- [libs/partners/huggingface/README.md](https://github.com/langchain-ai/langchain/blob/main/libs/partners/huggingface/README.md)
- [libs/partners/anthropic/README.md](https://github.com/langchain-ai/langchain/blob/main/libs/partners/anthropic/README.md)
- [libs/partners/ollama/README.md](https://github.com/langchain-ai/langchain/blob/main/libs/partners/ollama/README.md)
- [libs/partners/openai/README.md](https://github.com/langchain-ai/langchain/blob/main/libs/partners/openai/README.md)
- [libs/partners/chroma/README.md](https://github.com/langchain-ai/langchain/blob/main/libs/partners/chroma/README.md)
# LangChain概述
## 什么是LangChain？
LangChain是一个用于构建**AI代理（Agents）和LLM驱动应用**的框架。它通过将可互操作的组件与第三方集成串联在一起，简化了AI应用程序的开发过程，同时为底层技术演进提供了前瞻性支持。
LangChain的主要特性包括：
- **灵活抽象层** — 从用于快速起步的高级链（Chains）到用于细粒度控制的低级组件，LangChain可以随着应用程序复杂度的增长而扩展
- **模块化设计** — 可互换的组件便于定制和扩展
- **丰富的集成生态** — 支持多种LLM提供商、向量存储、工具和外部服务
资料来源：[README.md:1-25](https://github.com/langchain-ai/langchain/blob/main/README.md)
## 核心架构
LangChain采用模块化架构设计，主要包含以下核心层次：
```mermaid
graph TD
A[LangChain主框架] --> B[langchain_core 核心库]
A --> C[langchain_classic 经典组件库]
A --> D[Partner Packages 合作伙伴包]
B --> B1[基础抽象层]
B --> B2[核心API]
B --> B3[回调系统]
C --> C1[Chains 链]
C --> C2[工具扩展]
C --> C3[索引API]
D --> D1[OpenAI集成]
D --> D2[HuggingFace集成]
D --> D3[Anthropic集成]
D --> D4[Ollama集成]
D --> D5[向量存储集成]
```
### 核心库 (langchain_core)
`langchain_core`是LangChain的基础库，提供了所有其他组件依赖的核心抽象和接口。
| 模块 | 功能描述 |
|------|----------|
| `language_models` | 语言模型基础接口 |
| `prompts` | 提示词模板抽象 |
| `callbacks` | 异步/同步回调管理系统 |
| `tools` | 工具接口定义 |
### 经典组件库 (langchain_classic)
`langchain_classic`包含传统的链式组件、代理逻辑、API调用链等经典功能。
| 组件类型 | 说明 |
|----------|------|
| Chains | 可组合的任务执行链 |
| Agents | AI代理实现 |
| Tools | 可扩展的工具集 |
| Deprecated APIs | 已废弃但仍可用的历史接口 |
资料来源：[libs/langchain/langchain_classic/chains/api/base.py:1-30](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/langchain_classic/chains/api/base.py)
## 合作伙伴集成生态
LangChain提供了丰富的**合作伙伴包（Partner Packages）**集成，支持主流AI服务提供商和工具。
### 支持的LLM提供商
| 包名称 | 提供商 | 安装命令 |
|--------|--------|----------|
| langchain-openai | OpenAI | `pip install langchain-openai` |
| langchain-anthropic | Anthropic (Claude) | `pip install langchain-anthropic` |
| langchain-huggingface | Hugging Face | `pip install langchain-huggingface` |
| langchain-ollama | Ollama (本地模型) | `pip install langchain-ollama` |
| langchain-groq | Groq | `pip install langchain-groq` |
| langchain-deepseek | DeepSeek | `pip install langchain-deepseek` |
| langchain-perplexity | Perplexity | `pip install langchain-perplexity` |
| langchain-xai | xAI (Grok) | `pip install langchain-xai` |
| langchain-openrouter | OpenRouter | `pip install langchain-openrouter` |
| langchain-fireworks | Fireworks.ai | `pip install langchain-fireworks` |
资料来源：
- [libs/partners/openai/README.md:1-20](https://github.com/langchain-ai/langchain/blob/main/libs/partners/openai/README.md)
- [libs/partners/anthropic/README.md:1-15](https://github.com/langchain-ai/langchain/blob/main/libs/partners/anthropic/README.md)
- [libs/partners/huggingface/README.md:1-20](https://github.com/langchain-ai/langchain/blob/main/libs/partners/huggingface/README.md)
- [libs/partners/ollama/README.md:1-15](https://github.com/langchain-ai/langchain/blob/main/libs/partners/ollama/README.md)
### 向量存储与工具集成
| 包名称 | 用途 | 安装命令 |
|--------|------|----------|
| langchain-chroma | Chroma向量数据库 | `pip install langchain-chroma` |
| langchain-nomic | Nomic嵌入服务 | `pip install langchain-nomic` |
| langchain-exa | Exa网络搜索API | `pip install langchain-exa` |
资料来源：
- [libs/partners/chroma/README.md:1-15](https://github.com/langchain-ai/langchain/blob/main/libs/partners/chroma/README.md)
- [libs/partners/nomic/README.md:1-15](https://github.com/langchain-ai/langchain/blob/main/libs/partners/nomic/README.md)
## 快速开始
### 安装
```bash
pip install langchain
# 或使用 uv
uv add langchain
```
### 基础使用示例
```python
from langchain.chat_models import init_chat_model
# 初始化聊天模型
model = init_chat_model("openai:gpt-5.4")
# 调用模型
result = model.invoke("Hello, world!")
```
### 使用HuggingFace本地模型
如需使用本地推理功能，需安装完整依赖：
```bash
pip install langchain-huggingface[full]
```
> **注意**：完整安装包含 `sentence-transformers>=5.2.0` 和 `transformers>=5.0.0`。从 `langchain-community` 迁移的用户需要注意版本要求变化。
资料来源：
- [README.md:30-45](https://github.com/langchain-ai/langchain/blob/main/README.md)
- [libs/partners/huggingface/README.md:10-20](https://github.com/langchain-ai/langchain/blob/main/libs/partners/huggingface/README.md)
## 技术特性
### 安全机制
LangChain的APIChain组件包含详细的安全警告和最佳实践：
- **网络安全** — 使用requests工具包发起HTTP请求，需控制访问权限
- **域限制** — 支持`limit_to_domains`参数限制可访问的域名范围
- **请求类型** — 支持GET、POST、PATCH、PUT和DELETE等HTTP方法
```python
def _check_in_allowed_domain(url: str, limit_to_domains: Sequence[str]) -> bool:
"""检查URL是否在允许的域名列表中"""
scheme, domain = _extract_scheme_and_domain(url)
for allowed_domain in limit_to_domains:
allowed_scheme, allowed_domain_ = _extract_scheme_and_domain(allowed_domain)
if scheme == allowed_scheme and domain == allowed_domain_:
return True
return False
```
资料来源：[libs/langchain/langchain_classic/chains/api/base.py:35-50](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/langchain_classic/chains/api/base.py)
### 废弃与迁移策略
LangChain采用渐进式废弃策略，标记废弃的功能会：
- 指定废弃版本号 (`since`)
- 指定移除版本号 (`removal`)
- 提供替代方案建议 (`alternative`)
```python
@deprecated(
since="0.2.13",
removal="2.0.0",
alternative="langchain.agents.create_agent",
addendum="使用 create_agent 构建新代理..."
)
class APIChain(Chain):
...
```
## 文档与资源
| 资源类型 | 链接 |
|----------|------|
| 官方文档 | [docs.langchain.com](https://docs.langchain.com/oss/python/langchain/overview) |
| API参考 | [reference.langchain.com/python](https://reference.langchain.com/python) |
| 社区论坛 | [forum.langchain.com](https://forum.langchain.com) |
| LangChain.js | [GitHub - langchainjs](https://github.com/langchain-ai/langchainjs) |
| Deep Agents | [Deep Agents文档](http://docs.langchain.com/oss/python/deepagents/) |
| LangSmith | [开发调试平台](https://docs.langchain.com/langsmith/home) |
| LangChain Academy | [免费学习课程](https://academy.langchain.com/) |
## 相关工具与扩展
### LangGraph
对于需要更高级自定义或代理编排的场景，推荐使用**LangGraph** — LangChain官方的工作流控制框架。
```mermaid
graph LR
A[LangChain] -->|基础链式调用| B[简单任务]
A -->|复杂编排| C[LangGraph]
C -->|可控工作流| D[复杂Agent系统]
```
### Deep Agents
针对常见使用模式（规划、子代理、文件系统操作等）开箱即用的解决方案：
- 内置规划能力
- 多子代理协调
- 文件系统集成
- 常见模式封装
## 版本与发布策略
LangChain遵循语义化版本控制规范：
- **版本历史** — 详见 [Releases Policy](https://docs.langchain.com/oss/python/release-policy)
- **版本规范** — 详见 [Versioning Policy](https://docs.langchain.com/oss/python/versioning)
- **贡献指南** — 详见 [Contributing Guide](https://docs.langchain.com/oss/python/contributing/overview)
## 许可证
LangChain采用 **MIT许可证** 开源发布。
资料来源：[README.md:10-15](https://github.com/langchain-ai/langchain/blob/main/README.md)
---
## 核心架构与组件
### 相关页面
相关主题：[Runnable与链式调用](#page-runnable), [回调与追踪系统](#page-callbacks-tracing)
相关源码文件
以下源码文件用于生成本页说明：
- [README.md](https://github.com/langchain-ai/langchain/blob/main/README.md)
- [libs/langchain/README.md](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/README.md)
- [libs/langchain/langchain_classic/chains/api/base.py](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/langchain_classic/chains/api/base.py)
- [libs/partners/anthropic/README.md](https://github.com/langchain-ai/langchain/blob/main/libs/partners/anthropic/README.md)
- [libs/partners/openai/README.md](https://github.com/langchain-ai/langchain/blob/main/libs/partners/openai/README.md)
- [libs/partners/huggingface/README.md](https://github.com/langchain-ai/langchain/blob/main/libs/partners/huggingface/README.md)
- [libs/partners/ollama/README.md](https://github.com/langchain-ai/langchain/blob/main/libs/partners/ollama/README.md)
# 核心架构与组件
## 概述
LangChain 是一个用于构建智能体（Agent）和大语言模型（LLM）应用的框架。它通过将可互操作的组件与第三方集成串联在一起，简化了 AI 应用程序开发流程，同时在底层技术演进的背景下保障了决策的前瞻性。
资料来源：[README.md:1-10]()
## 系统架构分层
LangChain 的架构采用分层设计，主要包括以下几个核心层次：
```mermaid
graph TD
A[用户应用层] --> B[LangChain 核心层]
B --> C[Partner 包集成层]
C --> D[第三方服务 API]
B --> E[langchain_classic 经典组件]
B --> F[langchain_core 核心抽象]
F --> G[Messages 消息系统]
F --> H[Prompts 提示词模板]
F --> I[Output Parsers 输出解析器]
F --> J[Callbacks 回调机制]
```
## 核心模块解析
### langchain_core 核心抽象层
`langchain_core` 是整个框架的基础层，提供了所有组件赖以实现的抽象接口和核心类型系统。
#### 消息系统（Messages）
消息系统负责在应用各组件之间传递结构化的对话信息。LangChain 定义了标准化的消息格式，使得不同模型提供商之间的交互具有一致性。
#### 提示词模板（Prompts）
提示词模板系统提供了构建动态提示的能力，支持变量绑定和复用。模板系统与模型无关，可在不同的 LLM 提供商之间无缝切换。
#### 输出解析器（Output Parsers）
输出解析器负责将 LLM 的原始输出转换为结构化数据。这一机制使得 LLM 输出能够被下游组件直接消费和处理。
#### 回调机制（Callbacks）
回调系统提供了可扩展的事件通知机制，允许开发者在链式调用和智能体执行过程中插入自定义逻辑，如日志记录、监控和中间结果处理。
资料来源：[libs/langchain/README.md:1-8]()
### langchain_classic 经典组件
`langchain_classic` 包包含了 LangChain 的经典组件，包括链式调用（Chains）、API 集成工具和已弃用的遗留功能。
#### APIChain 组件
`APIChain` 是 LangChain 提供的一个经典链式组件，用于根据用户问题构建 API 请求并总结响应。其核心工作流程如下：
```mermaid
graph LR
A[用户问题] --> B[API_URL_PROMPT]
B --> C[LLM 生成 API 调用]
C --> D[TextRequestsWrapper 执行请求]
D --> E[API_RESPONSE_PROMPT]
E --> F[LLM 生成最终答案]
```
**安全特性**：APIChain 内置了域名限制机制，仅允许向预配置的域名发送请求：
```python
def _check_in_allowed_domain(url: str, limit_to_domains: Sequence[str]) -> bool:
scheme, domain = _extract_scheme_and_domain(url)
for allowed_domain in limit_to_domains:
allowed_scheme, allowed_domain_ = _extract_scheme_and_domain(allowed_domain)
if scheme == allowed_scheme and domain == allowed_domain_:
return True
return False
```
资料来源：[libs/langchain/langchain_classic/chains/api/base.py:1-45]()
## Partner 集成包架构
LangChain 采用模块化的 Partner 包设计，每个 LLM 或服务提供商都有独立的包进行维护。这种设计带来了以下优势：
| 优势 | 说明 |
|------|------|
| 独立版本管理 | 各集成包可独立发布和版本控制 |
| 按需依赖 | 用户只需安装实际使用的集成包 |
| 灵活升级 | 可独立升级特定提供商而无需影响其他组件 |
### Partner 包列表
| 包名 | 集成对象 | 安装命令 |
|------|---------|---------|
| langchain-openai | OpenAI GPT 系列 | `pip install langchain-openai` |
| langchain-anthropic | Anthropic Claude 系列 | `pip install langchain-anthropic` |
| langchain-huggingface | HuggingFace 模型和工具 | `pip install langchain-huggingface` |
| langchain-ollama | Ollama 本地模型 | `pip install langchain-ollama` |
| langchain-deepseek | DeepSeek 模型 | `pip install langchain-deepseek` |
| langchain-groq | Groq 推理服务 | `pip install langchain-groq` |
| langchain-chroma | Chroma 向量数据库 | `pip install langchain-chroma` |
资料来源：[libs/partners/openai/README.md:1-20]()
## 快速入门与初始化模式
LangChain 提供了统一的模型初始化接口 `init_chat_model`，支持通过字符串标识符动态加载和配置模型：
```python
from langchain.chat_models import init_chat_model
model = init_chat_model("openai:gpt-5.4")
result = model.invoke("Hello, world!")
```
这种初始化模式的核心优势在于：
1. **配置集中化**：所有模型配置通过单一入口管理
2. **提供商无关**：代码不依赖特定服务提供商的 SDK 接口
3. **环境适配**：自动处理不同环境的认证和连接设置
资料来源：[README.md:30-45]()
## 生态系统架构图
```mermaid
graph BT
LangChain --> LangGraph
LangChain --> LangSmith
LangGraph --> LangSmith
LangChain --> Partner_Packages
Partner_Packages --> OpenAI
Partner_Packages --> Anthropic
Partner_Packages --> HuggingFace
Partner_Packages --> Ollama
Partner_Packages --> DeepSeek
Partner_Packages --> Groq
Partner_Packages --> Chroma
LangGraph --> Agents
Agents --> Tools
Tools --> APIs
```
## 组件依赖关系
### 核心依赖链
```
langchain (主包)
├── langchain_core (强制依赖)
│ ├── messages (消息类型定义)
│ ├── prompts (模板系统)
│ ├── output_parsers (输出解析)
│ └── callbacks (事件回调)
├── langchain_classic (核心链式组件)
│ ├── chains (经典链)
│ ├── agents (经典代理)
│ └── tools (工具集)
└── Partner 包 (可选依赖)
├── langchain-openai
├── langchain-anthropic
├── langchain-huggingface
└── ... (其他提供商)
```
### 依赖版本要求
各 Partner 包对依赖库有不同的版本要求：
| 包 | 核心依赖 | 版本要求 |
|----|---------|---------|
| langchain-huggingface[full] | sentence-transformers | >=5.2.0 |
| langchain-huggingface[full] | transformers | >=5.0.0 |
资料来源：[libs/partners/huggingface/README.md:1-20]()
## 安全与权限控制
### APIChain 域名限制
APIChain 实现了严格的域名白名单机制，防止应用向未授权的外部服务发送请求。这一机制通过 `_check_in_allowed_domain` 函数实现：
```mermaid
graph TD
A[API 请求 URL] --> B{检查域名是否在白名单}
B -->|是| C[允许请求]
B -->|否| D[拒绝请求]
```
**安全建议**：在生产环境中使用 APIChain 时，应严格限制 `limit_to_domains` 参数的范围，仅包含必要的服务域名。
资料来源：[libs/langchain/langchain_classic/chains/api/base.py:35-45]()
## 技术选型要点
### 何时使用 langchain_classic vs 新架构
| 场景 | 推荐方案 |
|------|---------|
| 新项目开发 | 使用 LangGraph 和 `create_agent` |
| 现有 APIChain 用户 | 迁移至新架构（已在 0.2.13 标记弃用） |
| 需要自定义链式逻辑 | 使用 langchain_classic |
| 标准 Agent 构建 | 使用 LangGraph |
**弃用说明**：`APIChain` 已在 0.2.13 版本标记为弃用，计划在 2.0.0 版本移除。推荐迁移至使用 `create_agent` 和自定义工具绑定的方式。
资料来源：[libs/langchain/langchain_classic/chains/api/base.py:45-60]()
## 总结
LangChain 的核心架构采用分层和模块化设计，通过 `langchain_core` 提供的基础抽象、`langchain_classic` 的经典组件以及丰富的 Partner 包生态，支持快速构建和部署 LLM 应用。开发者应优先使用最新的 LangGraph 架构进行新项目开发，同时了解经典组件的迁移路径以保障长期维护性。
---
## Runnable与链式调用
### 相关页面
相关主题：[核心架构与组件](#page-core-architecture), [Agent代理系统](#page-agents)
相关源码文件
以下源码文件用于生成本页说明：
- [libs/core/langchain_core/runnables/base.py](https://github.com/langchain-ai/langchain/blob/main/libs/core/langchain_core/runnables/base.py)
- [libs/core/langchain_core/runnables/config.py](https://github.com/langchain-ai/langchain/blob/main/libs/core/langchain_core/runnables/config.py)
- [libs/core/langchain_core/runnables/branch.py](https://github.com/langchain-ai/langchain/blob/main/libs/core/langchain_core/runnables/branch.py)
- [libs/core/langchain_core/runnables/retry.py](https://github.com/langchain-ai/langchain/blob/main/libs/core/langchain_core/runnables/retry.py)
# Runnable与链式调用
## 概述
Runnable 是 LangChain 框架的核心抽象接口，定义了所有可执行组件的统一契约。它代表了可以接收输入、产生输出的任意操作单元，包括语言模型、提示词模板、工具、检索器、链等。
LangChain 的核心设计理念是将复杂的 AI 应用分解为可组合的小单元，通过链式调用（Chaining）将多个 Runnable 连接在一起，形成端到端的工作流。
## 核心架构
### Runnable 在 LangChain 中的角色
```mermaid
graph TD
A[输入] --> B[Runnable 组件 1]
B --> C[Runnable 组件 2]
C --> D[Runnable 组件 N]
D --> E[输出]
F[BaseRunnable] --> G[RunnableSerializable]
F --> H[Chain 基类]
F --> I[PromptTemplate]
F --> J[BaseChatModel]
F --> K[BaseTool]
```
### 关键类层次结构
| 类名 | 层级 | 说明 |
|------|------|------|
| `BaseRunnable` | 抽象基类 | 定义所有 Runnable 的核心接口 |
| `RunnableSerializable` | 序列化层 | 提供 JSON/字典序列化能力 |
| `Chain` | 业务层 | 链式调用的具体实现基类 |
| 具体实现类 | 应用层 | `LLMChain`、`ConversationalRetrievalChain` 等 |
## BaseRunnable 抽象基类
### 核心方法定义
`BaseRunnable` 定义了 Runnable 的核心接口规范，所有实现类必须遵循这些契约。
```python
# libs/core/langchain_core/runnables/base.py
class BaseRunnable(Generic[Input, Output]):
"""所有 Runnable 对象的基类。"""
def invoke(self, input: Input, config: Optional[RunnableConfig] = None) -> Output:
"""同步调用 Runnable。"""
...
async def ainvoke(self, input: Input, config: Optional[RunnableConfig] = None) -> Output:
"""异步调用 Runnable。"""
...
def batch(self, inputs: List[Input], config: Optional[RunnableConfig] = None) -> List[Output]:
"""批量同步调用。"""
...
async def abatch(self, inputs: List[Input], config: Optional[RunnableConfig] = None) -> List[Output]:
"""批量异步调用。"""
...
```
### 输入输出类型
| 方法 | 输入类型 | 输出类型 | 说明 |
|------|----------|----------|------|
| `invoke` | `Input` | `Output` | 单次同步调用 |
| `ainvoke` | `Input` | `Output` | 单次异步调用 |
| `batch` | `List[Input]` | `List[Output]` | 批量同步调用 |
| `abatch` | `List[Input]` | `List[Output]` | 批量异步调用 |
| `stream` | `Input` | `Iterator[Output]` | 流式同步输出 |
| `astream` | `Input` | `AsyncIterator[Output]` | 流式异步输出 |
## 链式调用机制
### 管道操作符 `|`
链式调用通过管道操作符 `|` 实现，将前一个 Runnable 的输出作为后一个 Runnable 的输入。
```python
# libs/core/langchain_core/runnables/base.py
class BaseRunnable(Generic[Input, Output]):
def __or__(
self,
other: Union[Runnable[Any, Any], Callable[[Any], Any]]
) -> RunnableSerializable[Any, Any]:
"""使用 | 操作符组合两个 Runnable。"""
return RunnableSequence(first=self, last=_maybe_wrap_in_prompt(other))
```
### 链式调用示例
```mermaid
graph LR
A[PromptTemplate] -->|"管道操作符 |"| B[BaseChatModel]
B -->|"管道操作符 |"| C[StrOutputParser]
A1[输入: 问题] --> A
C --> R[最终输出]
```
**代码示例：**
```python
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
# 定义链的组件
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个专业的技术助手。"),
("human", "{question}")
])
model = ChatOpenAI(model="gpt-4")
output_parser = StrOutputParser()
# 链式组合
chain = prompt | model | output_parser
# 调用链
result = chain.invoke({"question": "什么是 LangChain？"})
```
### RunnableSequence 内部实现
```python
# libs/core/langchain_core/runnables/base.py
class RunnableSequence(RunnableSerializable[Input, Output]):
"""按顺序执行的一系列 Runnable。"""
first: Runnable[Any, Any]
middle: List[Runnable[Any, Any]] = Field(default_factory=list)
last: Runnable[Any, Any]
def invoke(self, input: Input, config: Optional[RunnableConfig] = None) -> Output:
# 逐步执行每个 Runnable，将输出传递给下一个
for Runnable in self.runnables:
input = runnable.invoke(input, config)
return input
```
### pipe() 方法
除了 `|` 操作符，还可以通过 `pipe()` 方法实现链式调用：
```python
chain = prompt.pipe(model).pipe(output_parser)
```
两种方式在功能上等价，开发者可根据偏好选择。
## Runnable 配置
### RunnableConfig 配置结构
```python
# libs/core/langchain_core/runnables/config.py
class RunnableConfig(TypedDict, total=False):
"""配置 Runnable 执行的参数。"""
tags: Optional[List[str]]
"""用于标识此执行的标签列表。"""
metadata: Optional[Dict[str, Any]]
"""附加到此执行的元数据。"""
callbacks: Optional[Union[CallbackManagerForChainRun, AsyncCallbackManagerForChainRun]]
"""回调管理器，用于监控和日志记录。"""
max_concurrency: Optional[int]
"""最大并发执行数。"""
recursion_limit: int
"""链中最大递归深度，默认 25。"""
configurable: Optional[Dict[str, Any]]
"""运行时可配置参数。"""
```
### 配置传递机制
```mermaid
graph TD
A[RunnableConfig] --> B[invoke/ainvoke]
B --> C{子 Runnable}
C --> D[config 参数]
C --> E[继承配置]
style A fill:#e1f5fe
style D fill:#c8e6c9
style E fill:#fff3e0
```
### 配置合并策略
```python
# libs/core/langchain_core/runnables/config.py
def merge_configs(*configs: Optional[RunnableConfig]) -> RunnableConfig:
"""合并多个配置，后面的配置优先。"""
merged: Dict[str, Any] = {}
for config in configs:
if config:
merged.update(config)
return merged
```
## 重试机制
### RetryRunnable 包装器
```python
# libs/core/langchain_core/runnables/retry.py
class RetryRunnable(Runnable[Input, Output]):
"""为 Runnable 添加重试逻辑的包装器。"""
def __init__(
self,
runnable: Runnable[Input, Output],
max_attempts: int = 3,
min_seconds: float = 1,
max_seconds: float = 10,
retry_on: Optional[Type[BaseException]] = None,
):
self.runnable = runnable
self.max_attempts = max_attempts
self.retry_on = retry_on or Exception
```
### 重试配置参数
| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `max_attempts` | `int` | `3` | 最大重试次数 |
| `min_seconds` | `float` | `1` | 最小等待时间（秒） |
| `max_seconds` | `float` | `10` | 最大等待时间（秒） |
| `retry_on` | `Type[Exception]` | `Exception` | 需要重试的异常类型 |
| `backoff_factor` | `float` | `2` | 退避因子（指数退避） |
### 重试策略使用示例
```python
from langchain_core.runnables.retry import RetryRunnable
# 创建带重试的链
chain_with_retry = RetryRunnable(
runnable=model,
max_attempts=3,
min_seconds=1,
max_seconds=10,
retry_on=RateLimitError # 只在速率限制错误时重试
)
```
### 指数退避算法
```mermaid
sequenceDiagram
participant C as 调用方
participant R as RetryRunnable
participant M as Model
C->>R: invoke()
R->>M: 第一次调用
M--xR: 异常/限流
R->>R: 等待 1s
R->>M: 第二次调用
M--xR: 异常/限流
R->>R: 等待 2s
R->>M: 第三次调用
M-->>R: 成功
R-->>C: 返回结果
```
## 分支逻辑
### RunnableBranch 条件分支
```python
# libs/core/langchain_core/runnables/branch.py
class RunnableBranch(RunnableSerializable[Any, Output]):
"""根据条件选择不同执行路径的 Runnable。"""
branches: Sequence[Tuple[Runnable[Input, bool], Runnable[Input, Output]]]
default: Optional[Runnable[Input, Output]]
```
### 分支配置参数
| 参数 | 类型 | 说明 |
|------|------|------|
| `branches` | `Sequence[Tuple[condition, runnable]]` | 条件-执行对列表 |
| `default` | `Runnable` | 默认执行的 Runnable |
### 分支使用示例
```python
from langchain_core.runnables.branch import RunnableBranch
# 定义分支条件
branches = [
(
lambda x: "技术" in x.get("category", ""),
technical_chain # 技术问题走此链
),
(
lambda x: "一般" in x.get("category", ""),
general_chain # 一般问题走此链
),
]
# 添加默认分支
branch_chain = RunnableBranch(
*branches,
general_chain # 默认链
)
```
### 分支执行流程
```mermaid
graph TD
A[输入] --> B{条件 1}
B -->|True| C[执行 Runnable 1]
B -->|False| D{条件 2}
D -->|True| E[执行 Runnable 2]
D -->|False| F{...}
F -->|...| G[默认 Runnable]
C --> H[输出]
E --> H
G --> H
```
## 常见使用模式
### 模式一：简单管道链
```python
chain = prompt | model | output_parser
result = chain.invoke({"question": "..."})
```
### 模式二：带配置的链
```python
from langchain_core.runnables.config import RunnableConfig
config = RunnableConfig(
tags=["production", "v1"],
metadata={"user_id": "123"},
recursion_limit=50
)
result = chain.invoke({"question": "..."}, config=config)
```
### 模式三：并行分支合并
```python
from langchain_core.runnables import RunnableParallel
# 并行执行多个任务
parallel_chain = RunnableParallel(
summary=summary_prompt | model,
keywords=keyword_prompt | model
)
result = parallel_chain.invoke({"document": doc})
# 结果: {"summary": "...", "keywords": "..."}
```
### 模式四：动态链
```python
from langchain_core.runnables import RunnableLambda
# 使用函数创建 Runnable
def route_by_topic(input_dict):
topic = input_dict["topic"]
if topic == "技术":
return tech_chain
elif topic == "商业":
return business_chain
else:
return general_chain
dynamic_chain = RunnableLambda(route_by_topic)
```
### 模式五：完整的 LangChain 代理链
```python
from langchain.agents import create_agent
from langchain_core.tools import tool
@tool
def search_api(query: str) -> str:
"""搜索外部 API。"""
# API 调用逻辑
return result
# 创建代理
agent = create_agent(tools=[search_api], model=model)
result = agent.invoke({"input": "帮我搜索最新的 AI 新闻"})
```
## 高级特性
### 回调机制
```python
from langchain_core.callbacks import BaseCallbackHandler
class CustomHandler(BaseCallbackHandler):
def on_chain_start(self, serialized, inputs, **kwargs):
print(f"链开始执行，输入: {inputs}")
def on_chain_end(self, outputs, **kwargs):
print(f"链执行完成，输出: {outputs}")
handler = CustomHandler()
result = chain.invoke({"question": "..."}, config={"callbacks": [handler]})
```
### 流式处理
```python
# 流式调用
for chunk in chain.stream({"question": "..."}):
print(chunk, end="", flush=True)
```
### 批量处理
```python
# 批量调用
questions = [
{"question": "问题1"},
{"question": "问题2"},
{"question": "问题3"}
]
results = chain.batch(questions)
```
## 与旧版 Chain 的关系
| 特性 | `BaseRunnable` | `Chain` (旧版) |
|------|----------------|----------------|
| 统一接口 | ✅ | ❌ |
| 异步支持 | ✅ | 部分 |
| 流式支持 | ✅ | ❌ |
| 序列化 | ✅ | ✅ |
| 推荐程度 | 首选 | 已废弃 |
> **迁移提示**：`Chain` 类在 LangChain 0.2.13 版本被标记为废弃，将在 2.0.0 版本移除。建议使用基于 `Runnable` 的新 API。资料来源：[libs/langchain/langchain_classic/chains/api/base.py](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/langchain_classic/chains/api/base.py)
## 最佳实践
1. **优先使用 Runnable API**：新代码应使用 `Runnable` 接口和链式调用
2. **合理设置递归限制**：复杂链可能需要调高 `recursion_limit`
3. **使用配置管理**：通过 `RunnableConfig` 管理标签和元数据
4. **添加重试机制**：对不稳定操作使用 `RetryRunnable`
5. **利用类型提示**：`BaseRunnable` 是泛型类，指定输入输出类型可提高代码质量
## 总结
Runnable 与链式调用是 LangChain 框架的核心设计范式，通过统一的接口抽象和灵活组合机制，使开发者能够构建从简单到复杂的各种 AI 应用。掌握这一体系对于高效使用 LangChain 至关重要。
---
## Agent代理系统
### 相关页面
相关主题：[工具与工具包](#page-tools), [Runnable与链式调用](#page-runnable)
相关源码文件
以下源码文件用于生成本页说明：
- [libs/langchain/langchain_classic/agents/agent.py](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/langchain_classic/agents/agent.py)
- [libs/langchain/langchain_classic/agents/react/agent.py](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/langchain_classic/agents/react/agent.py)
- [libs/langchain/langchain_classic/agents/openai_functions_agent/base.py](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/langchain_classic/agents/openai_functions_agent/base.py)
- [libs/langchain/langchain_classic/agents/structured_chat/base.py](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/langchain_classic/agents/structured_chat/base.py)
# Agent代理系统
## 概述
Agent（代理）系统是LangChain框架中用于构建自主智能体的核心模块。Agent能够根据用户输入动态决定执行哪些操作（Tool），通过循环调用工具并处理结果来完成任务。与静态的Chain不同，Agent具有推理和决策能力，能够在运行时根据上下文选择下一步行动。 资料来源：[libs/langchain/langchain_classic/agents/agent.py:1-50]()
Agent系统主要由以下几个核心组件构成：
| 组件 | 说明 |
|------|------|
| **Agent** | 核心抽象类，定义Agent的基本接口和执行循环 |
| **Tool** | 可执行的功能单元，Agent调用以完成特定任务 |
| **Toolkit** | 工具集合，便于批量加载相关工具 |
| **AgentExecutor** | Agent的执行器，负责运行Agent并处理输出 |
```mermaid
graph TD
A[用户输入] --> B[Agent推理]
B --> C{需要调用工具?}
C -->|是| D[选择合适Tool]
D --> E[执行Tool]
E --> F[处理Tool结果]
F --> B
C -->|否| G[返回最终结果]
```
## Agent架构
### 基础Agent类
`Agent`类是所有Agent类型的基类，继承自`Chain`。它定义了Agent的核心接口，包括推理逻辑和执行循环。 资料来源：[libs/langchain/langchain_classic/agents/agent.py]()
```python
class Agent(Chain):
"""Agent基类，定义推理和执行接口"""
@property
def observation_prefix(self) -> str:
"""观察结果前缀"""
@property
def llm_prefix(self) -> str:
"""LLM输出前缀"""
```
### Agent执行流程
```mermaid
sequenceDiagram
participant U as 用户
participant A as Agent
participant LLM as LLM
participant T as Tool
U->>A: 输入任务
loop 执行循环
A->>LLM: 发送提示
LLM-->>A: 返回Action/Action Input
A->>T: 调用Tool
T-->>A: 返回Observation
end
A-->>U: 返回最终结果
```
## Agent类型
### ReAct Agent
ReAct（Reasoning + Acting）是一种结合推理和行动的Agent模式。Agent首先进行推理，确定需要执行的操作，然后执行该操作，最后根据结果更新推理状态。 资料来源：[libs/langchain/langchain_classic/agents/react/agent.py:1-30]()
```python
class BaseReActAgent(Agent):
"""ReAct Agent基类
采用Reasoning + Acting循环模式:
1. Thought: 思考下一步行动
2. Action: 选择要执行的动作
3. Observation: 获取动作执行结果
"""
```
**ReAct Agent特点：**
- 显式的推理过程，每一步都有`Thought`
- 适合需要多步推理的复杂任务
- 通过prompt模板控制输出格式
### OpenAI Functions Agent
OpenAI Functions Agent专门用于支持OpenAI function calling功能的模型。它利用模型的原生函数调用能力来生成结构化的工具调用。 资料来源：[libs/langchain/langchain_classic/agents/openai_functions_agent/base.py:1-50]()
```python
class OpenAIFunctionsAgent(Agent):
"""基于OpenAI function calling的Agent
使用模型原生能力生成函数调用，
无需解析文本输出
"""
@property
def allowed_tools(self) -> list[str]:
"""允许调用的工具列表"""
```
**优势：**
| 特性 | 优势 |
|------|------|
| 结构化输出 | 直接生成JSON格式的工具调用 |
| 可靠性 | 减少解析错误 |
| 类型安全 | 支持参数类型验证 |
### Structured Chat Agent
Structured Chat Agent支持复杂的多轮对话场景，能够处理包含多个参数的函数调用。它结合了聊天模型的能力和结构化工具调用。 资料来源：[libs/langchain/langchain_classic/agents/structured_chat/base.py:1-50]()
```python
class StructuredChatAgent(Agent):
"""结构化聊天Agent
适用于需要调用复杂工具的场景，
支持多参数和嵌套参数
"""
```
## Tool集成
### Tool定义
Tool是Agent执行具体操作的接口。每个Tool包含名称、描述和执行函数。
```python
from langchain_core.tools import Tool
search_tool = Tool(
name="search",
description="搜索互联网获取最新信息",
func=search_function
)
```
### Tool选择机制
Agent根据Tool的描述来决定调用哪个Tool。这一过程由LLM完成：
1. **描述匹配**：Agent将任务与Tool描述进行匹配
2. **参数提取**：从用户输入中提取Tool所需的参数
3. **执行验证**：验证参数是否符合Tool的schema要求
## 配置选项
### 关键参数
| 参数 | 类型 | 说明 | 默认值 |
|------|------|------|--------|
| `verbose` | bool | 是否输出详细执行信息 | False |
| `max_iterations` | int | 最大迭代次数 | 15 |
| `max_execution_time` | float | 最大执行时间（秒） | None |
| `early_stopping_method` | str | 提前停止策略 | "force" |
### 错误处理
Agent执行过程中可能遇到各种错误：
```python
from langchain_core.agents import AgentFinish, AgentStep
class AgentExecutor:
"""Agent执行器，处理错误和重试"""
def _handle_tool_error(self, error: Exception) -> str:
"""处理工具执行错误"""
def _handle_parsing_error(self, error: Exception) -> str:
"""处理输出解析错误"""
```
## 迁移指南
### 从APIChain迁移
`APIChain`已在0.2.13版本标记为废弃，将在2.0.0版本移除。建议迁移到新的Agent架构：
```python
# 旧方式 (已废弃)
from langchain.chains import APIChain
chain = APIChain.from_url_and_prompt(...)
# 新方式
from langchain.agents import create_agent
agent = create_agent(tools=[http_request_tool])
```
资料来源：[libs/langchain/langchain_classic/chains/api/base.py:30-50]()
## 最佳实践
### 1. Tool设计原则
- **清晰描述**：Tool描述应准确说明功能和用途
- **参数验证**：使用pydantic模型定义参数schema
- **错误返回**：Tool应返回有意义的错误信息
### 2. Agent配置建议
- 设置合理的`max_iterations`防止无限循环
- 启用`verbose=True`便于调试
- 考虑设置`max_execution_time`限制执行时间
### 3. Prompt优化
```python
# 使用系统提示词优化Agent行为
agent = create_agent(
tools=my_tools,
system_message="""
你是一个专业的助手。
总是先思考再行动。
如果不确定，请返回最终答案。
"""
)
```
## 相关资源
- [LangChain官方文档](https://docs.langchain.com/oss/python/langchain/agents)
- [API参考](https://reference.langchain.com/python/langchain_classic/agents.html)
- [示例代码库](https://github.com/langchain-ai/langchain/tree/main/examples)
## 总结
Agent代理系统是LangChain的核心能力之一，通过结合LLM的推理能力和Tool的执行能力，实现了灵活的自主任务执行。框架提供了多种Agent类型以适应不同场景，开发者应根据具体需求选择合适的Agent模式。
---
## 工具与工具包
### 相关页面
相关主题：[Agent代理系统](#page-agents), [向量存储与检索](#page-vectorstores)
相关源码文件
以下源码文件用于生成本页说明：
- [libs/core/langchain_core/tools/base.py](https://github.com/langchain-ai/langchain/blob/main/libs/core/langchain_core/tools/base.py)
- [libs/core/langchain_core/tools/structured.py](https://github.com/langchain-ai/langchain/blob/main/libs/core/langchain_core/tools/structured.py)
- [libs/langchain/langchain_classic/tools/base.py](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/langchain_classic/tools/base.py)
- [libs/langchain/langchain_classic/agents/agent_toolkits/sql/toolkit.py](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/langchain_classic/agents/agent_toolkits/sql/toolkit.py)
# 工具与工具包
## 概述
LangChain 中的**工具（Tools）**和**工具包（Toolkits）**是框架的核心组件，它们使 LLM（大型语言模型）能够与外部世界交互、执行具体操作、访问实时信息以及操控各种系统和数据源。工具是可执行的原子单元，而工具包则是相关工具的集合，为特定领域或任务提供完整的功能支持。
在 LangChain 的架构中，工具系统位于 `langchain_core` 层，这表明工具是跨多个包共享的基础抽象。`langchain-core` 定义了工具的核心接口和基类，而 `langchain-classic` 在此基础上提供了更丰富的实现和特定场景的工具包。
LangChain 的工具系统设计遵循以下核心原则：类型安全、异步支持、结构化参数验证、以及与主流 LLM 提供商的广泛兼容性。通过统一的工具接口，开发者可以轻松创建自定义工具或使用预构建的工具包来加速开发。
## 工具的核心架构
### 工具的基类层次
LangChain 的工具系统采用分层架构设计。`BaseTool` 是所有工具的抽象基类，定义了工具的基本行为和接口。`StructuredTool` 则在 `BaseTool` 基础上添加了结构化参数支持，允许工具接受复杂类型的输入参数。
工具的核心方法包括：
- **invoke**：同步执行工具
- **ainvoke**：异步执行工具
- **run**：兼容旧版 API 的执行方法
- **arun**：异步版本的旧 API
这种分层设计允许工具在保持一致接口的同时，根据具体需求提供不同的功能实现。基础层负责定义契约，而具体实现可以继承和扩展这些基类以添加特定功能。
```mermaid
graph TD
A[BaseTool] --> B[StructuredTool]
A --> C[Tool]
B --> D[自定义结构化工具]
C --> E[自定义简单工具]
D --> F[工具执行层]
E --> F
F --> G[LLM 交互]
```
### 工具的参数定义
工具使用 Pydantic 模型定义参数，这确保了参数的类型安全和自动验证。每个工具通过 `args_schema` 属性声明其接受的参数，包括参数名称、类型、描述和默认值。
参数定义的关键字段包括：
- **name**：工具的唯一标识符
- **description**：工具功能的自然语言描述，供 LLM 理解何时使用
- **args_schema**：Pydantic 模型类，定义参数结构
- **return_direct**：控制返回值是否直接返回给用户
参数描述在工具选择过程中至关重要，因为 LLM 需要根据描述来决定调用哪个工具以及如何传递参数。良好的描述应该清晰说明工具的用途、参数含义以及返回值格式。
### 工具的执行流程
工具的执行涉及多个步骤，从接收调用请求到返回结果。当 LLM 决定调用工具时，系统会经过以下流程：参数验证、权限检查、实际执行、结果处理和返回。
异步执行支持是 LangChain 工具系统的重要特性。通过 `ainvoke` 和 `arun` 方法，工具可以执行 I/O 密集型操作而不会阻塞主线程。这对于需要调用外部 API、访问数据库或执行网络请求的工具尤为重要。
错误处理机制贯穿整个执行流程。工具可以捕获和处理各种异常情况，并根据错误类型返回有意义的消息或重试操作。工具的错误处理设计允许开发者定义降级策略和恢复逻辑。
## StructuredTool 的结构化参数支持
`StructuredTool` 是 LangChain 中用于处理复杂参数的工具基类。与简单的字符串参数不同，`StructuredTool` 支持嵌套对象、列表和其他复杂数据类型，这使得工具能够接收更丰富和结构化的输入。
结构化参数通过 Pydantic 模型定义，这带来了几个关键优势：类型检查在运行时自动执行，参数验证规则可以集中定义，JSON Schema 可以自动生成以供 LLM 理解参数结构。
`StructuredTool` 的核心特性包括：
- **类型安全的参数传递**：参数值在传递前经过类型验证
- **自动生成的参数文档**：基于 Pydantic 模型自动生成参数描述
- **嵌套结构支持**：允许定义复杂的多层参数结构
- **默认值处理**：自动应用默认值，减少调用方的负担
## 工具包的设计模式
### 工具包的概念与用途
工具包（Toolkit）是一组相关工具的集合，专门设计用于处理特定领域或完成特定任务。例如，SQL 工具包包含用于数据库查询、表列表获取、架构检查等的相关工具，它们协同工作以支持完整的数据库交互场景。
工具包的设计遵循组合原则：将相关的原子功能打包在一起，使开发者可以一次性引入完整的功能集合。这种设计减少了集成工作量，同时确保了工具之间的一致性和互操作性。
### SQL 工具包详解
SQL 工具包是 LangChain 提供的标准工具包之一，专门用于数据库操作场景。该工具包通常包含以下核心工具：
| 工具名称 | 功能描述 | 输入参数 |
|---------|---------|---------|
| query_sql_db | 执行 SQL 查询 | query: str |
| list_sql_tables | 列出数据库中的表 | - |
| get_sql_table_schema | 获取表结构信息 | table_names: List[str] |
| query_checker | 验证和修正 SQL 语法 | query: str |
SQL 工具包通过统一的接口与数据库交互，支持多种数据库后端。工具包初始化时需要数据库连接信息，如连接字符串或连接池配置。工具包内部处理连接管理、查询执行和结果转换等细节，对使用者屏蔽了底层复杂性。
工具包的设计还考虑了安全性。SQL 工具包可以配置允许访问的表列表，防止 LLM 执行未经授权的数据库操作。参数验证和 SQL 语法检查功能有助于防止 SQL 注入等安全威胁。
## 工具与 LLM 的集成
### 工具选择机制
在 LangChain 中，LLM 可以通过多种方式选择和调用工具。最常见的模式是让 LLM 输出结构化的工具调用指令，这些指令包含工具名称和参数。LangChain 提供了解析和执行这些调用的基础设施。
工具选择涉及几个关键因素：
- **工具描述**：LLM 根据描述判断工具的适用性
- **参数约束**：LLM 需要生成符合参数 schema 的调用
- **上下文信息**：当前对话状态和可用工具列表影响选择
### 工具调用的执行
当 LLM 输出工具调用时，LangChain 的执行层负责：
1. 解析 LLM 输出，提取工具名称和参数
2. 验证参数是否符合工具的 schema
3. 查找并实例化目标工具
4. 执行工具并捕获结果或错误
5. 将结果反馈给 LLM 进行后续处理
执行层支持同步和异步两种模式。同步模式下，工具按顺序依次执行；异步模式下，支持并行执行多个独立的工具调用以提高效率。
## 自定义工具开发
### 创建简单工具
创建自定义工具的基本流程相对直接。首先定义工具函数，然后使用 `@tool` 装饰器或继承 `BaseTool` 类将其转换为 LangChain 工具。工具函数应专注于单一职责，便于测试和维护。
自定义工具的关键考虑因素包括：
- **描述的清晰性**：描述应帮助 LLM 理解何时以及如何调用工具
- **参数命名的直观性**：使用清晰、描述性的参数名称
- **错误处理**：定义良好的错误消息，帮助调试和恢复
- **返回值格式**：确保返回值易于 LLM 理解和处理
### 创建结构化工具
对于需要复杂参数的工兴，应使用 `StructuredTool` 或直接继承 `BaseTool` 并定义 Pydantic 格式的 `args_schema`。这种方法允许定义嵌套参数、类型验证规则和默认值。
结构化工具开发的关键步骤：
1. 定义 Pydantic 模型表示工具参数
2. 继承 `StructuredTool` 或相应基类
3. 实现核心执行逻辑
4. 配置工具元数据（名称、描述等）
## 安全与权限控制
### 域限制机制
LangChain 提供了 URL 域限制功能以防止工具被滥用。例如，API 调用工具可以配置允许访问的域名白名单，系统会在执行前验证目标 URL 是否在允许列表中。
域限制通过以下函数实现：
- `_extract_scheme_and_domain`：从 URL 提取协议和域名
- `_check_in_allowed_domain`：检查 URL 是否在允许的域名列表中
这种机制对于安全敏感的用例至关重要，特别是在允许 LLM 发起网络请求的场景中。
### 工具权限管理
工具系统支持多种权限控制机制：
- **表级访问控制**：数据库工具可以限制可访问的表
- **操作类型限制**：限制工具可执行的操作类型（如只读）
- **速率限制**：防止过度使用或资源耗尽
## 生态集成
### 合作伙伴包中的工具
LangChain 的生态系统中包含多个合作伙伴包，每个包都提供了针对特定服务或平台的工具集成。这些包位于 `libs/partners/` 目录下，包括：
| 合作伙伴 | 包名称 | 主要功能 |
|---------|-------|---------|
| OpenAI | langchain-openai | OpenAI API 集成 |
| Anthropic | langchain-anthropic | Claude 模型支持 |
| HuggingFace | langchain-huggingface | 模型推理与嵌入 |
| Ollama | langchain-ollama | 本地 LLM 支持 |
| DeepSeek | langchain-deepseek | DeepSeek API |
这些合作伙伴包通常包含针对特定平台优化的工具实现，充分利用各平台的特性和能力。
### 社区工具与扩展
除了官方维护的包外，LangChain 还支持社区贡献的工具和工具包。`langchain-community` 包曾经是社区工具的主要存放地，但新版本中建议使用专门的合作伙伴包或自定义工具实现。
## 最佳实践
### 工具设计原则
设计良好的工具应遵循以下原则：
- **单一职责**：每个工具应专注于完成一项具体任务
- **清晰命名**：工具和参数名称应自描述
- **完整文档**：提供详尽的描述说明工具用途和使用方式
- **错误恢复**：工具应能优雅地处理错误并提供有用的反馈
- **类型安全**：使用类型注解和 Pydantic 验证确保参数正确
### 性能优化建议
在生产环境中使用工具时，应考虑以下性能优化策略：
- **连接池复用**：对于需要数据库或网络连接的工兴，应复用连接而非每次调用创建新连接
- **异步执行**：I/O 密集型操作应使用异步 API
- **结果缓存**：对于不经常变化的查询结果，可以进行缓存以减少重复调用
- **批处理**：将多个小操作合并为批处理以减少开销
### 调试与监控
工具执行过程中的调试和监控对于维护系统可靠性至关重要。建议的实现方式包括：
- 配置回调处理器以记录工具调用
- 实现详细的日志记录以便问题排查
- 设置监控指标以追踪工具使用模式和性能
- 建立告警机制以在工具失败时及时响应
## 相关资源
- 官方文档：https://docs.langchain.com/oss/python/langchain/overview
- API 参考：https://reference.langchain.com/python
- 贡献指南：https://docs.langchain.com/oss/python/contributing/overview
---
## 向量存储与检索
### 相关页面
相关主题：[文档加载与处理](#page-document-loaders), [工具与工具包](#page-tools)
相关源码文件
以下源码文件用于生成本页说明：
- [libs/core/langchain_core/vectorstores/base.py](https://github.com/langchain-ai/langchain/blob/main/libs/core/langchain_core/vectorstores/base.py)
- [libs/core/langchain_core/embeddings/embeddings.py](https://github.com/langchain-ai/langchain/blob/main/libs/core/langchain_core/embeddings/embeddings.py)
- [libs/langchain/langchain_classic/retrievers/__init__.py](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/langchain_classic/retrievers/__init__.py)
- [libs/langchain/langchain_classic/retrievers/self_query/base.py](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/langchain_classic/retrievers/self_query/base.py)
# 向量存储与检索
## 概述
向量存储与检索是 LangChain 框架中用于处理语义搜索和相似性匹配的核心功能模块。在大型语言模型（LLM）应用中，向量存储通过将文本、文档或其他数据转换为高维向量表示，实现高效的语义检索能力。当用户提出查询时，系统会将查询转换为向量，并在向量数据库中查找最相似的文档或文本片段。LangChain 的向量存储架构设计遵循模块化原则，核心接口定义在 `langchain-core` 包中，而具体实现则分布在各个集成包中，例如 `langchain-chroma` 用于 Chroma 向量数据库的集成。这种架构使得开发者可以灵活地在不同向量存储提供商之间切换，同时保持上层应用代码的一致性。
## 核心架构
LangChain 的向量存储系统采用分层架构设计，从底层到上层依次包括向量存储实现层、嵌入模型层和检索器层。向量存储层负责实际的数据存储和相似性搜索操作，支持多种后端包括 Chroma、FAISS、Pinecone 等主流向量数据库。嵌入模型层提供文本到向量的转换能力，通过 `Embeddings` 接口定义标准化的嵌入生成方法。检索器层则封装了各种检索策略，包括直接相似性检索、自查询检索、时间加权检索等高级功能。这种分层设计使得系统具有良好的可扩展性和可维护性，开发者可以根据需求自由组合不同层次的组件。
```mermaid
graph TD
A[用户查询] --> B[嵌入模型]
B --> C[查询向量]
C --> D[向量存储]
D --> E[相似文档]
E --> F[检索结果]
G[文档] --> H[嵌入模型]
H --> I[文档向量]
I --> D
J[元数据] --> D
```
## 向量存储基类
LangChain 的向量存储功能由 `VectorStore` 基类提供核心抽象。该基类定义了一组标准接口，包括 `add_texts`、`add_documents`、`similarity_search`、`similarity_search_by_vector` 等核心方法。向量存储实现需要继承这个基类并实现相应的抽象方法，同时可以提供特定于后端的优化功能。基类还提供了异步支持，通过 `AsyncVectorStore` 接口支持异步操作，这对于构建高性能应用至关重要。向量存储的构造函数通常接受嵌入模型实例、连接参数和配置选项，使得初始化过程灵活且易于配置。
| 方法 | 参数 | 返回值 | 说明 |
|------|------|--------|------|
| `add_texts` | texts, metadatas, ids | List[str] | 添加文本到向量存储 |
| `add_documents` | documents | List[str] | 添加文档对象到向量存储 |
| `similarity_search` | query, k, filter | List[Document] | 执行相似性搜索 |
| `similarity_search_by_vector` | embedding, k, filter | List[Document] | 按向量执行相似性搜索 |
| `delete` | ids | None | 删除指定ID的文档 |
| `save_local` | folder_path | None | 将向量存储保存到本地 |
## 嵌入模型接口
嵌入模型是向量存储系统的核心组件，负责将文本数据转换为高维向量表示。LangChain 通过 `Embeddings` 基类定义了跨不同嵌入提供商的统一接口，该接口包含 `embed_query` 和 `embed_documents` 两个核心方法。`embed_query` 方法用于单个查询文本的嵌入生成，而 `embed_documents` 方法支持批量处理多个文档的嵌入计算。这个设计允许开发者轻松切换不同的嵌入提供商，无论是使用 OpenAI 的 text-embedding-ada-002、HuggingFace 的 sentence-transformers，还是本地部署的嵌入模型。LangChain 还提供了缓存和批处理优化，以减少 API 调用开销和提高整体性能。
```python
from langchain_core.embeddings import Embeddings
class Embeddings(ABC):
"""嵌入模型抽象基类"""
@abstractmethod
def embed_query(self, text: str) -> List[float]:
"""嵌入单个查询文本"""
pass
@abstractmethod
def embed_documents(self, texts: List[str]) -> List[List[float]]:
"""嵌入多个文档文本"""
pass
async def aembed_query(self, text: str) -> List[float]:
"""异步嵌入单个查询"""
return self.embed_query(text)
async def aembed_documents(self, texts: List[str]) -> List[List[float]]:
"""异步嵌入多个文档"""
return self.embed_documents(texts)
```
## 检索器模块
检索器（Retriever）是 LangChain 中用于从向量存储中获取相关文档的高级抽象。`BaseRetriever` 类定义了标准化的检索接口，使得不同检索策略的实现可以互换使用。LangChain 提供了多种内置检索器，包括 `VectorStoreRetriever` 进行基础向量相似性检索、`SelfQueryRetriever` 支持基于元数据的结构化查询、`ContextualCompressionRetriever` 在检索后进行结果压缩、`MultiVectorRetriever` 支持多向量检索策略，以及 `TimeWeightedVectorStoreRetriever` 结合时间因素的加权检索。这种丰富的检索器生态使得开发者可以根据具体应用场景选择最合适的检索策略。
## 自查询检索器
自查询检索器（Self-Query Retriever）是一种高级检索能力，它能够自动从用户查询中提取语义搜索条件和元数据过滤条件。这种检索器特别适用于结构化元数据丰富的场景，例如文档带有时间戳、类别、作者等属性。当用户提出一个问题时，自查询检索器会同时生成两个部分：一是用于语义相似性匹配的文本嵌入向量，二是用于精确过滤的元数据条件表达式。这种双重处理机制使得检索结果既具有语义相关性，又满足精确的元数据约束条件。
| 组件 | 说明 |
|------|------|
| `query_constructor` | 从用户查询中提取结构和语义信息 |
| `allowed_operators` | 支持的元数据比较运算符 |
| `default_operator` | 默认的逻辑运算符（AND/OR） |
| `attribute_infomer` | 属性名称的信息提取器 |
自查询检索器的工作流程涉及多个步骤：首先，大语言模型（LLM）分析用户查询，识别其中的语义搜索意图和元数据过滤条件；然后，系统将提取的元数据条件转换为结构化查询表达式；最后，向量存储同时执行语义搜索和元数据过滤，返回满足双重条件的结果文档。这种设计将自然语言处理和结构化查询有机结合，大大增强了检索系统的表达能力和精确度。
## 相似性搜索策略
LangChain 支持多种相似性搜索策略，以满足不同性能和精度需求。基础的相似性搜索默认使用余弦相似度（Cosine Similarity）作为距离度量，这是语义搜索中最常用的度量方式。系统还支持欧氏距离（L2 Distance）和点积（Dot Product）等其他度量方式，开发者可以根据具体场景和嵌入模型特性选择最合适的度量方法。高级搜索选项包括设置搜索返回的文档数量 `k` 参数、元数据过滤条件 `filter` 参数，以及是否返回相似性分数 `include_embeddings` 参数。这些参数的合理配置对于平衡搜索性能和结果质量至关重要。
## 与向量数据库的集成
LangChain 与多种主流向量数据库提供了深度集成支持。在合作伙伴包（partners）中，`langchain-chroma` 提供了与 Chroma 向量数据库的官方集成，Chroma 是一个专为 LLM 应用设计的开源嵌入数据库，提供了简单的安装和使用体验。除此之外，LangChain 还通过社区包支持 Pinecone、Weaviate、Qdrant、Milvus、FAISS、Elasticsearch 等多种向量存储后端。每种集成都遵循统一的 `VectorStore` 接口规范，确保应用代码在不同后端之间具有良好的可移植性。
```mermaid
graph LR
A[应用层] --> B[VectorStore 接口]
B --> C[Chroma 集成]
B --> D[Pinecone 集成]
B --> E[FAISS 集成]
B --> F[其他集成]
C --> G[Chroma 数据库]
D --> H[Pinecone 云服务]
E --> I[本地 FAISS 索引]
F --> J[其他后端]
```
## 文档处理与索引
在将文档添加到向量存储之前，LangChain 提供了完整的文档处理流水线支持。文档首先通过文本分割器（Text Splitter）被拆分成较小的块（chunks），这是因为大多数嵌入模型对输入长度有限制，而且较小的块能够提供更精确的检索结果。LangChain 的 `text-splitters` 包提供了多种分割策略，包括按字符数分割、按句子分割、递归字符分割等。分割后的文档块包含文本内容（page_content）和元数据（metadata）两个部分，元数据可以包含来源信息、页码、创建时间等任何有助于检索的信息。
| 分割策略 | 适用场景 | 参数说明 |
|----------|----------|----------|
| CharacterTextSplitter | 简单文本分割 | chunk_size, chunk_overlap, separator |
| RecursiveCharacterTextSplitter | 保留段落结构 | 同上，增加 separators 参数 |
| TokenTextSplitter | 按 token 数量分割 | 需要 tokenizer 参数 |
## 异步操作支持
现代 LLM 应用对性能和并发处理能力有较高要求，LangChain 的向量存储系统提供了完整的异步操作支持。异步接口包括 `aadd_texts`、`aadd_documents`、`aasync_similarity_search` 等方法，允许开发者在异步环境中高效地处理批量操作。异步设计特别适用于需要同时查询多个向量存储、或者在高并发场景下处理大量请求的应用。LangChain 的异步实现遵循 Python 的 asyncio 规范，可以与 FastAPI 等现代 Web 框架无缝集成。
## 安全性与权限控制
在构建向量存储应用时，安全性是需要重点考虑的方面。LangChain 提供了域名限制（Domain Limiting）功能，可以配置向量存储只允许查询特定域名的数据源，防止潜在的安全风险。这一功能在 APIChain 等可能暴露给最终用户的组件中尤为重要，通过限制可访问的域名范围，可以防止恶意查询或数据泄露。开发者应该仔细评估应用的安全需求，合理配置访问控制策略，确保向量存储系统在提供便利的同时不引入安全漏洞。
## 最佳实践
在实际应用中，向量存储与检索系统的性能优化需要关注多个方面。首先是嵌入模型的选择，不同的嵌入模型在精度、速度和成本方面各有权衡，应该根据具体场景选择最合适的模型。其次是文档块大小的确定，过大的块可能导致检索精度下降，过小的块则可能丢失上下文信息，建议通过实验找到最佳的块大小和重叠参数。批量操作时应该利用批处理 API 减少网络开销，同时注意避免超过 API 的速率限制。在生产环境中，应该实现适当的错误处理和重试机制，以应对网络波动和服务不可用等情况。
```mermaid
graph TD
A[开始] --> B[选择嵌入模型]
B --> C[设计文档分割策略]
C --> D[配置向量存储]
D --> E[构建检索器]
E --> F[添加安全限制]
F --> G[性能测试]
G --> H{是否满足需求}
H -->|否| I[调整参数]
I --> D
H -->|是| J[部署到生产环境]
```
## 资料来源
- `libs/core/langchain_core/vectorstores/base.py` - 向量存储基类定义
- `libs/core/langchain_core/embeddings/embeddings.py` - 嵌入模型接口定义
- `libs/langchain/langchain_classic/retrievers/__init__.py` - 检索器模块初始化
- `libs/langchain/langchain_classic/retrievers/self_query/base.py` - 自查询检索器实现
- `libs/partners/chroma/README.md` - Chroma 集成文档
---
## 文档加载与处理
### 相关页面
相关主题：[向量存储与检索](#page-vectorstores), [内存与状态管理](#page-memory)
相关源码文件
以下源码文件用于生成本页说明：
- [libs/core/langchain_core/document_loaders/base.py](https://github.com/langchain-ai/langchain/blob/main/libs/core/langchain_core/document_loaders/base.py)
- [libs/langchain/langchain_classic/document_loaders/base.py](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/langchain_classic/document_loaders/base.py)
- [libs/langchain/langchain_classic/document_loaders/parsers/registry.py](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/langchain_classic/document_loaders/parsers/registry.py)
- [libs/text-splitters/langchain_text_splitters/base.py](https://github.com/langchain-ai/langchain/blob/main/libs/text-splitters/langchain_text_splitters/base.py)
# 文档加载与处理
## 概述
文档加载与处理是 LangChain 框架中用于将各种格式的外部数据转换为标准化 `Document` 对象的核心模块。该模块支持从多种数据源加载内容，包括 PDF、HTML、Markdown、JSON、CSV、Word 文档等，并提供了灵活的分词（Text Splitting）功能以支持大规模文档处理场景。
LangChain 的文档处理架构采用分层设计，核心接口定义在 `langchain-core` 包中，具体实现则分散在 `langchain-classic` 和各集成包中。
## 核心架构
LangChain 文档处理流程包含三个主要阶段：**加载（Loading）**、**解析（Parsing）** 和 **分块（Chunking/Splitting）**。
```mermaid
graph TD
A[原始数据源] --> B[DocumentLoader]
B --> C[Document 对象]
C --> D[TextSplitter]
D --> E[Document 列表]
E --> F[Vector Store / Chain]
B1[文件系统] --> B
B2[Web URL] --> B
B3[数据库] --> B
B4[API 响应] --> B
C --> C1[page_content: str]
C --> C2[metadata: dict]
```
## 核心数据模型
### Document 对象
`Document` 是 LangChain 中表示文档的基础数据结构，包含以下核心属性：
| 属性 | 类型 | 说明 |
|------|------|------|
| `page_content` | `str` | 文档的文本内容 |
| `metadata` | `dict` | 文档元数据，包含来源、页码等信息 |
```python
class Document:
page_content: str
metadata: dict = Field(default_factory=dict)
```
资料来源：[libs/core/langchain_core/document_loaders/base.py]()
### BaseLoader 抽象基类
所有文档加载器必须继承自 `BaseLoader` 基类：
```python
class BaseLoader(ABC):
"""文档加载器基类"""
@abstractmethod
def lazy_load(self) -> Iterator[Document]:
"""懒加载文档，按需生成 Document 对象"""
...
def load(self) -> list[Document]:
"""同步加载所有文档"""
return list(self.lazy_load())
async def aload(self) -> list[Document]:
"""异步加载所有文档"""
return [doc async for doc in self.alazy_load()]
@abstractmethod
async def alazy_load(self) -> AsyncIterator[Document]:
"""异步懒加载文档"""
...
```
资料来源：[libs/core/langchain_core/document_loaders/base.py]()
## 文档加载器类型
### 按数据源分类
| 类型 | 说明 | 典型实现 |
|------|------|----------|
| 文件加载器 | 从本地文件系统读取文件 | `TextLoader`, `CSVLoader`, `PDFLoader` |
| Web 加载器 | 从 URL 或网页抓取内容 | `WebBaseLoader`, `WebUnstructuredLoader` |
| 数据库加载器 | 从数据库查询结果加载 | `SQLDatabaseLoader` |
| API 加载器 | 从 API 响应解析文档 | `APIBaseLoader` |
### 内置加载器列表
`langchain-classic` 包提供了丰富的内置加载器实现：
| 加载器类 | 支持格式 | 包路径 |
|----------|----------|--------|
| `TextLoader` | .txt, 纯文本 | `langchain_classic.document_loaders` |
| `CSVLoader` | .csv | `langchain_classic.document_loaders` |
| `UnstructuredFileLoader` | 多种格式 | `langchain_classic.document_loaders` |
| `PyPDFLoader` | PDF | `langchain_classic.document_loaders` |
| `WebBaseLoader` | 网页 | `langchain_classic.document_loaders` |
| `NotionLoader` | Notion 页面 | `langchain_partners` |
## 解析器注册机制
LangChain 使用解析器注册表（Registry）来管理不同文档类型的解析逻辑。
```mermaid
graph LR
A[文件输入] --> B[ParserRegistry]
B --> C{文件类型检测}
C -->|.pdf| D[PDFParser]
C -->|.html| E[HTMLParser]
C -->|.md| F[MarkdownParser]
C -->|.json| G[JSONParser]
D --> H[Document 对象]
E --> H
F --> H
G --> H
```
### ParserRegistry 类
```python
class ParserRegistry:
"""文档解析器注册表"""
def __init__(self, parsers: dict[str, BaseParser] | None = None):
self._parsers = parsers or {}
def register(self, extension: str, parser: BaseParser) -> None:
"""注册解析器"""
self._parsers[extension.lower()] = parser
def get(self, extension: str) -> BaseParser | None:
"""获取指定扩展名对应的解析器"""
return self._parsers.get(extension.lower())
def get_extractors(self) -> dict[str, BaseParser]:
"""获取所有注册的解析器"""
return self._parsers.copy()
```
资料来源：[libs/langchain/langchain_classic/document_loaders/parsers/registry.py]()
## 文本分块（Text Splitting）
对于长文档，需要将其分割成较小的块（chunks）以便嵌入（embedding）和检索。`langchain-text-splitters` 包提供了灵活的分块策略。
### BaseCharacterTextSplitter
基于字符的分块器，按指定分隔符分割文本：
```python
class BaseCharacterTextSplitter(TextSplitter):
"""基于字符的分块器基类"""
def __init__(
self,
separator: str = "\n\n",
is_separator_regex: bool = False,
chunk_size: int = 4000,
chunk_overlap: int = 200,
length_function: Callable[[str], int] = len,
add_start_index: bool = False,
strip_whitespace: bool = True,
):
...
```
### 关键参数说明
| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `separator` | `str` | `"\n\n"` | 分隔符字符串 |
| `is_separator_regex` | `bool` | `False` | 是否将分隔符作为正则表达式 |
| `chunk_size` | `int` | `4000` | 每个块的最大字符数 |
| `chunk_overlap` | `int` | `200` | 相邻块之间的重叠字符数 |
| `length_function` | `Callable` | `len` | 计算文本长度的函数 |
| `add_start_index` | `bool` | `False` | 是否在元数据中添加起始位置 |
| `strip_whitespace` | `bool` | `True` | 是否去除首尾空白 |
资料来源：[libs/text-splitters/langchain_text_splitters/base.py]()
### 分块策略
LangChain 支持多种分块策略：
```mermaid
graph TD
A[输入文本] --> B{分块策略}
B -->|字符分割| C[CharacterTextSplitter]
B -->|递归分割| D[RecursiveCharacterTextSplitter]
B -->|代码分割| E[LanguageSplitter]
B -->|语义分割| F[SemanticChunker]
C --> G[按固定分隔符分割]
D --> H[优先小分隔符
失败则回退大分隔符]
E --> I[按编程语言语法分割]
F --> J[基于嵌入相似度分割]
G --> K[Document 列表]
H --> K
I --> K
J --> K
```
### RecursiveCharacterTextSplitter
递归字符分块器尝试按优先级顺序使用不同的分隔符：
```python
class RecursiveCharacterTextSplitter(BaseSeparatorTextSplitter):
"""递归字符分块器"""
def __init__(
self,
separators: list[str] | None = None,
**kwargs,
):
if separators is None:
separators = ["\n\n", "\n", " ", ""]
super().__init__(separators=separators, **kwargs)
```
**分割优先级：**
1. `\n\n` - 段落分隔符
2. `\n` - 换行符
3. ` ` - 空格
4. `""` - 单字符
资料来源：[libs/text-splitters/langchain_text_splitters/base.py]()
## 使用示例
### 基本加载流程
```python
from langchain_core.document_loaders import TextLoader
from langchain_core.documents import Document
# 使用 TextLoader 加载文本文件
loader = TextLoader(file_path="example.txt")
documents = loader.load()
# 访问文档内容
for doc in documents:
print(f"内容: {doc.page_content}")
print(f"元数据: {doc.metadata}")
```
### 异步加载
```python
from langchain_core.document_loaders import BaseLoader
class AsyncDocumentLoader(BaseLoader):
"""自定义异步加载器"""
async def alazy_load(self) -> AsyncIterator[Document]:
# 异步获取数据
content = await self._fetch_content()
yield Document(page_content=content)
```
### 文档分块
```python
from langchain_text_splitters import RecursiveCharacterTextSplitter
# 初始化分块器
splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
separators=["\n\n", "\n", " ", ""]
)
# 对文档进行分块
texts = ["这是第一段文本...", "这是第二段文本..."]
docs = [Document(page_content=t) for t in texts]
chunks = splitter.split_documents(docs)
print(f"原始文档数: {len(docs)}")
print(f"分块后块数: {len(chunks)}")
```
## 元数据管理
文档元数据用于存储与内容相关的附加信息：
| 元数据键 | 说明 | 示例值 |
|----------|------|--------|
| `source` | 文档来源路径或 URL | `"./data/document.pdf"` |
| `page` | 页码（PDF 等多页文档） | `1` |
| `total_pages` | 总页数 | `42` |
| `creation_date` | 创建时间 | `"2024-01-15"` |
| `author` | 作者 | `"John Doe"` |
```python
# 创建带元数据的文档
doc = Document(
page_content="这是文档内容",
metadata={
"source": "report.pdf",
"page": 5,
"author": "Alice"
}
)
```
## 与其他模块的集成
```mermaid
graph LR
A[DocumentLoader] --> B[Document]
B --> C[TextSplitter]
C --> D[Chunked Documents]
D --> E[Embeddings]
E --> F[Vector Store]
D --> G[Cache]
D --> H[Chain Input]
```
文档处理模块与 LangChain 其他核心组件的协作：
| 组件 | 集成方式 |
|------|----------|
| **Embeddings** | 分块后的文本块传递给嵌入模型生成向量 |
| **Vector Store** | 嵌入向量存储到向量数据库 |
| **Retrieval** | 向量检索用于问答系统的上下文增强 |
| **Chain** | 文档内容作为 LLM Chain 的输入 |
## 最佳实践
### 1. 选择合适的分块大小
| 使用场景 | 推荐 chunk_size | 说明 |
|----------|-----------------|------|
| 问答系统 | 500-1000 | 短块提高检索精度 |
| 摘要生成 | 2000-4000 | 足够上下文生成准确摘要 |
| 对话系统 | 1000-2000 | 平衡上下文和相关性 |
### 2. 合理设置重叠
- **chunk_overlap** 应设置为 **chunk_size 的 10-20%**
- 足够的重叠确保跨块边界的语义连续性
### 3. 处理特殊格式
对于结构化文档（PDF、Word），优先使用专用加载器以保留布局和元信息：
```python
# PDF 加载（保留页码）
from langchain_core.document_loaders import PyPDFLoader
loader = PyPDFLoader("document.pdf")
pages = loader.load_and_split() # 每页返回一个 Document
```
## 扩展自定义加载器
```python
from langchain_core.document_loaders import BaseLoader
from langchain_core.documents import Document
from typing import Iterator
class CustomLoader(BaseLoader):
"""自定义文档加载器"""
def __init__(self, file_path: str):
self.file_path = file_path
def lazy_load(self) -> Iterator[Document]:
"""实现懒加载逻辑"""
with open(self.file_path, 'r', encoding='utf-8') as f:
content = f.read()
yield Document(
page_content=content,
metadata={"source": self.file_path}
)
async def alazy_load(self) -> AsyncIterator[Document]:
"""实现异步懒加载"""
content = await self._async_read()
yield Document(
page_content=content,
metadata={"source": self.file_path}
)
```
## 相关资源
| 资源类型 | 链接 |
|----------|------|
| API 文档 | [LangChain Core Document Loaders](https://reference.langchain.com/python/langchain_core/document_loaders) |
| 使用指南 | [LangChain Docs - Document Loaders](https://docs.langchain.com/oss/python/integrations/providers/huggingface) |
| GitHub 仓库 | [langchain-ai/langchain](https://github.com/langchain-ai/langchain) |
---
## 内存与状态管理
### 相关页面
相关主题：[文档加载与处理](#page-document-loaders), [Agent代理系统](#page-agents)
相关源码文件
以下源码文件用于生成本页说明：
- [libs/langchain/langchain_classic/memory/__init__.py](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/langchain_classic/memory/__init__.py)
- [libs/langchain/langchain_classic/memory/buffer.py](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/langchain_classic/memory/buffer.py)
- [libs/langchain/langchain_classic/memory/summary.py](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/langchain_classic/memory/summary.py)
- [libs/langchain/langchain_classic/memory/chat_message_histories/__init__.py](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/langchain_classic/memory/chat_message_histories/__init__.py)
- [libs/langchain/langchain_classic/memory/vectorstore.py](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/langchain_classic/memory/vectorstore.py)
- [libs/langchain/langchain_classic/chains/base.py](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/langchain_classic/chains/base.py)
# 内存与状态管理
## 概述
在 LangChain 框架中，**内存与状态管理**是构建上下文感知（context-aware）应用的核心能力。内存模块负责在多轮对话或复杂任务执行过程中持久化、检索和管理交互历史，使大型语言模型（LLM）能够访问先前的上下文信息，从而生成连贯且符合上下文的响应。
LangChain 的内存系统被设计为可插拔（pluggable）架构，允许开发者根据不同的应用场景选择或自定义合适的内存实现方案。这些方案涵盖从简单的缓冲区存储到复杂的向量检索等多种策略。
## 核心架构
### 系统组件关系
LangChain 的内存系统主要由以下几个核心组件构成，它们之间通过标准化接口进行交互：
```mermaid
graph TD
A[用户交互 / Chain 执行] --> B[Memory 模块]
B --> C[缓冲区内存
BufferMemory]
B --> D[摘要内存
SummaryMemory]
B --> E[向量存储内存
VectorStoreMemory]
B --> F[聊天消息历史
ChatMessageHistory]
C --> G[BaseMemory 接口]
D --> G
E --> G
F --> G
G --> H[Chain 上下文]
G --> I[LLM 上下文窗口]
```
### 内存类型分类
LangChain 内存模块支持多种内存实现，以满足不同场景的需求：
| 内存类型 | 用途描述 | 适用场景 |
|---------|---------|---------|
| `BufferMemory` | 简单缓冲区存储原始消息 | 短对话、测试环境 |
| `SummaryMemory` | 自动生成对话摘要 | 长对话、资源受限环境 |
| `VectorStoreMemory` | 使用向量数据库存储和检索 | 大规模历史检索、语义搜索 |
| `ChatMessageHistory` | 管理聊天消息的生命周期 | 独立的消息历史管理 |
| `CombinedMemory` | 组合多种内存策略 | 复杂应用、需要多种检索方式 |
资料来源：[libs/langchain/langchain_classic/memory/__init__.py]()
## 缓冲区内存（BufferMemory）
### 功能说明
缓冲区内存是 LangChain 中最简单的内存实现，它按顺序存储所有对话消息，并在需要时将完整的历史记录传递给下游组件。这种方式保留了所有交互细节，但会随着对话长度增加而消耗更多令牌（token）。
### 核心实现
```python
class BufferMemory(BaseMemory):
"""简单的缓冲区内存实现"""
chat_memory: ChatMessageHistory = Field(default_factory=ChatMessageHistory)
output_key: Optional[str] = None
input_key: Optional[str] = None
return_messages: bool = False
```
资料来源：[libs/langchain/langchain_classic/memory/buffer.py]()
### 消息缓冲机制
缓冲区内存通过 `ChatMessageHistory` 管理消息的添加和检索。消息被组织为一系列 `HumanMessage` 和 `AIMessage` 对象，按时间顺序排列。
```mermaid
sequenceDiagram
participant U as 用户
participant M as BufferMemory
participant H as ChatMessageHistory
U->>M: 添加用户消息
M->>H: store_message(content, type="human")
H-->>M: 确认存储
U->>M: 添加AI响应
M->>H: store_message(content, type="ai")
H-->>M: 确认存储
M->>H: load_memory_variables()
H-->>M: 返回完整消息历史
M-->>U: 传递给Chain/LLM
```
### 使用示例
```python
from langchain.memory import BufferMemory
from langchain.chains import ConversationChain
memory = BufferMemory(
return_messages=True,
output_key="response",
input_key="input"
)
chain = ConversationChain(
llm=llm,
memory=memory,
verbose=True
)
# 多轮对话
chain.invoke({"input": "我叫张三"})
chain.invoke({"input": "我叫什么名字？"}) # 能够记住之前的信息
```
## 摘要内存（SummaryMemory）
### 设计目的
摘要内存在处理长对话时更为高效，它不会存储每一条原始消息，而是维护一个不断更新的对话摘要。这种设计显著减少了令牌消耗，同时保留了对话的核心信息。
### 工作原理
摘要内存采用增量更新策略：每当收到新消息时，系统会调用 LLM 分析当前对话历史并生成更新的摘要。这个摘要替换了原有的历史记录，成为新的上下文基础。
```mermaid
graph LR
A[新消息到达] --> B{对话历史为空？}
B -->|是| C[生成初始摘要]
B -->|否| D[加载当前摘要]
D --> E[分析新旧内容]
E --> F[生成更新摘要]
C --> G[存储摘要]
F --> G
```
资料来源：[libs/langchain/langchain_classic/memory/summary.py]()
### 配置参数
| 参数名 | 类型 | 默认值 | 说明 |
|-------|------|--------|------|
| `chat_memory` | ChatMessageHistory | 自动创建 | 消息存储后端 |
| `memory_key` | str | "history" | 变量名，用于模板 |
| `output_key` | Optional[str] | None | 输出提取键 |
| `input_key` | Optional[str] | None | 输入提取键 |
| `return_messages` | bool | False | 是否返回消息对象 |
| `summary_message_num` | int | 1 | 摘要生成触发消息数 |
## 聊天消息历史（ChatMessageHistory）
### 架构角色
`ChatMessageHistory` 是 LangChain 内存系统的核心数据层，负责管理消息的持久化存储和检索。它作为一个抽象层，支持多种后端存储实现。
```mermaid
classDiagram
class BaseChatMessageHistory {
<>
+messages: List[BaseMessage]
+add_user_message()*
+add_ai_message()*
+clear()*
}
class ChatMessageHistory {
+add_user_message()
+add_ai_message()
+add_message()
+clear()
+get_messages()
}
class FileChatMessageHistory {
+add_user_message()
+add_ai_message()
+clear()
+get_messages()
}
class RedisChatMessageHistory {
+add_user_message()
+add_ai_message()
+clear()
+get_messages()
}
BaseChatMessageHistory <|-- ChatMessageHistory
BaseChatMessageHistory <|-- FileChatMessageHistory
BaseChatMessageHistory <|-- RedisChatMessageHistory
```
资料来源：[libs/langchain/langchain_classic/memory/chat_message_histories/__init__.py]()
### 消息类型
系统支持多种消息类型以适应不同的交互模式：
| 消息类型 | 标识符 | 用途 |
|---------|-------|------|
| `HumanMessage` | human | 用户输入消息 |
| `AIMessage` | ai | AI 生成的响应 |
| `SystemMessage` | system | 系统级指令 |
| `FunctionMessage` | function | 函数调用结果 |
| `ToolMessage` | tool | 工具执行结果 |
## 向量存储内存（VectorStoreMemory）
### 语义检索能力
向量存储内存利用嵌入（embedding）技术将对话内容转换为高维向量，然后存储在向量数据库中。这种方法支持语义相似性搜索，允许系统根据含义而非精确关键词来检索相关历史。
```mermaid
graph TD
A[对话消息] --> B[Embedding 模型]
B --> C[向量表示]
C --> D[向量数据库存储]
E[查询请求] --> F[查询 Embedding]
F --> G[向量相似度搜索]
D --> G
G --> H[检索相关历史]
H --> I[返回上下文片段]
```
资料来源：[libs/langchain/langchain_classic/memory/vectorstore.py]()
### 配置选项
| 参数 | 类型 | 说明 |
|-----|------|------|
| `vectorstore` | VectorStore | 向量存储后端实例 |
| `memory_key` | str | 内存变量标识 |
| `input_key` | Optional[str] | 输入键 |
| `search_kwargs` | dict | 搜索参数（如 k 值） |
| `embedding` | Optional[Embedding] | 嵌入模型 |
## 与 Chain 的集成
### 数据流处理
LangChain 的 Chain（链）是内存系统的主要消费者。Chain 通过标准化的内存接口与各种内存实现进行交互：
```mermaid
graph LR
A[用户输入] --> B[Chain.input_keys]
B --> C[Memory.load_memory_variables]
C --> D[与输入合并]
D --> E[LLM 处理]
E --> F[Chain.output_keys]
G[Memory] --> C
H[历史消息] --> G
F --> I[Memory.save_context]
I --> J[更新内存状态]
```
资料来源：[libs/langchain/langchain_classic/chains/base.py]()
### 内存绑定机制
在创建 Chain 实例时，通过 `memory` 参数将内存模块绑定到 Chain：
```python
from langchain.chains import LLMChain
from langchain.memory import ConversationBufferMemory
# 创建内存实例
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
# 创建链并绑定内存
chain = LLMChain(
llm=llm,
prompt=prompt,
memory=memory
)
```
## 最佳实践
### 场景选择指南
| 应用场景 | 推荐内存类型 | 原因 |
|---------|-------------|------|
| 简短测试/演示 | BufferMemory | 实现简单、易于调试 |
| 长对话、有限上下文 | SummaryMemory | 节省令牌、保留关键信息 |
| 需要语义检索 | VectorStoreMemory | 支持模糊匹配、智能召回 |
| 持久化需求 | 配合 Redis/File 等后端 | 跨会话保持状态 |
| 复杂企业应用 | CombinedMemory | 组合多种策略 |
### 性能优化建议
1. **消息分页**：对于超长对话，实施基于时间或数量的消息分页策略
2. **异步处理**：使用异步内存实现提高 I/O 密集型操作的吞吐量
3. **缓存策略**：对频繁访问的摘要结果实施缓存
4. **后端选择**：根据数据量选择合适的存储后端（内存/文件/数据库/向量库）
### 安全考量
在生产环境中部署内存系统时，应注意以下安全要点：
- **数据隔离**：确保不同用户/会话的内存数据严格隔离
- **敏感信息处理**：对包含敏感内容的对话实施适当的脱敏或加密
- **访问控制**：限制对历史消息的读取和修改权限
- **数据清理**：定期清理不再需要的会话历史，遵守数据保留政策
## 扩展与自定义
### 自定义内存实现
开发者可以通过继承 `BaseMemory` 类来创建自定义内存实现：
```python
from langchain.memory import BaseMemory
from langchain.schema import BaseMessage
class CustomMemory(BaseMemory):
"""自定义内存实现示例"""
memory_key: str = "custom_history"
@property
def memory_variables(self) -> List[str]:
return [self.memory_key]
def load_memory_variables(
self, inputs: Dict[str, Any]
) -> Dict[str, Any]:
# 自定义加载逻辑
return {self.memory_key: []}
def save_context(
self, inputs: Dict[str, Any], outputs: Dict[str, str]
) -> None:
# 自定义保存逻辑
pass
def clear(self) -> None:
# 自定义清理逻辑
pass
```
### 与外部系统集成
LangChain 内存系统支持与多种外部系统集成，以实现更强大的状态管理能力：
| 外部系统 | 集成方式 | 典型用途 |
|---------|---------|---------|
| Redis | `RedisChatMessageHistory` | 分布式环境、缓存 |
| MongoDB | `MongoDBChatMessageHistory` | 文档存储、灵活性 |
| PostgreSQL | 数据库后端 | 企业级持久化 |
| Elasticsearch | 向量搜索 | 高级检索能力 |
## 总结
LangChain 的内存与状态管理系统为构建上下文感知的 LLM 应用提供了强大而灵活的基础设施。通过模块化的设计和丰富的内置实现，开发者可以轻松选择或组合适合特定场景的内存策略。从简单的缓冲区到复杂的向量检索，系统提供了全面的解决方案来管理对话状态和历史信息。
核心优势包括：
- **统一接口**：所有内存类型遵循 `BaseMemory` 接口，便于替换和扩展
- **多种策略**：支持缓冲、摘要、向量检索等多种内存管理策略
- **持久化支持**：可与多种存储后端集成，满足生产环境需求
- **无缝集成**：与 Chain、Agent 等组件深度集成，开箱即用
---
## Chat模型与LLM集成
### 相关页面
相关主题：[核心架构与组件](#page-core-architecture), [LangChain概述](#page-overview)
相关源码文件
以下源码文件用于生成本页说明：
- [libs/partners/openai/README.md](https://github.com/langchain-ai/langchain/blob/main/libs/partners/openai/README.md)
- [libs/partners/anthropic/README.md](https://github.com/langchain-ai/langchain/blob/main/libs/partners/anthropic/README.md)
- [libs/partners/huggingface/README.md](https://github.com/langchain-ai/langchain/blob/main/libs/partners/huggingface/README.md)
- [libs/langchain/langchain_classic/chains/api/base.py](https://github.com/langchain-ai/langchain/blob/main/libs/langchain/langchain_classic/chains/api/base.py)
- [libs/partners/ollama/README.md](https://github.com/langchain-ai/langchain/blob/main/libs/partners/ollama/README.md)
- [README.md](https://github.com/langchain-ai/langchain/blob/main/README.md)
# Chat模型与LLM集成
LangChain 是一个用于构建 LLM 驱动应用的框架，它提供了模块化和可互操作的组件架构。Chat 模型与 LLM 集成是 LangChain 生态系统中的核心功能模块，提供了与各种大语言模型提供商的标准化接口。
## 概述
LangChain 的 Chat 模型与 LLM 集成系统为开发者提供了统一的方式来接入不同提供商的大语言模型服务。该系统基于 `langchain-core` 库构建的核心抽象层，通过合作伙伴包（Partner Packages）的形式支持多种 LLM 提供商。
## 核心架构
### 集成包结构
LangChain 采用模块化的包结构组织其 LLM 集成：
| 包名称 | 用途 | 核心功能 |
|--------|------|----------|
| `langchain-openai` | OpenAI 模型集成 | GPT-4o、GPT-4 等模型调用 |
| `langchain-anthropic` | Anthropic 模型集成 | Claude 系列模型调用 |
| `langchain-huggingface` | HuggingFace 集成 | 本地推理和嵌入模型 |
| `langchain-ollama` | Ollama 本地模型 | 本地部署的开源模型 |
| `langchain-groq` | Groq 加速推理 | 高速 LLM 推理服务 |
| `langchain-deepseek` | DeepSeek 模型 | 深度求索模型集成 |
| `langchain-fireworks` | Fireworks.ai | 高性能推理平台 |
### 类层次结构
LangChain 的 LLM 集成采用分层设计：
```
BaseLanguageModel (langchain-core)
├── BaseChatModel
│ ├── ChatOpenAI (langchain-openai)
│ ├── ChatAnthropic (langchain-anthropic)
│ └── ...
└── BaseLLM
├── OpenAI (langchain-openai)
└── ...
```
资料来源：[README.md:31-36]()
## 快速开始
### 基础安装
使用 pip 安装 LangChain 主包：
```bash
pip install langchain
```
### 初始化 Chat 模型
LangChain 提供了统一的模型初始化接口：
```python
from langchain.chat_models import init_chat_model
model = init_chat_model("openai:gpt-5.4")
result = model.invoke("Hello, world!")
```
资料来源：[README.md:31-36]()
### OpenAI 集成
安装 langchain-openai 包：
```bash
pip install langchain-openai
```
该包包含与 OpenAI GPT 模型交互的 LangChain 集成，提供对 GPT-4o、GPT-4 等模型的支持。
资料来源：[libs/partners/openai/README.md:1-20]()
### Anthropic 集成
安装 langchain-anthropic 包：
```bash
pip install langchain-anthropic
```
该包提供与 Anthropic Claude 模型的集成功能，支持通过 LangChain 框架调用 Claude 系列生成模型。
资料来源：[libs/partners/anthropic/README.md:1-18]()
### HuggingFace 集成
基础安装：
```bash
pip install langchain-huggingface
```
如果需要使用 `HuggingFaceEmbeddings` 或 `HuggingFacePipeline` 进行本地推理，应安装完整版本：
```bash
pip install langchain-huggingface[full]
```
> **注意**：基础安装不包含 `sentence-transformers` 或 `transformers`。`[full]` 版本包含 `sentence-transformers>=5.2.0` 和 `transformers>=5.0.0`。
> **迁移提示**：从 `langchain-community` 迁移的用户需要注意，旧版接受 `sentence-transformers>=2.2.0`，但 `langchain-huggingface[full]` 要求 `>=5.2.0`。
资料来源：[libs/partners/huggingface/README.md:1-25]()
### Ollama 本地模型
安装 langchain-ollama 包：
```bash
pip install langchain-ollama
```
该包提供与 Ollama 的集成，支持本地部署的开源大语言模型。
资料来源：[libs/partners/ollama/README.md:1-18]()
## APIChain 与工具调用
### Chain 架构
APIChain 是 LangChain Classic 中的一个重要组件，它实现了调用 API 并总结响应以回答问题的功能。该链使用请求工具包发送 GET、POST、PATCH、PUT 和 DELETE 请求。
```python
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
toolkit = RequestsToolkit(
requests_wrapper=TextRequestsWrapper(headers={}),
allow_dangerous_requests=ALLOW_DANGEROUS_REQUESTS,
)
tools = toolkit.get_tools()
api_request_chain = (
API_URL_PROMPT.partial(api_docs=api_spec)
| model.bind_tools(tools, tool_choice="any")
)
```
资料来源：[libs/langchain/langchain_classic/chains/api/base.py:1-85]()
### 状态管理
LangChain 使用 LangGraph 进行状态管理：
```python
class ChainState(TypedDict):
"""LangGraph state."""
messages: Annotated[Sequence[BaseMessage], add_messages]
```
### 工作流图
```mermaid
graph TD
A[用户问题] --> B[call_tool]
B --> C[execute_tool]
C --> D[call_model]
D --> E[返回结果]
B -->|API调用| F[RequestsToolkit]
C -->|执行工具| G[HTTP请求]
D -->|LLM处理| H[ChatOpenAI]
```
## 安全性考虑
APIChain 包含重要的安全警告：
> **安全注意**：此 API 链使用请求工具包向 API 发出 GET、POST、PATCH、PUT 和 DELETE 请求。
>
> 请谨慎控制允许使用此链的人员。如果向最终用户公开，用户将能够代表托管代码的服务器发出任意请求。例如，用户可以请求服务器向只能从服务器访问的私有 API 发出请求。
>
> 请控制对此工具包的使用权限和网络访问权限。
资料来源：[libs/langchain/langchain_classic/chains/api/base.py:85-100]()
## 领域限制
### URL 验证机制
LangChain 提供了 URL 域限制功能，防止恶意请求：
```python
def _extract_scheme_and_domain(url: str) -> tuple[str, str]:
"""Extract the scheme + domain from a given URL."""
parsed_uri = urlparse(url)
return parsed_uri.scheme, parsed_uri.netloc
def _check_in_allowed_domain(url: str, limit_to_domains: Sequence[str]) -> bool:
"""Check if a URL is in the allowed domains."""
scheme, domain = _extract_scheme_and_domain(url)
for allowed_domain in limit_to_domains:
allowed_scheme, allowed_domain_ = _extract_scheme_and_domain(allowed_domain)
if scheme == allowed_scheme and domain == allowed_domain_:
return True
return False
```
资料来源：[libs/langchain/langchain_classic/chains/api/base.py:30-60]()
## 弃用与迁移
LangChain Classic 中的某些组件已被标记为弃用：
```python
@deprecated(
since="0.2.13",
removal="2.0.0",
alternative="langchain.agents.create_agent",
addendum=(
"Build new agents with `create_agent` and bind a tool that issues "
"the HTTP request."
),
)
class APIChain(Chain):
"""Chain that makes API calls..."""
```
推荐使用新的 `langchain.agents.create_agent` 方法构建代理。
资料来源：[libs/langchain/langchain_classic/chains/api/base.py:60-85]()
## 更多提供商
LangChain 生态系统还支持众多其他 LLM 提供商：
| 提供商 | 包名 | 特点 |
|--------|------|------|
| DeepSeek | `langchain-deepseek` | 深度求索模型 |
| xAI | `langchain-xai` | xAI 官方集成 |
| Perplexity | `langchain-perplexity` | Perplexity AI |
| Fireworks | `langchain-fireworks` | 高性能推理 |
| OpenRouter | `langchain-openrouter` | 统一 API 支持数百模型 |
| Exa | `langchain-exa` | Web 搜索 API |
## 文档资源
| 资源类型 | 链接 |
|----------|------|
| API 参考 | [reference.langchain.com](https://reference.langchain.com/python/langchain_classic) |
| 概念指南 | [docs.langchain.com](https://docs.langchain.com/oss/python/langchain/overview) |
| 发布策略 | [docs.langchain.com](https://docs.langchain.com/oss/python/release-policy) |
| 版本控制 | [docs.langchain.com](https://docs.langchain.com/oss/python/versioning) |
| 贡献指南 | [docs.langchain.com](https://docs.langchain.com/oss/python/contributing/overview) |
资料来源：[libs/langchain/README.md:1-20]()
资料来源：[libs/partners/openai/README.md:1-20]()
## 相关链接
- **LangGraph**：用于构建可控代理工作流的框架 - [docs.langchain.com/langgraph](https://docs.langchain.com/oss/python/langgraph/overview)
- **LangSmith**：用于开发、调试和部署 AI 代理的工具 - [docs.langchain.com/langsmith](https://docs.langchain.com/langsmith/home)
- **Deep Agents**：基于 LangChain 的高级代理包 - [docs.langchain.com/deepagents](http://docs.langchain.com/oss/python/deepagents/)
- **LangChain.js**：等效的 JavaScript/TypeScript 库 - [github.com/langchain-ai/langchainjs](https://github.com/langchain-ai/langchainjs)
---
## 回调与追踪系统
### 相关页面
相关主题：[核心架构与组件](#page-core-architecture)
相关源码文件
以下源码文件用于生成本页说明：
- [libs/core/langchain_core/tracers/__init__.py](https://github.com/langchain-ai/langchain/blob/main/libs/core/langchain_core/tracers/__init__.py)
- [libs/core/langchain_core/tracers/base.py](https://github.com/langchain-ai/langchain/blob/main/libs/core/langchain_core/tracers/base.py)
- [libs/core/langchain_core/tracers/event_stream.py](https://github.com/langchain-ai/langchain/blob/main/libs/core/langchain_core/tracers/event_stream.py)
- [libs/core/langchain_core/callbacks/manager.py](https://github.com/langchain-ai/langchain/blob/main/libs/core/langchain_core/callbacks/manager.py)
# 回调与追踪系统
## 概述
LangChain 的回调与追踪系统是框架的核心基础设施之一，为 LangChain 组件提供了一套统一的生命周期事件通知和执行追踪机制。该系统使开发者能够在 LangChain 运行期间捕获关键事件、监控执行流程、收集性能指标，并实现自定义的日志记录和监控功能。
回调与追踪系统的主要职责包括：
- **事件传播**：在 Chain、LLM、Tool、Agent 等核心组件的生命周期关键时刻触发回调
- **执行追踪**：记录组件的输入、输出、执行时间和中间状态
- **异步支持**：提供异步和同步两种回调处理模式
- **上下文传递**：通过回调管理器维护和传递调用上下文
资料来源：[libs/core/langchain_core/tracers/__init__.py:1-50]()
## 系统架构
### 核心组件关系
LangChain 的回调与追踪系统由以下几个核心组件构成，它们之间形成清晰的层次结构：
```mermaid
graph TD
A[CallbackHandler 回调处理器] --> B[BaseCallbackManager 基础回调管理器]
A --> C[Tracer 追踪器]
C --> D[BaseTracer 基础追踪器]
C --> E[EventStreamTracer 事件流追踪器]
B --> F[CallbackManager 同步回调管理器]
B --> G[AsyncCallbackManager 异步回调管理器]
H[Run 运行时对象] --> C
```
### 组件职责
| 组件类型 | 职责描述 | 位置 |
|---------|---------|------|
| CallbackHandler | 定义回调方法接口 | langchain_core.callbacks |
| Tracer | 追踪器基类，生成 Run 对象 | langchain_core.tracers |
| CallbackManager | 管理同步回调的执行 | langchain_core.callbacks.manager |
| AsyncCallbackManager | 管理异步回调的执行 | langchain_core.callbacks.manager |
| Run | 代表一次执行追踪的运行时对象 | langchain_core.tracers.base |
资料来源：[libs/core/langchain_core/tracers/base.py:1-100]()
## 回调处理器
### CallbackHandler 接口
CallbackHandler 是所有回调处理器的基类接口，定义了组件生命周期中可能被触发的方法。处理器可以继承此接口并选择性重写特定方法来实现自定义行为。
```python
class CallbackHandler:
"""回调处理器基类接口"""
def on_llm_start(self, **kwargs) -> None:
"""LLM 调用开始时触发"""
pass
def on_llm_end(self, **kwargs) -> None:
"""LLM 调用结束时触发"""
pass
def on_chain_start(self, **kwargs) -> None:
"""Chain 执行开始时触发"""
pass
def on_chain_end(self, **kwargs) -> None:
"""Chain 执行结束时触发"""
pass
def on_tool_start(self, **kwargs) -> None:
"""Tool 调用开始时触发"""
pass
def on_tool_end(self, **kwargs) -> None:
"""Tool 调用结束时触发"""
pass
```
资料来源：[libs/core/langchain_core/callbacks/manager.py:1-80]()
### 内置回调处理器
LangChain 提供了多种内置回调处理器，适用于不同的监控和日志需求：
| 处理器名称 | 用途 | 关键功能 |
|-----------|------|---------|
| StdOutCallbackHandler | 标准输出日志 | 在控制台打印执行信息 |
| TracingCallbackHandler | 分布式追踪 | 集成第三方追踪系统 |
| LangChainTracer | LangChain 原生追踪 | 记录 Run 到追踪后端 |
| FileCallbackHandler | 文件日志 | 写入日志到文件 |
## 追踪器系统
### Tracer 抽象基类
Tracer 是专门用于创建和记录 Run 对象的组件。每个 Run 代表一次被追踪的执行单元，包含执行过程中的元数据、输入输出等信息。
```mermaid
graph LR
A[开始追踪] --> B[创建 Run 对象]
B --> C[记录事件]
C --> D{事件类型}
D -->|start| E[on_run_start]
D -->|end| F[on_run_end]
D -->|error| G[on_run_error]
E --> H[更新 Run 状态]
F --> H
G --> H
H --> I[持久化或发送]
```
资料来源：[libs/core/langchain_core/tracers/base.py:1-150]()
### BaseTracer 实现
BaseTracer 提供了 Tracer 接口的基础实现，处理 Run 对象的创建和状态管理：
```python
class BaseTracer(BaseCallbackHandler, ABC):
"""追踪器基类实现"""
def __init__(self):
self.run_dict: dict[str, Run] = {}
self._active_runs: set[str] = set()
@abstractmethod
def _generate_run_id(self) -> str:
"""生成唯一的 Run 标识符"""
pass
```
资料来源：[libs/core/langchain_core/tracers/base.py:50-100]()
### EventStreamTracer 事件流追踪器
EventStreamTracer 是一种特殊的追踪器实现，它通过事件流的方式持续输出追踪事件，适用于需要实时监控的场景：
```mermaid
graph TD
A[执行组件] --> B[EventStreamTracer]
B --> C[事件序列化器]
C --> D[事件缓冲区]
D --> E[流式输出]
E --> F[消费者/监控仪表盘]
```
资料来源：[libs/core/langchain_core/tracers/event_stream.py:1-100]()
EventStreamTracer 的核心特性包括：
- **流式输出**：实时将事件推送至消费者
- **事件缓冲**：批量处理事件以优化性能
- **序列化支持**：将 Run 对象转换为可传输格式
## 回调管理器
### CallbackManager
CallbackManager 是同步回调执行的管理器，负责注册、调用和清理回调处理器：
```mermaid
graph TD
A[请求进入] --> B[CallbackManager]
B --> C[遍历已注册处理器]
C --> D{按顺序触发方法}
D --> E[on_chain_start]
D --> F[on_llm_start]
D --> G[on_tool_start]
E --> H[执行组件逻辑]
H --> I[on_chain_end]
I --> J[on_llm_end]
J --> K[on_tool_end]
K --> L[返回结果]
```
资料来源：[libs/core/langchain_core/callbacks/manager.py:100-200]()
### 回调管理器核心方法
| 方法名 | 参数 | 返回类型 | 功能描述 |
|--------|------|---------|---------|
| add_handler | handler, name | None | 注册新的回调处理器 |
| remove_handler | name | None | 移除指定名称的处理器 |
| set_handlers | handlers | None | 替换所有已注册的处理器 |
| clear | None | None | 清空所有处理器 |
| get_handler | name | CallbackHandler | 根据名称获取处理器 |
### 异步回调管理
AsyncCallbackManager 提供与 CallbackManager 类似的功能，但专门用于异步执行环境：
```python
class AsyncCallbackManager(BaseCallbackManager):
"""异步回调管理器"""
async def ahandle_chain_start(
self,
name: str,
serialized: dict[str, Any],
inputs: dict[str, Any],
run_id: UUID,
parent_run_id: UUID | None = None,
**kwargs: Any,
) -> AsyncManagerGetResults:
"""异步处理 Chain 开始事件"""
pass
async def ahandle_chain_end(
self,
outputs: dict[str, Any],
run_id: UUID,
**kwargs: Any,
) -> None:
"""异步处理 Chain 结束事件"""
pass
```
资料来源：[libs/core/langchain_core/callbacks/manager.py:200-300]()
## Run 对象模型
### Run 数据结构
Run 是追踪系统的核心数据模型，代表一次可追踪的执行单元：
```mermaid
classDiagram
class Run {
+str id
+str name
+str session_id
+str run_type
+dict~str, Any~ inputs
+dict~str, Any~ outputs
+datetime start_time
+datetime end_time
+str|None parent_run_id
+dict~str, Any~ extra
+str|None error
}
class ChainRun {
+str run_type = "chain"
}
class LLMRun {
+str run_type = "llm"
+int token_usage
}
class ToolRun {
+str run_type = "tool"
}
Run <|-- ChainRun
Run <|-- LLMRun
Run <|-- ToolRun
```
### Run 追踪字段
| 字段名 | 类型 | 可选 | 说明 |
|-------|------|------|------|
| id | str | 否 | Run 的唯一标识符 |
| name | str | 否 | Run 的名称 |
| run_type | str | 否 | Run 类型（chain/llm/tool） |
| session_id | str | 是 | 所属会话标识 |
| parent_run_id | str | 是 | 父级 Run 标识 |
| inputs | dict | 是 | 输入参数 |
| outputs | dict | 是 | 输出结果 |
| start_time | datetime | 否 | 开始时间 |
| end_time | datetime | 是 | 结束时间 |
| error | str | 是 | 错误信息 |
资料来源：[libs/core/langchain_core/tracers/base.py:100-200]()
## 使用示例
### 基本用法
```python
from langchain_core.callbacks import CallbackManager, StdOutCallbackHandler
from langchain_openai import ChatOpenAI
# 创建回调管理器
callback_manager = CallbackManager(handlers=[StdOutCallbackHandler()])
# 创建带回调的 LLM 实例
llm = ChatOpenAI(
model="gpt-4",
callback_manager=callback_manager
)
# 调用时自动触发回调
response = llm.invoke("Hello, world!")
```
### 自定义回调处理器
```python
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.tracers.base import BaseTracer
class CustomTracer(BaseTracer):
"""自定义追踪器"""
def __init__(self):
super().__init__()
self.events = []
def _generate_run_id(self) -> str:
return str(uuid.uuid4())
def on_chain_start(self, serialized, inputs, **kwargs):
self.events.append({
"type": "chain_start",
"inputs": inputs
})
def on_chain_end(self, outputs, **kwargs):
self.events.append({
"type": "chain_end",
"outputs": outputs
})
```
资料来源：[libs/core/langchain_core/callbacks/manager.py:300-400]()
### 父子 Run 追踪
```python
from langchain_core.callbacks import CallbackManager
# 在嵌套调用中自动维护父子关系
callback_manager = CallbackManager(handlers=[tracer])
# 父 Chain
parent_chain = Chain(callback_manager=callback_manager)
# 子 Chain（自动关联 parent_run_id）
child_chain = Chain(callback_manager=callback_manager)
result = parent_chain.invoke(input, config={"callbacks": callback_manager})
```
## 配置选项
### 全局回调配置
| 配置项 | 类型 | 默认值 | 说明 |
|-------|------|-------|------|
| callbacks | CallbackManager | None | 全局回调管理器 |
| verbose | bool | False | 启用详细输出 |
| tracing_enabled | bool | True | 启用追踪功能 |
| tracing_endpoint | str | None | 追踪服务端点 |
| tracing_headers | dict | None | 追踪请求头 |
### 环境变量
| 变量名 | 值 | 作用 |
|-------|-----|------|
| LANGCHAIN_TRACING_V2 | true/false | 启用 V2 追踪 |
| LANGCHAIN_ENDPOINT | URL | 追踪服务端点 |
| LANGCHAIN_API_KEY | string | API 认证密钥 |
| LANGCHAIN_PROJECT | string | 追踪项目名称 |
## 安全注意事项
在 LangChain 的 APIChain 文档中明确指出回调与追踪系统的安全考量：
> **安全注意**：API 链使用 requests 工具包向 API 发起 `GET`、`POST`、`PATCH`、`PUT` 和 `DELETE` 请求。
>
> 请谨慎控制谁可以使用此链以及它具有何种网络访问权限。
>
> 如果向最终用户公开，请考虑用户将能够代表服务器发起任意请求。
资料来源：[libs/langchain/langchain_classic/chains/api/base.py:40-60]()
### 安全最佳实践
- 限制回调中发送的敏感数据范围
- 在生产环境中使用 HTTPS 传输追踪数据
- 实施回调数据的访问控制策略
- 定期审计追踪日志内容
## 生态集成
### 合作伙伴包中的追踪支持
多个 LangChain 合作伙伴包内置了回调与追踪系统支持：
| 集成包 | 追踪支持 | 主要功能 |
|-------|---------|---------|
| langchain-openai | ✅ | OpenAI API 调用追踪 |
| langchain-anthropic | ✅ | Anthropic API 调用追踪 |
| langchain-huggingface | ✅ | Transformers 本地推理追踪 |
| langchain-ollama | ✅ | Ollama 本地模型追踪 |
| langchain-deepseek | ✅ | DeepSeek API 调用追踪 |
资料来源：[libs/partners/openai/README.md](), [libs/partners/anthropic/README.md](), [libs/partners/huggingface/README.md]()
## 总结
LangChain 的回调与追踪系统为框架提供了强大的可观测性和扩展性能力。通过统一的回调接口、灵活的追踪机制和完善的 Run 数据模型，开发者可以轻松实现：
- 执行过程监控和日志记录
- 性能指标收集和分析
- 分布式追踪集成
- 自定义监控和告警系统
该系统设计遵循了模块化原则，允许开发者根据需求选择性地使用回调或追踪功能，或将两者结合使用以获得完整的应用可观测性。
---
---
## Doramagic 踩坑日志
项目：langchain-ai/langchain
摘要：发现 6 个潜在踩坑项，其中 0 个为 high/blocking；最高优先级：能力坑 - 能力判断依赖假设。
## 1. 能力坑 · 能力判断依赖假设
- 严重度：medium
- 证据强度：source_linked
- 发现：README/documentation is current enough for a first validation pass.
- 对用户的影响：假设不成立时，用户拿不到承诺的能力。
- 建议检查：将假设转成下游验证清单。
- 防护动作：假设必须转成验证项；没有验证结果前不能写成事实。
- 证据：capability.assumptions | github_repo:552661142 | https://github.com/langchain-ai/langchain | README/documentation is current enough for a first validation pass.
## 2. 维护坑 · 维护活跃度未知
- 严重度：medium
- 证据强度：source_linked
- 发现：未记录 last_activity_observed。
- 对用户的影响：新项目、停更项目和活跃项目会被混在一起，推荐信任度下降。
- 建议检查：补 GitHub 最近 commit、release、issue/PR 响应信号。
- 防护动作：维护活跃度未知时，推荐强度不能标为高信任。
- 证据：evidence.maintainer_signals | github_repo:552661142 | https://github.com/langchain-ai/langchain | last_activity_observed missing
## 3. 安全/权限坑 · 下游验证发现风险项
- 严重度：medium
- 证据强度：source_linked
- 发现：no_demo
- 对用户的影响：下游已经要求复核，不能在页面中弱化。
- 建议检查：进入安全/权限治理复核队列。
- 防护动作：下游风险存在时必须保持 review/recommendation 降级。
- 证据：downstream_validation.risk_items | github_repo:552661142 | https://github.com/langchain-ai/langchain | no_demo; severity=medium
## 4. 安全/权限坑 · 存在评分风险
- 严重度：medium
- 证据强度：source_linked
- 发现：no_demo
- 对用户的影响：风险会影响是否适合普通用户安装。
- 建议检查：把风险写入边界卡，并确认是否需要人工复核。
- 防护动作：评分风险必须进入边界卡，不能只作为内部分数。
- 证据：risks.scoring_risks | github_repo:552661142 | https://github.com/langchain-ai/langchain | no_demo; severity=medium
## 5. 维护坑 · issue/PR 响应质量未知
- 严重度：low
- 证据强度：source_linked
- 发现：issue_or_pr_quality=unknown。
- 对用户的影响：用户无法判断遇到问题后是否有人维护。
- 建议检查：抽样最近 issue/PR，判断是否长期无人处理。
- 防护动作：issue/PR 响应未知时，必须提示维护风险。
- 证据：evidence.maintainer_signals | github_repo:552661142 | https://github.com/langchain-ai/langchain | issue_or_pr_quality=unknown
## 6. 维护坑 · 发布节奏不明确
- 严重度：low
- 证据强度：source_linked
- 发现：release_recency=unknown。
- 对用户的影响：安装命令和文档可能落后于代码，用户踩坑概率升高。
- 建议检查：确认最近 release/tag 和 README 安装命令是否一致。
- 防护动作：发布节奏未知或过期时，安装说明必须标注可能漂移。
- 证据：evidence.maintainer_signals | github_repo:552661142 | https://github.com/langchain-ai/langchain | release_recency=unknown