模块 nonebot_plugin_marshoai.plugin.func_call.caller
class Caller
func __init__(self, name: str = '', description: str | None = None)
源代码 或 在GitHub上查看
python
def __init__(self, name: str='', description: str | None=None):
self._name: str = name
'函数名称'
self._description = description
'函数描述'
self._plugin: Plugin | None = None
'所属插件对象,装饰时声明'
self.func: ASYNC_FUNCTION_CALL_FUNC | None = None
'函数对象'
self.module_name: str = ''
'模块名,仅为父级模块名,不一定是插件顶级模块名'
self._parameters: dict[str, Any] = {}
'声明参数'
self.di: SessionContextDepends = SessionContextDepends()
'依赖注入的参数信息'
self.default: dict[str, Any] = {}
'默认值'
self.ctx: SessionContext | None = None
self._permission: Permission | None = None
self._rule: Rule | None = None
func params(self, **kwargs: Any) -> Caller
源代码 或 在GitHub上查看
python
def params(self, **kwargs: Any) -> 'Caller':
self._parameters.update(kwargs)
return self
func permission(self, permission: Permission) -> Caller
源代码 或 在GitHub上查看
python
def permission(self, permission: Permission) -> 'Caller':
self._permission = self._permission or permission
return self
async func pre_check(self) -> tuple[bool, str]
源代码 或 在GitHub上查看
python
async def pre_check(self) -> tuple[bool, str]:
if self.ctx is None:
return (False, '上下文为空')
if self.ctx.bot is None or self.ctx.event is None:
return (False, 'Context is None')
if self._permission and (not await self._permission(self.ctx.bot, self.ctx.event)):
return (False, '告诉用户 Permission Denied 权限不足')
if self.ctx.state is None:
return (False, 'State is None')
if self._rule and (not await self._rule(self.ctx.bot, self.ctx.event, self.ctx.state)):
return (False, '告诉用户 Rule Denied 规则不匹配')
return (True, '')
func rule(self, rule: Rule) -> Caller
源代码 或 在GitHub上查看
python
def rule(self, rule: Rule) -> 'Caller':
self._rule = self._rule and rule
return self
func name(self, name: str) -> Caller
说明: 设置函数名称
参数:
- name (str): 函数名称
返回: Caller: Caller对象
源代码 或 在GitHub上查看
python
def name(self, name: str) -> 'Caller':
self._name = name
return self
func description(self, description: str) -> Caller
源代码 或 在GitHub上查看
python
def description(self, description: str) -> 'Caller':
self._description = description
return self
func self () func: F => F
说明: 装饰函数,注册为一个可被AI调用的function call函数
参数:
- func (F): 函数对象
返回: F: 函数对象
源代码 或 在GitHub上查看
python
def __call__(self, func: F) -> F:
global _caller_data
if not self._name:
self._name = func.__name__
sig = inspect.signature(func)
for name, param in sig.parameters.items():
if issubclass(param.annotation, Event) or isinstance(param.annotation, Event):
self.di.event = name
if issubclass(param.annotation, Caller) or isinstance(param.annotation, Caller):
self.di.caller = name
if issubclass(param.annotation, Bot) or isinstance(param.annotation, Bot):
self.di.bot = name
if issubclass(param.annotation, Matcher) or isinstance(param.annotation, Matcher):
self.di.matcher = name
if param.annotation == T_State:
self.di.state = name
for name, param in sig.parameters.items():
if param.default is not inspect.Parameter.empty:
self.default[name] = param.default
if is_coroutine_callable(func):
self.func = func
else:
self.func = async_wrap(func)
if (module := inspect.getmodule(func)):
module_name = module.__name__.split('.')[-1]
else:
module_name = ''
self.module_name = module_name
_caller_data[self.aifc_name] = self
logger.opt(colors=True).debug(f'<y>加载函数 {self.full_name}: {self._description}</y>')
return func
func data(self) -> dict[str, Any]
返回: dict[str, Any]: 函数的json数据
源代码 或 在GitHub上查看
python
def data(self) -> dict[str, Any]:
properties = {key: value.data() for key, value in self._parameters.items()}
if not properties:
properties['placeholder'] = {'type': 'string', 'description': '占位符,用于显示在对话框中'}
return {'type': 'function', 'function': {'name': self.aifc_name, 'description': self._description, 'parameters': {'type': 'object', 'properties': properties}, 'required': [key for key, value in self._parameters.items() if value.default is None]}}
func set_ctx(self, ctx: SessionContext) -> None
说明: 设置依赖注入上下文
参数:
- ctx (SessionContext): 依赖注入上下文
源代码 或 在GitHub上查看
python
def set_ctx(self, ctx: SessionContext) -> None:
ctx.caller = self
self.ctx = ctx
for type_name, arg_name in self.di.model_dump().items():
if arg_name:
self.default[arg_name] = ctx.__getattribute__(type_name)
func with_ctx(self, ctx: SessionContext) -> Caller
说明: 设置依赖注入上下文
参数:
- ctx (SessionContext): 依赖注入上下文
返回: Caller: Caller对象
源代码 或 在GitHub上查看
python
def with_ctx(self, ctx: SessionContext) -> 'Caller':
self.set_ctx(ctx)
return self
async func call(self, *args: Any, **kwargs: Any) -> Any
说明: 调用函数
返回: Any: 函数返回值
源代码 或 在GitHub上查看
python
async def call(self, *args: Any, **kwargs: Any) -> Any:
y, r = await self.pre_check()
if not y:
logger.debug(f'Function {self._name} pre_check failed: {r}')
return r
if self.func is None:
raise ValueError('未注册函数对象')
for name, value in self.default.items():
if name not in kwargs:
kwargs[name] = value
return await self.func(*args, **kwargs)
func short_name(self) -> str
说明: 函数本名
源代码 或 在GitHub上查看
python
@property
def short_name(self) -> str:
return self._name.split('.')[-1]
func aifc_name(self) -> str
说明: AI调用名,没有点
源代码 或 在GitHub上查看
python
@property
def aifc_name(self) -> str:
return self.full_name.replace('.', '-')
func full_name(self) -> str
说明: 完整名
源代码 或 在GitHub上查看
python
@property
def full_name(self) -> str:
return self.module_name + '.' + self._name
func short_info(self) -> str
源代码 或 在GitHub上查看
python
@property
def short_info(self) -> str:
return f'{self.full_name}({self._description})'
func on_function_call(name: str = '', description: str | None = None) -> Caller
参数:
- description: 函数描述,若为None则从函数的docstring中获取
返回: Caller: Caller对象
源代码 或 在GitHub上查看
python
def on_function_call(name: str='', description: str | None=None) -> Caller:
caller = Caller(name=name, description=description)
return caller
func get_function_calls() -> dict[str, Caller]
说明: 获取所有已注册的function call函数
返回: dict[str, Caller]: 所有已注册的function call函数
源代码 或 在GitHub上查看
python
def get_function_calls() -> dict[str, Caller]:
return _caller_data