Spaces:
Runtime error
Runtime error
File size: 5,741 Bytes
ed4d993 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 |
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):
"""Service for interacting with the Connery Runner API.
It gets the list of available actions from the Connery Runner,
wraps them in ConneryAction Tools and returns them to the user.
It also provides a method for running the actions.
"""
runner_url: Optional[str] = None
api_key: Optional[str] = None
@root_validator()
def validate_attributes(cls, values: Dict) -> Dict:
"""
Validate the attributes of the ConneryService class.
Parameters:
values (dict): The arguments to validate.
Returns:
dict: The validated arguments.
"""
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]:
"""
Returns the list of actions available in the Connery Runner.
Returns:
List[ConneryAction]: The list of actions available in the Connery Runner.
"""
return [
ConneryAction.create_instance(action, self)
for action in self._list_actions()
]
def get_action(self, action_id: str) -> ConneryAction:
"""
Returns the specified action available in the Connery Runner.
Parameters:
action_id (str): The ID of the action to return.
Returns:
ConneryAction: The action with the specified 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]:
"""
Runs the specified Connery Action with the provided input.
Parameters:
action_id (str): The ID of the action to run.
input (Dict[str, str]): The input object expected by the action.
Returns:
Dict[str, str]: The output of the action.
"""
return self._run_action(action_id, input)
def _list_actions(self) -> List[Action]:
"""
Returns the list of actions available in the Connery Runner.
Returns:
List[Action]: The list of actions available in the 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:
"""
Returns the specified action available in the Connery Runner.
Parameters:
action_id (str): The ID of the action to return.
Returns:
Action: The action with the specified 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]:
"""
Runs the specified Connery Action with the provided input.
Parameters:
action_id (str): The ID of the action to run.
prompt (str): This is a plain English prompt
with all the information needed to run the action.
input (Dict[str, str]): The input object expected by the action.
If provided together with the prompt,
the input takes precedence over the input specified in the prompt.
Returns:
Dict[str, str]: The output of the action.
"""
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]:
"""
Returns a standard set of HTTP headers
to be used in API calls to the Connery runner.
Returns:
Dict[str, str]: The standard set of HTTP headers.
"""
return {"Content-Type": "application/json", "x-api-key": self.api_key or ""}
|