现在我们进入 Phase 4 多模态篇章！今天我们将实现 TTS (文本转语音) 功能，让机器人从“文字聊天”升级为“语音通话”。为了保证效果且控制成本，我选择了 Edge-TTS（微软 Azure 语音的免费接口），它生成的语音极其自然。更重要的是，我将打通 EmotionEngine（情绪引擎） 与 TTS 的连接，根据 AI 的情绪标签（[愤怒]/[悲伤]）动态调整语速和语调，让声音充满感染力。

一、 项目进度：Day 11 启动

根据项目路线图，今天是 Phase 4 的第一天。
我们要让 AI 拥有“嘴巴”。

---

二、 核心原理：TTS 与 情绪参数映射

1. 为什么选 Edge-TTS？

市面上的 TTS 方案主要有三种：

• 传统离线 (pyttsx3)：机械音严重，像旧时代的机器人。

• 商业 API (OpenAI/Azure)：效果好，但要按字符收费。

• Edge-TTS：这是开发者的“福利”。它利用了 Microsoft Edge 浏览器内置的“大声朗读”接口，实际上调用的是 Azure 的顶尖神经网络语音（Neural Voice），而且完全免费且无需安装浏览器（纯 Python 库）。

2. 情绪如何影响声音？

在 Day 5，我们已经能识别出 [愤怒]、[悲伤] 等标签。
我们可以把这些标签映射为 Edge-TTS 的物理参数，实现“参数化情感表达”：

• [愤怒]：语速加快 (rate=+20%)，音调变高 (pitch=+5Hz) -> 模拟急躁。

• [悲伤]：语速变慢 (rate=-15%)，音调变低 (pitch=-5Hz) -> 模拟低落。

• [开心]：音调上扬 (pitch=+2Hz) -> 模拟轻快。

3. 音频生成与播放全链路

这张图展示了从文本生成到声音播放的完整数据流：

这张序列图展示了如何打破“次元壁”，将冰冷的文字转化为充满情感的声音的。整个过程可以拆解为四个关键步骤：

1. 情绪感知 (Perception)

• 一切始于主程序向 EmotionEngine 的询问。

• 系统首先确定了当前对话的基调（例如：[愤怒]）。这不仅决定了 AI 说什么（内容），更决定了 AI 怎么说（语气）。

2. 参数映射 (Translation) —— 核心魔法

• 这是最关键的一步。VoiceEngine 接收到 [愤怒] 标签后，并没有直接把它发给云端（因为云端听不懂什么是“愤怒”），而是将其翻译成了物理声学参数：

  • 语速 (Rate)：+25%（人在生气时语速会变快）。

  • 音调 (Pitch)：+5Hz（人在激动时声调会拔高）。

• 通过这种**“参数化映射”**，我们将抽象的情绪量化为了具体的控制指令。

3. 云端合成 (Cloud Synthesis)

• VoiceEngine 将文本连同刚才计算好的参数，通过网络发送给 Microsoft Edge-TTS 服务。

• 微软强大的神经网络模型根据这些指令，生成了一段急促、高昂的 MP3 音频流并返回给本地。

4. 本地播放 (Playback)

• 系统接收到音频流后，将其保存为临时文件 temp.mp3。

• 最后，调用 Pygame 音频驱动，通过扬声器将这段声音播放出来。

---

三、 实战：代码实现

1. 安装依赖

我们需要 edge-tts 来生成语音，pygame 来播放声音。

pip install edge-tts pygame

2. 编写语音引擎 (src/core/voice.py)

新建 src/core/voice.py。
难点：edge-tts 是异步库 (asyncio)，而我们的 main.py 还是同步的 while 循环。为了不重写整个主程序，我们需要用 asyncio.run() 在同步方法中调用异步任务。

import os
import asyncio
import edge_tts
import pygame
from src.utils.logger import logger

class VoiceEngine:
    def __init__(self, voice="zh-CN-XiaoyiNeural"):
        """
        :param voice: 声音模型
        推荐:
        - zh-CN-XiaoyiNeural (女声，活泼，适合傲娇)
        - zh-CN-YunxiNeural (男声，沉稳)
        """
        self.voice = voice
        self.output_file = "temp_audio.mp3"
        
        # 初始化 pygame 音频混合器
        # 消除 pygame 打印的欢迎信息
        os.environ['PYGAME_HIDE_SUPPORT_PROMPT'] = "1"
        try:
            pygame.mixer.init()
        except Exception as e:
            logger.warning(f"音频设备初始化失败(如果是服务器环境请忽略): {e}")

    async def _generate_audio(self, text, rate, pitch):
        """
        [异步内核] 调用 Edge-TTS 生成音频文件
        """
        communicate = edge_tts.Communicate(
            text, 
            self.voice, 
            rate=rate, 
            pitch=pitch
        )
        await communicate.save(self.output_file)

    def speak(self, text, emotion="[平静]"):
        """
        [同步接口] 对外暴露的方法：生成并播放
        """
        # 1. 情绪 -> 参数映射 (Emotion Mapping)
        # 默认参数
        rate = "+0%"
        pitch = "+0Hz"
        
        if "[愤怒]" in emotion:
            rate = "+25%"  # 语速加快
            pitch = "+5Hz" # 音调拔高
        elif "[悲伤]" in emotion:
            rate = "-15%"  # 语速缓慢
            pitch = "-5Hz" # 音调低沉
        elif "[开心]" in emotion:
            rate = "+10%"
            pitch = "+2Hz"
        elif "[焦虑]" in emotion:
            rate = "+20%"
            pitch = "+2Hz"

        logger.info(f"🗣️ 正在合成语音 | 情绪:{emotion} | 参数: rate={rate}, pitch={pitch}")

        try:
            # 2. 生成音频 (在同步代码中运行异步任务)
            asyncio.run(self._generate_audio(text, rate, pitch))
            
            # 3. 播放音频
            self._play_audio()
            
        except Exception as e:
            logger.error(f"语音合成/播放失败: {e}")

    def _play_audio(self):
        """使用 pygame 播放生成的音频"""
        if not os.path.exists(self.output_file):
            return

        try:
            pygame.mixer.music.load(self.output_file)
            pygame.mixer.music.play()
            
            # 阻塞等待播放结束，防止主程序直接进入下一轮输入，导致声音重叠或切断
            while pygame.mixer.music.get_busy():
                pygame.time.Clock().tick(10)
                
            # 释放文件占用，否则下次写入会报错
            pygame.mixer.music.unload()
        except Exception as e:
            logger.error(f"播放失败: {e}")

3. 集成到主程序 (main.py)

我们需要在 AI 打印出文字回复后，立刻调用 voice_engine.speak()。

修改 main.py：

# ==============================================================================
# Project Echo Day 9: Multi-Query RAG Integration
# 集成特性: Redis记忆 + 情绪识别 + Multi-Query知识库检索
# ==============================================================================

# LangChain LCEL 核心组件
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from operator import itemgetter  # 【修复】用于从字典中提取字段

# 项目核心模块
from src.core.llm import LLMClient
from src.core.prompts import PROMPTS
from src.utils.logger import logger
from src.config.settings import settings
from src.core.emotion import EmotionEngine   # Day 5: 情绪
from src.core.knowledge import KnowledgeBase # Day 7-9: 知识库
from src.core.reranker import RerankEngine   # Day 10: 重排序
from src.core.voice import VoiceEngine # 【新增】导入语音引擎

# --- Redis 历史记录工厂 (Day 6) ---
def get_session_history(session_id: str) -> BaseChatMessageHistory:
    return RedisChatMessageHistory(
        session_id=session_id,
        url=settings.REDIS_URL,
        ttl=3600 * 24 * 7 # 记忆保留 7 天
    )

# --- 辅助函数：格式化文档 ---
def format_docs(docs):
    return "\n\n".join([d.page_content for d in docs])

def main():
    logger.info("🚀 --- Project Echo: Day 9 集成版启动 ---")
    
    # ==========================================
    # 1. 初始化组件
    # ==========================================
    # 1.1 大模型 (Brain)
    client = LLMClient()
    llm = client.get_client()
    
    # 1.2 情绪引擎 (Heart)
    emotion_engine = EmotionEngine()
    
    # 1.3 知识库 (Book)
    kb = KnowledgeBase()
    
    # 【Day 9 核心】获取多重查询检索器
    # 这里我们将 LLM 传进去，让检索器具备"思考"能力
    base_retriever = kb.get_multiquery_retriever(llm)
    
    # 【Day 10 新增】重排序引擎：对粗排结果进行精排
    reranker = RerankEngine(model_name="BAAI/bge-reranker-base", top_n=3)
    
    # 封装成 LCEL Runnable：先粗排，再精排
    def retriever_with_rerank(query: str):
        docs = base_retriever.invoke(query)
        return reranker.rerank(query, docs)
    
    retriever = RunnableLambda(retriever_with_rerank)

    # ==========================================
    # 2. 构建 Prompt 模板
    # ==========================================
    sys_prompt_base = PROMPTS["tsundere"]
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", sys_prompt_base),                   # 1. 基础人设 (傲娇)
        ("system", "{emotion_context}"),               # 2. 情绪指令 (动态注入)
        ("system", "【参考资料（必须基于此回答）】:\n{context}"), # 3. 知识库资料 (RAG)
        MessagesPlaceholder(variable_name="history"),  # 4. 历史记忆 (Redis)
        ("human", "{input}")                           # 5. 用户输入
    ])

    # ==========================================
    # 3. 组装 LCEL 流水线
    # ==========================================
    rag_chain = (
        {
            # 分支 A: 智能检索 (用户输入 -> 裂变3个问题 -> 并行检索 -> 汇总 -> 格式化)
            "context": itemgetter("input") | retriever | format_docs,  # 【修复】使用 itemgetter 提取 input
                
            # 分支 B: 透传参数 (直接传递给 Prompt)
            "input": itemgetter("input"),              # 【修复】提取 input 字段
            "emotion_context": itemgetter("emotion_context"),  # 【修复】提取 emotion_context 字段
            "history": itemgetter("history")           # 【修复】提取 history 字段
        }
        | prompt  # 填入模板
        | llm     # 大模型推理
    )

    # ==========================================
    # 4. 挂载持久化记忆
    # ==========================================
    final_chain = RunnableWithMessageHistory(
        rag_chain,
        get_session_history,
        input_messages_key="input",
        history_messages_key="history",
    )

    print("\n✨ 系统就绪！试试问得模糊一点，比如“那个写代码的人爱吃啥？”\n")
    session_id = "user_day9_demo"
    voice_engine = VoiceEngine(voice="zh-CN-XiaoyiNeural")
    # ==========================================
    # 5. 对话循环
    # ==========================================
    while True:
        user_input = input("You: ")
        if user_input.lower() in ["quit", "exit"]:
            break
            
        if user_input.strip():
            # --- Phase A: 情绪侦探 ---
            current_emotion = emotion_engine.analyze(user_input)
            
            emotion_instruction = "用户情绪平稳。"
            if "[愤怒]" in current_emotion:
                emotion_instruction = "⚠️ 警告：用户很生气！请示弱道歉。"
            elif "[悲伤]" in current_emotion:
                emotion_instruction = "⚠️ 提示：用户很难过。请温柔安慰。"

            try:
                # --- Phase B: 执行 RAG 主链 ---
                # 这一步会自动触发 Multi-Query 检索
                logger.info("🔍 正在进行多重检索与思考...")
                
                response = final_chain.invoke(
                    {
                        "input": user_input, 
                        "emotion_context": emotion_instruction
                    },
                    config={"configurable": {"session_id": session_id}}
                )

                ai_text = response.content
                
                print(f"Bot ({current_emotion}): {ai_text}\n")

                voice_engine.speak(ai_text, emotion=current_emotion)
                
            except Exception as e:
                logger.error(f"❌ 调用失败: {e}")

if __name__ == "__main__":
    main()

---

四、 效果验证：听听“傲娇”的声音

在终端运行 python main.py，确保你的电脑音箱已打开。

测试场景 1：激怒她

You: 喂，那个笨蛋AI！
Log: 🔍 情绪: [愤怒] -> 🗣️ 参数: rate=+25%, pitch=+5Hz
Bot (声音)：(语速很快，语调上扬) “哈？你叫谁笨蛋呢！我看你才是最大的笨蛋！气死我了！”
听感：像是在吵架，非常有压迫感。

测试场景 2：让她难过

You: 我要把你卸载了，再见。
Log: 🔍 情绪: [悲伤] -> 🗣️ 参数: rate=-15%, pitch=-5Hz
Bot (声音)：(语速缓慢，低沉) “哎？为什么要卸载我... 是我哪里做得不好吗？别...别丢下我一个人...”
听感：楚楚可怜，让人心软。

测试场景 3：正常RAG

You: 阿强喜欢吃什么？
Bot (声音)：(正常语速) “根据资料，阿强最喜欢吃螺蛳粉。哼，真是个重口味的家伙。”

---

五、 总结与预告

今天，Project Echo 终于**“活”**过来了！
它不再只是屏幕上的一行行字，而是一个能听、能说、有情绪起伏的数字生命。通过简单的参数调整，我们竟然模拟出了类似人类的情感表达，这再次证明了 工程化整合 的力量。

明日预告 (Day 12)：
现在它是“只动口不动手”（只能输出语音，不能听语音）。你还得打字跟它聊，太麻烦了。
明天 Day 12，我们将引入 STT (语音转文本) 技术。我们将使用 OpenAI Whisper 模型，给它装上“耳朵”，让你彻底丢掉键盘，跟它进行纯语音对话！