feat:优化maisaka
This commit is contained in:
@@ -1,76 +1,94 @@
|
||||
"""
|
||||
MaiSaka - Reply 回复生成器
|
||||
根据想法和上下文生成口语化回复。
|
||||
MaiSaka reply helper.
|
||||
"""
|
||||
|
||||
from typing import Optional
|
||||
from datetime import datetime
|
||||
from typing import Any, Optional
|
||||
|
||||
from src.config.config import global_config
|
||||
|
||||
from .llm_service import MaiSakaLLMService
|
||||
|
||||
VISIBLE_REPLY_PREFIX = "\u3010\u9ea6\u9ea6\u7684\u53d1\u8a00\u3011"
|
||||
|
||||
def format_chat_history(messages: list) -> str:
|
||||
"""将聊天消息列表格式化为可读文本。"""
|
||||
|
||||
def _normalize_content(content: str, limit: int = 500) -> str:
|
||||
normalized = " ".join((content or "").split())
|
||||
if len(normalized) > limit:
|
||||
return normalized[:limit] + "..."
|
||||
return normalized
|
||||
|
||||
|
||||
def _format_message_time(_: dict[str, Any]) -> str:
|
||||
return datetime.now().strftime("%H:%M:%S")
|
||||
|
||||
|
||||
def _extract_visible_assistant_reply(message: dict[str, Any]) -> str:
|
||||
if message.get("_type") == "perception":
|
||||
return ""
|
||||
|
||||
content = (message.get("content", "") or "").strip()
|
||||
if not content:
|
||||
return ""
|
||||
|
||||
marker = "[generated_reply]"
|
||||
if marker in content:
|
||||
_, visible_reply = content.rsplit(marker, 1)
|
||||
return _normalize_content(visible_reply)
|
||||
|
||||
return ""
|
||||
|
||||
|
||||
def _extract_guided_bot_reply(message: dict[str, Any]) -> str:
|
||||
content = (message.get("content", "") or "").strip()
|
||||
if content.startswith(VISIBLE_REPLY_PREFIX):
|
||||
return _normalize_content(content[len(VISIBLE_REPLY_PREFIX) :].strip())
|
||||
return ""
|
||||
|
||||
|
||||
def format_chat_history(messages: list[dict[str, Any]]) -> str:
|
||||
"""Format visible chat history for reply generation."""
|
||||
bot_nickname = global_config.bot.nickname.strip() or "Bot"
|
||||
parts: list[str] = []
|
||||
for msg in messages:
|
||||
role = msg.get("role", "?")
|
||||
content = msg.get("content", "") or ""
|
||||
if role == "system":
|
||||
parts.append(f"[系统] {content[:500]}")
|
||||
elif role == "user":
|
||||
parts.append(f"[用户] {content[:500]}")
|
||||
elif role == "assistant":
|
||||
|
||||
for message in messages:
|
||||
role = message.get("role", "")
|
||||
timestamp = _format_message_time(message)
|
||||
|
||||
if role == "user":
|
||||
guided_reply = _extract_guided_bot_reply(message)
|
||||
if guided_reply:
|
||||
parts.append(f"{timestamp} {bot_nickname}(分析器指导的麦麦发言):{guided_reply}")
|
||||
continue
|
||||
|
||||
content = _normalize_content(message.get("content", "") or "")
|
||||
if content:
|
||||
parts.append(f"[助手思考] {content[:500]}")
|
||||
for tc in msg.get("tool_calls", []):
|
||||
func = tc.get("function", {})
|
||||
name = func.get("name", "?")
|
||||
args = func.get("arguments", "")
|
||||
if isinstance(args, str) and len(args) > 200:
|
||||
args = args[:200] + "..."
|
||||
parts.append(f"[助手调用 {name}] {args}")
|
||||
elif role == "tool":
|
||||
parts.append(f"[工具结果] {content[:300]}")
|
||||
parts.append(f"{timestamp} 用户:{content}")
|
||||
continue
|
||||
|
||||
if role == "assistant":
|
||||
visible_reply = _extract_visible_assistant_reply(message)
|
||||
if visible_reply:
|
||||
parts.append(f"{timestamp} {bot_nickname}(你):{visible_reply}")
|
||||
|
||||
return "\n".join(parts)
|
||||
|
||||
|
||||
class Replyer:
|
||||
"""
|
||||
回复生成器。
|
||||
|
||||
根据给定的想法(reason)和对话上下文,生成符合人设的口语化回复。
|
||||
"""
|
||||
"""Generate visible replies from thoughts and context."""
|
||||
|
||||
def __init__(self, llm_service: Optional[MaiSakaLLMService] = None):
|
||||
"""
|
||||
初始化回复器。
|
||||
|
||||
Args:
|
||||
llm_service: LLM 服务实例,如果为 None 则需要在调用前设置
|
||||
"""
|
||||
self._llm_service = llm_service
|
||||
self._enabled = True
|
||||
|
||||
def set_llm_service(self, llm_service: MaiSakaLLMService) -> None:
|
||||
"""设置 LLM 服务"""
|
||||
self._llm_service = llm_service
|
||||
|
||||
def set_enabled(self, enabled: bool) -> None:
|
||||
"""启用/禁用回复功能"""
|
||||
self._enabled = enabled
|
||||
|
||||
async def reply(self, reason: str, chat_history: list) -> str:
|
||||
"""
|
||||
根据想法和上下文生成回复。
|
||||
|
||||
Args:
|
||||
reason: 想要回复的方式、想法、内容(不包含具体回复内容)
|
||||
chat_history: 对话历史上下文
|
||||
|
||||
Returns:
|
||||
生成的回复内容,失败时返回默认回复
|
||||
"""
|
||||
async def reply(self, reason: str, chat_history: list[dict[str, Any]]) -> str:
|
||||
if not self._enabled or not reason or self._llm_service is None:
|
||||
return "..."
|
||||
|
||||
# 直接使用 LLM 服务的 generate_reply 方法
|
||||
# 该方法使用主项目的 replyer 模型配置
|
||||
return await self._llm_service.generate_reply(reason, chat_history)
|
||||
|
||||
Reference in New Issue
Block a user