File size: 1,975 Bytes
031beb8 |
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 |
import json
from typing import Any, Dict, List, Union
import requests
from pydantic import BaseModel
class ReActStep(BaseModel):
"""RaAct 推理步骤"""
thought: Dict[str, Union[Dict[str, str], str]]
action: str
action_input: Dict[str, str]
observation: Dict[str, Any] = {}
def to_str(self) -> str:
s = f"Thought: {self.thought}\n"
s += f"Action: {self.action}\n"
s += f"Action Input: {self.action_input}\n"
if self.observation:
s += f"Observation: {self.observation}\n"
return s
class RequstField(BaseModel):
"""请求体字段"""
enum: List[str]
name: str
description: str
is_required: bool
parament_type: str
@property
def to_simple_dict(self) -> Dict[str, Any]:
data = {
"description": self.description,
"required": self.is_required,
}
if self.enum:
data["enum"] = self.enum
return data
class Plugin(BaseModel):
"""插件"""
url: str
method: str
headers: Dict[str, str]
request_body: List[RequstField]
name_for_human: str
description_for_human: str
description_for_model: str
unique_name_for_model: str
@property
def parameter_schema(self) -> str:
parameter = {}
for field in self.request_body:
if not field.is_required:
continue
parameter[field.name] = field.to_simple_dict
return json.dumps(parameter, ensure_ascii=False)
def run(self, **kwargs):
"""运行插件"""
response = requests.request(
self.method.upper(),
self.url,
headers=self.headers,
params=kwargs,
json=kwargs,
)
return response.text
@property
def required_parameters(self) -> List[RequstField]:
"""必填参数"""
return [field for field in self.request_body if field.is_required]
|