模块 nonebot_plugin_marshoai.marsho
async func at_enable()
源代码 或 在GitHub上查看
python
async def at_enable():
return config.marshoai_at@add_usermsg_cmd.handle()
async func add_usermsg(target: MsgTarget, arg: Message = CommandArg())
源代码 或 在GitHub上查看
python
@add_usermsg_cmd.handle()
async def add_usermsg(target: MsgTarget, arg: Message=CommandArg()):
if (msg := arg.extract_plain_text()):
context.append(UserMessage(content=msg).as_dict(), target.id, target.private)
await add_usermsg_cmd.finish('已添加用户消息')@add_assistantmsg_cmd.handle()
async func add_assistantmsg(target: MsgTarget, arg: Message = CommandArg())
源代码 或 在GitHub上查看
python
@add_assistantmsg_cmd.handle()
async def add_assistantmsg(target: MsgTarget, arg: Message=CommandArg()):
if (msg := arg.extract_plain_text()):
context.append(AssistantMessage(content=msg).as_dict(), target.id, target.private)
await add_assistantmsg_cmd.finish('已添加助手消息')@praises_cmd.handle()
async func praises()
源代码 或 在GitHub上查看
python
@praises_cmd.handle()
async def praises():
await praises_cmd.finish(build_praises())@contexts_cmd.handle()
async func contexts(target: MsgTarget)
源代码 或 在GitHub上查看
python
@contexts_cmd.handle()
async def contexts(target: MsgTarget):
backup_context = await get_backup_context(target.id, target.private)
if backup_context:
context.set_context(backup_context, target.id, target.private)
await contexts_cmd.finish(str(context.build(target.id, target.private)))@save_context_cmd.handle()
async func save_context(target: MsgTarget, arg: Message = CommandArg())
源代码 或 在GitHub上查看
python
@save_context_cmd.handle()
async def save_context(target: MsgTarget, arg: Message=CommandArg()):
contexts_data = context.build(target.id, target.private)
if not context:
await save_context_cmd.finish('暂无上下文可以保存')
if (msg := arg.extract_plain_text()):
await save_context_to_json(msg, contexts_data, 'contexts')
await save_context_cmd.finish('已保存上下文')@load_context_cmd.handle()
async func load_context(target: MsgTarget, arg: Message = CommandArg())
源代码 或 在GitHub上查看
python
@load_context_cmd.handle()
async def load_context(target: MsgTarget, arg: Message=CommandArg()):
if (msg := arg.extract_plain_text()):
await get_backup_context(target.id, target.private)
context.set_context(await load_context_from_json(msg, 'contexts'), target.id, target.private)
await load_context_cmd.finish('已加载并覆盖上下文')@resetmem_cmd.handle()
async func resetmem(target: MsgTarget)
源代码 或 在GitHub上查看
python
@resetmem_cmd.handle()
async def resetmem(target: MsgTarget):
if [target.id, target.private] not in target_list:
target_list.append([target.id, target.private])
context.reset(target.id, target.private)
await resetmem_cmd.finish('上下文已重置')@changemodel_cmd.handle()
async func changemodel(arg: Message = CommandArg())
源代码 或 在GitHub上查看
python
@changemodel_cmd.handle()
async def changemodel(arg: Message=CommandArg()):
global model_name
if (model := arg.extract_plain_text()):
model_name = model
await changemodel_cmd.finish('已切换')@nickname_cmd.handle()
async func nickname(event: Event, name = None)
源代码 或 在GitHub上查看
python
@nickname_cmd.handle()
async def nickname(event: Event, name=None):
nicknames = await get_nicknames()
user_id = event.get_user_id()
if not name:
if user_id not in nicknames:
await nickname_cmd.finish('你未设置昵称')
await nickname_cmd.finish('你的昵称为：' + str(nicknames[user_id]))
if name == 'reset':
await set_nickname(user_id, '')
await nickname_cmd.finish('已重置昵称')
else:
if len(name) > config.marshoai_nickname_limit:
await nickname_cmd.finish('昵称超出长度限制：' + str(config.marshoai_nickname_limit))
await set_nickname(user_id, name)
await nickname_cmd.finish('已设置昵称为：' + name)@refresh_data_cmd.handle()
async func refresh_data()
源代码 或 在GitHub上查看
python
@refresh_data_cmd.handle()
async def refresh_data():
await refresh_nickname_json()
await refresh_praises_json()
await refresh_data_cmd.finish('已刷新数据')@marsho_help_cmd.handle()
async func marsho_help()
源代码 或 在GitHub上查看
python
@marsho_help_cmd.handle()
async def marsho_help():
await marsho_help_cmd.finish(metadata.usage)@marsho_status_cmd.handle()
async func marsho_status(bot: Bot)
源代码 或 在GitHub上查看
python
@marsho_status_cmd.handle()
async def marsho_status(bot: Bot):
await marsho_status_cmd.finish(f'当前适配器：{bot.adapter.get_name()}\n当前使用的模型：{model_name}\n当前支持图片的模型：{str(SUPPORT_IMAGE_MODELS + config.marshoai_additional_image_models)}')@marsho_at.handle()@marsho_cmd.handle()
async func marsho(target: MsgTarget, event: Event, bot: Bot, state: T_State, matcher: Matcher, text: Optional[UniMsg] = None)
源代码 或 在GitHub上查看
python
@marsho_at.handle()
@marsho_cmd.handle()
async def marsho(target: MsgTarget, event: Event, bot: Bot, state: T_State, matcher: Matcher, text: Optional[UniMsg]=None):
global target_list
if event.get_message().extract_plain_text() and (not text and event.get_message().extract_plain_text() != config.marshoai_default_name):
text = event.get_message()
if not text:
await marsho_cmd.finish(INTRODUCTION)
try:
user_id = event.get_user_id()
nicknames = await get_nicknames()
user_nickname = nicknames.get(user_id, '')
if user_nickname != '':
nickname_prompt = f'\n*此消息的说话者id为:{user_id}，名字为:{user_nickname}*'
else:
nickname_prompt = ''
if config.marshoai_enforce_nickname:
await UniMessage('※你未设置自己的昵称。你**必须**使用「nickname [昵称]」命令设置昵称后才能进行对话。').send()
return
if config.marshoai_enable_nickname_tip:
await UniMessage('※你未设置自己的昵称。推荐使用「nickname [昵称]」命令设置昵称来获得个性化(可能）回答。').send()
is_support_image_model = model_name.lower() in SUPPORT_IMAGE_MODELS + config.marshoai_additional_image_models
is_reasoning_model = model_name.lower() in NO_SYSPROMPT_MODELS
usermsg = [] if is_support_image_model else ''
for i in text:
if i.type == 'text':
if is_support_image_model:
usermsg += [TextContentItem(text=i.data['text'] + nickname_prompt)]
else:
usermsg += str(i.data['text'] + nickname_prompt)
elif i.type == 'image':
if is_support_image_model:
usermsg.append(ImageContentItem(image_url=ImageUrl(url=str(await get_image_b64(i.data['url'])))))
elif config.marshoai_enable_support_image_tip:
await UniMessage('*此模型不支持图片处理或管理员未启用此模型的图片支持。图片将被忽略。').send()
backup_context = await get_backup_context(target.id, target.private)
if backup_context:
context.set_context(backup_context, target.id, target.private)
logger.info(f'已恢复会话 {target.id} 的上下文备份~')
context_msg = context.build(target.id, target.private)
if not is_reasoning_model:
context_msg = [get_prompt()] + context_msg
tools_lists = tools.tools_list + list(map(lambda v: v.data(), get_function_calls().values()))
response = await make_chat(client=client, model_name=model_name, msg=context_msg + [UserMessage(content=usermsg)], tools=tools_lists if tools_lists else None)
choice = response.choices[0]
if choice.message.tool_calls != None:
choice['finish_reason'] = CompletionsFinishReason.TOOL_CALLS
if choice['finish_reason'] == CompletionsFinishReason.STOPPED:
context.append(UserMessage(content=usermsg).as_dict(), target.id, target.private)
context.append(choice.message.as_dict(), target.id, target.private)
if [target.id, target.private] not in target_list:
target_list.append([target.id, target.private])
if config.marshoai_enable_richtext_parse:
await (await parse_richtext(str(choice.message.content))).send(reply_to=True)
else:
await UniMessage(str(choice.message.content)).send(reply_to=True)
elif choice['finish_reason'] == CompletionsFinishReason.CONTENT_FILTERED:
await UniMessage('*已被内容过滤器过滤。请调整聊天内容后重试。').send(reply_to=True)
return
elif choice['finish_reason'] == CompletionsFinishReason.TOOL_CALLS:
tool_msg = []
while choice.message.tool_calls != None:
tool_msg.append(AssistantMessage(tool_calls=response.choices[0].message.tool_calls))
for tool_call in choice.message.tool_calls:
if isinstance(tool_call, ChatCompletionsToolCall):
try:
function_args = json.loads(tool_call.function.arguments)
except json.JSONDecodeError:
function_args = json.loads(tool_call.function.arguments.replace("'", '"'))
if 'placeholder' in function_args:
del function_args['placeholder']
logger.info(f"调用函数 {tool_call.function.name.replace('-', '.')}\n参数:" + '\n'.join([f'{k}={v}' for k, v in function_args.items()]))
await UniMessage(f"调用函数 {tool_call.function.name.replace('-', '.')}\n参数:" + '\n'.join([f'{k}={v}' for k, v in function_args.items()])).send()
if tools.has_function(tool_call.function.name):
logger.debug(f'调用工具函数 {tool_call.function.name}')
func_return = await tools.call(tool_call.function.name, function_args)
elif (caller := get_function_calls().get(tool_call.function.name)):
logger.debug(f'调用插件函数 {caller.full_name}')
func_return = await caller.with_ctx(SessionContext(bot=bot, event=event, state=state, matcher=matcher)).call(**function_args)
else:
logger.error(f"未找到函数 {tool_call.function.name.replace('-', '.')}")
func_return = f"未找到函数 {tool_call.function.name.replace('-', '.')}"
tool_msg.append(ToolMessage(tool_call_id=tool_call.id, content=func_return).as_dict())
request_msg = context_msg + [UserMessage(content=usermsg).as_dict()] + tool_msg
response = await make_chat(client=client, model_name=model_name, msg=request_msg, tools=tools_lists if tools_lists else None)
choice = response.choices[0]
if choice.message.tool_calls != None:
choice['finish_reason'] = CompletionsFinishReason.TOOL_CALLS
if choice['finish_reason'] == CompletionsFinishReason.STOPPED:
context.append(UserMessage(content=usermsg).as_dict(), target.id, target.private)
context.append(choice.message.as_dict(), target.id, target.private)
if config.marshoai_enable_richtext_parse:
await (await parse_richtext(str(choice.message.content))).send(reply_to=True)
else:
await UniMessage(str(choice.message.content)).send(reply_to=True)
else:
await marsho_cmd.finish(f"意外的完成原因:{choice['finish_reason']}")
else:
await marsho_cmd.finish(f"意外的完成原因:{choice['finish_reason']}")
except Exception as e:
await UniMessage(str(e) + suggest_solution(str(e))).send()
traceback.print_exc()
return@poke_notify.handle()
async func poke(event: Event)
源代码 或 在GitHub上查看
python
@poke_notify.handle()
async def poke(event: Event):
user_id = event.get_user_id()
nicknames = await get_nicknames()
user_nickname = nicknames.get(user_id, '')
try:
if config.marshoai_poke_suffix != '':
response = await make_chat(client=client, model_name=model_name, msg=[get_prompt(), UserMessage(content=f'*{user_nickname}{config.marshoai_poke_suffix}')])
choice = response.choices[0]
if choice['finish_reason'] == CompletionsFinishReason.STOPPED:
await UniMessage(' ' + str(choice.message.content)).send(at_sender=True)
except Exception as e:
await UniMessage(str(e) + suggest_solution(str(e))).send()
traceback.print_exc()
returnvar text
说明: type: ignore
默认值:
event.get_message()
var request_msg
说明: type: ignore
默认值:
context_msg + [UserMessage(content=usermsg).as_dict()] + tool_msg