小智终端 5种 场景模式 OK

wangzhibo

2026-09-11 659c470eba542383987aad227b7f918100f131fa

 xiaozhi-esp32-server/main/xiaozhi-server/config/manage_api_client.py

@@ -221,31 +221,55 @@


async def report(
    mac_address: str, session_id: str, chat_type: int, content: str, audio, report_time
    mac_address: str, session_id: str, chat_type: int, content: str, audio, report_time,
    task_instance_id: str = None, step_index: int = None,
) -> Optional[Dict]:
    """异步聊天记录上报"""
    if not content or not ManageApiClient._instance:
        return None
    try:
        payload = {
            "macAddress": mac_address,
            "sessionId": session_id,
            "chatType": chat_type,
            "content": content,
            "reportTime": report_time,
            "audioBase64": (
                base64.b64encode(audio).decode("utf-8") if audio else None
            ),
        }
        if task_instance_id:
            payload["taskInstanceId"] = task_instance_id
        if step_index is not None:
            payload["stepIndex"] = step_index
        return await ManageApiClient._instance._execute_async_request(
            "POST",
            f"/agent/chat-history/report",
            json={
                "macAddress": mac_address,
                "sessionId": session_id,
                "chatType": chat_type,
                "content": content,
                "reportTime": report_time,
                "audioBase64": (
                    base64.b64encode(audio).decode("utf-8") if audio else None
                ),
            },
            json=payload,
        )
    except Exception as e:
        print(f"TTS上报失败: {e}")
        return None


async def upsert_task_instance(payload: Dict) -> Optional[Dict]:
    """上报 TaskSession 全文。失败返回 None，不抛给语音热路径。未配置管理端视为成功。"""
    if not payload or not ManageApiClient._instance:
        return {"skipped": True}
    instance_id = payload.get("task_session_id") or ""
    if not instance_id:
        return {"skipped": True}
    try:
        return await ManageApiClient._instance._async_request(
            "PUT",
            f"/task/instance/{instance_id}",
            json=payload,
        )
    except Exception as e:
        print(f"TaskInstance 上报失败: {e}")
        return None


async def lookup_address_book(caller_mac: str, nickname: str) -> Optional[Dict]:
    """根据昵称查找目标设备"""
    if not ManageApiClient._instance: