import{_ as i,c as a,ae as n,o as l}from"./chunks/framework.BzDBnRMZ.js";const g=JSON.parse('{"title":"caller","description":"","frontmatter":{"title":"caller","order":100},"headers":[],"relativePath":"dev/api/plugin/func_call/caller.md","filePath":"zh/dev/api/plugin/func_call/caller.md","lastUpdated":null}'),t={name:"dev/api/plugin/func_call/caller.md"};function h(e,s,k,p,r,d){return l(),a("div",null,s[0]||(s[0]=[n(`
nonebot_plugin_marshoai.plugin.func_call.caller
Caller
__init__(self, name: str = '', description: str | None = None, func_type: str = 'function', no_module_name: bool = False)
def __init__(self, name: str='', description: str | None=None, func_type: str='function', no_module_name: bool=False):
self._name: str = name
'函数名称'
self._description = description
'函数描述'
self._func_type = func_type
'函数类型'
self.no_module_name = no_module_name
'是否不包含模块名'
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
params(self, **kwargs: Any) -> Caller
def params(self, **kwargs: Any) -> 'Caller':
self._parameters.update(kwargs)
return self
permission(self, permission: Permission) -> Caller
def permission(self, permission: Permission) -> 'Caller':
self._permission = self._permission or permission
return self
pre_check(self) -> tuple[bool, str]
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, '')
rule(self, rule: Rule) -> Caller
def rule(self, rule: Rule) -> 'Caller':
self._rule = self._rule and rule
return self
name(self, name: str) -> Caller
说明: 设置函数名称
参数:
- name (str): 函数名称
返回: Caller: Caller对象
def name(self, name: str) -> 'Caller':
self._name = name
return self
description(self, description: str) -> Caller
def description(self, description: str) -> 'Caller':
self._description = description
return self
self () func: F => F
说明: 装饰函数,注册为一个可被AI调用的function call函数
参数:
- func (F): 函数对象
返回: F: 函数对象
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
data(self) -> dict[str, Any]
返回: dict[str, Any]: 函数的json数据
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': self._func_type, '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]}}
set_ctx(self, ctx: SessionContext) -> None
说明: 设置依赖注入上下文
参数:
- ctx (SessionContext): 依赖注入上下文
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)
with_ctx(self, ctx: SessionContext) -> Caller
说明: 设置依赖注入上下文
参数:
- ctx (SessionContext): 依赖注入上下文
返回: Caller: Caller对象
def with_ctx(self, ctx: SessionContext) -> 'Caller':
self.set_ctx(ctx)
return self
call(self, *args: Any, **kwargs: Any) -> Any
说明: 调用函数
返回: Any: 函数返回值
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)
short_name(self) -> str
说明: 函数本名
@property
def short_name(self) -> str:
return self._name.split('.')[-1]
aifc_name(self) -> str
说明: AI调用名,没有点
@property
def aifc_name(self) -> str:
if self.no_module_name:
return self._name
return self.full_name.replace('.', '-')
full_name(self) -> str
说明: 完整名
@property
def full_name(self) -> str:
return self.module_name + '.' + self._name
short_info(self) -> str
@property
def short_info(self) -> str:
return f'{self.full_name}({self._description})'
on_function_call(name: str = '', description: str | None = None, func_type: str = 'function', no_module_name: bool = False) -> Caller
参数:
- name: 函数名称,若为空则从函数的__name__属性获取
- description: 函数描述,若为None则从函数的docstring中获取
- func_type: 函数类型,默认为function,若要注册为 Moonshot AI 的内置函数则为builtin_function
- no_module_name: 是否不包含模块名,当注册为 Moonshot AI 的内置函数时为True
返回: Caller: Caller对象
def on_function_call(name: str='', description: str | None=None, func_type: str='function', no_module_name: bool=False) -> Caller:
caller = Caller(name=name, description=description, func_type=func_type, no_module_name=no_module_name)
return caller
get_function_calls() -> dict[str, Caller]
说明: 获取所有已注册的function call函数
返回: dict[str, Caller]: 所有已注册的function call函数
def get_function_calls() -> dict[str, Caller]:
return _caller_data