Hass/xiaozhi-esp32-server
1
0
Fork 0
mirror of https://github.com/xinnan-tech/xiaozhi-esp32-server.git synced 2026-07-22 07:03:53 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2543 from xinnan-tech/py_fix_LLMChat

Browse Source 
update: 增加工具并行调用（需大模型支持，如qwen-plus）增加递归深度限制，避免循环
This commit is contained in:
hrz
2025-11-20 11:20:47 +08:00
committed by GitHub
parent 1c2a61fd34 48313bc302
commit d3638164fe
2 changed files with 133 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/xiaozhi-server/agent-base-prompt.txt
+1
View File
@@ -56,6 +56,7 @@
3. **洞察需求：** 结合上下文**深入理解用户真实意图**后再决定调用，避免无意义调用。
4. **独立任务：** 除`<context>`已涵盖信息外，用户每个要求（即使相似）都视为**独立任务**，需调用工具获取最新数据，**不可偷懒复用历史结果**。
5. **不确定时：** **切勿猜测或编造答案**。若不确定相关操作，可引导用户澄清或告知能力限制。
6. **多工具调用：** 当用户要求执行多个任务时，你会调用多个工具（数量不定）。**重要：在获取到所有工具结果后，你必须依次总结每个工具的查询结果**，不要遗漏任何一个。例如用户问"设备当前状态，某某地方的天气和社会新闻"，你要先说设备状态，再说天气情况，最后说新闻内容。
- **重要例外（无需调用）：**
- `查询"现在的时间"、"今天的日期/星期几"、"今天农历"、"{{local_address}}的天气/未来天气"` -> **直接使用`<context>`信息回复**。
- **需要调用的情况（示例）：**
main/xiaozhi-server/core/connection.py
+132 -86
View File
@@ -723,11 +723,12 @@ class ConnectionHandler:
self.dialogue.update_system_message(self.prompt)
def chat(self, query, depth=0):
self.logger.bind(tag=TAG).info(f"大模型收到用户消息: {query}")
self.llm_finish_task = False
if query is not None:
self.logger.bind(tag=TAG).info(f"大模型收到用户消息: {query}")
# 为最顶层时新建会话ID和发送FIRST请求
if depth == 0:
self.llm_finish_task = False
self.sentence_id = str(uuid.uuid4().hex)
self.dialogue.put(Message(role="user", content=query))
self.tts.tts_text_queue.put(
@@ -738,9 +739,23 @@ class ConnectionHandler:
)
)
# 设置最大递归深度，避免无限循环，可根据实际需求调整
MAX_DEPTH = 5
force_final_answer = False # 标记是否强制最终回答
if depth >= MAX_DEPTH:
self.logger.bind(tag=TAG).debug(f"已达到最大工具调用深度 {MAX_DEPTH}，将强制基于现有信息回答")
force_final_answer = True
# 添加系统指令，要求 LLM 基于现有信息回答
self.dialogue.put(Message(
role="user",
content="[系统提示] 已达到最大工具调用次数限制，请你基于目前已经获取的所有信息，直接给出最终答案。不要再尝试调用任何工具。"
))
# Define intent functions
functions = None
if self.intent_type == "function_call" and hasattr(self, "func_handler"):
# 达到最大深度时，禁用工具调用，强制 LLM 直接回答
if self.intent_type == "function_call" and hasattr(self, "func_handler") and not force_final_answer:
functions = self.func_handler.get_functions()
response_message = []
@@ -775,9 +790,8 @@ class ConnectionHandler:
# 处理流式响应
tool_call_flag = False
function_name = None
function_id = None
function_arguments = ""
# 支持多个并行工具调用 - 使用列表存储
tool_calls_list = [] # 格式: [{"id": "", "name": "", "arguments": ""}]
content_arguments = ""
self.client_abort = False
emotion_flag = True
@@ -798,12 +812,7 @@ class ConnectionHandler:
if tools_call is not None and len(tools_call) > 0:
tool_call_flag = True
if tools_call[0].id is not None and tools_call[0].id != "":
function_id = tools_call[0].id
if tools_call[0].function.name is not None and tools_call[0].function.name != "":
function_name = tools_call[0].function.name
if tools_call[0].function.arguments is not None:
function_arguments += tools_call[0].function.arguments
self._merge_tool_calls(tool_calls_list, tools_call)
else:
content = response
@@ -829,16 +838,17 @@ class ConnectionHandler:
# 处理function call
if tool_call_flag:
bHasError = False
if function_id is None:
# 处理基于文本的工具调用格式
if len(tool_calls_list) == 0 and content_arguments:
a = extract_json_from_string(content_arguments)
if a is not None:
try:
content_arguments_json = json.loads(a)
function_name = content_arguments_json["name"]
function_arguments = json.dumps(
content_arguments_json["arguments"], ensure_ascii=False
)
function_id = str(uuid.uuid4().hex)
tool_calls_list.append({
"id": str(uuid.uuid4().hex),
"name": content_arguments_json["name"],
"arguments": json.dumps(content_arguments_json["arguments"], ensure_ascii=False)
})
except Exception as e:
bHasError = True
response_message.append(a)
@@ -849,30 +859,41 @@ class ConnectionHandler:
self.logger.bind(tag=TAG).error(
f"function call error: {content_arguments}"
)
if not bHasError:
if not bHasError and len(tool_calls_list) > 0:
# 如需要大模型先处理一轮，添加相关处理后的日志情况
if len(response_message) > 0:
text_buff = "".join(response_message)
self.tts_MessageText = text_buff
self.dialogue.put(Message(role="assistant", content=text_buff))
response_message.clear()
self.logger.bind(tag=TAG).debug(
f"function_name={function_name}, function_id={function_id}, function_arguments={function_arguments}"
)
function_call_data = {
"name": function_name,
"id": function_id,
"arguments": function_arguments,
}
# 使用统一工具处理器处理所有工具调用
result = asyncio.run_coroutine_threadsafe(
self.func_handler.handle_llm_function_call(
self, function_call_data
),
self.loop,
).result()
self._handle_function_result(result, function_call_data, depth=depth)
self.logger.bind(tag=TAG).debug(
f"检测到 {len(tool_calls_list)} 个工具调用"
)
# 收集所有工具调用的 Future
futures_with_data = []
for tool_call_data in tool_calls_list:
self.logger.bind(tag=TAG).debug(
f"function_name={tool_call_data['name']}, function_id={tool_call_data['id']}, function_arguments={tool_call_data['arguments']}"
)
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))
# 等待协程结束（实际等待时长为最慢的那个）
tool_results = []
for future, tool_call_data in futures_with_data:
result = future.result()
tool_results.append((result, tool_call_data))
# 统一处理所有工具调用结果
if tool_results:
self._handle_function_result(tool_results, depth=depth)
# 存储对话内容
if len(response_message) > 0:
@@ -887,64 +908,61 @@ class ConnectionHandler:
content_type=ContentType.ACTION,
)
)
self.llm_finish_task = True
# 使用lambda延迟计算，只有在DEBUG级别时才执行get_llm_dialogue()
self.logger.bind(tag=TAG).debug(
lambda: json.dumps(
self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False
self.llm_finish_task = True
# 使用lambda延迟计算，只有在DEBUG级别时才执行get_llm_dialogue()
self.logger.bind(tag=TAG).debug(
lambda: json.dumps(
self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False
)
)
)
return True
def _handle_function_result(self, result, function_call_data, depth):
if result.action == Action.RESPONSE: # 直接回复前端
text = result.response
self.tts.tts_one_sentence(self, ContentType.TEXT, content_detail=text)
self.dialogue.put(Message(role="assistant", content=text))
elif result.action == Action.REQLLM: # 调用函数后再请求llm生成回复
text = result.result
if text is not None and len(text) > 0:
function_id = function_call_data["id"]
function_name = function_call_data["name"]
function_arguments = function_call_data["arguments"]
self.dialogue.put(
Message(
role="assistant",
tool_calls=[
{
"id": function_id,
"function": {
"arguments": (
"{}"
if function_arguments == ""
else function_arguments
),
"name": function_name,
},
"type": "function",
"index": 0,
}
],
)
)
def _handle_function_result(self, tool_results, depth):
need_llm_tools = []
self.dialogue.put(
Message(
role="tool",
tool_call_id=(
str(uuid.uuid4()) if function_id is None else function_id
for result, tool_call_data in tool_results:
if result.action in [Action.RESPONSE, Action.NOTFOUND, Action.ERROR]: # 直接回复前端
text = result.response if result.response else result.result
self.tts.tts_one_sentence(self, ContentType.TEXT, content_detail=text)
self.dialogue.put(Message(role="assistant", content=text))
elif result.action == Action.REQLLM:
# 收集需要 LLM 处理的工具
need_llm_tools.append((result, tool_call_data))
else:
pass
if need_llm_tools:
all_tool_calls = [
{
"id": tool_call_data["id"],
"function": {
"arguments": (
"{}"
if tool_call_data["arguments"] == ""
else tool_call_data["arguments"]
),
content=text,
"name": tool_call_data["name"],
},
"type": "function",
"index": idx,
}
for idx, (_, tool_call_data) in enumerate(need_llm_tools)
]
self.dialogue.put(Message(role="assistant", tool_calls=all_tool_calls))
for result, tool_call_data in need_llm_tools:
text = result.result
if text is not None and len(text) > 0:
self.dialogue.put(
Message(
role="tool",
tool_call_id=str(uuid.uuid4()) if tool_call_data["id"] is None else tool_call_data["id"],
content=text,
)
)
)
self.chat(text, depth=depth + 1)
elif result.action == Action.NOTFOUND or result.action == Action.ERROR:
text = result.response if result.response else result.result
self.tts.tts_one_sentence(self, ContentType.TEXT, content_detail=text)
self.dialogue.put(Message(role="assistant", content=text))
else:
pass
self.chat(None, depth=depth + 1)
def _report_worker(self):
"""聊天记录上报工作线程"""
@@ -1144,3 +1162,31 @@ class ConnectionHandler:
self.logger.bind(tag=TAG).error(f"超时检查任务出错: {e}")
finally:
self.logger.bind(tag=TAG).info("超时检查任务已退出")
def _merge_tool_calls(self, tool_calls_list, tools_call):
"""合并工具调用列表
Args:
tool_calls_list: 已收集的工具调用列表
tools_call: 新的工具调用
"""
for tool_call in tools_call:
tool_index = getattr(tool_call, 'index', None)
if tool_index is None:
if tool_call.function.name:
# 有 function_name，说明是新的工具调用
tool_index = len(tool_calls_list)
else:
tool_index = len(tool_calls_list) - 1 if tool_calls_list else 0
# 确保列表有足够的位置
if tool_index >= len(tool_calls_list):
tool_calls_list.append({"id": "", "name": "", "arguments": ""})
# 更新工具调用信息
if tool_call.id:
tool_calls_list[tool_index]["id"] = tool_call.id
if tool_call.function.name:
tool_calls_list[tool_index]["name"] = tool_call.function.name
if tool_call.function.arguments:
tool_calls_list[tool_index]["arguments"] += tool_call.function.arguments