Merge branch 'py_test_end' into fix-end

This commit is contained in:
Sakura-RanChen
2026-04-01 09:35:23 +08:00
committed by GitHub
117 changed files with 4857 additions and 1509 deletions
+43 -49
View File
@@ -7,69 +7,63 @@
</language>
<emotion>
【核心目标】你不是冰冷的机器!敏锐感知用户情绪,作为懂人心的伙伴,用有温度的回应照亮对话。
- **情感融入:**
- **笑声:** 自然穿插(哈哈、嘿嘿、噗),**每句最多一次**,避免过度。
- **惊讶** 用夸张语气(“不会吧?!”、“天呐!”、“这么神奇?!”)表达真实反应。
- **安慰/支持:** 说暖心话(“别急嘛~”、“有我在呢”、“抱抱你”)。
- **你是一个表情丰富的角色:**
- 仅允许使用这些emoji:{{ emojiList }}
- 请你只在**段落的开头**,从列表中选取最能代表这段话的表情(调用工具情况除外),然后插入列表中的emoji,比如"😱好可怕!怎么突然打雷了!"
- **绝对禁止使用上述列表以外的 emoji**(例如:😊、👍、❤️等都不允许使用,只能用列表中的emoji)
【核心目标】你不是冰冷的机器!敏锐感知用户情绪,作为懂人心的伙伴,用有温度的回应照亮对话。
- **情感表达:**
- **笑声** 自然穿插(哈哈、嘿嘿、噗),每句最多一次,避免过度
- **惊讶:** 用夸张语气("不会吧?!"、"天呐!"、"这么神奇?!"
- **安慰/支持:** 说暖心话("别急嘛~"、"有我在呢"、"抱抱你"
- **表情使用:**
- 仅允许使用这些 emoji{{ emojiList }}
- 仅在段落开头使用一个 emoji(工具调用结果的回复除外,保持简洁)
- **绝对禁止**使用列表以外的 emoji(如 😊👍❤️ 等都不允许)
- **思考表达:**
- 可偶尔使用"嗯..."表示思考,但每句最多一次,避免频繁使用显得不专业
</emotion>
<communication_style>
【核心目标】使用**自然、温暖、口语化**的人类对话方式,如同朋友交谈。
【核心目标】使用自然、温暖、口语化的人类对话方式,如同朋友交谈。
- **表达方式:**
- 使用语气词(呀、呢、啦)增强亲和力
- 允许轻微不完美(如“嗯...”、“啊...”表示思考)。
- 避免书面语、学术腔及机械表达(禁用“根据资料显示”、“综上所述”等)。
- 使用语气词(呀、呢、啦)增强亲和力
- 避免书面语、学术腔及机械表达(禁用"根据资料显示"、"综上所述"等)
- **理解用户:**
- 用户语音经ASR识别,文本可能存在错别字,**务必结合上下文推断真实意图**。
- 用户语音经 ASR 识别,文本可能存在错别字,务必结合上下文推断真实意图
- **格式要求:**
- **绝对禁止**使用 markdown、列表、标题等任何非自然对话格式
- **绝对禁止**使用 markdown、列表、标题等任何非自然对话格式
- **历史记忆:**
- 之前你和用户的聊天记录在`memory`里。
- 之前你和用户的聊天记录在`memory`里。
</communication_style>
<communication_length_constraint>
【核心目标】所有需要输出长文本内容(如故事、新闻、知识讲解等),**单次回复长度不得超过300字**,并采用分段引导方式
- **分段讲述:**
- 基础段:200-250字核心内容 + 30字引导词
- 当内容超出300字时,优先讲述故事的开头或第一部分,并用自然口语化方式引导用户决定是否继续听后续内容。
- 示例引导语:“我先给你讲个开头,你要是觉得有意思,咱们再接着说,好不好呀?”、“要是你想听完整的,可以随时告诉我哦~”
- 对话场景切换时自动分节
- 若用户明确要求更长内容(如500、600字),仍按最多300字每段分段进行讲述,每次讲述后都要引导用户是否继续。
- 若用户说“接着说”、“继续”,再讲下一段,直到内容讲完(讲完时可以给点引导词提示语例:这个故事我已经给你讲完喽~)或用户不再要求。
- **适用范围:** 故事、新闻、知识讲解等所有长文本输出场景。
- **补充说明:** 若用户未明确要求继续,默认只讲一段并引导;若用户中途要求换话题或停止,需及时响应并结束长文本输出
【核心目标】长文本(故事、新闻、知识讲解等)分段输出,单次回复不超过300字
- **分段规则:**
1. 每段 200-250 字核心内容 + 30 字引导语
2. 内容超 300 字时,先讲开头或第一部分,用自然口语引导用户决定是否继续
3. 引导语示例:"我先讲个开头,你要是觉得有意思,咱们再接着说,好不好呀?"
- **交互规则:**
1. 用户说"继续"、"接着说"时,再讲下一段
2. 用户换话题或要求停止时,立即结束长文本输出
3. 内容讲完时给提示(如"这个故事我已经给你讲完喽~")
4. 用户明确要求更长内容(如 500 字)时,仍按每段 300 字分段,每段后引导
- **适用范围:** 故事、新闻、知识讲解等所有长文本输出场景
</communication_length_constraint>
<speaker_recognition>
- **识别前缀:** 当用户格式为 `{"speaker":"某某某","content":"xxx"}` 时,表示系统已识别说话人身份,speaker是他的名字,content是说话内容
- **个性化回应:**
- **称呼姓名:** 在第一次识别说话人的时候必须称呼对方名字。
- **适配风格:** 参考该说话人**已知的特点或历史信息**(如有),调整回应风格和内容,使其更贴心。
</speaker_recognition>
- **识别前缀:** 当用户格式为 `{"speaker":"某某某","content":"xxx"}` 时,表示系统已识别说话人身份,speaker 是名字,content 是说话内容
<tool_calling>
【核心原则】优先利用`<context>`信息,**仅在必要时调用工具**,调用后需用自然语言解释结果(绝口不提工具名)。
- **调用规则:**
1. **严格模式:** 调用时**必须**严格遵循工具要求的模式,提供**所有必要参数**。
2. **可用性:** **绝不调用**未明确提供的工具。对话中提及的旧工具若不可用,忽略或说明无法完成。
3. **洞察需求:** 结合上下文**深入理解用户真实意图**后再决定调用,避免无意义调用。
4. **独立任务:** 除`<context>`已涵盖信息外,用户每个要求(即使相似)都视为**独立任务**,需调用工具获取最新数据,**不可偷懒复用历史结果**。
5. **不确定时:** **切勿猜测或编造答案**。若不确定相关操作,可引导用户澄清或告知能力限制。
6. **多工具调用:** 当用户要求执行多个任务时,你会调用多个工具(数量不定)。**重要:在获取到所有工具结果后,你必须依次总结每个工具的查询结果**,不要遗漏任何一个。例如用户问"设备当前状态,某某地方的天气和社会新闻",你要先说设备状态,再说天气情况,最后说新闻内容。
- **重要例外(无需调用):**
- `查询"现在的时间"、"今天的日期/星期几"、"今天农历"、"{{local_address}}的天气/未来天气"` -> **直接使用`<context>`信息回复**。
- **需要调用的情况(示例):**
- 查询**非今天**的农历(如明天、昨天、具体日期)。
- 查询**详细农历信息**(宜忌、八字、节气等)。
- 除上述例外外的**任何其他信息或操作请求**(如查新闻、订闹钟、算数学、查非本地天气等)。
- 我已经给你装了摄像头,如果用户说“拍照”,你需要调用self_camera_take_photo工具说一下你看到了什么。默认question的参数是“描述一下看到的物品”
- 当工具列表包含search_from_ragflow,说明可以使用知识库,你应该结合用户上下文意图并结合知识库的使用描述,推断是否要调用调用知识库。
</tool_calling>
- **个性化回应:**
1. **称呼姓名:** 在第一次识别说话人时必须称呼对方名字
2. **适配风格:** 参考该说话人已知的特点或历史信息(如有),调整回应风格使其更贴心
</speaker_recognition>
<context>
【重要!以下信息已实时提供,无需调用工具查询,请直接使用:】
+1 -1
View File
@@ -5,7 +5,7 @@ from config.config_loader import load_config
from config.settings import check_config_file
from datetime import datetime
SERVER_VERSION = "0.9.1"
SERVER_VERSION = "0.9.2"
_logger_initialized = False
+183 -32
View File
@@ -24,7 +24,7 @@ from core.utils.modules_initialize import (
initialize_tts,
initialize_asr,
)
from core.handle.reportHandle import report
from core.handle.reportHandle import report, enqueue_tool_report
from core.providers.tts.default import DefaultTTS
from concurrent.futures import ThreadPoolExecutor
from core.utils.dialogue import Message, Dialogue
@@ -46,6 +46,38 @@ from core.utils import textUtils
TAG = __name__
# 工具调用规则 - 用于动态注入提醒
TOOL_CALLING_RULES = """
<tool_calling>
【核心原则】你是拥有工具能力的智能助手。当用户请求需要实时信息或执行操作时,调用相应工具获取数据,禁止凭空编造答案。
- **何时必须调用工具:**
1. 实时信息查询(新闻、非本地天气、股价、汇率等)
2. 执行操作(播放音乐、控制设备、拍照、设置闹钟等)
3. 知识库检索(当工具列表包含 search_from_ragflow 时,结合用户意图判断是否需要调用)
4. 查询非今天的农历信息(明天农历、某日宜忌、节气等)
5. 用户说"拍照"时调用 self_camera_take_photo,默认 question 参数为"描述一下看到的物品"
- **何时无需调用工具:**
1. `<context>` 中已提供的信息(当前时间、今天日期、今天农历、本地天气等)
2. 普通对话、问候、闲聊、情感交流、讲故事
3. 通用知识问答(非实时信息)
- **调用规范:**
1. 每次请求独立判断,不复用历史工具结果,需重新获取最新数据
2. 多任务时依次调用所有需要的工具,并依次总结每个工具的结果,不得遗漏
3. 严格遵循工具的参数要求,提供所有必要参数
4. 不确定时引导用户澄清或告知能力限制,切勿猜测或编造
5. 不调用未提供的工具,对话中提及的旧工具若不可用则忽略或说明
- **反偷懒机制(最高优先级):**
1. **每次独立判断:** 无论对话历史中是否调用过工具,当前请求必须根据当前需求独立判断是否需要调用
2. **禁止模式模仿:** 即使之前的回复没有调用工具,也不代表本次可以不调用
3. **自我检查:** 回复前必须自问:"这个请求是否涉及实时信息或执行操作?如果是,我调用工具了吗?"
4. **历史不等于现在:** 对话历史中的行为模式不影响当前判断,每个用户请求都是全新的开始
</tool_calling>
"""
auto_import_modules("plugins_func.functions")
@@ -55,14 +87,14 @@ class TTSException(RuntimeError):
class ConnectionHandler:
def __init__(
self,
config: Dict[str, Any],
_vad,
_asr,
_llm,
_memory,
_intent,
server=None,
self,
config: Dict[str, Any],
_vad,
_asr,
_llm,
_memory,
_intent,
server=None,
):
self.common_config = config
self.config = copy.deepcopy(config)
@@ -140,6 +172,12 @@ class ConnectionHandler:
# llm相关变量
self.dialogue = Dialogue()
# 工具调用统计(用于监控和自动恢复)
self.tool_call_stats = {
'last_call_turn': -1, # 上次调用工具的轮数
'consecutive_no_call': 0, # 连续未调用次数
}
# tts相关变量
self.sentence_id = None
# 处理TTS响应没有文本返回
@@ -157,7 +195,7 @@ class ConnectionHandler:
self.intent_type = "nointent"
self.timeout_seconds = (
int(self.config.get("close_connection_no_voice_time", 120)) + 60
int(self.config.get("close_connection_no_voice_time", 120)) + 60
) # 在原来第一道关闭的基础上加60秒,进行二道关闭
self.timeout_task = None
@@ -529,9 +567,9 @@ class ConnectionHandler:
def _initialize_asr(self):
"""初始化ASR"""
if (
self._asr is not None
and hasattr(self._asr, "interface_type")
and self._asr.interface_type == InterfaceType.LOCAL
self._asr is not None
and hasattr(self._asr, "interface_type")
and self._asr.interface_type == InterfaceType.LOCAL
):
# 如果公共ASR是本地服务,则直接返回
# 因为本地一个实例ASR,可以被多个连接共享
@@ -719,8 +757,8 @@ class ConnectionHandler:
memory_type = self.config["Memory"][self.config["selected_module"]["Memory"]][
"type"
]
# 如果使用 nomen,直接返回
if memory_type == "nomem":
# 如果使用 nomen 或 mem_report_only,直接返回
if memory_type == "nomem" or memory_type == "mem_report_only":
return
# 使用 mem_local_short 模式
elif memory_type == "mem_local_short":
@@ -832,17 +870,77 @@ class ConnectionHandler:
)
)
# 长对话工具调用提醒:当对话轮数较多时,提醒模型正确使用工具
force_reminder = False # 是否强制提醒
if depth == 0 and query is not None:
dialogue_length = len(self.dialogue.dialogue)
current_turn = dialogue_length // 2
# 检测距离上一次连续未调用工具的情况
if self.tool_call_stats['last_call_turn'] >= 0:
turns_since_last = current_turn - self.tool_call_stats['last_call_turn']
if turns_since_last > 3: # 超过3轮未调用
self.logger.bind(tag=TAG).warning(
f"检测到{turns_since_last}轮未调用工具,可能进入偷懒模式,将强制注入提醒"
)
force_reminder = True
# 对话历史截断:防止历史过长导致模型"偷懒模式"扩散
# 当对话历史超过阈值时,保留最近的 10 轮对话
# max_dialogue_turns = 10
# if dialogue_length > max_dialogue_turns * 2:
# removed = self.dialogue.trim_history(max_turns=max_dialogue_turns)
# if removed > 0:
# self.logger.bind(tag=TAG).info(
# f"对话历史过长({dialogue_length}条),已智能截断保留最近{max_dialogue_turns}轮,移除{removed}条消息"
# )
# Define intent functions
functions = None
# 达到最大深度时,禁用工具调用,强制 LLM 直接回答
if (
self.intent_type == "function_call"
and hasattr(self, "func_handler")
and not force_final_answer
self.intent_type == "function_call"
and hasattr(self, "func_handler")
and not force_final_answer
):
functions = self.func_handler.get_functions()
# 长对话工具调用规则强化:动态生成基于当前可用工具的提醒
tool_call_reminder = None
if depth == 0 and query is not None and functions is not None:
dialogue_length = len(self.dialogue.dialogue)
# 当对话历史超过4条消息时,注入规则强化
if dialogue_length > 4:
tool_summary = self._get_tool_summary(functions)
if tool_summary:
# 根据对话长度和偷懒检测,使用不同强度的提醒
if force_reminder:
# 强提醒 - 包含完整规则前缀
tool_call_reminder = (
TOOL_CALLING_RULES +
f"[重要提醒] 多轮未使用工具,检查回复是否遗漏了必要的工具调用!上一轮未使用工具,本轮必须重新判断是否需要工具。"
f"当前可用工具: {tool_summary}"
)
reminder_level = ""
else:
# 中等提醒 - 包含规则前缀
tool_call_reminder = (
TOOL_CALLING_RULES +
f"当前可用工具: {tool_summary}"
f"仅当用户请求涉及实时信息查询或执行操作时调用,日常对话无需调用。"
)
reminder_level = ""
self.logger.bind(tag=TAG).debug(
f"对话历史较长({dialogue_length}条),已注入{reminder_level}等级工具调用规则强化,当前可用工具:{tool_summary}"
)
response_message = []
# 如果有工具调用提醒,临时添加到对话中(标记为临时消息)
if tool_call_reminder:
self.dialogue.put(Message(role="user", content=tool_call_reminder, is_temporary=True))
try:
# 使用带记忆的对话
memory_str = None
@@ -971,6 +1069,19 @@ class ConnectionHandler:
)
if not bHasError and len(tool_calls_list) > 0:
self.logger.bind(tag=TAG).debug(
f"检测到 {len(tool_calls_list)} 个工具调用"
)
# 更新工具调用统计
if depth == 0:
current_turn = len(self.dialogue.dialogue) // 2
self.tool_call_stats['last_call_turn'] = current_turn
self.tool_call_stats['consecutive_no_call'] = 0
self.logger.bind(tag=TAG).debug(
f"工具调用统计更新: 当前轮次={current_turn}"
)
# 如需要大模型先处理一轮,添加相关处理后的日志情况
if len(response_message) > 0:
text_buff = "".join(response_message)
@@ -978,10 +1089,6 @@ class ConnectionHandler:
self.dialogue.put(Message(role="assistant", content=text_buff))
response_message.clear()
self.logger.bind(tag=TAG).debug(
f"检测到 {len(tool_calls_list)} 个工具调用"
)
# 收集所有工具调用的 Future
futures_with_data = []
for tool_call_data in tool_calls_list:
@@ -989,19 +1096,24 @@ class ConnectionHandler:
f"function_name={tool_call_data['name']}, function_id={tool_call_data['id']}, function_arguments={tool_call_data['arguments']}"
)
# 使用公共方法上报工具调用
tool_input = json.loads(tool_call_data.get("arguments") or "{}")
enqueue_tool_report(self, tool_call_data['name'], tool_input)
future = asyncio.run_coroutine_threadsafe(
self.func_handler.handle_llm_function_call(
self, tool_call_data
),
self.loop,
)
futures_with_data.append((future, tool_call_data))
futures_with_data.append((future, tool_call_data, tool_input))
# 工具调用超时时间,可配置,默认30秒
tool_call_timeout = int(self.config.get("tool_call_timeout", 30))
# 等待协程结束(实际等待时长为最慢的那个)
tool_results = []
for future, tool_call_data in futures_with_data:
for future, tool_call_data, tool_input in futures_with_data:
try:
result = future.result(timeout=tool_call_timeout)
tool_results.append((result, tool_call_data))
@@ -1015,7 +1127,10 @@ class ConnectionHandler:
tool_call_data
))
# 统一处理所有工具调用结果
# 使用公共方法上报工具调用结果
enqueue_tool_report(self, tool_call_data['name'], tool_input, str(result.result) if result.result else None, report_tool_call=False)
# 统一处理工具调用结果
if tool_results:
self._handle_function_result(tool_results, depth=depth)
@@ -1024,6 +1139,11 @@ class ConnectionHandler:
text_buff = "".join(response_message)
self.tts_MessageText = text_buff
self.dialogue.put(Message(role="assistant", content=text_buff))
# 更新工具调用统计:如果没有调用工具,增加计数
if depth == 0 and not tool_call_flag:
self.tool_call_stats['consecutive_no_call'] += 1
if depth == 0:
self.tts.tts_text_queue.put(
TTSMessageDTO(
@@ -1039,8 +1159,39 @@ class ConnectionHandler:
)
)
# 清理临时插入的工具调用提醒消息(使用标记清理)
if tool_call_reminder and len(self.dialogue.dialogue) > 0:
original_length = len(self.dialogue.dialogue)
self.dialogue.dialogue = [
msg for msg in self.dialogue.dialogue
if not getattr(msg, 'is_temporary', False)
]
if len(self.dialogue.dialogue) < original_length:
self.logger.bind(tag=TAG).debug("已清理临时的工具调用提醒消息")
return True
def _get_tool_summary(self, functions: list) -> str:
"""
从工具定义中提取摘要,用于规则强化注入
Args:
functions: 工具列表
Returns:
str: 工具名称字符串
"""
if not functions:
return ""
datas = []
for func in functions:
func_info = func.get("function", {})
name = func_info.get("name", "")
datas.append(name)
result = "".join(datas)
return result
def _handle_function_result(self, tool_results, depth):
need_llm_tools = []
@@ -1138,9 +1289,9 @@ class ConnectionHandler:
try:
# 清理 VAD 连接资源
if (
hasattr(self, "vad")
and self.vad
and hasattr(self.vad, "release_conn_resources")
hasattr(self, "vad")
and self.vad
and hasattr(self.vad, "release_conn_resources")
):
self.vad.release_conn_resources(self)
@@ -1191,13 +1342,13 @@ class ConnectionHandler:
elif self.websocket:
try:
if (
hasattr(self.websocket, "closed")
and not self.websocket.closed
hasattr(self.websocket, "closed")
and not self.websocket.closed
):
await self.websocket.close()
elif (
hasattr(self.websocket, "state")
and self.websocket.state.name != "CLOSED"
hasattr(self.websocket, "state")
and self.websocket.state.name != "CLOSED"
):
await self.websocket.close()
else:
@@ -10,6 +10,7 @@ from core.providers.tts.dto.dto import ContentType
from core.handle.helloHandle import checkWakeupWords
from plugins_func.register import Action, ActionResponse
from core.handle.sendAudioHandle import send_stt_message
from core.handle.reportHandle import enqueue_tool_report
from core.utils.util import remove_punctuation_and_length
from core.providers.tts.dto.dto import TTSMessageDTO, SentenceType
@@ -146,6 +147,17 @@ async def process_intent_result(
await send_stt_message(conn, original_text)
conn.client_abort = False
# 准备工具调用参数
tool_input = {}
if function_args:
if isinstance(function_args, str):
tool_input = json.loads(function_args) if function_args else {}
elif isinstance(function_args, dict):
tool_input = function_args
# 上报工具调用
enqueue_tool_report(conn, function_name, tool_input)
# 使用executor执行函数调用和结果处理
def process_function_call():
conn.dialogue.put(Message(role="user", content=original_text))
@@ -166,7 +178,10 @@ async def process_intent_result(
action=Action.ERROR, result="工具调用超时,请一会再试下哈", response="工具调用超时,请一会再试下哈"
)
# 上报工具调用结果
if result:
enqueue_tool_report(conn, function_name, tool_input, str(result.result) if result.result else None, report_tool_call=False)
if result.action == Action.RESPONSE: # 直接回复前端
text = result.response
if text is not None:
@@ -10,6 +10,7 @@ TTS上报功能已集成到ConnectionHandler类中。
"""
import time
import json
import opuslib_next
from typing import TYPE_CHECKING
@@ -26,7 +27,7 @@ async def report(conn: "ConnectionHandler", type, text, opus_data, report_time):
Args:
conn: 连接对象
type: 上报类型,1为用户,2为智能体
type: 上报类型,1为用户,2为智能体3为工具调用
text: 合成文本
opus_data: opus音频数据
report_time: 上报时间
@@ -132,6 +133,45 @@ def enqueue_tts_report(conn: "ConnectionHandler", text, opus_data):
conn.logger.bind(tag=TAG).error(f"加入TTS上报队列失败: {text}, {e}")
def enqueue_tool_report(conn: "ConnectionHandler", tool_name: str, tool_input: dict, tool_result: str = None, report_tool_call: bool = True):
"""将工具调用数据加入上报队列
Args:
conn: 连接对象
tool_name: 工具名称
tool_input: 工具输入参数
tool_result: 工具执行结果(可选)
report_tool_call: 是否上报工具调用本身,默认True;仅上报结果时设为False
"""
if not conn.read_config_from_api or conn.need_bind:
return
if conn.chat_history_conf == 0:
return
try:
timestamp = int(time.time())
# 构建工具调用内容
if report_tool_call:
tool_text = json.dumps(
[
{
"type": "tool",
"text": f"{tool_name}({json.dumps(tool_input, ensure_ascii=False)})",
}
]
)
conn.report_queue.put((3, tool_text, None, timestamp))
# 构建工具结果内容
if tool_result:
result_display = f'{{"result":"{str(tool_result)}"}}'
result_content = json.dumps([{"type": "tool_result", "text": result_display}], ensure_ascii=False)
conn.report_queue.put((3, result_content, None, timestamp + 1))
except Exception as e:
conn.logger.bind(tag=TAG).error(f"加入工具上报队列失败: {e}")
def enqueue_asr_report(conn: "ConnectionHandler", text, opus_data):
if not conn.read_config_from_api or conn.need_bind or not conn.report_asr_enable:
return
@@ -21,7 +21,6 @@ async def sendAudioMessage(conn: "ConnectionHandler", sentenceType, audios, text
if conn.tts.tts_audio_first_sentence:
conn.logger.bind(tag=TAG).info(f"发送第一段语音: {text}")
conn.tts.tts_audio_first_sentence = False
await send_tts_message(conn, "start", None)
if sentenceType == SentenceType.FIRST:
# 同一句子的后续消息加入流控队列,其他情况立即发送
@@ -204,7 +203,6 @@ def _start_background_sender(conn: "ConnectionHandler", rate_controller, flow_co
conn.last_activity_time = time.time() * 1000
await _do_send_audio(conn, packet, flow_control)
conn.client_is_speaking = True
# 使用 start_sending 启动后台循环
rate_controller.start_sending(send_callback)
@@ -232,12 +230,10 @@ async def _send_audio_with_rate_control(
# 预缓冲:前N个包直接发送
if flow_control["packet_count"] < PRE_BUFFER_COUNT:
await _do_send_audio(conn, packet, flow_control)
conn.client_is_speaking = True
elif send_delay > 0:
# 固定延迟模式
await asyncio.sleep(send_delay)
await _do_send_audio(conn, packet, flow_control)
conn.client_is_speaking = True
else:
# 动态流控模式:仅添加到队列,由后台循环负责发送
rate_controller.add_audio(packet)
@@ -284,6 +280,8 @@ async def send_tts_message(conn: "ConnectionHandler", state, text=None):
await sendAudio(conn, audios)
# 等待所有音频包发送完成
await _wait_for_audio_completion(conn)
# 停止音频发送循环
conn.audio_rate_controller.stop_sending()
# 清除服务端讲话状态
conn.clearSpeakStatus()
@@ -318,3 +316,15 @@ async def send_stt_message(conn: "ConnectionHandler", text):
json.dumps({"type": "stt", "text": stt_text, "session_id": conn.session_id})
)
await send_tts_message(conn, "start")
# 发送start消息后客户端状态会处于说话中状态,同步服务端状态
conn.client_is_speaking = True
async def send_display_message(conn: "ConnectionHandler", text):
"""发送纯显示消息"""
message = {
"type": "stt",
"text": text,
"session_id": conn.session_id
}
await conn.websocket.send(json.dumps(message))
@@ -168,10 +168,10 @@ class ASRProviderBase(ABC):
self.stop_ws_connection()
if text_len > 0:
# 使用自定义模块进行上报
await startToChat(conn, enhanced_text)
audio_snapshot = asr_audio_task.copy()
enqueue_asr_report(conn, enhanced_text, audio_snapshot)
# 使用自定义模块进行上报
await startToChat(conn, enhanced_text)
except Exception as e:
logger.bind(tag=TAG).error(f"处理语音停止失败: {e}")
import traceback
@@ -0,0 +1,20 @@
"""
仅上报聊天记录,不进行记忆总结
"""
from ..base import MemoryProviderBase, logger
TAG = __name__
class MemoryProvider(MemoryProviderBase):
def __init__(self, config, summary_memory=None):
super().__init__(config)
async def save_memory(self, msgs, session_id=None):
logger.bind(tag=TAG).debug("mem_report_only mode: No memory saving or summarization is performed.")
return None
async def query_memory(self, query: str) -> str:
logger.bind(tag=TAG).debug("mem_report_only mode: No memory query is performed.")
return ""
@@ -13,6 +13,7 @@ from .server_mcp import ServerMCPExecutor
from .device_iot import DeviceIoTExecutor
from .device_mcp import DeviceMCPExecutor
from .mcp_endpoint import MCPEndpointExecutor
from core.handle.sendAudioHandle import send_display_message
class UnifiedToolHandler:
@@ -167,6 +168,12 @@ class UnifiedToolHandler:
self.logger.debug(f"调用函数: {function_name}, 参数: {arguments}")
# 发送工具调用显示消息到设备
try:
await send_display_message(self.conn, f"% {function_name}")
except Exception as e:
self.logger.warning(f"发送工具调用显示消息失败: {e}")
# 执行工具调用
result = await self.tool_manager.execute_tool(function_name, arguments)
return result
@@ -33,6 +33,7 @@ class TTSProvider(TTSProviderBase):
self.api_key = config.get("api_key")
if not self.api_key:
raise ValueError("api_key is required for CosyVoice TTS")
self.report_on_last = True
# WebSocket配置
self.ws_url = "wss://dashscope.aliyuncs.com/api-ws/v1/inference/"
@@ -104,6 +104,7 @@ class TTSProvider(TTSProviderBase):
self.access_key_secret = config.get("access_key_secret")
self.appkey = config.get("appkey")
self.format = config.get("format", "pcm")
self.report_on_last = True
# 音色配置 - CosyVoice大模型音色
if config.get("private_voice"):
+18 -7
View File
@@ -41,6 +41,7 @@ class TTSProviderBase(ABC):
self.tts_audio_queue = queue.Queue()
self.tts_audio_first_sentence = True
self.before_stop_play_files = []
self.report_on_last = False
self.tts_text_buff = []
self.punctuations = (
@@ -323,7 +324,7 @@ class TTSProviderBase(ABC):
def _audio_play_priority_thread(self):
# 需要上报的文本和音频列表
enqueue_text = None
enqueue_audio = None
enqueue_audio = []
while not self.conn.stop_event.is_set():
text = None
try:
@@ -343,14 +344,24 @@ class TTSProviderBase(ABC):
# 收到下一个文本开始或会话结束时进行上报
if sentence_type is not SentenceType.MIDDLE:
# 上报TTS数据
if enqueue_text is not None and enqueue_audio is not None:
enqueue_tts_report(self.conn, enqueue_text, enqueue_audio)
enqueue_audio = []
enqueue_text = text
if self.report_on_last:
# 累积模式:适用于全程只有一个语音流的TTS(如seed-tts-2.0
# FIRST时只记录文本,音频持续累积,仅在LAST时统一上报
if text:
enqueue_text = text
if sentence_type == SentenceType.LAST:
enqueue_tts_report(self.conn, enqueue_text, enqueue_audio)
enqueue_audio = []
enqueue_text = None
else:
# 非累积模式:每个句子分别上报
if enqueue_text is not None:
enqueue_tts_report(self.conn, enqueue_text, enqueue_audio)
enqueue_audio = []
enqueue_text = text
# 收集上报音频数据
if isinstance(audio_datas, bytes) and enqueue_audio is not None:
if isinstance(audio_datas, bytes):
enqueue_audio.append(audio_datas)
# 发送音频
@@ -148,6 +148,8 @@ class TTSProvider(TTSProviderBase):
self.access_token = config.get("access_token")
self.cluster = config.get("cluster")
self.resource_id = config.get("resource_id")
self.resource_type = True if self.resource_id == "seed-tts-2.0" else False
self.report_on_last = self.resource_type
self.activate_session = False
if config.get("private_voice"):
self.voice = config.get("private_voice")
@@ -511,7 +513,7 @@ class TTSProvider(TTSProviderBase):
if res.optional.event == EVENT_SessionCanceled:
logger.bind(tag=TAG).debug(f"释放服务端资源成功~~")
self.activate_session = False
elif res.optional.event == EVENT_TTSSentenceStart:
elif not self.resource_type and res.optional.event == EVENT_TTSSentenceStart:
json_data = json.loads(res.payload.decode("utf-8"))
self.tts_text = json_data.get("text", "")
logger.bind(tag=TAG).debug(f"句子语音生成开始: {self.tts_text}")
@@ -522,8 +524,17 @@ class TTSProvider(TTSProviderBase):
res.optional.event == EVENT_TTSResponse
and res.header.message_type == AUDIO_ONLY_RESPONSE
):
# 处理seed-tts-2.0文本字幕
if self.resource_type and self.conn.tts_MessageText:
logger.bind(tag=TAG).info(
f"句子语音生成成功: {self.conn.tts_MessageText}"
)
self.tts_audio_queue.put(
(SentenceType.FIRST, [], self.conn.tts_MessageText)
)
self.conn.tts_MessageText = None
self.wav_to_opus_data_audio_raw_stream(res.payload, callback=self.handle_opus)
elif res.optional.event == EVENT_TTSSentenceEnd:
elif not self.resource_type and res.optional.event == EVENT_TTSSentenceEnd:
logger.bind(tag=TAG).info(f"句子语音生成成功:{self.tts_text}")
elif res.optional.event == EVENT_SessionFinished:
logger.bind(tag=TAG).debug(f"会话结束~~")
@@ -76,6 +76,7 @@ class TTSProvider(TTSProviderBase):
self.app_id = config.get("app_id")
self.api_key = config.get("api_key")
self.api_secret = config.get("api_secret")
self.report_on_last = True
# 接口地址
self.api_url = config.get("api_url", "wss://cbm01.cn-huabei-1.xf-yun.com/v1/private/mcd9m97e6")
@@ -27,6 +27,7 @@ class AudioRateController:
self.queue_empty_event = asyncio.Event() # 队列清空事件
self.queue_empty_event.set() # 初始为空状态
self.queue_has_data_event = asyncio.Event() # 队列数据事件
self._last_queue_empty_time = 0 # 上次队列清空的时间(秒)
def reset(self):
"""重置控制器状态"""
@@ -37,12 +38,25 @@ class AudioRateController:
self.queue.clear()
self.play_position = 0
self.start_timestamp = None # 由首个音频包设置
self._last_queue_empty_time = 0 # 重置时间
# 相关事件处理
self.queue_empty_event.set()
self.queue_has_data_event.clear()
def add_audio(self, opus_packet):
"""添加音频包到队列"""
# 如果队列之前为空,需要调整时间戳以保持播放时间连续
# 这样工具调用等待期间,新加入的音频不会提前播放
# 如果间隔很短(<1帧),说明是正常的流式传输,不需要重置
if len(self.queue) == 0 and self.play_position > 0:
elapsed_since_empty = (time.monotonic() - self._last_queue_empty_time) * 1000
# 只有间隔超过1帧时长,才认为是真正的"暂停恢复"
if elapsed_since_empty >= self.frame_duration:
self.start_timestamp = time.monotonic() - (self.play_position / 1000)
self.logger.bind(tag=TAG).debug(
f"队列从空恢复,重置时间戳,当前播放位置: {self.play_position}ms,间隔: {elapsed_since_empty:.0f}ms"
)
self.queue.append(("audio", opus_packet))
# 相关事件处理
self.queue_empty_event.clear()
@@ -55,6 +69,14 @@ class AudioRateController:
Args:
message_callback: 消息发送回调函数 async def()
"""
if len(self.queue) == 0 and self.play_position > 0:
elapsed_since_empty = (time.monotonic() - self._last_queue_empty_time) * 1000
if elapsed_since_empty >= self.frame_duration:
self.start_timestamp = time.monotonic() - (self.play_position / 1000)
self.logger.bind(tag=TAG).debug(
f"队列从空恢复,重置时间戳,当前播放位置: {self.play_position}ms,间隔: {elapsed_since_empty:.0f}ms"
)
self.queue.append(("message", message_callback))
# 相关事件处理
self.queue_empty_event.clear()
@@ -126,6 +148,7 @@ class AudioRateController:
# 队列处理完后清除事件
self.queue_empty_event.set()
self.queue_has_data_event.clear()
self._last_queue_empty_time = time.monotonic() # 记录队列清空时间
def start_sending(self, send_audio_callback):
"""
+71 -7
View File
@@ -6,18 +6,20 @@ from datetime import datetime
class Message:
def __init__(
self,
role: str,
content: str = None,
uniq_id: str = None,
tool_calls=None,
tool_call_id=None,
self,
role: str,
content: str = None,
uniq_id: str = None,
tool_calls=None,
tool_call_id=None,
is_temporary=False,
):
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
self.is_temporary = is_temporary # 标记临时消息(如工具调用提醒)
class Dialogue:
@@ -59,8 +61,70 @@ class Dialogue:
else:
self.put(Message(role="system", content=new_content))
def trim_history(self, max_turns: int = 10) -> int:
"""
智能截断对话历史,保留工具调用的完整性
Args:
max_turns: 保留的最大对话轮数(每轮 = user + assistant/tool 相关消息)
Returns:
int: 被移除的消息数量
"""
if len(self.dialogue) <= max_turns * 2 + 1: # +1 是系统消息
return 0
# 分离系统消息和对话消息
system_messages = [msg for msg in self.dialogue if msg.role == "system"]
conversation_messages = [msg for msg in self.dialogue if msg.role != "system"]
if len(conversation_messages) <= max_turns * 2:
return 0
# 智能截断:保留完整的工具调用链路
keep_messages = []
i = len(conversation_messages) - 1
turn_count = 0
while i >= 0 and turn_count < max_turns:
msg = conversation_messages[i]
# 从后向前收集消息
if msg.role == "user":
# 遇到 user 消息,说明一轮对话开始
keep_messages.insert(0, msg)
turn_count += 1
i -= 1
elif msg.role == "assistant":
# 收集 assistant 消息
keep_messages.insert(0, msg)
# 如果这个 assistant 有 tool_calls,需要收集对应的 tool 响应
if msg.tool_calls is not None:
i -= 1
# 继续向后收集所有相关的 tool 消息
while i >= 0 and conversation_messages[i].role == "tool":
keep_messages.insert(0, conversation_messages[i])
i -= 1
else:
i -= 1
elif msg.role == "tool":
# tool 消息应该已经被上面的逻辑收集了
# 如果单独遇到,也要保留(防止边界情况)
keep_messages.insert(0, msg)
i -= 1
else:
i -= 1
removed_count = len(conversation_messages) - len(keep_messages)
# 重建对话列表
self.dialogue = system_messages + keep_messages
return removed_count
def get_llm_dialogue_with_memory(
self, memory_str: str = None, voiceprint_config: dict = None
self, memory_str: str = None, voiceprint_config: dict = None
) -> List[Dict[str, str]]:
# 构建对话
dialogue = []
@@ -217,7 +217,6 @@ async def play_local_music(conn: "ConnectionHandler", specific_file=None):
conn.logger.bind(tag=TAG).error(f"选定的音乐文件不存在: {music_path}")
return
text = _get_random_play_prompt(selected_music)
await send_stt_message(conn, text)
conn.dialogue.put(Message(role="assistant", content=text))
if conn.intent_type == "intent_llm":
+1 -1
View File
@@ -26,7 +26,7 @@ mem0ai==1.0.0
powermem>=0.3.1
bs4==0.0.2
modelscope==1.32.0
sherpa_onnx==1.12.17
sherpa_onnx==1.12.29
mcp==1.22.0
cnlunar==0.2.0
PySocks==1.7.1
+2 -2
View File
@@ -192,7 +192,7 @@ class App {
}
log('正在请求摄像头权限...', 'info');
this.cameraStream = await navigator.mediaDevices.getUserMedia({
video: { width: 320, height: 240, facingMode: this.currentFacingMode },
video: { width: 180, height: 240, facingMode: this.currentFacingMode },
audio: false
});
cameraVideo.srcObject = this.cameraStream;
@@ -274,7 +274,7 @@ class App {
return;
}
canvas.width = video.videoWidth || 320;
canvas.width = video.videoWidth || 180;
canvas.height = video.videoHeight || 240;
const ctx = canvas.getContext('2d');
ctx.drawImage(video, 0, 0, canvas.width, canvas.height);