nonebot_plugin_marshoai.handler MarshoHandler __init__(self, client: AsyncOpenAI, context: MarshoContext) def __init__(self, client: AsyncOpenAI, context: MarshoContext):
self.client = client
self.context = context
self.bot: Bot = current_bot.get()
self.event: Event = current_event.get()
self.matcher: Matcher = current_matcher.get()
self.message_id: str = UniMessage.get_message_id(self.event)
self.target = UniMessage.get_target(self.event)process_user_input(self, user_input: UniMsg, model_name: str) -> Union[str, list] 说明: 处理用户输入为可输入 API 的格式，并添加昵称提示
async def process_user_input(self, user_input: UniMsg, model_name: str) -> Union[str, list]:
is_support_image_model = model_name.lower() in SUPPORT_IMAGE_MODELS + config.marshoai_additional_image_models
usermsg = [] if is_support_image_model else ''
user_nickname = await get_nickname_by_user_id(self.event.get_user_id())
if user_nickname:
nickname_prompt = f'\\n此消息的说话者为: {user_nickname}'
else:
nickname_prompt = ''
for i in user_input:
if i.type == 'text':
if is_support_image_model:
usermsg += [TextContentItem(text=i.data['text'] + nickname_prompt).as_dict()]
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'])))).as_dict())
logger.info(f"输入图片 {i.data['url']}")
elif config.marshoai_enable_support_image_tip:
await UniMessage('*此模型不支持图片处理或管理员未启用此模型的图片支持。图片将被忽略。').send()
return usermsghandle_single_chat(self, user_message: Union[str, list], model_name: str, tools_list: list, tool_message: Optional[list] = None) -> ChatCompletion 说明: 处理单条聊天
async def handle_single_chat(self, user_message: Union[str, list], model_name: str, tools_list: list, tool_message: Optional[list]=None) -> ChatCompletion:
context_msg = get_prompt(model_name) + self.context.build(self.target.id, self.target.private)
response = await make_chat_openai(client=self.client, msg=context_msg + [UserMessage(content=user_message).as_dict()] + (tool_message if tool_message else []), model_name=model_name, tools=tools_list if tools_list else None)
return responsehandle_function_call(self, completion: ChatCompletion, user_message: Union[str, list], model_name: str, tools_list: list) async def handle_function_call(self, completion: ChatCompletion, user_message: Union[str, list], model_name: str, tools_list: list):
tool_msg = []
choice = completion.choices[0]
tool_calls = choice.message.tool_calls
tool_msg.append(choice.message)
for tool_call in tool_calls:
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 (caller := get_function_calls().get(tool_call.function.name)):
logger.debug(f'调用插件函数 {caller.full_name}')
func_return = await caller.with_ctx(SessionContext(bot=self.bot, event=self.event, matcher=self.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())
return await self.handle_common_chat(user_message=user_message, model_name=model_name, tools_list=tools_list, tool_message=tool_msg)handle_common_chat(self, user_message: Union[str, list], model_name: str, tools_list: list, stream: bool = False, tool_message: Optional[list] = None) -> Union[Tuple[UserMessage, ChatCompletionMessage], None] 说明: 处理一般聊天
async def handle_common_chat(self, user_message: Union[str, list], model_name: str, tools_list: list, stream: bool=False, tool_message: Optional[list]=None) -> Union[Tuple[UserMessage, ChatCompletionMessage], None]:
global target_list
if stream:
raise NotImplementedError
response = await self.handle_single_chat(user_message=user_message, model_name=model_name, tools_list=tools_list, tool_message=tool_message)
choice = response.choices[0]
if choice.message.tool_calls is not None and config.marshoai_fix_toolcalls:
choice.finish_reason = 'tool_calls'
logger.info(f'完成原因：{choice.finish_reason}')
if choice.finish_reason == CompletionsFinishReason.STOPPED:
choice_msg_content, choice_msg_thinking, choice_msg_after = extract_content_and_think(choice.message)
if choice_msg_thinking and config.marshoai_send_thinking:
await UniMessage('思维链：\\n' + choice_msg_thinking).send()
if [self.target.id, self.target.private] not in target_list:
target_list.append([self.target.id, self.target.private])
if config.marshoai_enable_richtext_parse:
await (await parse_richtext(str(choice_msg_content))).send(reply_to=True)
else:
await UniMessage(str(choice_msg_content)).send(reply_to=True)
return (UserMessage(content=user_message), choice_msg_after)
elif choice.finish_reason == CompletionsFinishReason.CONTENT_FILTERED:
await UniMessage('*已被内容过滤器过滤。请调整聊天内容后重试。').send(reply_to=True)
return None
elif choice.finish_reason == CompletionsFinishReason.TOOL_CALLS:
return await self.handle_function_call(response, user_message, model_name, tools_list)
else:
await UniMessage(f'意外的完成原因:{choice.finish_reason}').send()
return None