Merge pull request #3196 from xinnan-tech/py-device-calling

Py device calling

main/xiaozhi-server/config/config_loader.py

@@ -7,6 +7,7 @@ from config.manage_api_client import (
     get_server_config,
     get_agent_models,
     get_correct_words,
+    lookup_address_book,
     DeviceNotFoundException,
     DeviceBindException,
 )

main/xiaozhi-server/config/manage_api_client.py

@@ -245,6 +245,20 @@ async def report(
         return None
 
 
+async def lookup_address_book(caller_mac: str, nickname: str) -> Optional[Dict]:
+    """根据昵称查找目标设备"""
+    if not ManageApiClient._instance:
+        return None
+    try:
+        return await ManageApiClient._instance._execute_async_request(
+            "GET",
+            f"/device/address-book/lookup?callerMac={caller_mac}&nickname={nickname}",
+        )
+    except Exception as e:
+        print(f"通讯录查找失败: {e}")
+        return None
+
+
 def init_service(config):
     ManageApiClient(config)
 

main/xiaozhi-server/core/connection.py

@@ -190,6 +190,9 @@ class ConnectionHandler:
         # 初始化提示词管理器
         self.prompt_manager = PromptManager(self.config, self.logger)
 
+        # 初始化通话状态
+        self.calling = False
+
     async def handle_connection(self, ws: websockets.ServerConnection):
         try:
             # 获取运行中的事件循环（必须在异步上下文中）

main/xiaozhi-server/core/handle/sendAudioHandle.py

@@ -47,7 +47,8 @@ async def sendAudioMessage(conn: "ConnectionHandler", sentenceType, audios, text
         conn.logger.bind(tag=TAG).info(f"发送音频消息: {sentenceType}, {text}")
 
     # 发送结束消息（如果是最后一个文本）
-    if sentenceType == SentenceType.LAST:
+    # 通话需要维持speaking状态
+    if not conn.calling and sentenceType == SentenceType.LAST:
         await send_tts_message(conn, "stop", None)
         if conn.close_after_chat:
             await conn.close()

main/xiaozhi-server/core/handle/textHandler/listenMessageHandler.py

@@ -1,17 +1,21 @@
 import time
+import uuid
 import asyncio
 from typing import Dict, Any, TYPE_CHECKING
 
 if TYPE_CHECKING:
     from core.connection import ConnectionHandler
 
+from core.utils.dialogue import Message
+from core.providers.asr.dto.dto import InterfaceType
 from core.handle.receiveAudioHandle import startToChat
 from core.handle.reportHandle import enqueue_asr_report
 from core.handle.sendAudioHandle import send_stt_message, send_tts_message
 from core.handle.textMessageHandler import TextMessageHandler
 from core.handle.textMessageType import TextMessageType
 from core.utils.util import remove_punctuation_and_length
-from core.providers.asr.dto.dto import InterfaceType
+from core.providers.tts.dto.dto import ContentType, TTSMessageDTO, SentenceType
+
 
 TAG = __name__
 
@@ -54,6 +58,25 @@ class ListenTextMessageHandler(TextMessageHandler):
                     original_text
                 )
 
+                # 检查是否是设备呼叫指令 [device_call]
+                if original_text.startswith("[device_call]"):
+                    # 提取 tag 后的文本
+                    call_text = original_text[len("[device_call]"):].strip()
+                    conn.logger.bind(tag=TAG).info(f"收到设备呼叫指令: {call_text}")
+
+                    # 准备开始新会话
+                    conn.sentence_id = uuid.uuid4().hex
+
+                    await send_stt_message(conn, call_text)
+                    conn.tts.store_tts_text(conn.sentence_id, call_text)
+                    conn.tts.tts_text_queue.put(TTSMessageDTO(sentence_id=conn.sentence_id, sentence_type=SentenceType.FIRST, content_type=ContentType.ACTION))
+                    conn.tts.tts_one_sentence(conn, ContentType.TEXT, content_detail=call_text)
+                    conn.tts.tts_text_queue.put(TTSMessageDTO(sentence_id=conn.sentence_id, sentence_type=SentenceType.LAST, content_type=ContentType.ACTION))
+
+                    # 添加到对话历史，让模型理解上下文
+                    conn.dialogue.put(Message(role="assistant", content=call_text))
+                    return
+
                 # 识别是否是唤醒词
                 is_wakeup_words = filtered_text in conn.config.get("wakeup_words")
                 # 是否开启唤醒词回复

main/xiaozhi-server/plugins_func/functions/call_device.py

@@ -0,0 +1,108 @@
+"""呼叫设备工具"""
+import requests
+from typing import TYPE_CHECKING
+
+from config.logger import setup_logging
+from plugins_func.register import register_function, ToolType, ActionResponse, Action
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
+
+TAG = __name__
+logger = setup_logging()
+
+call_device_function_desc = {
+    "type": "function",
+    "function": {
+        "name": "call_device",
+        "description": (
+            "用于设备之间建立语音通话连接。"
+            "当用户说出以下意图时调用此工具：\n"
+            "1. 主动呼叫：用户说”呼叫XX/打电话给XX/连线XX/打给XX/帮我呼叫XX”时调用，nickname取XX。"
+            "例如：”呼叫张三”→nickname=”张三”、”帮我连线小陈”→nickname=”小陈”；\n"
+            "2. 接听来电：系统刚提示”您收到来自XX的来电，是否接听？”后，用户说”接听/接通/同意接听/同意连线/同意对话”时调用，"
+            "nickname取提示中的XX。\n"
+            "如果用户输入不是明确接听，也不是明确拒绝，不得调用call_device，必须先追问一次"
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "nickname": {"type": "string", "description": "目标设备的备注名，例如：小陈、小翁"},
+            },
+            "required": ["nickname"],
+        },
+    },
+}
+
+
+def _request_api(url: str, params: dict, headers: dict) -> requests.Response:
+    return requests.get(url, params=params, headers=headers, timeout=10)
+
+
+def _failed_reply(msg: str) -> ActionResponse:
+    return ActionResponse(action=Action.RESPONSE, response=msg)
+
+
+@register_function("call_device", call_device_function_desc, ToolType.SYSTEM_CTL)
+def call_device(conn: "ConnectionHandler", nickname: str):
+    caller_mac = conn.headers.get("device-id")
+    if not caller_mac:
+        return _failed_reply("无法获取本机MAC地址")
+
+    api_config = conn.config.get("manager-api", {})
+    api_url = api_config.get("url")
+    api_secret = api_config.get("secret")
+    if not api_url or not api_secret:
+        logger.bind(tag=TAG).error("manager-api配置缺失")
+        return _failed_reply("配置错误，请稍后再试")
+
+    headers = {"Authorization": f"Bearer {api_secret}"}
+
+    # 查询通讯录
+    try:
+        resp = _request_api(
+            f"{api_url}/device/address-book/lookup",
+            params={"callerMac": caller_mac, "nickname": nickname},
+            headers=headers,
+        )
+        result = resp.json()
+    except requests.RequestException as e:
+        logger.bind(tag=TAG).error(f"通讯录查找请求失败: {e}")
+        return _failed_reply("通讯录查询失败，请稍后再试")
+
+    if result.get("code") != 0 or not result.get("data"):
+        return _failed_reply(f"未找到备注为'{nickname}'的设备")
+
+    data = result["data"]
+    target_mac = data.get("targetMac")
+    caller_nickname = data.get("callerNickname")
+    has_permission = data.get("hasPermission")
+
+    if not target_mac:
+        return _failed_reply(f"未找到备注为'{nickname}'的设备")
+    if not caller_nickname:
+        return _failed_reply("呼叫失败，您不是对方的联系人")
+    if not has_permission:
+        return _failed_reply("呼叫失败，您没有权限呼叫该设备")
+
+    # 通过Java中转调用网关
+    try:
+        resp = _request_api(
+            f"{api_url}/device/call/forward",
+            params={"callerMac": caller_mac, "targetMac": target_mac, "callerNickname": caller_nickname},
+            headers=headers,
+        )
+        result = resp.json()
+    except requests.RequestException as e:
+        logger.bind(tag=TAG).error(f"呼叫请求转发失败: {e}")
+        return _failed_reply("呼叫失败，请稍后再试")
+
+    if result.get("code") != 0:
+        return _failed_reply(result.get("msg", "呼叫失败"))
+
+    call_data = result.get("data", {})
+    if call_data.get("status") == "offline":
+        return _failed_reply(call_data.get("message"))
+
+    conn.calling = True
+    return ActionResponse(action=Action.NONE, response=f"正在呼叫{nickname}，请等待对方接听")