什么是LangChain？

LangChain是一个用于构建AI代理（Agents）和LLM驱动应用的框架。它通过将可互操作的组件与第三方集成串联在一起，简化了AI应用程序的开发过程，同时为底层技术演进提供了前瞻性支持。

LangChain的主要特性包括：

• 灵活抽象层 — 从用于快速起步的高级链（Chains）到用于细粒度控制的低级组件，LangChain可以随着应用程序复杂度的增长而扩展
• 模块化设计 — 可互换的组件便于定制和扩展
• 丰富的集成生态 — 支持多种LLM提供商、向量存储、工具和外部服务

资料来源：README.md:1-25

核心架构

LangChain采用模块化架构设计，主要包含以下核心层次：

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的基础库，提供了所有其他组件依赖的核心抽象和接口。

经典组件库 (langchain_classic)

langchain_classic包含传统的链式组件、代理逻辑、API调用链等经典功能。

资料来源：libs/langchain/langchain_classic/chains/api/base.py:1-30

合作伙伴集成生态

LangChain提供了丰富的合作伙伴包（Partner Packages）集成，支持主流AI服务提供商和工具。

支持的LLM提供商

资料来源：

• libs/partners/openai/README.md:1-20
• libs/partners/anthropic/README.md:1-15
• libs/partners/huggingface/README.md:1-20
• libs/partners/ollama/README.md:1-15

向量存储与工具集成

资料来源：

• libs/partners/chroma/README.md:1-15
• libs/partners/nomic/README.md:1-15

快速开始

安装

pip install langchain
# 或使用 uv
uv add langchain

基础使用示例

from langchain.chat_models import init_chat_model

# 初始化聊天模型
model = init_chat_model("openai:gpt-5.4")

# 调用模型
result = model.invoke("Hello, world!")

使用HuggingFace本地模型

如需使用本地推理功能，需安装完整依赖：

pip install langchain-huggingface[full]

注意：完整安装包含 sentence-transformers>=5.2.0 和 transformers>=5.0.0。从 langchain-community 迁移的用户需要注意版本要求变化。

资料来源：

• README.md:30-45
• libs/partners/huggingface/README.md:10-20

技术特性

安全机制

LangChain的APIChain组件包含详细的安全警告和最佳实践：

• 网络安全 — 使用requests工具包发起HTTP请求，需控制访问权限
• 域限制 — 支持limit_to_domains参数限制可访问的域名范围
• 请求类型 — 支持GET、POST、PATCH、PUT和DELETE等HTTP方法

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

废弃与迁移策略

LangChain采用渐进式废弃策略，标记废弃的功能会：

• 指定废弃版本号 (since)
• 指定移除版本号 (removal)
• 提供替代方案建议 (alternative)

@deprecated(
    since="0.2.13",
    removal="2.0.0",
    alternative="langchain.agents.create_agent",
    addendum="使用 create_agent 构建新代理..."
)
class APIChain(Chain):
    ...

文档与资源

相关工具与扩展

LangGraph

对于需要更高级自定义或代理编排的场景，推荐使用LangGraph — LangChain官方的工作流控制框架。

graph LR
    A[LangChain] -->|基础链式调用| B[简单任务]
    A -->|复杂编排| C[LangGraph]
    C -->|可控工作流| D[复杂Agent系统]

Deep Agents

针对常见使用模式（规划、子代理、文件系统操作等）开箱即用的解决方案：

• 内置规划能力
• 多子代理协调
• 文件系统集成
• 常见模式封装

版本与发布策略

LangChain遵循语义化版本控制规范：

• 版本历史 — 详见 Releases Policy
• 版本规范 — 详见 Versioning Policy
• 贡献指南 — 详见 Contributing Guide

许可证

LangChain采用 MIT许可证 开源发布。

资料来源：README.md:10-15

概述

LangChain 是一个用于构建智能体（Agent）和大语言模型（LLM）应用的框架。它通过将可互操作的组件与第三方集成串联在一起，简化了 AI 应用程序开发流程，同时在底层技术演进的背景下保障了决策的前瞻性。

资料来源：README.md:1-10

系统架构分层

LangChain 的架构采用分层设计，主要包括以下几个核心层次：

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 请求并总结响应。其核心工作流程如下：

graph LR
    A[用户问题] --> B[API_URL_PROMPT]
    B --> C[LLM 生成 API 调用]
    C --> D[TextRequestsWrapper 执行请求]
    D --> E[API_RESPONSE_PROMPT]
    E --> F[LLM 生成最终答案]

安全特性：APIChain 内置了域名限制机制，仅允许向预配置的域名发送请求：

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 包列表

资料来源：libs/partners/openai/README.md:1-20

快速入门与初始化模式

LangChain 提供了统一的模型初始化接口 init_chat_model，支持通过字符串标识符动态加载和配置模型：

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

生态系统架构图

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 包对依赖库有不同的版本要求：

资料来源：libs/partners/huggingface/README.md:1-20

安全与权限控制

APIChain 域名限制

APIChain 实现了严格的域名白名单机制，防止应用向未授权的外部服务发送请求。这一机制通过 _check_in_allowed_domain 函数实现：

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 新架构

弃用说明：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 是 LangChain 框架的核心抽象接口，定义了所有可执行组件的统一契约。它代表了可以接收输入、产生输出的任意操作单元，包括语言模型、提示词模板、工具、检索器、链等。

LangChain 的核心设计理念是将复杂的 AI 应用分解为可组合的小单元，通过链式调用（Chaining）将多个 Runnable 连接在一起，形成端到端的工作流。

核心架构

Runnable 在 LangChain 中的角色

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 抽象基类

核心方法定义

BaseRunnable 定义了 Runnable 的核心接口规范，所有实现类必须遵循这些契约。

# 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]:
        """批量异步调用。"""
        ...

输入输出类型

链式调用机制

管道操作符 `|`

链式调用通过管道操作符 | 实现，将前一个 Runnable 的输出作为后一个 Runnable 的输入。

# 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))

链式调用示例

graph LR
    A[PromptTemplate] -->|"管道操作符 |"| B[BaseChatModel]
    B -->|"管道操作符 |"| C[StrOutputParser]
    
    A1[输入: 问题] --> A
    C --> R[最终输出]

代码示例：

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 内部实现

# 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() 方法实现链式调用：

chain = prompt.pipe(model).pipe(output_parser)

两种方式在功能上等价，开发者可根据偏好选择。

Runnable 配置

RunnableConfig 配置结构

# 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]]
    """运行时可配置参数。"""

配置传递机制

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

配置合并策略

# 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 包装器

# 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

重试配置参数

重试策略使用示例

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  # 只在速率限制错误时重试
)

指数退避算法

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 条件分支

# 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]]

分支配置参数

分支使用示例

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  # 默认链
)

分支执行流程

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

常见使用模式

模式一：简单管道链

chain = prompt | model | output_parser
result = chain.invoke({"question": "..."})

模式二：带配置的链

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)

模式三：并行分支合并

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": "..."}

模式四：动态链

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 代理链

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 新闻"})

高级特性

回调机制

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]})

流式处理

# 流式调用
for chunk in chain.stream({"question": "..."}):
    print(chunk, end="", flush=True)

批量处理

# 批量调用
questions = [
    {"question": "问题1"},
    {"question": "问题2"},
    {"question": "问题3"}
]

results = chain.batch(questions)

与旧版 Chain 的关系

迁移提示：Chain 类在 LangChain 0.2.13 版本被标记为废弃，将在 2.0.0 版本移除。建议使用基于 Runnable 的新 API。资料来源：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（代理）系统是LangChain框架中用于构建自主智能体的核心模块。Agent能够根据用户输入动态决定执行哪些操作（Tool），通过循环调用工具并处理结果来完成任务。与静态的Chain不同，Agent具有推理和决策能力，能够在运行时根据上下文选择下一步行动。 资料来源：libs/langchain/langchain_classic/agents/agent.py:1-50

Agent系统主要由以下几个核心组件构成：

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

class Agent(Chain):
    """Agent基类，定义推理和执行接口"""
    
    @property
    def observation_prefix(self) -> str:
        """观察结果前缀"""
        
    @property
    def llm_prefix(self) -> str:
        """LLM输出前缀"""

Agent执行流程

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

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

class OpenAIFunctionsAgent(Agent):
    """基于OpenAI function calling的Agent
    
    使用模型原生能力生成函数调用，
    无需解析文本输出
    """
    
    @property
    def allowed_tools(self) -> list[str]:
        """允许调用的工具列表"""

优势：

Structured Chat Agent

Structured Chat Agent支持复杂的多轮对话场景，能够处理包含多个参数的函数调用。它结合了聊天模型的能力和结构化工具调用。 资料来源：libs/langchain/langchain_classic/agents/structured_chat/base.py:1-50

class StructuredChatAgent(Agent):
    """结构化聊天Agent
    
    适用于需要调用复杂工具的场景，
    支持多参数和嵌套参数
    """

Tool集成

Tool定义

Tool是Agent执行具体操作的接口。每个Tool包含名称、描述和执行函数。

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要求

配置选项

关键参数

错误处理

Agent执行过程中可能遇到各种错误：

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架构：

# 旧方式 (已废弃)
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优化

# 使用系统提示词优化Agent行为
agent = create_agent(
    tools=my_tools,
    system_message="""
    你是一个专业的助手。
    总是先思考再行动。
    如果不确定，请返回最终答案。
    """
)

相关资源

• LangChain官方文档
• API参考
• 示例代码库

总结

Agent代理系统是LangChain的核心能力之一，通过结合LLM的推理能力和Tool的执行能力，实现了灵活的自主任务执行。框架提供了多种Agent类型以适应不同场景，开发者应根据具体需求选择合适的Agent模式。

概述

LangChain 中的工具（Tools）和工具包（Toolkits）是框架的核心组件，它们使 LLM（大型语言模型）能够与外部世界交互、执行具体操作、访问实时信息以及操控各种系统和数据源。工具是可执行的原子单元，而工具包则是相关工具的集合，为特定领域或任务提供完整的功能支持。

在 LangChain 的架构中，工具系统位于 langchain_core 层，这表明工具是跨多个包共享的基础抽象。langchain-core 定义了工具的核心接口和基类，而 langchain-classic 在此基础上提供了更丰富的实现和特定场景的工具包。

LangChain 的工具系统设计遵循以下核心原则：类型安全、异步支持、结构化参数验证、以及与主流 LLM 提供商的广泛兼容性。通过统一的工具接口，开发者可以轻松创建自定义工具或使用预构建的工具包来加速开发。

工具的核心架构

工具的基类层次

LangChain 的工具系统采用分层架构设计。BaseTool 是所有工具的抽象基类，定义了工具的基本行为和接口。StructuredTool 则在 BaseTool 基础上添加了结构化参数支持，允许工具接受复杂类型的输入参数。

工具的核心方法包括：

• invoke：同步执行工具
• ainvoke：异步执行工具
• run：兼容旧版 API 的执行方法
• arun：异步版本的旧 API

这种分层设计允许工具在保持一致接口的同时，根据具体需求提供不同的功能实现。基础层负责定义契约，而具体实现可以继承和扩展这些基类以添加特定功能。

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 提供的标准工具包之一，专门用于数据库操作场景。该工具包通常包含以下核心工具：

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/ 目录下，包括：

这些合作伙伴包通常包含针对特定平台优化的工具实现，充分利用各平台的特性和能力。

社区工具与扩展

除了官方维护的包外，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

概述

向量存储与检索是 LangChain 框架中用于处理语义搜索和相似性匹配的核心功能模块。在大型语言模型（LLM）应用中，向量存储通过将文本、文档或其他数据转换为高维向量表示，实现高效的语义检索能力。当用户提出查询时，系统会将查询转换为向量，并在向量数据库中查找最相似的文档或文本片段。LangChain 的向量存储架构设计遵循模块化原则，核心接口定义在 langchain-core 包中，而具体实现则分布在各个集成包中，例如 langchain-chroma 用于 Chroma 向量数据库的集成。这种架构使得开发者可以灵活地在不同向量存储提供商之间切换，同时保持上层应用代码的一致性。

核心架构

LangChain 的向量存储系统采用分层架构设计，从底层到上层依次包括向量存储实现层、嵌入模型层和检索器层。向量存储层负责实际的数据存储和相似性搜索操作，支持多种后端包括 Chroma、FAISS、Pinecone 等主流向量数据库。嵌入模型层提供文本到向量的转换能力，通过 Embeddings 接口定义标准化的嵌入生成方法。检索器层则封装了各种检索策略，包括直接相似性检索、自查询检索、时间加权检索等高级功能。这种分层设计使得系统具有良好的可扩展性和可维护性，开发者可以根据需求自由组合不同层次的组件。

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 接口支持异步操作，这对于构建高性能应用至关重要。向量存储的构造函数通常接受嵌入模型实例、连接参数和配置选项，使得初始化过程灵活且易于配置。

嵌入模型接口

嵌入模型是向量存储系统的核心组件，负责将文本数据转换为高维向量表示。LangChain 通过 Embeddings 基类定义了跨不同嵌入提供商的统一接口，该接口包含 embed_query 和 embed_documents 两个核心方法。embed_query 方法用于单个查询文本的嵌入生成，而 embed_documents 方法支持批量处理多个文档的嵌入计算。这个设计允许开发者轻松切换不同的嵌入提供商，无论是使用 OpenAI 的 text-embedding-ada-002、HuggingFace 的 sentence-transformers，还是本地部署的嵌入模型。LangChain 还提供了缓存和批处理优化，以减少 API 调用开销和提高整体性能。

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）是一种高级检索能力，它能够自动从用户查询中提取语义搜索条件和元数据过滤条件。这种检索器特别适用于结构化元数据丰富的场景，例如文档带有时间戳、类别、作者等属性。当用户提出一个问题时，自查询检索器会同时生成两个部分：一是用于语义相似性匹配的文本嵌入向量，二是用于精确过滤的元数据条件表达式。这种双重处理机制使得检索结果既具有语义相关性，又满足精确的元数据约束条件。

自查询检索器的工作流程涉及多个步骤：首先，大语言模型（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 接口规范，确保应用代码在不同后端之间具有良好的可移植性。

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）两个部分，元数据可以包含来源信息、页码、创建时间等任何有助于检索的信息。

异步操作支持

现代 LLM 应用对性能和并发处理能力有较高要求，LangChain 的向量存储系统提供了完整的异步操作支持。异步接口包括 aadd_texts、aadd_documents、aasync_similarity_search 等方法，允许开发者在异步环境中高效地处理批量操作。异步设计特别适用于需要同时查询多个向量存储、或者在高并发场景下处理大量请求的应用。LangChain 的异步实现遵循 Python 的 asyncio 规范，可以与 FastAPI 等现代 Web 框架无缝集成。

安全性与权限控制

在构建向量存储应用时，安全性是需要重点考虑的方面。LangChain 提供了域名限制（Domain Limiting）功能，可以配置向量存储只允许查询特定域名的数据源，防止潜在的安全风险。这一功能在 APIChain 等可能暴露给最终用户的组件中尤为重要，通过限制可访问的域名范围，可以防止恶意查询或数据泄露。开发者应该仔细评估应用的安全需求，合理配置访问控制策略，确保向量存储系统在提供便利的同时不引入安全漏洞。

最佳实践

在实际应用中，向量存储与检索系统的性能优化需要关注多个方面。首先是嵌入模型的选择，不同的嵌入模型在精度、速度和成本方面各有权衡，应该根据具体场景选择最合适的模型。其次是文档块大小的确定，过大的块可能导致检索精度下降，过小的块则可能丢失上下文信息，建议通过实验找到最佳的块大小和重叠参数。批量操作时应该利用批处理 API 减少网络开销，同时注意避免超过 API 的速率限制。在生产环境中，应该实现适当的错误处理和重试机制，以应对网络波动和服务不可用等情况。

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 集成文档

概述

文档加载与处理是 LangChain 框架中用于将各种格式的外部数据转换为标准化 Document 对象的核心模块。该模块支持从多种数据源加载内容，包括 PDF、HTML、Markdown、JSON、CSV、Word 文档等，并提供了灵活的分词（Text Splitting）功能以支持大规模文档处理场景。

LangChain 的文档处理架构采用分层设计，核心接口定义在 langchain-core 包中，具体实现则分散在 langchain-classic 和各集成包中。

核心架构

LangChain 文档处理流程包含三个主要阶段：加载（Loading）、解析（Parsing） 和 分块（Chunking/Splitting）。

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 中表示文档的基础数据结构，包含以下核心属性：

class Document:
    page_content: str
    metadata: dict = Field(default_factory=dict)

资料来源：libs/core/langchain_core/document_loaders/base.py

BaseLoader 抽象基类

所有文档加载器必须继承自 BaseLoader 基类：

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

文档加载器类型

按数据源分类

内置加载器列表

langchain-classic 包提供了丰富的内置加载器实现：

解析器注册机制

LangChain 使用解析器注册表（Registry）来管理不同文档类型的解析逻辑。

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 类

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

基于字符的分块器，按指定分隔符分割文本：

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,
    ):
        ...

关键参数说明

资料来源：libs/text-splitters/langchain_text_splitters/base.py

分块策略

LangChain 支持多种分块策略：

graph TD
    A[输入文本] --> B{分块策略}
    
    B -->|字符分割| C[CharacterTextSplitter]
    B -->|递归分割| D[RecursiveCharacterTextSplitter]
    B -->|代码分割| E[LanguageSplitter]
    B -->|语义分割| F[SemanticChunker]
    
    C --> G[按固定分隔符分割]
    D --> H[优先小分隔符<br/>失败则回退大分隔符]
    E --> I[按编程语言语法分割]
    F --> J[基于嵌入相似度分割]
    
    G --> K[Document 列表]
    H --> K
    I --> K
    J --> K

RecursiveCharacterTextSplitter

递归字符分块器尝试按优先级顺序使用不同的分隔符：

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

使用示例

基本加载流程

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}")

异步加载

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)

文档分块

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)}")

元数据管理

文档元数据用于存储与内容相关的附加信息：

# 创建带元数据的文档
doc = Document(
    page_content="这是文档内容",
    metadata={
        "source": "report.pdf",
        "page": 5,
        "author": "Alice"
    }
)

与其他模块的集成

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 其他核心组件的协作：

最佳实践

1. 选择合适的分块大小

2. 合理设置重叠

• chunk_overlap 应设置为 chunk_size 的 10-20%
• 足够的重叠确保跨块边界的语义连续性

3. 处理特殊格式

对于结构化文档（PDF、Word），优先使用专用加载器以保留布局和元信息：

# PDF 加载（保留页码）
from langchain_core.document_loaders import PyPDFLoader

loader = PyPDFLoader("document.pdf")
pages = loader.load_and_split()  # 每页返回一个 Document

扩展自定义加载器

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}
        )

相关资源

概述

在 LangChain 框架中，内存与状态管理是构建上下文感知（context-aware）应用的核心能力。内存模块负责在多轮对话或复杂任务执行过程中持久化、检索和管理交互历史，使大型语言模型（LLM）能够访问先前的上下文信息，从而生成连贯且符合上下文的响应。

LangChain 的内存系统被设计为可插拔（pluggable）架构，允许开发者根据不同的应用场景选择或自定义合适的内存实现方案。这些方案涵盖从简单的缓冲区存储到复杂的向量检索等多种策略。

核心架构

系统组件关系

LangChain 的内存系统主要由以下几个核心组件构成，它们之间通过标准化接口进行交互：

graph TD
    A[用户交互 / Chain 执行] --> B[Memory 模块]
    B --> C[缓冲区内存<br/>BufferMemory]
    B --> D[摘要内存<br/>SummaryMemory]
    B --> E[向量存储内存<br/>VectorStoreMemory]
    B --> F[聊天消息历史<br/>ChatMessageHistory]
    
    C --> G[BaseMemory 接口]
    D --> G
    E --> G
    F --> G
    
    G --> H[Chain 上下文]
    G --> I[LLM 上下文窗口]

内存类型分类

LangChain 内存模块支持多种内存实现，以满足不同场景的需求：

资料来源：libs/langchain/langchain_classic/memory/__init__.py

缓冲区内存（BufferMemory）

功能说明

缓冲区内存是 LangChain 中最简单的内存实现，它按顺序存储所有对话消息，并在需要时将完整的历史记录传递给下游组件。这种方式保留了所有交互细节，但会随着对话长度增加而消耗更多令牌（token）。

核心实现

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 对象，按时间顺序排列。

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

使用示例

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 分析当前对话历史并生成更新的摘要。这个摘要替换了原有的历史记录，成为新的上下文基础。

graph LR
    A[新消息到达] --> B{对话历史为空？}
    B -->|是| C[生成初始摘要]
    B -->|否| D[加载当前摘要]
    D --> E[分析新旧内容]
    E --> F[生成更新摘要]
    C --> G[存储摘要]
    F --> G

资料来源：libs/langchain/langchain_classic/memory/summary.py

配置参数

聊天消息历史（ChatMessageHistory）

架构角色

ChatMessageHistory 是 LangChain 内存系统的核心数据层，负责管理消息的持久化存储和检索。它作为一个抽象层，支持多种后端存储实现。

classDiagram
    class BaseChatMessageHistory {
        <<abstract>>
        +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

消息类型

系统支持多种消息类型以适应不同的交互模式：

向量存储内存（VectorStoreMemory）

语义检索能力

向量存储内存利用嵌入（embedding）技术将对话内容转换为高维向量，然后存储在向量数据库中。这种方法支持语义相似性搜索，允许系统根据含义而非精确关键词来检索相关历史。

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

配置选项

与 Chain 的集成

数据流处理

LangChain 的 Chain（链）是内存系统的主要消费者。Chain 通过标准化的内存接口与各种内存实现进行交互：

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：

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
)

最佳实践

场景选择指南

性能优化建议

1. 消息分页：对于超长对话，实施基于时间或数量的消息分页策略
2. 异步处理：使用异步内存实现提高 I/O 密集型操作的吞吐量
3. 缓存策略：对频繁访问的摘要结果实施缓存
4. 后端选择：根据数据量选择合适的存储后端（内存/文件/数据库/向量库）

安全考量

在生产环境中部署内存系统时，应注意以下安全要点：

• 数据隔离：确保不同用户/会话的内存数据严格隔离
• 敏感信息处理：对包含敏感内容的对话实施适当的脱敏或加密
• 访问控制：限制对历史消息的读取和修改权限
• 数据清理：定期清理不再需要的会话历史，遵守数据保留政策

扩展与自定义

自定义内存实现

开发者可以通过继承 BaseMemory 类来创建自定义内存实现：

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 内存系统支持与多种外部系统集成，以实现更强大的状态管理能力：

总结

LangChain 的内存与状态管理系统为构建上下文感知的 LLM 应用提供了强大而灵活的基础设施。通过模块化的设计和丰富的内置实现，开发者可以轻松选择或组合适合特定场景的内存策略。从简单的缓冲区到复杂的向量检索，系统提供了全面的解决方案来管理对话状态和历史信息。

核心优势包括：

• 统一接口：所有内存类型遵循 BaseMemory 接口，便于替换和扩展
• 多种策略：支持缓冲、摘要、向量检索等多种内存管理策略
• 持久化支持：可与多种存储后端集成，满足生产环境需求
• 无缝集成：与 Chain、Agent 等组件深度集成，开箱即用

概述

LangChain 的 Chat 模型与 LLM 集成系统为开发者提供了统一的方式来接入不同提供商的大语言模型服务。该系统基于 langchain-core 库构建的核心抽象层，通过合作伙伴包（Partner Packages）的形式支持多种 LLM 提供商。

核心架构

集成包结构

LangChain 采用模块化的包结构组织其 LLM 集成：

类层次结构

LangChain 的 LLM 集成采用分层设计：

BaseLanguageModel (langchain-core)
    ├── BaseChatModel
    │   ├── ChatOpenAI (langchain-openai)
    │   ├── ChatAnthropic (langchain-anthropic)
    │   └── ...
    └── BaseLLM
        ├── OpenAI (langchain-openai)
        └── ...

资料来源：README.md:31-36

快速开始

基础安装

使用 pip 安装 LangChain 主包：

pip install langchain

初始化 Chat 模型

LangChain 提供了统一的模型初始化接口：

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 包：

pip install langchain-openai

该包包含与 OpenAI GPT 模型交互的 LangChain 集成，提供对 GPT-4o、GPT-4 等模型的支持。

资料来源：libs/partners/openai/README.md:1-20

Anthropic 集成

安装 langchain-anthropic 包：

pip install langchain-anthropic

该包提供与 Anthropic Claude 模型的集成功能，支持通过 LangChain 框架调用 Claude 系列生成模型。

资料来源：libs/partners/anthropic/README.md:1-18

HuggingFace 集成

基础安装：

pip install langchain-huggingface

如果需要使用 HuggingFaceEmbeddings 或 HuggingFacePipeline 进行本地推理，应安装完整版本：

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 包：

pip install langchain-ollama

该包提供与 Ollama 的集成，支持本地部署的开源大语言模型。

资料来源：libs/partners/ollama/README.md:1-18

APIChain 与工具调用

Chain 架构

APIChain 是 LangChain Classic 中的一个重要组件，它实现了调用 API 并总结响应以回答问题的功能。该链使用请求工具包发送 GET、POST、PATCH、PUT 和 DELETE 请求。

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 进行状态管理：

class ChainState(TypedDict):
    """LangGraph state."""
    messages: Annotated[Sequence[BaseMessage], add_messages]

工作流图

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 域限制功能，防止恶意请求：

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 中的某些组件已被标记为弃用：

@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 提供商：

文档资源

资料来源：libs/langchain/README.md:1-20 资料来源：libs/partners/openai/README.md:1-20

相关链接

• LangGraph：用于构建可控代理工作流的框架 - docs.langchain.com/langgraph
• LangSmith：用于开发、调试和部署 AI 代理的工具 - docs.langchain.com/langsmith
• Deep Agents：基于 LangChain 的高级代理包 - docs.langchain.com/deepagents
• LangChain.js：等效的 JavaScript/TypeScript 库 - github.com/langchain-ai/langchainjs

概述

LangChain 的回调与追踪系统是框架的核心基础设施之一，为 LangChain 组件提供了一套统一的生命周期事件通知和执行追踪机制。该系统使开发者能够在 LangChain 运行期间捕获关键事件、监控执行流程、收集性能指标，并实现自定义的日志记录和监控功能。

回调与追踪系统的主要职责包括：

• 事件传播：在 Chain、LLM、Tool、Agent 等核心组件的生命周期关键时刻触发回调
• 执行追踪：记录组件的输入、输出、执行时间和中间状态
• 异步支持：提供异步和同步两种回调处理模式
• 上下文传递：通过回调管理器维护和传递调用上下文

资料来源：libs/core/langchain_core/tracers/__init__.py:1-50

系统架构

核心组件关系

LangChain 的回调与追踪系统由以下几个核心组件构成，它们之间形成清晰的层次结构：

graph TD
    A[CallbackHandler 回调处理器] --> B[BaseCallbackManager 基础回调管理器]
    A --> C[Tracer 追踪器]
    C --> D[BaseTracer 基础追踪器]
    C --> E[EventStreamTracer 事件流追踪器]
    B --> F[CallbackManager 同步回调管理器]
    B --> G[AsyncCallbackManager 异步回调管理器]
    H[Run 运行时对象] --> C

组件职责

资料来源：libs/core/langchain_core/tracers/base.py:1-100

回调处理器

CallbackHandler 接口

CallbackHandler 是所有回调处理器的基类接口，定义了组件生命周期中可能被触发的方法。处理器可以继承此接口并选择性重写特定方法来实现自定义行为。

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 提供了多种内置回调处理器，适用于不同的监控和日志需求：

追踪器系统

Tracer 抽象基类

Tracer 是专门用于创建和记录 Run 对象的组件。每个 Run 代表一次被追踪的执行单元，包含执行过程中的元数据、输入输出等信息。

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 对象的创建和状态管理：

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 是一种特殊的追踪器实现，它通过事件流的方式持续输出追踪事件，适用于需要实时监控的场景：

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 是同步回调执行的管理器，负责注册、调用和清理回调处理器：

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

回调管理器核心方法

异步回调管理

AsyncCallbackManager 提供与 CallbackManager 类似的功能，但专门用于异步执行环境：

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 是追踪系统的核心数据模型，代表一次可追踪的执行单元：

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 追踪字段

资料来源：libs/core/langchain_core/tracers/base.py:100-200

使用示例

基本用法

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!")

自定义回调处理器

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 追踪

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})

配置选项

全局回调配置

环境变量

安全注意事项

在 LangChain 的 APIChain 文档中明确指出回调与追踪系统的安全考量：

安全注意：API 链使用 requests 工具包向 API 发起 GET、POST、PATCH、PUT 和 DELETE 请求。

请谨慎控制谁可以使用此链以及它具有何种网络访问权限。

如果向最终用户公开，请考虑用户将能够代表服务器发起任意请求。

资料来源：libs/langchain/langchain_classic/chains/api/base.py:40-60

安全最佳实践

• 限制回调中发送的敏感数据范围
• 在生产环境中使用 HTTPS 传输追踪数据
• 实施回调数据的访问控制策略
• 定期审计追踪日志内容

生态集成

合作伙伴包中的追踪支持

多个 LangChain 合作伙伴包内置了回调与追踪系统支持：

资料来源：libs/partners/openai/README.md, libs/partners/anthropic/README.md, libs/partners/huggingface/README.md

总结

LangChain 的回调与追踪系统为框架提供了强大的可观测性和扩展性能力。通过统一的回调接口、灵活的追踪机制和完善的 Run 数据模型，开发者可以轻松实现：

• 执行过程监控和日志记录
• 性能指标收集和分析
• 分布式追踪集成
• 自定义监控和告警系统

该系统设计遵循了模块化原则，允许开发者根据需求选择性地使用回调或追踪功能，或将两者结合使用以获得完整的应用可观测性。

Pitfall Log / 踩坑日志

项目：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