# VibeVoice 支持多种 VAD 引擎
class VADProviders:
    - Silero_VAD      # 轻量高效，CPU 友好
    - WebRTC_VAD      # 实时性好，业界广泛使用#
    - Maus_VAD         # 高精度，适合研究场景

# config/asr.yaml
asr:
  provider: "whisper"
  model: "large-v3"
  language: "auto"  # 自动检测语言
  vad: "silero"
  
  # 或使用 Azure
  provider: "azure"
  speech_key: "${AZURE_SPEECH_KEY}"
  speech_region: "eastus"

# LLM 配置示例
llm:
  provider: "openai"
  model: "gpt-4o-audio-preview"  # 支持音频的 GPT-4o
  temperature: 0.7
  streaming: true
  
  # 或使用 Claude
  provider: "anthropic"
  model: "claude-sonnet-4-20260219"
  audio_output: true  # Claude 的音频输出模式

# 流式 TTS 配置
tts:
  provider: "edge"
  voice: "zh-CN-XiaoxiaoNeural"  # 中文音色
  rate: "+0%"      # 语速调整
  pitch: "+0Hz"    # 音调调整
  
  # 流式播放配置
  stream_chunk_ms: 100  # 每 100ms 发送一个音频块

# skill_example.py
from vibevoice.skills import Skill, register

@register("weather")
class WeatherSkill(Skill):
    name = "天气查询"
    description = "查询指定城市的天气情况"
    
    async def execute(self, context: dict) -> str:
        city = context.get("params", {}).get("city", "北京")
        # 调用天气 API
        weather = await self.call_api(f"/weather?city={city}")
        return f"{city}今天天气：{weather['desc']}，气温{weather['temp']}度"
    
    def get_schema(self) -> dict:
        return {
            "name": "weather",
            "description": "查询城市天气",
            "parameters": {
                "city": {"type": "string", "description": "城市名称"}
            }
        }

# 支持多轮对话，自动维护上下文
conversation = await vibevoice.create_session(
    user_id="user123",
    system_prompt="你是小微，一个友好的语音助手。"
)

# 语音输入 → 自动识别 → LLM 推理 → 语音输出
result = await conversation.voice_chat(audio_stream=microphone_stream)
print(result.text)  # 文本记录

# 多语言配置
language:
  detection: "auto"  # 自动检测
  supported:
    - zh-CN    # 简体中文
    - en-US    # 英语
    - ja-JP    # 日语
    - ko-KR    # 韩语
  default: "zh-CN"

# 注册工具函数
@vibevoice.tool("calculate")
def calculator(expression: str) -> float:
    """计算数学表达式"""
    return eval(expression)

# AI 自动调用工具
user: "帮我计算 123 加 456 乘以 2"
# AI 自动调用 calculator 工具
# 返回：1035

# RAG 知识库问答
await conversation.enable_rag(
    vector_store="your-vector-db",
    top_k=5,
    similarity_threshold=0.7
)

# 情绪识别配置
emotion:
  enabled: true
  model: "emotion-classifier-v1"
  
# AI 根据情绪调整回复风格
user_tone = "焦急"
# AI 回复风格自动调整为：语速加快、语气安抚、简洁直接

pip install vibevoice
vibevoice --version

git clone https://github.com/microsoft/VibeVoice.git
cd VibeVoice
pip install -e .

# 拉取镜像
docker pull vibevoice/vibevoice:latest

# 运行容器
docker run -d \
  --name vibevoice \
  -p 8080:8080 \
  -v ~/.vibevoice:/root/.vibevoice \
  --device /dev/snd:/dev/snd \
  vibevoice/vibevoice:latest

# 创建配置文件
mkdir -p ~/.vibevoice
cat > ~/.vibevoice/config.yaml << EOF
llm:
  provider: "openai"
  api_key: "${OPENAI_API_KEY}"
  
asr:
  provider: "whisper"
  model: "large-v3"

tts:
  provider: "edge"
  voice: "zh-CN-XiaoxiaoNeural"
EOF

vibevoice web --port 8080
# 打开浏览器访问 http://localhost:8080

# 直接语音对话
vibevoice chat --voice

# 或文本对话
vibevoice chat --text

import asyncio
from vibevoice import VibeVoice

async def main():
    # 初始化
    vv = VibeVoice(config_path="~/.vibevoice/config.yaml")
    
    # 创建对话会话
    session = await vv.create_session(
        user_id="user_001",
        system_prompt="你是一个专业的健身教练。"
    )
    
    # 语音对话
    print("开始对话（按 Ctrl+C 退出）...")
    async for result in session.voice_loop():
        print(f"用户：{result.user_text}")
        print(f"AI：{result.response_text}")
        print(f"置信度：{result.confidence:.2%}")

if __name__ == "__main__":
    asyncio.run(main())

# 精度优先（需要 GPU）
asr:
  model: "large-v3"  # ~3GB 显存
  
# 速度优先（CPU 可用）
asr:
  model: "base"  # ~140MB

# 使用本地 TTS 减少延迟
tts:
  provider: "coqui"
  model: "xtts_v2"  # 本地运行
  # 无需网络请求，延迟降低 50%+

# 使用 GPU 加速 Whisper
export CUDA_VISIBLE_DEVICES=0
vibevoice chat --device cuda

from vibevoice.asr.base import BaseASR

class MyASR(BaseASR):
    name = "my_asr"
    
    async def recognize(self, audio_chunk: bytes) -> str:
        # 实现自己的 ASR 逻辑
        result = await self.my_asr_api(audio_chunk)
        return result.text
    
    async def detect_speech_end(self, audio_chunk: bytes) -> bool:
        # 实现 VAD 逻辑
        return await self.my_vad_model(audio_chunk)

# 注册引擎
vv.register_asr("my_asr", MyASR())

from vibevoice.tts.base import BaseTTS

class MyTTS(BaseTTS):
    name = "my_tts"
    
    async def synthesize(self, text: str) -> bytes:
        # 返回 WAV/MP3 格式音频
        audio = await self.my_tts_api(text)
        return audio
    
    def stream_audio(self, text: str):
        # 流式音频生成
        for chunk in self.my_streaming_tts(text):
            yield chunk

vv.register_tts("my_tts", MyTTS())

from vibevoice.llm.base import BaseLLM

class MyLLM(BaseLLM):
    name = "my_llm"
    
    async def generate(self, messages: list, **kwargs):
        response = await self.my_llm_api(messages)
        return response.text
    
    async def stream_generate(self, messages: list):
        async for chunk in self.my_streaming_api(messages):
            yield chunk

vv.register_llm("my_llm", MyLLM())

# 开发自定义 WebSocket 接口
from vibevoice.api.websocket import WebSocketHandler

class CustomWSHandler(WebSocketHandler):
    async def on_voice_frame(self, frame: bytes):
        # 处理自定义音频帧
        pass
    
    async def on_llm_token(self, token: str):
        # 处理 LLM 流式输出
        await self.send_json({"token": token})

# 注册到 API 服务器
vv.api_server.register("/custom", CustomWSHandler)

# 减少音频延迟
audio_config = {
    "chunk_size_ms": 100,      # 减小到 100ms
    "sample_rate": 16000, "channels": 1,              # 单声道
    "codec": "pcm_s16le"       # 无压缩 PCM
}

# 多用户并发支持
vv = VibeVoice(
    max_concurrent_sessions=100,
    session_timeout=300  # 5分钟超时
)

# 生产环境安全配置
security:
  api_key_required: true
  rate_limit:
    requests_per_minute: 60
    sessions_per_user: 5
    
  audio:
    max_duration_seconds: 300  # 最大语音时长
    allowed_formats: ["pcm", "wav"]
    
  logging:
    log_audio: false  # 生产环境关闭音频日志

# 接入监控系统
vv.monitor = {
    "prometheus_port": 9090,
    "metrics": [
        "latency.asr",
        "latency.llm",
        "latency.tts",
        "latency.end_to_end",
        "sessions.active",
        "sessions.total"
    ]
}

# docker-compose.yml
version: '3.8'
services:
  vibevoice:
    image: vibevoice/vibevoice:latest
    ports:
      - "8080:8080"
    volumes:
      - ./config:/root/.vibevoice
      - ./data:/root/.vibevoice/data
    devices:
      - /dev/snd:/dev/snd
    environment:
      - CUDA_VISIBLE_DEVICES=0
    restart: unless-stopped
    
  # 可选：本地 LLM
  ollama:
    image: ollama/ollama:latest
    ports:
      - "11434:11434"
    volumes:
      - ollama:/root/.ollama
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]

# 微信公众平台对接示例
from vibevoice.integrations import WeChatAdapter

adapter = WeChatAdapter(
    app_id="your_app_id",
    app_secret="your_secret"
)

@adapter.on_voice_message()
async def handle_voice(msg):
    audio = await msg.download_voice()
    response = await vibevoice.process(audio)
    await msg.reply_voice(response.audio)

# 安装
pip install vibevoice

# 配置
vibevoice config init
vibevoice config set llm.provider openai
vibevoice config set llm.api_key YOUR_KEY

# 运行
vibevoice web --port 8080      # Web UI
vibevoice chat --voice         # 语音对话
vibevoice chat --text          # 文本对话

# 开发
vibevoice dev server           # 开发服务器
vibevoice dev test            # 运行测试

# 管理
vibevoice sessions list        # 查看会话
vibevoice sessions kill SESSION_ID  # 关闭会话