AIcarus TTS Worker

这是 AIcarusForQQ 的独立 TTS Worker 仓库。主程序作为 WebSocket Server，Worker 作为 WebSocket Client 连接主程序，接收文本任务并流式返回 PCM 音频。

这个仓库的目标不是内置所有 TTS 引擎，而是提供统一协议客户端和可插拔后端框架。用户可以保留协议层，只替换 backends 里的合成逻辑。

当前能力

• 向主程序注册 plugin_id、协议版本、音频格式和 LLM 参数 schema
• 接收 task 指令
• 上报 start / streaming / completed / error 状态
• 使用二进制帧返回 PCM 音频 chunk
• 响应主程序 ping 心跳
• 自动重连
• 限制 Worker 侧并发任务数
• 内置 dummy 后端，用于联调协议
• 内置 aliyun_cosyvoice 后端，用于调用阿里云百炼 CosyVoice TTS

快速开始

pip install -r requirements.txt
copy config.example.yaml config.yaml
python run.py --config config.yaml

主程序配置中需要启用 TTS：

tts:
  enabled: true
  host: "127.0.0.1"
  port: 8765
  secret_token: ""
  max_concurrent_tasks_per_plugin: 8

如果主程序设置了 secret_token，这里的 config.yaml 必须填写相同值。

配置说明

main_app_uri: "ws://127.0.0.1:8765"
secret_token: ""
plugin_id: "dev.aicarus.aliyun_cosyvoice"
protocol_version: "1.0"
reconnect_delay_seconds: 5
max_concurrent_tasks: 8

backend:
  name: "aliyun_cosyvoice"
  config:
    api_key_env: "DASHSCOPE_API_KEY"
    model: "cosyvoice-v3-flash"
    voice: "longanyang"
    format: "PCM_16000HZ_MONO_16BIT"
    sample_rate: 16000
    timeout_millis: 60000

阿里云 API Key 请配置到环境变量，例如 Windows CMD：

set DASHSCOPE_API_KEY=你的 DashScope API Key

不要把 API Key 写入仓库配置文件。若需要使用 CosyVoice v3.5 系列，请先在阿里云侧准备声音设计或复刻音色，并把 model / voice 改成对应值。

plugin_id 建议使用反域名格式。主程序发现同一个 plugin_id 重连时，会关闭旧连接并保留新连接。

新增自己的 TTS 后端

新增文件，例如 aicq_tts_worker/backends/my_tts.py：

from collections.abc import AsyncIterator
from typing import Any

from .base import TTSBackend


class MyTTSBackend(TTSBackend):
    async def synthesize(self, text: str, parameters: dict[str, Any]) -> AsyncIterator[bytes]:
        async for pcm_chunk in your_tts_client.stream(text, **parameters):
            yield pcm_chunk

然后在 aicq_tts_worker/backends/__init__.py 里注册：

from .my_tts import MyTTSBackend

BACKENDS = {
    "dummy": DummyBackend,
    "my_tts": MyTTSBackend,
}

最后把配置改成：

backend:
  name: "my_tts"
  config: {}

协议摘要

JSON 控制帧：

• Worker -> Main: register
• Main -> Worker: register_ack
• Main -> Worker: task
• Worker -> Main: status
• Main -> Worker: ping
• Worker -> Main: pong

音频二进制帧：

[4 bytes: task_id length, UInt32 BE] [task_id UTF-8] [PCM bytes]

task_id 最大 256 字节。

测试

python -m unittest