实现 LangChain 集成 - AI 开发中文文档

集成包是用户可以为其项目安装的 Python 包。它们实现了一个或多个遵循 LangChain 接口标准的组件。 LangChain 组件是 langchain-core 中基础类的子类。例如包括对话模型、工具、检索器等。您的集成包通常会实现至少一个这些组件的子类。展开下方的选项卡以查看每个组件的详细信息。

对话模型是 @[BaseChatModel] 类的子类。它们实现了生成对话补全、处理消息格式化和管理模型参数的方法。

:::python 对话模型集成指南目前处于开发中。在此期间，请阅读对话模型概念指南了解 LangChain 对话模型的工作原理。您还可以参考 LangChain 仓库中的现有集成 ::: :::js 对话模型集成指南目前处于开发中。在此期间，请阅读对话模型概念指南了解 LangChain 对话模型的工作原理。您还可以参考 LangChain 仓库中的现有集成 :::

嵌入模型是 @[Embeddings] 类的子类。

嵌入模型集成指南目前处于开发中。在此期间，请阅读嵌入模型概念指南了解 LangChain 嵌入模型的工作原理。

工具主要有两种使用方式：

定义一个“输入模式”或“参数模式”，将其与文本请求一起传递给对话模型的工具调用功能，以便对话模型可以生成一个“工具调用”，或调用工具的参数。
接收上述生成的“工具调用”，执行某些操作并返回一个响应，该响应可以作为一个 ToolMessage 传递回对话模型。

Tools 类必须继承自 @[BaseTool] 基类。此接口有 3 个属性和 2 个方法需要在子类中实现。

工具集成指南目前处于开发中。在此期间，请阅读工具概念指南了解 LangChain 工具的工作原理。

中间件允许您通过挂钩模型调用、工具调用和智能体生命周期事件来自定义智能体行为。中间件类是 @[AgentMiddleware] 基类的子类。在构建集成之前，请先阅读自定义中间件指南了解挂钩、状态更新和中间件模式。中间件集成通常分为两类：

类型	描述	示例
特定提供者	利用提供者的独特功能	提示词缓存、原生工具执行、内容审核
跨提供者	适用于任何模型或工具	速率限制、PII 检测、日志记录、安全限制

特定提供者的中间件位于提供者的集成包中（例如 langchain-anthropic）。跨提供者的中间件可以作为独立包发布。您还可以使用这些现有的中间件集成作为参考：:::python

OpenAI content moderation

带有配置选项和退出行为的单个中间件。

Anthropic middleware

用于提示词缓存、工具、记忆和文件搜索的多个中间件类。

AWS prompt caching

带有模型行为表的特定提供者提示词缓存。

Custom middleware guide

挂钩、状态更新和模式的完整参考。

::: :::js

Anthropic middleware

用于提示词缓存、工具、记忆和文件搜索的多个中间件类。

Custom middleware guide

挂钩、状态更新和模式的完整参考。

:::

沙盒集成使深度智能体能够在隔离环境中运行代码。:::python 实现 Deep Agents 的 @[SandboxBackendProtocol]。此协议包括 execute()、异步变体以及文件系统工具方法，例如 ls、read、write、edit、glob 和 grep。实际上，如果您的沙盒环境可以运行 Shell 命令并且有 python3 可用，您通常应该从 @[BaseSandbox] 继承。BaseSandbox 通过 python3 提供文件系统操作，因此您主要需要实现 execute()、upload_files()、download_files() 和 id。

Example BaseSandbox scaffold

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:
        # 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 固定装置，该装置生成一个干净的 SandboxBackendProtocol 实例。

Example sandbox standard test setup

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:
            # Replace this with your provider's cleanup logic.
            client.delete_sandbox(backend.id)

将此内容放在文件中，例如 tests/integration_tests/test_sandbox.py。标准套件将为您处理实际的文件系统和命令执行断言。参考实现： 查看日潭合作伙伴集成，它继承自 BaseSandbox 并实现了 execute()、upload_files()、download_files() 和 id。 :::

官方原文：src/oss/contributing/implement-langchain.mdx · 非官方翻译，译文由 AI 生成，以英文原文为准。

OpenAI content moderation

Anthropic middleware

AWS prompt caching

Custom middleware guide

Anthropic middleware

Custom middleware guide

​测试您的集成

测试您的集成