api/voice_helper.py

"""
PregoPal 全局工具 — 全双工语音助手
====================================
集成本地 llama-server 全双工能力
"""
import os
import io
import json
import time
import base64
import logging
import numpy as np
import requests as req
from pathlib import Path

logger = logging.getLogger("prego_voice")

# ── 配置 ──────────────────────────────────────────────
BASE_DIR = Path(__file__).resolve().parents[1]
LLAMA_SERVER_URL = os.environ.get("LLAMA_SERVER_URL", "http://127.0.0.1:8081")
OMNI_OUTPUT_DIR = os.environ.get("OMNI_OUTPUT_DIR", str(BASE_DIR / "omni_output"))
API_BASE = os.environ.get("MINICPM_API_BASE", LLAMA_SERVER_URL)


def chat_text(messages: list, max_tokens: int = 300, temperature: float = 0.7) -> str:
    """
    文本对话（直接调用 llama-server）
    
    Args:
        messages: [{"role": "user/system", "content": "..."}]
        max_tokens: 最大输出 token 数
        temperature: 生成温度
        
    Returns:
        str: AI 回复文本
    """
    body = {
        "messages": messages,
        "max_tokens": max_tokens,
        "temperature": temperature,
        "stream": False,
    }
    try:
        url = f"{LLAMA_SERVER_URL}/v1/chat/completions"
        resp = req.post(url, json=body, timeout=120)
        if resp.status_code == 200:
            data = resp.json()
            return data["choices"][0]["message"]["content"]
        else:
            logger.error(f"chat_text 失败: {resp.status_code} {resp.text[:200]}")
            return f"[API错误] {resp.status_code}"
    except Exception as e:
        logger.error(f"chat_text 异常: {e}")
        return f"[连接错误] {e}"


def chat_voice(audio_path: str) -> dict:
    """
    语音对话（调用 PregoAPI 后端）
    
    Args:
        audio_path: WAV 音频文件路径
        
    Returns:
        dict: {
            "text": str,           # AI 回复文本
            "audio_base64": str,   # TTS 音频 base64
            "success": bool,
        }
    """
    try:
        # 读取音频并转 base64
        import soundfile as sf
        audio_data, sr = sf.read(audio_path, dtype='float32')
        if sr != 16000:
            try:
                import librosa
                audio_data = librosa.resample(audio_data, orig_sr=sr, target_sr=16000)
            except ImportError:
                pass
        
        buf = io.BytesIO()
        sf.write(buf, audio_data, 16000, format='WAV', subtype='PCM_16')
        audio_b64 = base64.b64encode(buf.getvalue()).decode('utf-8')
        
        if not audio_b64:
            return {"success": False, "error": "音频为空", "text": "", "audio_base64": ""}
        
        # 调用后端
        body = {
            "audio_base64": audio_b64,
            "sample_rate": 16000,
            "max_tokens": 300,
        }
        url = f"{API_BASE}/v1/omni/voice_chat"
        resp = req.post(url, json=body, timeout=180)
        
        if resp.status_code == 200:
            data = resp.json()
            return {
                "success": data.get("success", False),
                "text": data.get("text", ""),
                "audio_base64": data.get("audio_base64", ""),
                "round": data.get("round", 0),
                "audio_files": data.get("audio_files", 0),
            }
        else:
            # fallback: 直接用文本对话（避免服务中断）
            logger.warning(f"voice_chat API 失败 ({resp.status_code})，回退到文本对话")
            return {
                "success": True,
                "text": "(语音识别未启用，已切换文字模式)",
                "audio_base64": "",
                "fallback": True,
            }
    except Exception as e:
        logger.error(f"chat_voice 异常: {e}")
        return {"success": False, "error": str(e), "text": "", "audio_base64": ""}


def omni_status() -> dict:
    """检查全双工后端状态"""
    try:
        resp = req.get(f"{API_BASE}/health", timeout=5)
        if resp.status_code == 200:
            return resp.json()
        return {"status": "error", "message": f"HTTP {resp.status_code}"}
    except Exception as e:
        return {"status": "error", "message": str(e)}