Source code for langchain_community.tools.plugin

from __future__ import annotations

import json
from typing import Optional, Type

import requests
import yaml
from langchain_core.callbacks import (
    AsyncCallbackManagerForToolRun,
    CallbackManagerForToolRun,
)
from langchain_core.pydantic_v1 import BaseModel
from langchain_core.tools import BaseTool


class ApiConfig(BaseModel):
    """API配置。"""

    type: str
    url: str
    has_user_authentication: Optional[bool] = False


class AIPlugin(BaseModel):
    """AI插件定义。"""

    schema_version: str
    name_for_model: str
    name_for_human: str
    description_for_model: str
    description_for_human: str
    auth: Optional[dict] = None
    api: ApiConfig
    logo_url: Optional[str]
    contact_email: Optional[str]
    legal_info_url: Optional[str]

    @classmethod
    def from_url(cls, url: str) -> AIPlugin:
        """从URL实例化AIPlugin。"""
        response = requests.get(url).json()
        return cls(**response)


def marshal_spec(txt: str) -> dict:
    """将yaml或json序列化的规范转换为字典。

参数：
    txt：yaml或json序列化的规范。

返回：
    dict：规范的字典形式。
"""
    try:
        return json.loads(txt)
    except json.JSONDecodeError:
        return yaml.safe_load(txt)


class AIPluginToolSchema(BaseModel):
    """AIPluginTool的架构。"""

    tool_input: Optional[str] = ""


class AIPluginTool(BaseTool):
    """用于获取AI插件的OpenAPI规范的工具。"""

    plugin: AIPlugin
    api_spec: str
    args_schema: Type[AIPluginToolSchema] = AIPluginToolSchema

    @classmethod
    def from_plugin_url(cls, url: str) -> AIPluginTool:
        plugin = AIPlugin.from_url(url)
        description = (
            f"Call this tool to get the OpenAPI spec (and usage guide) "
            f"for interacting with the {plugin.name_for_human} API. "
            f"You should only call this ONCE! What is the "
            f"{plugin.name_for_human} API useful for? "
        ) + plugin.description_for_human
        open_api_spec_str = requests.get(plugin.api.url).text
        open_api_spec = marshal_spec(open_api_spec_str)
        api_spec = (
            f"Usage Guide: {plugin.description_for_model}\n\n"
            f"OpenAPI Spec: {open_api_spec}"
        )

        return cls(
            name=plugin.name_for_model,
            description=description,
            plugin=plugin,
            api_spec=api_spec,
        )

    def _run(
        self,
        tool_input: Optional[str] = "",
        run_manager: Optional[CallbackManagerForToolRun] = None,
    ) -> str:
        """使用这个工具。"""
        return self.api_spec

    async def _arun(
        self,
        tool_input: Optional[str] = None,
        run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
    ) -> str:
        """使用该工具进行异步操作。"""
        return self.api_spec