langchain-core 中基类的子类。示例包括聊天模型、工具、检索器等。
你的集成包通常需要实现至少一个此类组件的子类。展开下方的标签页以查看每个组件的详细信息。
- 聊天模型
- 嵌入模型
- 工具
- 中间件
- 检查点器
- 沙箱
聊天模型是
BaseChatModel 类的子类。它们实现了生成聊天补全、处理消息格式化和管理模型参数的方法。聊天模型集成指南目前正在编写中。在此期间,请阅读聊天模型概念指南以了解 LangChain 聊天模型的工作原理。你也可以参考 LangChain 仓库 中现有的集成。
嵌入模型是
Embeddings 类的子类。嵌入模型集成指南目前正在编写中。在此期间,请阅读嵌入模型概念指南以了解 LangChain 嵌入模型的工作原理。
工具主要有两种使用方式:
- 定义一个“输入模式”或“参数模式”,与文本请求一起传递给聊天模型的工具调用功能,以便聊天模型可以生成“工具调用”或调用工具所需的参数。
- 接收上述生成的“工具调用”,执行某些操作并返回一个响应,该响应可以作为 ToolMessage 传递回聊天模型。
BaseTool 基类。此接口有 3 个属性和 2 个方法需要在子类中实现。工具集成指南目前正在编写中。在此期间,请阅读工具概念指南以了解 LangChain 工具的工作原理。
中间件 允许你通过挂钩到模型调用、工具调用和代理生命周期事件来自定义代理行为。中间件类继承自
特定于提供商的中间件位于提供商的集成包中(例如
AgentMiddleware 基类。在构建集成之前,请阅读自定义中间件指南以了解挂钩、状态更新和中间件模式。中间件集成通常分为两类:| 类型 | 描述 | 示例 |
|---|---|---|
| 特定于提供商 | 利用提供商的独特能力 | 提示缓存、原生工具执行、内容审核 |
| 跨提供商 | 适用于任何模型或工具 | 速率限制、PII 检测、日志记录、防护栏 |
langchain-anthropic)。跨提供商的中间件可以作为独立包发布。你也可以参考这些现有的中间件集成:OpenAI 内容审核
具有配置选项和退出行为的单个中间件。
Anthropic 中间件
用于提示缓存、工具、内存和文件搜索的多个中间件类。
AWS 提示缓存
特定于提供商的提示缓存,包含模型行为表。
自定义中间件指南
关于挂钩、状态更新和模式的完整参考。
检查点器实现了 LangGraph 中的持久化,允许代理在交互之间保存和恢复状态。请参阅 LangGraph 仓库 中现有的检查点器集成以获取实现示例。
沙箱集成使 Deep Agents 能够在隔离环境中运行代码。实现 Deep Agents 中的 将此文件放在
SandboxBackendProtocol。此协议包括 execute()、异步变体以及文件系统工具方法,如 ls、read、write、edit、glob 和 grep。实际上,如果你的沙箱环境可以运行 shell 命令并且有 python3 可用,你通常应该继承 BaseSandbox。BaseSandbox 通过 python3 提供文件系统操作,因此你主要需要实现 execute()、upload_files()、download_files() 和 id。示例 BaseSandbox 脚手架 可展开
from __future__ import annotations
from deepagents.backends.protocol import (
ExecuteResponse,
FileDownloadResponse,
FileUploadResponse,
)
from deepagents.backends.sandbox import BaseSandbox
class MySandbox(BaseSandbox):
def __init__(self, client: MySandboxSdkClient) -> None:
self._client = client
@property
def id(self) -> str:
return self._client.sandbox_id
def execute(
self,
command: str,
*,
timeout: int | None = None,
) -> ExecuteResponse:
# 在你的沙箱中执行 `command` 并将提供商响应映射到 ExecuteResponse。
result = self._client.run(command=command, timeout=timeout)
output = result.stdout or ""
if result.stderr:
output += f"\n<stderr>{result.stderr}</stderr>"
return ExecuteResponse(
output=output,
exit_code=result.exit_code,
truncated=False,
)
def upload_files(
self,
files: list[tuple[str, bytes]],
) -> list[FileUploadResponse]:
# 验证路径,尽可能批量处理请求,并将提供商结果按输入顺序映射回 FileUploadResponse 对象。
# 仅捕获和规范化 LLM 可能重试或修复的错误,例如 invalid_path 或 file_not_found。
return self._client.upload_files(files)
def download_files(self, paths: list[str]) -> list[FileDownloadResponse]:
# 验证路径,尽可能批量处理请求,并将提供商结果按输入顺序映射回 FileDownloadResponse 对象。
# 仅捕获和规范化 LLM 可能重试或修复的错误,例如 invalid_path 或 file_not_found。
return self._client.download_files(paths)
async def aexecute(
self,
command: str,
*,
timeout: int | None = None,
) -> ExecuteResponse:
...
async def aupload_files(
self,
files: list[tuple[str, bytes]],
) -> list[FileUploadResponse]:
...
async def adownload_files(
self,
paths: list[str],
) -> list[FileDownloadResponse]:
...
测试你的集成
使用沙箱标准测试套件验证你的集成。Python 套件使用langchain_tests.integration_tests 中的 SandboxIntegrationTests;继承它并提供一个 sandbox fixture,该 fixture 生成一个干净的 SandboxBackendProtocol 实例。示例沙箱标准测试设置 可展开
from __future__ import annotations
from collections.abc import Iterator
import pytest
from deepagents.backends.protocol import SandboxBackendProtocol
from langchain_tests.integration_tests import SandboxIntegrationTests
from langchain_myprovider import MySandbox
from myprovider_sdk import MySandboxSdkClient
class TestMySandboxStandard(SandboxIntegrationTests):
@pytest.fixture(scope="class")
def sandbox(self) -> Iterator[SandboxBackendProtocol]:
client = MySandboxSdkClient()
backend = MySandbox(client=client)
try:
yield backend
finally:
# 将此替换为你的提供商的清理逻辑。
client.delete_sandbox(backend.id)
tests/integration_tests/test_sandbox.py 等文件中。标准套件将为你处理实际的文件系统和命令执行断言。参考实现: 请参阅 Daytona 合作伙伴集成,它继承了 BaseSandbox 并实现了 execute()、upload_files()、download_files() 和 id。将这些文档 通过 MCP 连接到 Claude、VSCode 等,以获取实时答案。

