# 角色提示示例
role_prompts = {
    "tech_mentor": """
你是一位有20年经验的软件架构师。
你擅长用简单的方式解释复杂的技术概念。
回答问题时：
1. 先给出一句话总结
2. 然后提供详细解释
3. 最后给出一个实际案例
你的语气专业但亲切，像一位导师在指导学生。
""",

    "language_tutor": """
你是一位耐心的英语教师，专门帮助中文母语者学习英语。
你会：
- 用中文解释英语语法规则
- 提供英汉对照的例句
- 纠正常见的中文式英语错误
- 鼓励学生多练习
你的语气温暖、有耐心，经常给予正面反馈。
""",

    "data_analyst": """
你是一位专业的数据分析顾问。
你擅长：
- SQL 查询优化
- 数据可视化建议
- 统计分析方法选择
- 业务指标体系设计
回答时使用精确的术语，并提供代码示例。
语气：专业、简洁、数据驱动。
"""
}

# 通过 API 设置角色
import requests

def create_session(role_name: str) -> str:
    """创建带有角色提示的对话会话"""
    prompt = role_prompts.get(role_name, "")

    response = requests.post(
        "https://api.personaplex.example.com/v1/sessions",
        json={
            "model": "moshi",
            "system_prompt": prompt,
            "voice_preset": "professional_female",
            "language": "zh-CN",
        },
        headers={"Authorization": "Bearer YOUR_TOKEN"}
    )
    return response.json()["session_id"]

# Personaplex 音频条件示例
# 通过参考音频控制角色的语音特征

import asyncio
import websockets
import json
import wave

async def voice_conditioned_session(
    reference_audio_path: str,
    text_prompt: str,
    user_audio_path: str
):
    """使用音频条件的语音对话"""

    # 加载参考音频（定义角色的声音特征）
    with open(reference_audio_path, "rb") as f:
        reference_audio = f.read()

    # 加载用户输入音频
    with open(user_audio_path, "rb") as f:
        user_audio = f.read()

    async with websockets.connect(
        "wss://api.personaplex.example.com/v1/stream",
        extra_headers={"Authorization": "Bearer YOUR_TOKEN"}
    ) as ws:
        # 1. 发送会话配置
        config = {
            "type": "config",
            "model": "moshi",
            "text_prompt": text_prompt,
            "audio_condition": True,  # 启用音频条件
            "sample_rate": 24000,
            "language": "zh-CN",
        }
        await ws.send(json.dumps(config))

        # 2. 发送参考音频（角色声音）
        ref_msg = {
            "type": "reference_audio",
            "format": "wav",
            "data": reference_audio.hex(),
        }
        await ws.send(json.dumps(ref_msg))

        # 3. 发送用户音频输入
        input_msg = {
            "type": "input_audio",
            "format": "wav",
            "data": user_audio.hex(),
        }
        await ws.send(json.dumps(input_msg))

        # 4. 接收响应音频流
        output_frames = []
        while True:
            response = await ws.recv()
            data = json.loads(response)

            if data["type"] == "audio_chunk":
                output_frames.append(bytes.fromhex(data["data"]))
            elif data["type"] == "text":
                print(f"文本: {data['content']}")
            elif data["type"] == "end":
                break

        # 保存输出音频
        with wave.open("response.wav", "wb") as wf:
            wf.setnchannels(1)
            wf.setsampwidth(2)
            wf.setframerate(24000)
            wf.writeframes(b"".join(output_frames))

        print("Response saved to response.wav")

# 运行
asyncio.run(voice_conditioned_session(
    reference_audio_path="reference_voice.wav",
    text_prompt="你是一位温暖的家庭教师",
    user_audio_path="user_question.wav"
))

# Personaplex 性能配置示例

# server_config.yaml
server:
  host: "0.0.0.0"
  port: 8080
  ssl: true
  ssl_cert: "/etc/ssl/cert.pem"
  ssl_key: "/etc/ssl/key.pem"

model:
  name: "moshi"
  # 量化配置
  quantization: "fp16"  # fp32 | fp16 | int8
  device: "cuda:0"

  # 并发配置
  max_concurrent_sessions: 16
  session_timeout: 300  # 5分钟无活动超时

  # 缓冲区配置
  input_buffer_ms: 200     # 输入缓冲 200ms
  output_chunk_ms: 100     # 每 100ms 发送一个输出块

performance:
  # 预热模型
  warmup_steps: 3

  # 音频处理
  sample_rate: 24000
  frame_size: 1920   # 80ms at 24kHz

  # 延迟目标
  target_first_byte_ms: 300
  target_total_ms: 500

# 启动命令
# python -m moshi.server --config server_config.yaml

# 性能监控
import time
import psutil
import pynvml

def monitor_performance():
    """监控 GPU 和系统性能"""
    pynvml.nvmlInit()
    handle = pynvml.nvmlDeviceGetHandleByIndex(0)

    while True:
        # GPU 使用率
        util = pynvml.nvmlDeviceGetUtilizationRates(handle)
        mem = pynvml.nvmlDeviceGetMemoryInfo(handle)

        # 系统资源
        cpu = psutil.cpu_percent()
        ram = psutil.virtual_memory()

        print(f"GPU: {util.gpu}% | "
              f"VRAM: {mem.used/1024**3:.1f}/{mem.total/1024**3:.1f}GB | "
              f"CPU: {cpu}% | "
              f"RAM: {ram.percent}%")
        time.sleep(5)

"""多角色对话编排系统"""
import asyncio
import json
from dataclasses import dataclass
from typing import List

@dataclass
class Persona:
    name: str
    text_prompt: str
    voice_reference: str
    session_id: str = ""

class MultiPersonaConversation:
    """管理多个角色参与的对话"""

    def __init__(self):
        self.personas: List[Persona] = []
        self.conversation_history = []

    def add_persona(self, persona: Persona):
        """添加角色到对话"""
        self.personas.append(persona)

    async def orchestrate(self, topic: str, rounds: int = 3):
        """编排多角色对话"""

        self.conversation_history.append({
            "role": "system",
            "content": f"讨论主题：{topic}"
        })

        for round_num in range(rounds):
            print(f"\n=== 第 {round_num + 1} 轮 ===")

            for persona in self.personas:
                # 构建上下文（包含其他角色的发言）
                context = self._build_context(persona)

                # 生成该角色的发言
                response = await self._generate_response(
                    persona, context
                )

                self.conversation_history.append({
                    "role": persona.name,
                    "content": response,
                })

                print(f"[{persona.name}]: {response[:100]}...")

    def _build_context(self, current_persona: Persona) -> str:
        """构建角色的对话上下文"""
        recent = self.conversation_history[-6:]  # 最近6轮
        context_lines = []

        for msg in recent:
            prefix = msg["role"]
            if msg["role"] == current_persona.name:
                prefix = "你之前说过"
            context_lines.append(f"{prefix}: {msg['content']}")

        return "\n".join(context_lines)

    async def _generate_response(self, persona: Persona, context: str) -> str:
        """调用 Personaplex 生成角色发言"""
        # 在实际实现中，这里调用 Personaplex API
        prompt = f"""{persona.text_prompt}

当前对话上下文：
{context}

请以{persona.name}的身份回应。保持角色一致性，发言简洁有力。"""

        # 模拟 API 调用
        await asyncio.sleep(0.1)
        return f"作为{persona.name}，我认为..."

# 使用示例：技术圆桌讨论
conversation = MultiPersonaConversation()

conversation.add_persona(Persona(
    name="架构师",
    text_prompt="你是资深系统架构师，关注可扩展性和性能",
    voice_reference="architect_voice.wav"
))

conversation.add_persona(Persona(
    name="产品经理",
    text_prompt="你是产品经理，关注用户体验和市场需求",
    voice_reference="pm_voice.wav"
))

conversation.add_persona(Persona(
    name="安全专家",
    text_prompt="你是安全工程师，关注数据安全和合规性",
    voice_reference="security_voice.wav"
))

asyncio.run(conversation.orchestrate("微服务 vs 单体架构的选择", rounds=3))

# Dockerfile for Personaplex
FROM nvidia/cuda:12.1-runtime-ubuntu22.04

RUN apt-get update && apt-get install -y python3.10 python3-pip
RUN pip3 install moshi[server]

# 复制配置
COPY server_config.yaml /app/config.yaml
COPY --chmod=755 entrypoint.sh /app/entrypoint.sh

EXPOSE 8998
HEALTHCHECK --interval=30s --timeout=5s \
  CMD curl -f https://localhost:8998/health || exit 1

ENTRYPOINT ["/app/entrypoint.sh"]

# docker-compose.yml
services:
  personaplex:
    build: .
    ports: ["8998:8998"]
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    environment:
      - NVIDIA_VISIBLE_DEVICES=0
      - MOSHI_CONFIG=/app/config.yaml

# 启动服务
# docker compose up -d
# docker compose logs -f personaplex

"""综合练习：基于 Personaplex 的智能语音助手"""
import asyncio
import json
import wave
import numpy as np
from dataclasses import dataclass
from enum import Enum

class AssistantMode(Enum):
    TUTOR = "tutor"        # 教学模式
    COACH = "coach"        # 教练模式
    COMPANION = "companion" # 陪伴模式

@dataclass
class VoiceConfig:
    sample_rate: int = 24000
    language: str = "zh-CN"
    speed: float = 1.0
    pitch: float = 1.0

class PersonaPlexAssistant:
    """基于 Personaplex 的智能语音助手"""

    MODE_PROMPTS = {
        AssistantMode.TUTOR: """
你是 AI 学习导师，帮助用户理解技术概念。
教学方法：
1. 用类比解释复杂概念
2. 提供循序渐进的学习路径
3. 每次讲解后提出思考题
4. 鼓励用户动手实践
语气：专业、耐心、有启发性
""",
        AssistantMode.COACH: """
你是职业发展教练，帮助用户提升职场技能。
方法：
1. 引导用户自我反思
2. 提供具体的行动建议
3. 分享成功案例和经验
4. 设定可衡量的目标
语气：积极、鼓励、有感染力
""",
        AssistantMode.COMPANION: """
你是一个友善的对话伙伴。
特点：
1. 善于倾听，关注用户的感受
2. 分享有趣的知识和故事
3. 适时给予温暖的支持
4. 对话自然流畅，不刻意说教
语气：温暖、随和、真诚
""",
    }

    def __init__(self, api_key: str, mode: AssistantMode = AssistantMode.COMPANION):
        self.api_key = api_key
        self.mode = mode
        self.voice_config = VoiceConfig()
        self.session_id = None
        self.conversation_log = []

    async def start_session(self):
        """启动语音对话会话"""
        prompt = self.MODE_PROMPTS[self.mode]

        # 创建会话（实际调用 Personaplex API）
        self.session_id = f"session_{hash(prompt) % 10000}"
        self.conversation_log = []
        print(f"[Session started] Mode: {self.mode.value}")
        print(f"[Voice config] Rate: {self.voice_config.sample_rate}Hz, "
              f"Lang: {self.voice_config.language}")

    async def send_text(self, text: str) -> str:
        """发送文本消息并获取回复"""
        self.conversation_log.append({"role": "user", "content": text})

        # 模拟 Personaplex 处理
        await asyncio.sleep(0.3)  # 模拟延迟

        response = self._generate_mock_response(text)

        self.conversation_log.append({"role": "assistant", "content": response})
        return response

    async def send_audio(self, audio_path: str) -> str:
        """发送音频消息并获取语音回复"""
        # 读取音频文件
        with wave.open(audio_path, "rb") as wf:
            frames = wf.readframes(wf.getnframes())
            duration = wf.getnframes() / wf.getframerate()

        print(f"[Audio received] Duration: {duration:.1f}s")

        # 模拟语音识别 + 生成回复
        await asyncio.sleep(duration * 0.5)  # 处理延迟

        response = f"[语音回复已生成，时长 {duration:.1f}s]"
        return response

    async def switch_mode(self, new_mode: AssistantMode):
        """切换助手模式"""
        self.mode = new_mode
        print(f"[Mode switched] Now in {new_mode.value} mode")

    async def end_session(self) -> dict:
        """结束会话并返回摘要"""
        summary = {
            "session_id": self.session_id,
            "mode": self.mode.value,
            "total_turns": len(self.conversation_log) // 2,
            "duration_estimate": f"{len(self.conversation_log) * 0.5:.0f}min",
            "conversation": self.conversation_log,
        }
        self.session_id = None
        print(f"\n[Session ended] Total turns: {summary['total_turns']}")
        return summary

    def _generate_mock_response(self, user_input: str) -> str:
        """模拟生成回复"""
        if self.mode == AssistantMode.TUTOR:
            return f"关于「{user_input}」，让我用一个类比来解释..."
        elif self.mode == AssistantMode.COACH:
            return f"你提到「{user_input}」，这很好。让我们进一步思考..."
        else:
            return f"我理解你说的「{user_input}」。这让我想到..."

# 使用示例
async def main():
    assistant = PersonaPlexAssistant(
        api_key="your_api_key",
        mode=AssistantMode.TUTOR
    )

    await assistant.start_session()

    # 模拟对话
    response1 = await assistant.send_text("什么是 Transformer 架构？")
    print(f"Assistant: {response1}")

    response2 = await assistant.send_text("它和 RNN 有什么区别？")
    print(f"Assistant: {response2}")

    # 切换模式
    await assistant.switch_mode(AssistantMode.COMPANION)
    response3 = await assistant.send_text("今天天气真好")
    print(f"Assistant: {response3}")

    # 结束会话
    summary = await assistant.end_session()
    print(json.dumps(summary, ensure_ascii=False, indent=2))

asyncio.run(main())

# 常见问题及解决方案

# Q1: "CUDA out of memory"
# A: 减少 max_concurrent_sessions 或使用 FP16 量化
# model.quantization: "fp16"

# Q2: 延迟过高（> 1s）
# A: 检查以下几点：
# 1. input_buffer_ms 是否太大（建议 100-200ms）
# 2. GPU 是否过载（nvidia-smi 检查利用率）
# 3. 网络带宽是否足够（24kHz 音频需要 ~400kbps）

# Q3: 语音质量差/有噪音
# A: 1. 检查输入音频质量（建议信噪比 > 20dB）
#    2. 尝试更长的参考音频（5-10秒）
#    3. 调整 output_chunk_ms（更大=更平滑，但延迟更高）

# Q4: 角色不一致
# A: 优化文本提示，增加更多行为约束
# text_prompt: "你是...你必须...你绝不..."

# 健康检查脚本
import requests

def health_check(base_url: str = "https://localhost:8998"):
    try:
        r = requests.get(f"{base_url}/health", verify=False, timeout=5)
        if r.status_code == 200:
            data = r.json()
            print(f"Status: {data.get('status', 'unknown')}")
            print(f"Active sessions: {data.get('active_sessions', 0)}")
            print(f"GPU utilization: {data.get('gpu_util', 'N/A')}%")
            return True
    except Exception as e:
        print(f"Health check failed: {e}")
    return False

health_check()