mirror of
https://github.com/xinnan-tech/xiaozhi-esp32-server.git
synced 2026-07-22 23:23:55 +08:00
119 lines
4.5 KiB
Python
119 lines
4.5 KiB
Python
import uuid
|
||
from typing import List, Dict
|
||
from datetime import datetime
|
||
from config.settings import load_config
|
||
|
||
|
||
class Message:
|
||
def __init__(
|
||
self,
|
||
role: str,
|
||
content: str = None,
|
||
uniq_id: str = None,
|
||
tool_calls=None,
|
||
tool_call_id=None,
|
||
):
|
||
self.uniq_id = uniq_id if uniq_id is not None else str(uuid.uuid4())
|
||
self.role = role
|
||
self.content = content
|
||
self.tool_calls = tool_calls
|
||
self.tool_call_id = tool_call_id
|
||
|
||
|
||
class Dialogue:
|
||
def __init__(self):
|
||
self.dialogue: List[Message] = []
|
||
# 获取当前时间
|
||
self.current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
def put(self, message: Message):
|
||
self.dialogue.append(message)
|
||
|
||
def getMessages(self, m, dialogue):
|
||
if m.tool_calls is not None:
|
||
dialogue.append({"role": m.role, "tool_calls": m.tool_calls})
|
||
elif m.role == "tool":
|
||
dialogue.append(
|
||
{
|
||
"role": m.role,
|
||
"tool_call_id": (
|
||
str(uuid.uuid4()) if m.tool_call_id is None else m.tool_call_id
|
||
),
|
||
"content": m.content,
|
||
}
|
||
)
|
||
else:
|
||
dialogue.append({"role": m.role, "content": m.content})
|
||
|
||
def get_llm_dialogue(self) -> List[Dict[str, str]]:
|
||
# 直接调用get_llm_dialogue_with_memory，传入None作为memory_str
|
||
# 这样确保说话人功能在所有调用路径下都生效
|
||
return self.get_llm_dialogue_with_memory(None)
|
||
|
||
def update_system_message(self, new_content: str):
|
||
"""更新或添加系统消息"""
|
||
# 查找第一个系统消息
|
||
system_msg = next((msg for msg in self.dialogue if msg.role == "system"), None)
|
||
if system_msg:
|
||
system_msg.content = new_content
|
||
else:
|
||
self.put(Message(role="system", content=new_content))
|
||
|
||
def get_llm_dialogue_with_memory(
|
||
self, memory_str: str = None
|
||
) -> List[Dict[str, str]]:
|
||
# 构建对话
|
||
dialogue = []
|
||
|
||
# 添加系统提示和记忆
|
||
system_message = next(
|
||
(msg for msg in self.dialogue if msg.role == "system"), None
|
||
)
|
||
|
||
if system_message:
|
||
# 基础系统提示
|
||
enhanced_system_prompt = system_message.content
|
||
|
||
# 添加说话人识别功能说明
|
||
speaker_guidance = "\n\n[说话人识别功能说明]\n" \
|
||
"当用户消息为JSON格式包含speaker字段时（如：{\"speaker\": \"张三\", \"content\": \"消息内容\"}），表示系统已识别出说话人身份。\n" \
|
||
"请根据说话人的身份特征来调整回应风格和内容。\n" \
|
||
"你可以称呼说话人的名字，并参考他们的特点进行个性化回应。"
|
||
enhanced_system_prompt += speaker_guidance
|
||
|
||
# 添加说话人个性化描述
|
||
try:
|
||
config = load_config()
|
||
voiceprint_config = config.get("voiceprint", {})
|
||
speakers = voiceprint_config.get("speakers", [])
|
||
|
||
if speakers:
|
||
enhanced_system_prompt += "\n\n[已知说话人信息]"
|
||
for speaker_str in speakers:
|
||
try:
|
||
parts = speaker_str.split(",", 2)
|
||
if len(parts) >= 2:
|
||
speaker_id = parts[0].strip()
|
||
name = parts[1].strip()
|
||
# 如果描述为空，则为""
|
||
description = parts[2].strip() if len(parts) >= 3 else ""
|
||
enhanced_system_prompt += f"\n- {name}：{description}"
|
||
except:
|
||
continue
|
||
except:
|
||
# 配置读取失败时忽略错误，不影响其他功能
|
||
pass
|
||
|
||
# 只有当有记忆时才添加记忆部分
|
||
if memory_str and len(memory_str) > 0:
|
||
enhanced_system_prompt += f"\n\n以下是用户的历史记忆：\n```\n{memory_str}\n```"
|
||
|
||
dialogue.append({"role": "system", "content": enhanced_system_prompt})
|
||
|
||
# 添加用户和助手的对话
|
||
for m in self.dialogue:
|
||
if m.role != "system": # 跳过原始的系统消息
|
||
self.getMessages(m, dialogue)
|
||
|
||
return dialogue
|