集成包是用户可以安装并在项目使用的 Python 包。它们实现一个或多个符合 LangChain 接口标准的组件。
LangChain 组件是中的基类的子类 langchain-core。示例包括 聊天模型, 工具, 检索器等。
您的集成包通常至少实现这些组件之一的子类。展开下面的标签页查看每个组件的详细信息。
Chat Models
聊天模型是 BaseChatModel 类的子类。它们实现用于生成聊天补全、处理消息格式和管理模型参数的方法。
Embeddings
嵌入模型是 Embeddings 类的子类。
Tools
工具有 2 种主要用途:
- 定义一个"输入模式"或"参数模式",将其与文本请求一起传递给聊天模型的工具调用功能,以便聊天模型可以生成"工具调用",即调用工具的参数。
- 接收如上生成的"工具调用",执行某些操作并返回可作为 ToolMessage 传递回聊天模型的响应。
工具类必须继承 BaseTool 基类。此接口有 3 个属性和 2 个方法,应在子类中实现。
Middleware
中间件 允许您通过钩子到模型调用、工具调用和代理生命周期事件来自定义代理行为。中间件类子类化 AgentMiddleware 基类。
阅读 自定义中间件指南 以了解在构建集成之前的钩子、状态更新和中间件模式。
中间件集成通常分为两类:
| 类型 | 描述 | 示例 |
|---|---|---|
| **Provider-specific** | 利用提供商独特能力 | 提示缓存、本地工具执行、内容审核 |
| **Cross-provider** | 适用于任何模型或工具 | 速率限制、PII 检测、日志记录、安全护栏 |
特定提供商的中间件位于该提供商的集成包中(例如 langchain-anthropic)。跨提供商中间件可以作为独立包发布。
您也可以将这些现有的中间件集成用作参考:
OpenAI content moderation
具有配置选项和退出行为的单个中间件。
Anthropic middleware
用于提示缓存、工具、内存和文件搜索的多个中间件类。
AWS prompt caching
具有模型行为表的特定提供商提示缓存。
Custom middleware guide
钩子、状态更新和模式的完整参考。
Checkpointers
检查点程序实现 持久化 在 LangGraph 中,允许智能体保存和恢复跨交互的状态。
请参阅 LangGraph 仓库 中现有的检查点程序集成,了解实现示例。
Sandboxes
沙箱集成使 Deep Agents 能够在隔离环境中运行代码。
实现 SandboxBackendProtocol 从 Deep Agents。此协议包括 execute()、异步变体和文件系统工具方法,如 ls, read, write, edit, glob和 grep.
实际上,如果您的沙箱环境可以运行 shell 命令并有 python3 可用,您通常应该继承 BaseSandbox. BaseSandbox 通过 python3提供文件系统操作,因此您主要需要实现 execute(), upload_files(), download_files()和 id.
from __future__ import annotations
from deepagents.backends.protocol import (
ExecuteResponse,
FileDownloadResponse,
FileUploadResponse,
)
from deepagents.backends.sandbox import BaseSandbox # [!code highlight]
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:
# Execute `command` in your sandbox and map the provider response
# into 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]:
# Validate paths, batch requests where possible, and map provider
# results back into FileUploadResponse objects in input order.
# Only catch and normalize errors that an LLM can plausibly retry
# or fix, such as invalid_path or file_not_found.
return self._client.upload_files(files)
def download_files(self, paths: list[str]) -> list[FileDownloadResponse]:
# Validate paths, batch requests where possible, and map provider
# results back into FileDownloadResponse objects in input order.
# Only catch and normalize errors that an LLM can plausibly retry
# or fix, such as invalid_path or 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;继承它并提供一个 sandbox fixture,返回一个干净的 SandboxBackendProtocol instance.
from __future__ import annotations
from collections.abc import Iterator
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:
# Replace this with your provider's cleanup logic.
client.delete_sandbox(backend.id)
将此放在文件中,例如 tests/integration_tests/test_sandbox.py。标准套件将为您处理实际的文件系统和命令执行断言。
参考实现: 参见 Daytona 合作伙伴集成,它继承自 BaseSandbox 并实现了 execute(), upload_files(), download_files()和 id.