Liu Zhicong
145 lines
3.6 KiB
Python
145 lines
3.6 KiB
Python
from abc import ABC, abstractmethod
|
|
from typing import Dict,Coroutine,Callable
|
|
|
|
class ParameterDefine:
|
|
def __init__(self) -> None:
|
|
self.name = None
|
|
self.type = None
|
|
self.description = None
|
|
|
|
|
|
class AIFunction:
|
|
def __init__(self) -> None:
|
|
self.description : str = None
|
|
|
|
@abstractmethod
|
|
def get_name(self) -> str:
|
|
"""
|
|
return the name of the function (should be snake case)
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_description(self) -> str:
|
|
"""
|
|
return a detailed description of what the function does
|
|
"""
|
|
return self.description
|
|
|
|
@abstractmethod
|
|
def get_parameters(self) -> Dict:
|
|
"""
|
|
Return the list of parameters to execute this function in the form of
|
|
JSON schema as specified in the OpenAI documentation:
|
|
https://platform.openai.com/docs/api-reference/chat/create#chat/create-parameters
|
|
|
|
str = run_code(code:str)
|
|
parameters = {
|
|
"type": "object",
|
|
"properties": {
|
|
"code": {
|
|
"type": "string",
|
|
"description": "Python code which needs to be executed"
|
|
}
|
|
}
|
|
}
|
|
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def execute(self, **kwargs) -> str:
|
|
"""
|
|
Execute the function and return a JSON serializable dict.
|
|
The parameters are passed in the form of kwargs
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def is_local(self) -> bool:
|
|
"""
|
|
is this function call need network?
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def is_in_zone(self) -> bool:
|
|
"""
|
|
is this function call in Lan?
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def is_ready_only(self) -> bool:
|
|
pass
|
|
|
|
#def load_from_config(self,config:dict) -> bool:
|
|
# pass
|
|
|
|
class FunctionItem:
|
|
def __init__(self,name,args) -> None:
|
|
self.name = name
|
|
self.args = args
|
|
self.body = None
|
|
|
|
def append_body(self,body:str) -> None:
|
|
if self.body is None:
|
|
self.body = body
|
|
else:
|
|
self.body += body
|
|
|
|
def dumps(self) -> str:
|
|
pass
|
|
|
|
# call chain is a combination of ai_function,group of ai_function.
|
|
class CallChain:
|
|
def __init__(self) -> None:
|
|
pass
|
|
|
|
def load_from_config(self,config:dict) -> bool:
|
|
pass
|
|
|
|
async def execute(self):
|
|
pass
|
|
|
|
class SimpleAIFunction(AIFunction):
|
|
def __init__(self,func_id:str,description:str,func_handler:Coroutine,parameters:Dict = None) -> None:
|
|
self.func_id = func_id
|
|
self.description = description
|
|
self.func_handler = func_handler
|
|
self.parameters = parameters
|
|
|
|
def get_name(self) -> str:
|
|
return self.func_id
|
|
|
|
def get_parameters(self) -> Dict:
|
|
if self.parameters is not None:
|
|
result = {}
|
|
result["type"] = "object"
|
|
parm_defines = {}
|
|
for parm,desc in self.parameters.items():
|
|
parm_item = {}
|
|
parm_item["type"] = "string"
|
|
parm_item["description"] = desc
|
|
parm_defines[parm] = parm_item
|
|
result["properties"] = parm_defines
|
|
return result
|
|
return {"type": "object", "properties": {}}
|
|
|
|
|
|
async def execute(self,**kwargs) -> str:
|
|
if self.func_handler is None:
|
|
return "error: function not implemented"
|
|
|
|
return await self.func_handler(**kwargs)
|
|
|
|
def is_local(self) -> bool:
|
|
return True
|
|
|
|
def is_in_zone(self) -> bool:
|
|
return True
|
|
|
|
def is_ready_only(self) -> bool:
|
|
return False
|
|
|