langchain-core 库中的基类派生的子类,例如 聊天模型、工具、检索器,等等。
您的集成包通常会实现至少一个这些组件的一个子类。展开下面的选项卡以查看每个组件的详细信息。
- 聊天模型
- 嵌入式表示
- 工具
- 中间件
- 检查点器
- 沙盒
聊天模型是
BaseChatModel 类(参见参考文档)的子类。它们实现了生成聊天完成、处理消息格式化和管理模型参数的方法。当前,聊天模型集成指南尚在开发中。在此期间,请阅读 聊天模型概念性指南 以了解 LangChain 聊天模型的功能。您还可以参考 LangChain 库中的现有集成
嵌入式表示模型是
Embeddings 类(参见参考文档)的子类。当前,嵌入式表示模型集成指南尚在开发中。在此期间,请阅读 嵌入式表示模型概念性指南 以了解 LangChain 嵌入式表示模型的功能。
工具主要用于以下两种方式:
- 定义一个“输入模式”或“参数模式”,作为与聊天模型的工具调用功能一起传递给文本请求,以便聊天模型可以生成一个“工具调用”,即用于调用该工具的参数。
- 接收上述生成的“工具调用”,执行某些操作并返回响应,这些响应可以传递回聊天模型作为 ToolMessage。
BaseTool 基类(参见参考文档)。此接口有三个属性和两个方法,应在子类中实现。当前,工具集成指南尚在开发中。在此期间,请阅读 工具概念性指南 以了解 LangChain 工具的功能。
中间件(参见概述)允许您通过钩子模型调用、工具调用和代理生命周期事件来自定义代理行为。中间件类是从
特定提供者的中间件存在于提供者的集成包中(例如
AgentMiddleware 基类派生的。在构建集成之前,请阅读 自定义中间件指南,以了解挂钩、状态更新和中间件模式。中间件集成通常分为两类:| 类型 | 描述 | 示例 |
|---|---|---|
| 特定提供者 | 利用提供商的独特功能 | 提示缓存,原生工具执行,内容审核 |
| 跨提供商 | 适用于任何模型或工具 | 速率限制、PII 检测、日志记录、护栏 |
langchain-anthropic)。跨提供商的中间件可以作为独立包发布。您还可以使用这些现有的中间件集成作为参考:OpenAI 内容审核
单个具有配置选项和退出行为的中间件。
Anthropic 中间件
多个用于提示缓存、工具、记忆和文件搜索的中间件类。
AWS 提示缓存
适用于特定提供者的提示缓存,带有模型行为表。
自定义中间件指南
对挂钩、状态更新和模式的完整参考。
检查点器使 持久性 在 LangGraph 中成为可能,允许代理在交互之间保存和恢复状态。查看 LangGraph 库中的现有检查点器集成 以获取实现示例。
沙盒集成使 深度代理 能够在隔离环境中运行代码。实现来自 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 套件使用SandboxIntegrationTests 从 langchain_tests.integration_tests;从中派生并提供一个返回干净的 SandboxBackendProtocol 实例的 sandbox 固定值。# 示例沙盒标准测试设置可扩展
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 等连接起来,以获得实时答案。

