repo_description: MaiSaka, an LLM-based intelligent agent, is a digital lifeform devoted to understanding you and interacting in the style of a real human. She does not pursue perfection, nor does she seek efficiency; instead, she values warmth, authenticity, and genuine connection.
repo_topics: agent chat chatbot llm llm-agent llms python qq-bot qqbot
code: """Maisaka 对话循环服务。"""
  
  from dataclasses import dataclass
  from datetime import datetime
  from typing import Any, List, Optional, Sequence
  
  import asyncio
  
  from rich.console import RenderableType
  from src.common.data_models.llm_service_data_models import LLMGenerationOptions
  from src.common.logger import get_logger
  from src.common.prompt_i18n import load_prompt
  from src.common.utils.utils_session import SessionUtils
  from src.config.config import global_config
  from src.core.tooling import ToolAvailabilityContext, ToolRegistry
  from src.llm_models.model_client.base_client import BaseClient
  from src.llm_models.payload_content.message import Message, MessageBuilder, RoleType
  from src.llm_models.payload_content.resp_format import RespFormat
  from src.llm_models.payload_content.tool_option import ToolCall, ToolDefinitionInput, ToolOption, normalize_tool_options
  from src.plugin_runtime.hook_payloads import (
      deserialize_prompt_messages,
      deserialize_tool_calls,
      serialize_prompt_messages,
      serialize_tool_calls,
      serialize_tool_definitions,
  )
  from src.plugin_runtime.hook_schema_utils import build_object_schema
  from src.plugin_runtime.host.hook_spec_registry import HookSpec, HookSpecRegistry
  from src.services.llm_service import LLMServiceClient
  
  from .builtin_tool import get_builtin_tools
  from .context_messages import (
      AssistantMessage,
      LLMContextMessage,
      ToolResultMessage,
      build_llm_message_from_context,
  )
  from .history_utils import drop_orphan_tool_results
  from .display.prompt_cli_renderer import PromptCLIVisualizer
  from .visual_mode_utils import resolve_enable_visual_planner
  
  TIMING_GATE_TOOL_NAMES = {"continue", "no_reply", "wait"}
  
  
  @dataclass(slots=True)
  class ChatResponse:
      """LLM 对话循环单步响应。"""
  
      content: Optional[str]
      tool_calls: List[ToolCall]
      request_messages: List[Message]
      raw_message: AssistantMessage
      selected_history_count: int
      tool_count: int
      prompt_tokens: int
      built_message_count: int
      completion_tokens: int
      total_tokens: int
      prompt_section: Optional[RenderableType] = None
  
  
  logger = get_logger("maisaka_chat_loop")
  
  
  def register_maisaka_hook_specs(registry: HookSpecRegistry) -> List[HookSpec]:
      """注册 Maisaka 规划器内置 Hook 规格。
  
      Args:
          registry: 目标 Hook 规格注册中心。
  
      Returns:
          List[HookSpec]: 实际注册的 Hook 规格列表。
      """
  
      return registry.register_hook_specs(
          [
              HookSpec(
                  name="maisaka.planner.before_request",
                  description="在 Maisaka 向模型发起规划请求前触发，可改写消息窗口与工具定义。",
                  parameters_schema=build_object_schema(
                      {
                          "messages": {
                              "type": "array",
                              "description": "即将发给模型的 PromptMessage 列表。",
                          },
                          "tool_definitions": {
                              "type": "array",
                              "description": "当前候选工具定义列表。",
                          },
                          "selected_history_count": {
                              "type": "integer",
                              "description": "当前选中的上下文消息数量。",
                          },
                          "built_message_count": {
                              "type": "integer",
                              "description": "实际发送给模型的消息数量。",
                          },
                          "selection_reason": {
                              "type": "string",
                              "description": "上下文选择说明。",
                          },
                          "session_id": {
                              "type": "string",
                              "description": "当前会话 ID。",
                          },
                      },
                      required=[
                          "messages",
                          "tool_definitions",
                          "selected_history_count",
                          "built_message_count",
                          "selection_reason",
                          "session_id",
                      ],
                  ),
                  default_timeout_ms=6000,
                  allow_abort=False,
                  allow_kwargs_mutation=True,
              ),
              HookSpec(
                  name="maisaka.planner.after_response",
                  description="在 Maisaka 收到模型响应后触发，可调整文本结果与工具调用列表。",
                  parameters_schema=build_object_schema(
                      {
                          "response": {
                              "type": "string",
                              "description": "模型返回的文本内容。",
                          },
                          "tool_calls": {
                              "type": "array",
                              "description": "模型返回的工具调用列表。",
                          },
                          "selected_history_count": {
                              "type": "integer",
                              "description": "当前选中的上下文消息数量。",
                          },
                          "built_message_count": {
                              "type": "integer",
                              "description": "实际发送给模型的消息数量。",
                          },
                          "selection_reason": {
                              "type": "string",
                              "description": "上下文选择说明。",
                          },
                          "session_id": {
                              "type": "string",
                              "description": "当前会话 ID。",
                          },
                          "prompt_tokens": {
                              "type": "integer",
                              "description": "输入 Token 数。",
                          },
                          "completion_tokens": {
                              "type": "integer",
                              "description": "输出 Token 数。",
                          },
                          "total_tokens": {
                              "type": "integer",
                              "description": "总 Token 数。",
                          },
                      },
                      required=[
                          "response",
                          "tool_calls",
                          "selected_history_count",
                          "built_message_count",
                          "selection_reason",
                          "session_id",
                          "prompt_tokens",
                          "completion_tokens",
                          "total_tokens",
                      ],
                  ),
                  default_timeout_ms=6000,
                  allow_abort=False,
                  allow_kwargs_mutation=True,
              ),
          ]
      )
  
  
  class MaisakaChatLoopService:
      """负责 Maisaka 主对话循环、系统提示词和终端渲染。"""
  
      def __init__(
          self,
          chat_system_prompt: Optional[str] = None,
          session_id: Optional[str] = None,
          is_group_chat: Optional[bool] = None,
          max_tokens: int = 2048,
      ) -> None:
          """初始化 Maisaka 对话循环服务。
  
          Args:
              chat_system_prompt: 可选的系统提示词。
              session_id: 当前会话 ID，用于匹配会话级额外提示。
              is_group_chat: 当前会话是否为群聊。
              max_tokens: 规划器最大输出长度。
          """
  
          self._max_tokens = max_tokens
          self._is_group_chat = is_group_chat
          self._session_id = session_id or ""
          self._extra_tools: List[ToolOption] = []
          self._interrupt_flag: asyncio.Event | None = None
          self._tool_registry: ToolRegistry | None = None
          self._prompts_loaded = chat_system_prompt is not None
          self._prompt_load_lock = asyncio.Lock()
          self._personality_prompt = self._build_personality_prompt()
          if chat_system_prompt is None:
              self._chat_system_prompt = f"{self._personality_prompt}\n\nYou are a helpful AI assistant."
          else:
              self._chat_system_prompt = chat_system_prompt
          self._llm_chat = LLMServiceClient(task_name="planner", request_type="maisaka_planner")
  
      @property
      def personality_prompt(self) -> str:
          """返回当前人格提示词。"""
  
          return self._personality_prompt
  
      @staticmethod
      def _get_runtime_manager() -> Any:
          """获取插件运行时管理器。
  
          Returns:
              Any: 插件运行时管理器单例。
          """
  
          from src.plugin_runtime.integration import get_plugin_runtime_manager
  
          return get_plugin_runtime_manager()
  
      @staticmethod
      def _coerce_int(value: Any, default: int) -> int:
          """将任意值安全转换为整数。
  
          Args:
              value: 待转换的输入值。
              default: 转换失败时的默认值。
  
          Returns:
              int: 转换后的整数结果。
          """
  
          try:
              return int(value)
          except (TypeError, ValueError):
              return default
  
      def _build_personality_prompt(self) -> str:
          """构造人格提示词。"""
  
          try:
              bot_name = global_config.bot.nickname
              if global_config.bot.alias_names:
                  bot_nickname = f", also known as {','.join(global_config.bot.alias_names)}"
              else:
                  bot_nickname = ""
  
              prompt_personality = global_config.personality.personality
  
              return f"Your name is {bot_name}{bot_nickname}; persona: {prompt_personality};"
          except Exception:
              return "Your name is MaiMai; persona: lively and cute AI assistant."
  
      async def ensure_chat_prompt_loaded(self, tools_section: str = "") -> None:
          """确保主聊天提示词已经加载完成。
  
          Args:
              tools_section: 额外注入到提示词中的工具说明片段。
          """
          async with self._prompt_load_lock:
              try:
                  self._chat_system_prompt = load_prompt("maisaka_chat", **self.build_prompt_template_context(tools_section))
              except Exception:
                  self._chat_system_prompt = f"{self._personality_prompt}\n\nYou are a helpful AI assistant."
  
              self._prompts_loaded = True
  
      def build_prompt_template_context(self, tools_section: str = "") -> dict[str, str]:
          """构造 Maisaka prompt 模板的公共渲染参数。"""
  
          return {
              "bot_name": global_config.bot.nickname,
              "file_tools_section": tools_section,
              "group_chat_attention_block": self._build_group_chat_attention_block(),
              "identity": self._personality_prompt,
              "time_block": self._build_time_block(),
          }
  
      @staticmethod
      def _build_time_block() -> str:
          """构建当前时间提示块。"""
  
          return f"当前时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
  
      def _build_group_chat_attention_block(self) -> str:
          """构建当前聊天场景下的额外注意事项块。"""
  
          prompt_lines: List[str] = []
  
          if self._is_group_chat is True:
              if group_chat_prompt := str(global_config.chat.group_chat_prompt or "").strip():
                  prompt_lines.append(f"通用注意事项：\n{group_chat_prompt}")
          elif self._is_group_chat is False:
              if private_chat_prompt := str(global_config.chat.private_chat_prompts or "").strip():
                  prompt_lines.append(f"通用注意事项：\n{private_chat_prompt}")
  
          if self._session_id:
              if chat_prompt := self._get_chat_prompt_for_chat(self._session_id, self._is_group_chat).strip():
                  prompt_lines.append(f"当前聊天额外注意事项：\n{chat_prompt}")
  
          if not prompt_lines:
              return ""
  
          return "在该聊天中的注意事项：\n" + "\n\n".join(prompt_lines) + "\n"
  
      @staticmethod
      def _get_chat_prompt_for_chat(chat_id: str, is_group_chat: Optional[bool]) -> str:
          """根据聊天流 ID 获取匹配的额外提示。"""
  
          if not global_config.chat.chat_prompts:
              return ""
  
          for chat_prompt_item in global_config.chat.chat_prompts:
              if hasattr(chat_prompt_item, "platform"):
                  platform = str(chat_prompt_item.platform or "").strip()
                  item_id = str(chat_prompt_item.item_id or "").strip()
                  rule_type = str(chat_prompt_item.rule_type or "").strip()
                  prompt_content = str(chat_prompt_item.prompt or "").strip()
              elif isinstance(chat_prompt_item, str):
                  parts = chat_prompt_item.split(":", 3)
                  if len(parts) != 4:
                      continue
  
                  platform, item_id, rule_type, prompt_content = parts
                  platform = platform.strip()
                  item_id = item_id.strip()
                  rule_type = rule_type.strip()
                  prompt_content = prompt_content.strip()
              else:
                  continue
  
              if not platform or not item_id or not prompt_content:
                  continue
  
              if rule_type == "group":
                  config_is_group = True
                  config_chat_id = SessionUtils.calculate_session_id(platform, group_id=item_id)
              elif rule_type == "private":
                  config_is_group = False
                  config_chat_id = SessionUtils.calculate_session_id(platform, user_id=item_id)
              else:
                  continue
  
              if is_group_chat is not None and config_is_group != is_group_chat:
                  continue
  
              if config_chat_id == chat_id:
                  logger.debug(f"匹配到 Maisaka 聊天额外提示，chat_id: {chat_id}, prompt: {prompt_content[:50]}...")
                  return prompt_content
  
          return ""
  
      def set_extra_tools(self, tools: Sequence[ToolDefinitionInput]) -> None:
          """设置额外工具定义。
  
          Args:
              tools: 兼容旧接口的额外工具定义列表。
          """
  
          self._extra_tools = normalize_tool_options(list(tools)) or []
  
      def set_tool_registry(self, tool_registry: ToolRegistry | None) -> None:
          """设置统一工具注册表。
  
          Args:
              tool_registry: 统一工具注册表；传入 ``None`` 时退回旧工具列表模式。
          """
  
          self._tool_registry = tool_registry
  
      def set_interrupt_flag(self, interrupt_flag: asyncio.Event | None) -> None:
          """设置当前 planner 请求使用的中断标记。"""
          self._interrupt_flag = interrupt_flag
  
      def _build_request_messages(
          self,
          selected_history: List[LLMContextMessage],
          *,
          enable_visual_message: bool,
          injected_user_messages: Sequence[str] | None = None,
          system_prompt: Optional[str] = None,
      ) -> List[Message]:
          """构造发给大模型的消息列表。
  
          Args:
              selected_history: 已选中的上下文消息列表。
  
          Returns:
              List[Message]: 发送给大模型的消息列表。
          """
  
          messages: List[Message] = []
          system_msg = MessageBuilder().set_role(RoleType.System)
          system_msg.add_text_content(system_prompt if system_prompt is not None else self._chat_system_prompt)
          messages.append(system_msg.build())
  
          for msg in selected_history:
              llm_message = build_llm_message_from_context(
                  msg,
                  enable_visual_message=enable_visual_message,
              )
              if llm_message is not None:
                  messages.append(llm_message)
  
          normalized_injected_messages: List[Message] = []
          for injected_message in injected_user_messages or []:
              normalized_message = str(injected_message or "").strip()
              if not normalized_message:
                  continue
              normalized_injected_messages.append(
                  MessageBuilder()
                  .set_role(RoleType.User)
                  .add_text_content(normalized_message)
                  .build()
              )
  
          if normalized_injected_messages:
              insertion_index = self._resolve_injected_user_messages_insertion_index(messages)
              messages[insertion_index:insertion_index] = normalized_injected_messages
  
          return messages
  
      @staticmethod
      def _resolve_injected_user_messages_insertion_index(messages: Sequence[Message]) -> int:
          """计算 injected meta user messages 在请求中的插入位置。
  
          规则与 deferred attachment 更接近：
          - 从尾部向前寻找最近的 stopping point；
          - stopping point 为 assistant 消息或 tool 结果消息；
          - 找到后插入到其后面；
          - 若不存在 stopping point，则退回到 system 消息之后。
          """
  
          for index in range(len(messages) - 1, -1, -1):
              message = messages[index]
              if message.role in {RoleType.Assistant, RoleType.Tool}:
                  return index + 1
  
          if messages and messages[0].role == RoleType.System:
              return 1
          return 0
  
      async def chat_loop_step(
          self,
          chat_history: List[LLMContextMessage],
          *,
          injected_user_messages: Sequence[str] | None = None,
          request_kind: str = "planner",
          response_format: RespFormat | None = None,
          tool_definitions: Sequence[ToolDefinitionInput] | None = None,
      ) -> ChatResponse:
          """执行一轮 Maisaka 规划器请求。
  
          Args:
              chat_history: 当前对话历史。
  
          Returns:
              ChatResponse: 本轮规划器返回结果。
          """
  
          if not self._prompts_loaded:
              await self.ensure_chat_prompt_loaded()
          enable_visual_message = self._resolve_enable_visual_message(request_kind)
          selected_history, selection_reason = self.select_llm_context_messages(
              chat_history,
              request_kind=request_kind,
              enable_visual_message=enable_visual_message,
          )
          built_messages = self._build_request_messages(
              selected_history,
              enable_visual_message=enable_visual_message,
              injected_user_messages=injected_user_messages,
          )
  
          def message_factory(_client: BaseClient) -> List[Message]:
              """返回当前轮次已经构建好的请求消息。
  
              Args:
                  _client: 当前模型客户端；此处不依赖客户端能力。
  
              Returns:
                  List[Message]: 已经构建好的消息列表。
              """
  
              del _client
              return built_messages
  
          all_tools: List[ToolDefinitionInput]
          if tool_definitions is not None:
              all_tools = list(tool_definitions)
          elif self._tool_registry is not None:
              tool_specs = await self._tool_registry.list_tools(
                  ToolAvailabilityContext(
                      session_id=self._session_id,
                      stream_id=self._session_id,
                      is_group_chat=self._is_group_chat,
                  )
              )
              all_tools = [tool_spec.to_llm_definition() for tool_spec in tool_specs]
          else:
              all_tools = [*get_builtin_tools(), *self._extra_tools]
  
          before_request_result = await self._get_runtime_manager().invoke_hook(
              "maisaka.planner.before_request",
              messages=serialize_prompt_messages(built_messages),
              tool_definitions=serialize_tool_definitions(all_tools),
              selected_history_count=len(selected_history),
              built_message_count=len(built_messages),
              selection_reason=selection_reason,
              session_id=self._session_id,
          )
          before_request_kwargs = before_request_result.kwargs
          raw_messages = before_request_kwargs.get("messages")
          if isinstance(raw_messages, list):
              try:
                  built_messages = deserialize_prompt_messages(raw_messages)
              except Exception as exc:
                  logger.warning(f"Hook maisaka.planner.before_request 返回的 messages 无法反序列化，已忽略: {exc}")
          raw_tool_definitions = before_request_kwargs.get("tool_definitions")
          if isinstance(raw_tool_definitions, list):
              all_tools = [item for item in raw_tool_definitions if isinstance(item, dict)]
  
          prompt_section: RenderableType | None = None
          if global_config.debug.show_maisaka_thinking:
              prompt_section = PromptCLIVisualizer.build_prompt_section(
                  built_messages,
                  category="planner" if request_kind != "timing_gate" else "timing_gate",
                  chat_id=self._session_id,
                  request_kind=request_kind,
                  selection_reason=selection_reason,
                  folded=global_config.debug.fold_maisaka_thinking,
                  tool_definitions=list(all_tools),
              )
  
          generation_result = await self._llm_chat.generate_response_with_messages(
              message_factory=message_factory,
              options=LLMGenerationOptions(
                  tool_options=all_tools if all_tools else None,
                  max_tokens=self._max_tokens,
                  response_format=response_format,
                  interrupt_flag=self._interrupt_flag,
              ),
          )
  
          final_response = generation_result.response or ""
          final_tool_calls = list(generation_result.tool_calls or [])
          after_response_result = await self._get_runtime_manager().invoke_hook(
              "maisaka.planner.after_response",
              response=final_response,
              tool_calls=serialize_tool_calls(final_tool_calls),
              selected_history_count=len(selected_history),
              built_message_count=len(built_messages),
              selection_reason=selection_reason,
              session_id=self._session_id,
              prompt_tokens=generation_result.prompt_tokens,
              completion_tokens=generation_result.completion_tokens,
              total_tokens=generation_result.total_tokens,
          )
          after_response_kwargs = after_response_result.kwargs
          if "response" in after_response_kwargs:
              final_response = str(after_response_kwargs.get("response") or "")
          raw_tool_calls = after_response_kwargs.get("tool_calls")
          if isinstance(raw_tool_calls, list):
              try:
                  final_tool_calls = deserialize_tool_calls(raw_tool_calls)
              except Exception as exc:
                  logger.warning(f"Hook maisaka.planner.after_response 返回的 tool_calls 无法反序列化，已忽略: {exc}")
          prompt_tokens = self._coerce_int(after_response_kwargs.get("prompt_tokens"), generation_result.prompt_tokens)
          completion_tokens = self._coerce_int(
              after_response_kwargs.get("completion_tokens"),
              generation_result.completion_tokens,
          )
          total_tokens = self._coerce_int(after_response_kwargs.get("total_tokens"), generation_result.total_tokens)
  
          raw_message = AssistantMessage(
              content=final_response,
              timestamp=datetime.now(),
              tool_calls=final_tool_calls,
          )
          return ChatResponse(
              content=final_response or None,
              tool_calls=final_tool_calls,
              request_messages=list(built_messages),
              raw_message=raw_message,
              selected_history_count=len(selected_history),
              tool_count=len(all_tools),
              prompt_tokens=prompt_tokens,
              built_message_count=len(built_messages),
              completion_tokens=completion_tokens,
              total_tokens=total_tokens,
              prompt_section=prompt_section,
          )
  
      @staticmethod
      def select_llm_context_messages(
          chat_history: List[LLMContextMessage],
          *,
          enable_visual_message: Optional[bool] = None,
          request_kind: str = "planner",
          max_context_size: Optional[int] = None,
      ) -> tuple[List[LLMContextMessage], str]:
          """选择LLM上下文消息"""
  
          filtered_history = MaisakaChatLoopService._filter_history_for_request_kind(
              chat_history,
              request_kind=request_kind,
          )
          effective_context_size = max(1, int(max_context_size or global_config.chat.max_context_size))
          selected_indices: List[int] = []
          counted_message_count = 0
  
          active_enable_visual_message = (
              enable_visual_message
              if enable_visual_message is not None
              else MaisakaChatLoopService._resolve_enable_visual_message(request_kind)
          )
  
          for index in range(len(filtered_history) - 1, -1, -1):
              message = filtered_history[index]
              if (
                  build_llm_message_from_context(
                      message,
                      enable_visual_message=active_enable_visual_message,
                  )
                  is None
              ):
                  continue
  
              selected_indices.append(index)
              if message.count_in_context:
                  counted_message_count += 1
                  if counted_message_count >= effective_context_size:
                      break
  
          if not selected_indices:
              return [], "实际发送 0 条消息（tool 0 条，普通消息 0 条）"
  
          selected_indices.reverse()
          selected_history = [filtered_history[index] for index in selected_indices]
          selected_history, _ = MaisakaChatLoopService._hide_early_assistant_messages(selected_history)
          selected_history, _ = drop_orphan_tool_results(selected_history)
          tool_message_count = sum(1 for message in selected_history if isinstance(message, ToolResultMessage))
          normal_message_count = len(selected_history) - tool_message_count
          selection_reason = (
              f"实际发送 {len(selected_history)} 条消息"
              f"|消息 {normal_message_count} 条|tool {tool_message_count} 条"
          )
          return (
              selected_history,
              selection_reason,
          )
  
      @staticmethod
      def _filter_history_for_request_kind(
          selected_history: List[LLMContextMessage],
          *,
          request_kind: str,
      ) -> List[LLMContextMessage]:
          """按请求类型过滤不应暴露的历史工具链。"""
  
          if request_kind != "planner":
              return selected_history
  
          filtered_history: List[LLMContextMessage] = []
          for message in selected_history:
              if isinstance(message, ToolResultMessage) and message.tool_name in TIMING_GATE_TOOL_NAMES:
                  continue
  
              if isinstance(message, AssistantMessage) and message.tool_calls:
                  kept_tool_calls = [
                      tool_call
                      for tool_call in message.tool_calls
                      if tool_call.func_name not in TIMING_GATE_TOOL_NAMES
                  ]
                  if not kept_tool_calls:
                      continue
                  if len(kept_tool_calls) != len(message.tool_calls):
                      filtered_history.append(
                          AssistantMessage(
                              content=message.content,
                              timestamp=message.timestamp,
                              tool_calls=kept_tool_calls,
                              source_kind=message.source_kind,
                          )
                      )
                      continue
  
              filtered_history.append(message)
  
          return filtered_history
  
      @staticmethod
      def _resolve_enable_visual_message(request_kind: str) -> bool:
          if request_kind in {"planner", "timing_gate"}:
              return resolve_enable_visual_planner()
          return True
  
      @staticmethod
      def _hide_early_assistant_messages(
          selected_history: List[LLMContextMessage],
      ) -> tuple[List[LLMContextMessage], int]:
          """隐藏上下文中最早 50% 的 assistant 文本消息，但保留工具调用链路。"""
  
          assistant_indices = [
              index
              for index, message in enumerate(selected_history)
              if isinstance(message, AssistantMessage)
          ]
          hidden_assistant_count = len(assistant_indices) // 2
          if hidden_assistant_count <= 0:
              return selected_history, 0
  
          removed_assistant_indices = set(assistant_indices[:hidden_assistant_count])
  
          filtered_history: List[LLMContextMessage] = []
          for index, message in enumerate(selected_history):
              if index in removed_assistant_indices:
                  if not message.tool_calls:
                      continue
                  filtered_history.append(
                      AssistantMessage(
                          content="",
                          timestamp=message.timestamp,
                          tool_calls=list(message.tool_calls),
                          source_kind=message.source_kind,
                      )
                  )
                  continue
              filtered_history.append(message)
  
          return filtered_history, hidden_assistant_count
  
  