以编程方式使用文档

集成包是用户可以安装并在项目使用的 Python 包。它们实现一个或多个符合 LangChain 接口标准的组件。

LangChain 组件是中的基类的子类 langchain-core。示例包括 聊天模型, 工具, 检索器等。

您的集成包通常至少实现这些组件之一的子类。展开下面的标签页查看每个组件的详细信息。

Chat Models

聊天模型是 BaseChatModel 类的子类。它们实现用于生成聊天补全、处理消息格式和管理模型参数的方法。

Embeddings

嵌入模型是 Embeddings 类的子类。

Tools

工具有 2 种主要用途:

  1. 定义一个"输入模式"或"参数模式",将其与文本请求一起传递给聊天模型的工具调用功能,以便聊天模型可以生成"工具调用",即调用工具的参数。
  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, globgrep.

实际上,如果您的沙箱环境可以运行 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.