Source code for langchain_community.tools.connery.service

import json
from typing import Dict, List, Optional

import requests
from langchain_core.pydantic_v1 import BaseModel, root_validator
from langchain_core.utils.env import get_from_dict_or_env

from langchain_community.tools.connery.models import Action
from langchain_community.tools.connery.tool import ConneryAction


class ConneryService(BaseModel):
    """与Connery Runner API交互的服务。

它从Connery Runner获取可用操作的列表，将它们封装在ConneryAction工具中并返回给用户。
它还提供了运行这些操作的方法。"""

    runner_url: Optional[str] = None
    api_key: Optional[str] = None

    @root_validator()
    def validate_attributes(cls, values: Dict) -> Dict:
        """验证 ConneryService 类的属性。
参数：
    values（字典）：需要验证的参数。
返回：
    字典：经过验证的参数。
"""

        runner_url = get_from_dict_or_env(values, "runner_url", "CONNERY_RUNNER_URL")
        api_key = get_from_dict_or_env(values, "api_key", "CONNERY_RUNNER_API_KEY")

        if not runner_url:
            raise ValueError("CONNERY_RUNNER_URL environment variable must be set.")
        if not api_key:
            raise ValueError("CONNERY_RUNNER_API_KEY environment variable must be set.")

        values["runner_url"] = runner_url
        values["api_key"] = api_key

        return values

    def list_actions(self) -> List[ConneryAction]:
        """返回Connery Runner中可用操作的列表。
返回：
    List[ConneryAction]：Connery Runner中可用操作的列表。
"""

        return [
            ConneryAction.create_instance(action, self)
            for action in self._list_actions()
        ]

    def get_action(self, action_id: str) -> ConneryAction:
        """返回Connery Runner中指定的操作。
参数:
    action_id (str): 要返回的操作的ID。
返回:
    ConneryAction: 指定ID的操作。
"""

        return ConneryAction.create_instance(self._get_action(action_id), self)

    def run_action(self, action_id: str, input: Dict[str, str] = {}) -> Dict[str, str]:
        """运行指定的Connery动作并提供输入。
参数：
    action_id（str）：要运行的动作的ID。
    input（Dict[str, str]）：动作期望的输入对象。
返回：
    Dict[str, str]：动作的输出。
"""

        return self._run_action(action_id, input)

    def _list_actions(self) -> List[Action]:
        """返回Connery Runner中可用操作的列表。
返回：
    List[Action]：Connery Runner中可用操作的列表。
"""

        response = requests.get(
            f"{self.runner_url}/v1/actions", headers=self._get_headers()
        )

        if not response.ok:
            raise ValueError(
                (
                    "Failed to list actions."
                    f"Status code: {response.status_code}."
                    f"Error message: {response.json()['error']['message']}"
                )
            )

        return [Action(**action) for action in response.json()["data"]]

    def _get_action(self, action_id: str) -> Action:
        """返回Connery Runner中指定动作。
参数：
    action_id（str）：要返回的动作的ID。
返回：
    Action：具有指定ID的动作。
"""

        actions = self._list_actions()
        action = next((action for action in actions if action.id == action_id), None)
        if not action:
            raise ValueError(
                (
                    f"The action with ID {action_id} was not found in the list"
                    "of available actions in the Connery Runner."
                )
            )
        return action

    def _run_action(self, action_id: str, input: Dict[str, str] = {}) -> Dict[str, str]:
        """运行指定的Connery动作并提供输入。
参数：
    action_id（str）：要运行的动作的ID。
    prompt（str）：这是一个纯英文提示，包含运行动作所需的所有信息。
    input（Dict[str, str]）：动作期望的输入对象。
    如果与提示一起提供，则输入优先于提示中指定的输入。
返回：
    Dict[str, str]：动作的输出。
"""

        response = requests.post(
            f"{self.runner_url}/v1/actions/{action_id}/run",
            headers=self._get_headers(),
            data=json.dumps({"input": input}),
        )

        if not response.ok:
            raise ValueError(
                (
                    "Failed to run action."
                    f"Status code: {response.status_code}."
                    f"Error message: {response.json()['error']['message']}"
                )
            )

        if not response.json()["data"]["output"]:
            return {}
        else:
            return response.json()["data"]["output"]

    def _get_headers(self) -> Dict[str, str]:
        """返回一个标准的HTTP头，用于在API调用中与Connery runner交互。
返回：
    Dict[str, str]: 标准的HTTP头。
"""

        return {"Content-Type": "application/json", "x-api-key": self.api_key or ""}