重构函数信息获取逻辑;移除示例测试用例文件

This commit is contained in:
远野千束(神羽) 2024-12-15 02:51:37 +08:00
parent 4083aba99f
commit f1064b65db
10 changed files with 399 additions and 166 deletions

View File

@ -1,7 +1,7 @@
"""该功能目前正在开发中,暂时不可用,受影响的文件夹 `plugin`, `plugins`
"""
from .func_call import *
from .load import *
from .models import *
from .register import *
from .utils import *

View File

@ -0,0 +1,3 @@
from .caller import *
from .params import *
from .register import *

View File

@ -0,0 +1,76 @@
from typing import Generic
from ..typing import FUNCTION_CALL_FUNC
from .params import P
from .register import F
class Caller(Generic[P]):
def __init__(self, name: str | None = None, description: str | None = None):
self._name = name
self._description = description
self._parameters: dict[str, P] = {}
self.func: FUNCTION_CALL_FUNC | None = None
def params(self, **kwargs: P) -> "Caller":
"""设置多个函数参数
Args:
**kwargs: 参数字典
Returns:
Caller: Caller对象
"""
self._parameters.update(kwargs)
return self
def param(self, name: str, param: P) -> "Caller":
"""设置一个函数参数
Args:
name (str): 参数名
param (P): 参数对象
Returns:
Caller: Caller对象
"""
self._parameters[name] = param
return self
def name(self, name: str) -> "Caller":
"""设置函数名称
Args:
name (str): 函数名称
Returns:
Caller: Caller对象
"""
self._name = name
return self
def description(self, description: str) -> "Caller":
"""设置函数描述
Args:
description (str): 函数描述
Returns:
Caller: Caller对象
"""
self._description = description
return self
def __call__(self, func: F) -> F:
self.func = func
return func
def on_function_call(name: str | None = None, description: str | None = None) -> Caller:
"""返回一个Caller类可用于装饰一个函数使其注册为一个可被AI调用的function call函数
Args:
description: 函数描述若为None则从函数的docstring中获取
Returns:
Caller: Caller对象
"""
return Caller(name=name, description=description)

View File

@ -0,0 +1,125 @@
from enum import Enum
from typing import Any, TypeVar
from pydantic import BaseModel, Field
from ..typing import FUNCTION_CALL_FUNC
P = TypeVar("P", bound="Parameter")
"""参数类型泛型"""
class ParamTypes:
STRING = "string"
INTEGER = "integer"
ARRAY = "array"
OBJECT = "object"
BOOLEAN = "boolean"
NUMBER = "number"
class Parameter(BaseModel):
"""
插件函数参数对象
Attributes:
----------
name: str
参数名称
type: str
参数类型 string integer等
description: str
参数描述
"""
type_: str
"""参数类型描述 string integer等"""
description: str
"""参数描述"""
default: Any = None
"""默认值"""
properties: dict[str, Any] = {}
"""参数定义属性,例如最大值最小值等"""
required: bool = False
"""是否必须"""
def data(self) -> dict[str, Any]:
return {
"type": self.type_,
"description": self.description,
**{k: v for k, v in self.properties.items() if v is not None},
}
class String(Parameter):
type_: str = ParamTypes.STRING
properties: dict[str, Any] = Field(default_factory=dict)
enum: list[str] | None = None
class Integer(Parameter):
type_: str = ParamTypes.INTEGER
properties: dict[str, Any] = Field(
default_factory=lambda: {"minimum": 0, "maximum": 100}
)
minimum: int | None = None
maximum: int | None = None
class Array(Parameter):
type_: str = ParamTypes.ARRAY
properties: dict[str, Any] = Field(
default_factory=lambda: {"items": {"type": "string"}}
)
items: str = Field("string", description="数组元素类型")
class FunctionCall(BaseModel):
"""
插件函数对象
Attributes:
----------
name: str
函数名称
func: "FUNCTION_CALL"
函数对象
"""
name: str
"""函数名称 module.func"""
description: str
"""函数描述 这个函数用于获取天气信息"""
arguments: dict[str, Parameter]
"""函数参数信息"""
function: FUNCTION_CALL_FUNC
"""函数对象"""
kwargs: dict[str, Any] = {}
"""扩展参数"""
class Config:
arbitrary_types_allowed = True
def __hash__(self) -> int:
return hash(self.name)
def data(self) -> dict[str, Any]:
"""生成函数描述信息
Returns:
dict[str, Any]: 函数描述信息 字典
"""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": {
"type": "object",
"properties": {k: v.data() for k, v in self.arguments.items()},
},
"required": [k for k, v in self.arguments.items() if v.default is None],
**self.kwargs,
},
}

View File

@ -0,0 +1,126 @@
"""此模块用于获取function call中函数定义信息以及注册函数
"""
from typing import TypeVar
from nonebot import logger
from ..docstring.parser import parse
from ..typing import (
ASYNC_FUNCTION_CALL_FUNC,
FUNCTION_CALL_FUNC,
SYNC_FUNCTION_CALL_FUNC,
)
from .params import *
F = TypeVar("F", bound=FUNCTION_CALL_FUNC)
_loaded_functions: dict[str, FUNCTION_CALL_FUNC] = {}
def async_wrapper(func: SYNC_FUNCTION_CALL_FUNC) -> ASYNC_FUNCTION_CALL_FUNC:
"""将同步函数包装为异步函数,但是不会真正异步执行,仅用于统一调用及函数签名
Args:
func: 同步函数
Returns:
ASYNC_FUNCTION_CALL: 异步函数
"""
async def wrapper(*args, **kwargs) -> str:
return func(*args, **kwargs)
return wrapper
def function_call(func: F) -> F:
"""返回一个装饰器,装饰一个函数, 使其注册为一个可被AI调用的function call函数
Args:
func: 函数对象要有完整的 Google Style Docstring
Returns:
str: 函数定义信息
"""
# TODO
# pre check docstring
if not func.__doc__:
logger.error(f"函数 {func.__name__} 没有文档字串,不被加载")
return func
else:
# 解析函数文档字串
result = parse(docstring=func.__doc__)
logger.debug(result.reduction())
return func
def caller(
description: str | None = None,
parameters: dict[str, P] | None = None,
):
"""返回一个装饰器,装饰一个函数, 使其注册为一个可被AI调用的function call函数
Args:
description: 函数描述
parameters: 函数参数
Returns:
str: 函数定义信息
"""
def decorator(func: FUNCTION_CALL_FUNC) -> FUNCTION_CALL_FUNC:
# TODO
# pre check docstring
if not func.__doc__:
logger.error(f"函数 {func.__name__} 没有文档字串,不被加载")
return func
else:
# 解析函数文档字串
result = parse(docstring=func.__doc__)
logger.debug(result.reduction())
return func
return decorator
# TODO 草案
# @caller(
# description="这个函数用来给你算命",
# parameters={
# "birthday": String(description="生日"),
# "gender": String(enum=["男", "女"], description="性别"),
# "name": String(description="姓名"),
# },
# )
# async def tell_fortune(birthday: str, name: str, gender: str) -> str:
# """这个函数用来给你算命
# Args:
# birthday: 生日
# name: 姓名
# Returns:
# str: 算命结果
# """
# return f"{name},你的生日是{birthday},你的运势是大吉大利"
@caller(
description="这个函数用来给你算命",
).parameters(
birthday=String(description="生日"),
name=String(enum=["", ""], description="性别"),
gender=String(description="姓名"),
)
async def tell_fortune(birthday: str, name: str, gender: str) -> str:
"""这个函数用来给你算命
Args:
birthday: 生日
name: 姓名
Returns:
str: 算命结果
"""
return f"{name},你的生日是{birthday},你的运势是大吉大利"

View File

@ -1,7 +1,7 @@
from types import ModuleType
from typing import Any
from pydantic import BaseModel
from pydantic import BaseModel, Field
from .typing import ASYNC_FUNCTION_CALL_FUNC, FUNCTION_CALL_FUNC
@ -69,75 +69,3 @@ class Plugin(BaseModel):
def __eq__(self, other: Any) -> bool:
return self.name == other.name
class FunctionCallArgument(BaseModel):
"""
插件函数参数对象
Attributes:
----------
name: str
参数名称
type: str
参数类型 string integer等
description: str
参数描述
"""
type_: str
"""参数类型描述 string integer等"""
description: str
"""参数描述"""
default: Any = None
"""默认值"""
def data(self) -> dict[str, Any]:
return {"type": self.type_, "description": self.description}
class FunctionCall(BaseModel):
"""
插件函数对象
Attributes:
----------
name: str
函数名称
func: "FUNCTION_CALL"
函数对象
"""
name: str
"""函数名称 module.func"""
description: str
"""函数描述 这个函数用于获取天气信息"""
arguments: dict[str, FunctionCallArgument]
"""函数参数信息"""
function: FUNCTION_CALL_FUNC
"""函数对象"""
class Config:
arbitrary_types_allowed = True
def __hash__(self) -> int:
return hash(self.name)
def data(self) -> dict[str, Any]:
"""生成函数描述信息
Returns:
dict[str, Any]: 函数描述信息 字典
"""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": {
"type": "object",
"properties": {k: v.data() for k, v in self.arguments.items()},
},
"required": [k for k, v in self.arguments.items() if v.default is None],
},
}

View File

@ -1,66 +0,0 @@
"""此模块用于获取function call中函数定义信息以及注册函数
"""
import inspect
import litedoc
from nonebot import logger
from nonebot_plugin_marshoai.plugin.utils import is_coroutine_callable
from .models import FunctionCall, FunctionCallArgument
from .typing import (
ASYNC_FUNCTION_CALL_FUNC,
FUNCTION_CALL_FUNC,
SYNC_FUNCTION_CALL_FUNC,
)
_loaded_functions: dict[str, FUNCTION_CALL_FUNC] = {}
def async_wrapper(func: SYNC_FUNCTION_CALL_FUNC) -> ASYNC_FUNCTION_CALL_FUNC:
"""将同步函数包装为异步函数,但是不会真正异步执行,仅用于统一调用及函数签名
Args:
func: 同步函数
Returns:
ASYNC_FUNCTION_CALL: 异步函数
"""
async def wrapper(*args, **kwargs) -> str:
return func(*args, **kwargs)
return wrapper
def function_call(*funcs: FUNCTION_CALL_FUNC) -> None:
"""返回一个装饰器,装饰一个函数, 使其注册为一个可被AI调用的function call函数
Args:
func: 函数对象要有完整的 Google Style Docstring
Returns:
str: 函数定义信息
"""
for func in funcs:
function_call = get_function_info(func)
# TODO: 注册函数
def get_function_info(func: FUNCTION_CALL_FUNC):
"""获取函数信息
Args:
func: 函数对象
Returns:
FunctionCall: 函数信息对象模型
"""
description = func.__doc__
# TODO: 获取函数参数信息
parameters = {} # type: ignore
# 使用inspect解析函数的传参及类型
sig = inspect.signature(func)
for name, param in sig.parameters.items():
logger.debug(name, param)

View File

@ -2,7 +2,13 @@ import traceback
import httpx
from nonebot_plugin_marshoai.plugin import PluginMetadata, function_call
from nonebot_plugin_marshoai.plugin import (
Integer,
PluginMetadata,
String,
function_call,
on_function_call,
)
__marsho_meta__ = PluginMetadata(
name="Bangumi 番剧信息",
@ -13,7 +19,9 @@ __marsho_meta__ = PluginMetadata(
)
@function_call
async def fetch_calendar():
"""获取今天日期"""
url = "https://api.bgm.tv/calendar"
headers = {
"User-Agent": "LiteyukiStudio/nonebot-plugin-marshoai (https://github.com/LiteyukiStudio/nonebot-plugin-marshoai)"
@ -26,11 +34,7 @@ async def fetch_calendar():
@function_call
async def get_bangumi_news() -> str:
"""获取今天的新番(动漫)列表,在调用之前,你需要知道今天星期几。
Returns:
_type_: _description_
"""
"""获取今天的新番(动漫)列表,在调用之前,你需要知道今天星期几。"""
result = await fetch_calendar()
info = ""
try:

View File

@ -0,0 +1,57 @@
from nonebot_plugin_marshoai.plugin import (
Integer,
Parameter,
PluginMetadata,
String,
on_function_call,
)
__marsho_meta__ = PluginMetadata(
name="SnowyKami 测试插件",
description="A test plugin for SnowyKami",
usage="SnowyKami Test Plugin",
)
@on_function_call(description="使用姓名,年龄,性别进行算命").params(
age=Integer(description="年龄"),
name=String(description="姓名"),
gender=String(enum=["", ""], description="性别"),
)
async def fortune_telling(age: int, name: str, gender: str) -> str:
"""使用姓名,年龄,性别进行算命
Args:
age (int): _description_
name (str): _description_
gender (str): _description_
Returns:
str: _description_
"""
# 进行一系列算命操作...
return f"{name},你的年龄是{age},你的性别很好"
@on_function_call(description="获取一个地点未来一段时间的天气").params(
location=String(description="地点名称,可以是城市名、地区名等"),
days=Integer(description="天数", minimum=1, maximum=30),
unit=String(enum=["摄氏度", "华氏度"], description="温度单位"),
)
async def get_weather(location: str, days: int, unit: str) -> str:
"""获取一个地点未来一段时间的天气
Args:
location (str): 地点名称可以是城市名地区名等
days (int): 天数
unit (str): 温度单位
Returns:
str: 天气信息
"""
# 进行一系列获取天气操作...
return f"{location}未来{days}天的天气信息..."

View File

@ -1,20 +0,0 @@
def example_function(num: int, text: str, is_a: bool) -> str:
"""这是一个示例描述
Args:
num (int): 描述整数
text (str): 文本类型
is_a (bool): 布尔类型
Returns:
str: 消息
"""
return "-"
class TestPlugin:
def test_get_function_info(self):
from nonebot_plugin_marshoai.plugin import get_function_info
func_info = get_function_info(example_function)
print(func_info)