Metadata-Version: 2.4
Name: aichain-sdk
Version: 1.0.7
Summary: AI Chain Python Client SDK for IIP Dispatch WebSocket protocol v2.0
Author-email: AI Chain SDK Team <justpluspro@gmail.com>
Project-URL: Homepage, https://aichain-sh.xfyun.cn
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: websockets>=14.0
Requires-Dist: pydantic<3.0,>=2.0
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21; extra == "dev"
Requires-Dist: build>=1.5.0; extra == "dev"
Requires-Dist: twine>=6.2.0; extra == "dev"
Requires-Dist: python-dotenv>=1.2.2; extra == "dev"
Requires-Dist: pydub>=0.25.0; extra == "dev"
Requires-Dist: numpy>=2.2.6; extra == "dev"
Requires-Dist: soundfile>=0.13.1; extra == "dev"
Requires-Dist: sounddevice>=0.5.5; extra == "dev"

# AIChain Dispatch SDK Python 集成指南

## 目录

+ [简介](#简介)
+ [快速开始](#快速开始)
+ [详细配置](#详细配置)
+ [API 参考](#api-参考)
+ [事件系统](#事件系统)
+ [完整集成流程](#完整集成流程)
+ [最佳实践](#最佳实践)
+ [错误处理](#错误处理)
+ [常见问题](#常见问题)

---

## 简介

AIChain Dispatch SDK 是一个基于 WebSocket 的实时通信 SDK，支持语音识别（STT）、自然语言理解（NLU）和语音合成（TTS）功能。本
SDK提供了完整的 Python 异步支持，基于 asyncio 和 websockets 实现。

> **协议参考**：本 SDK 基于 AIChain WebSocket 接口协议 v2.1
>
实现。完整的协议规范（连接鉴权、首帧配置、数据交互、响应格式等）见 [API 接口文档](https://www.yuque.com/aiui_open_platform/knowledge/hf7xdluok2yz3dib)。

### 主要特性

+ ✅ WebSocket 实时双向通信
+ ✅ 自动重连机制
+ ✅ 多模态消息支持（文本、语音、图片）
+ ✅ 流式数据传输
+ ✅ 完整的事件系统
+ ✅ 异步 I/O 设计（基于 asyncio）
+ ✅ 类型提示支持（Type Hints）

### 系统要求

+ **Python 版本**: Python 3.10+
+ **操作系统**: Windows / Linux / macOS
+ **依赖库**: websockets>=14.0, pydantic>=2.0,<3.0, numpy>=2.2.6, soundfile>=0.13.1, sounddevice>=0.5.5

---

## 快速开始

### 1. 安装依赖

使用 pip 安装 SDK：

```bash
pip install -i https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple aichain-sdk==1.0.7
```

或者使用uv管理依赖：

```bash
UV_DEFAULT_INDEX=https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple
uv add aichain-sdk==1.0.7
```

### 2. 初始化客户端

```python
import asyncio
from aichain_sdk import AIChainClient

client = AIChainClient(
    app_id="your-app-id",
    app_key="your-app-key",
    host="your-host",
    sn="your-device-sn"
)
```

> 初始化参数与连接鉴权相关，详见接口文档 §1.1 连接地址、§1.2 连接参数、§1.3 签名计算。

### 3. 连接服务器

```python
async def main():
    await client.connect()
    print("连接成功")


asyncio.run(main())
```

### 4. 发送消息

```python
@client.on("nlu.answer")
def on_answer(cid: str, result, is_last: bool):
    print(result.answer, end="", flush=True)
    if is_last:
        print()


async def main():
    await client.connect()
    await client.set_config({
        "nlu": {"enable": True}
    })

    await client.send_text("你好，世界！")
    await asyncio.sleep(5)
    await client.disconnect()


asyncio.run(main())
```

---

## 详细配置

> **协议参考**：SDK 的配置参数对应接口文档中 `session.config` 事件的 `config` 字段。全局配置见接口文档 §4.2
> 全局配置，STT/NLU/TTS 各能力配置分别见 §5、§6、§7。

### 客户端初始化参数

`AIChainClient` 构造函数提供了丰富的配置选项：

| 参数                       | 类型   | 必填   | 默认值    | 说明                                     |
|--------------------------|------|------|--------|----------------------------------------|
| `app_id`                 | str  | ✅    | -      | 应用 ID，由平台分配                            |
| `app_key`                | str  | ✅    | -      | 应用密钥，用于身份验证                            |
| `host`                   | str  | ✅    | -      | 服务器地址（可包含 `ws://` / `wss://` 前缀，会自动解析） |
| `sn`                     | str  | ❌    | ""     | 设备序列号，唯一标识设备                           |
| `scene`                  | str  | ❌    | "main" | 场景名称，用于多场景应用                           |
| `secure`                 | bool | None | ❌      | None                                   | 是否使用安全连接（wss://）。为 `None` 时根据 `host` 是否以 `wss://` 开头自动判断 |
| `auto_reconnect`         | bool | ❌    | True   | 是否启用自动重连                               |
| `reconnect_interval`     | int  | ❌    | 1000   | 初始重连间隔（毫秒）                             |
| `max_reconnect_interval` | int  | ❌    | 30000  | 最大重连间隔（毫秒）                             |
| `max_reconnect_attempts` | int  | ❌    | 10     | 最大重连次数                                 |
| `log_level`              | str  | ❌    | "INFO" | 日志级别（DEBUG/INFO/WARN/ERROR）            |
| `log_file`               | str  | None | ❌      | None                                   | 日志文件路径（None 则输出到控制台） |

#### 完整配置示例

```python
client = AIChainClient(
    app_id="your-app-id",
    app_key="your-app-key",
    host="wss://your-host.com",
    sn="device-12345",
    scene="main",
    auto_reconnect=True,
    reconnect_interval=2000,
    max_reconnect_interval=60000,
    max_reconnect_attempts=20,
    log_level="DEBUG",
    log_file="aichain.log"
)
```

### 会话配置（SessionConfig）

连接成功后，需要配置会话参数。配置可以使用字典或 `SessionConfig` 数据类。

> **协议参考**：会话配置对应接口文档 §2 首帧请求（session.config事件），详细参数介绍见 §4 首帧请求（详细参数介绍）。

#### 配置参数说明

**基础配置**

| 参数                   | 类型   | 必填 | 默认值           | 说明                                                                                                                                                                                                               |
|----------------------|------|----|---------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `mode`               | str  | ❌  | "full_duplex" | 工作模式：`"full_duplex"`（全双工）或 `"half_duplex"`（半双工）                                                                                                                                                                  |
| `simplifiedResponse` | bool | ❌  | True          | 是否简化响应内容                                                                                                                                                                                                         |
| `multiTurnEnabled`   | bool | ❌  | True          | 是否启用多轮对话                                                                                                                                                                                                         |
| `extraConfig`        | dict | ❌  | {}            | 会话级扩展参数，透传至服务端各处理器。v2.2 起系统内置参数已迁移至结构化字段：`interrupt_strategy` → `stt.interrupt.strategy`、`chunk_tts` → `tts.chunkTts`（建议改用 `tts.rateKbps`）、`language` → `nlu.replyLanguage`、`enableFiller` → `nlu.filler.enable` |

**STTConfig（语音识别配置）**

> 详见接口文档 §5 STT（语音识别）配置，含引擎参数（§5.1.6）、热词（§5.1.5）、打断策略（§5.1.4）等完整说明。

| 参数              | 类型                  | 必填 | 默认值    | 说明                                                                                     |
|-----------------|---------------------|----|--------|----------------------------------------------------------------------------------------|
| `enable`        | bool                | ❌  | False  | 是否启用 STT                                                                               |
| `sttEngineId`   | str                 | ❌  | ""     | STT 引擎 ID（如 `"2"` 讯飞中英、`"3"` 讯飞多语种、`"5"` 高性能听写引擎）                                      |
| `language`      | str                 | ❌  | "auto" | 语言代码（如 "zh"、"en"、"auto"）。多语种引擎（`sttEngineId=3/5`）支持使用 `\|` 分隔同时指定多个语种，如 `"zh\|en\|ja"` |
| `hotWords`      | list[dict]          | ❌  | []     | 热词列表（仅高性能听写引擎在 `language` 非 `auto` 时支持）                                                |
| `audioConfig`   | AudioConfig         | ❌  | -      | 音频配置                                                                                   |
| `vad`           | VadConfig           | ❌  | -      | 语音活动检测配置                                                                               |
| `turnDetection` | TurnDetectionConfig | ❌  | -      | 轮次检测配置                                                                                 |
| `interrupt`     | InterruptConfig     | ❌  | -      | 打断配置（v2.2: `enable` 已废弃，使用 `strategy` 字段）                                              |
| `engineParams`  | dict                | ❌  | {}     | 引擎额外参数                                                                                 |

**AudioConfig（音频配置）**

| 参数              | 类型  | 必填 | 默认值     | 说明                                                               |
|-----------------|-----|----|---------|------------------------------------------------------------------|
| `audioEncoding` | str | ❌  | "raw"   | 音频编码格式，取值范围由 STT/TTS 引擎决定（如 `raw`、`opus-wb`、`speex-wb`、`lame` 等） |
| `format`        | str | ❌  | "plain" | 音频格式，固定为 `"plain"`                                               |
| `sampleRate`    | int | ❌  | 16000   | 采样率（Hz）                                                          |
| `bitDepth`      | int | ❌  | 16      | 位深度                                                              |
| `channels`      | int | ❌  | 1       | 声道数（1=单声道, 2=立体声）                                                |

**VadConfig（语音活动检测配置）**

| 参数                    | 类型                   | 必填 | 默认值      | 说明                            |
|-----------------------|----------------------|----|----------|-------------------------------|
| `enable`              | bool                 | ❌  | True     | 是否启用 VAD                      |
| `vadType`             | str                  | ❌  | "silero" | VAD 引擎类型：`"silero"` 或 `"ten"` |
| `minSpeechDuration`   | int                  | ❌  | 50       | 最小语音持续时间（毫秒）                  |
| `minSilenceDuration`  | int                  | ❌  | 600      | 最小静音持续时间（毫秒），最小值 600          |
| `activationThreshold` | float                | ❌  | 0.5      | 激活阈值（0.0-1.0）                 |
| `farFieldFilter`      | FarFieldFilterConfig | ❌  | -        | 远场过滤配置，减少远距离音频误触发             |

**FarFieldFilterConfig（远场过滤配置）**

| 参数            | 类型    | 必填 | 默认值   | 说明                                     |
|---------------|-------|----|-------|----------------------------------------|
| `enable`      | bool  | ❌  | False | 是否启用远场过滤                               |
| `sensitivity` | float | ❌  | 0.5   | 远场过滤灵敏度，范围 0~1。0=最不灵敏（严格过滤），1=最灵敏（不过滤） |

**TurnDetectionConfig（轮次检测配置）**

| 参数                    | 类型   | 必填 | 默认值  | 说明         |
|-----------------------|------|----|------|------------|
| `enable`              | bool | ❌  | True | 是否启用轮次检测   |
| `minEndpointingDelay` | int  | ❌  | 500  | 最小端点延迟（毫秒） |
| `maxEndpointingDelay` | int  | ❌  | 3000 | 最大端点延迟（毫秒） |

**InterruptConfig（打断配置）**

| 参数         | 类型  | 必填 | 默认值                  | 说明                                                                                                               |
|------------|-----|----|----------------------|------------------------------------------------------------------------------------------------------------------|
| `strategy` | str | ❌  | "semantic_interrupt" | 打断策略：`"always_interrupt"`（强制打断）/ `"semantic_interrupt"`（语义判断）/ `"never_interrupt"`（不打断）。全双工不支持 `never_interrupt` |

**NLUConfig（自然语言理解配置）**

> 详见接口文档 §6 NLU（语义理解）配置，含可用模型列表（§6.1.1）、Tools 配置（§6.1.3）、RAG 配置（§6.1.4）等完整说明。

| 参数                      | 类型           | 必填 | 默认值   | 说明                                                        |
|-------------------------|--------------|----|-------|-----------------------------------------------------------|
| `enable`                | bool         | ❌  | False | 是否启用 NLU                                                  |
| `streamingBufferLength` | int          | ❌  | 10    | NLU 流式回复按文本长度分片下发的缓冲长度阈值（字符数）。值越小首段到达越快、事件更碎；值越大单帧越长、事件更少 |
| `replyLanguage`         | str          | ❌  | None  | 回复语种控制：`None`（不控制）/ `"auto"`（智能判断）/ `"ko"`、`"ja"` 等明确语种代码 |
| `welcomeReply`          | list[str]    | ❌  | []    | 欢迎语列表，服务端会随机选择一条返回                                        |
| `filler`                | FillerConfig | ❌  | None  | 垫话配置（`None` 表示未开启）。工具调用时智能根据耗时进行垫话                        |
| `tools`                 | ToolsConfig  | ❌  | None  | 工具调用配置（None 表示未开启）                                        |
| `rag`                   | RagConfig    | ❌  | None  | RAG 检索配置（None 表示未开启）                                      |

**FillerConfig（垫话配置）**

| 参数       | 类型   | 必填 | 默认值   | 说明       |
|----------|------|----|-------|----------|
| `enable` | bool | ❌  | False | 是否启用垫话功能 |

**ToolsConfig（工具配置）**

| 参数                 | 类型       | 必填 | 默认值              | 说明        |
|--------------------|----------|----|------------------|-----------|
| `model`            | LLMModel | ❌  | -                | LLM 模型配置  |
| `toolCallPrompt`   | str      | ❌  | "请根据用户需求选择合适的工具" | 工具调用提示词   |
| `toolPolishPrompt` | str      | ❌  | None             | 工具结果润色提示词 |

**RagConfig（RAG 配置）**

| 参数               | 类型             | 必填 | 默认值            | 说明        |
|------------------|----------------|----|----------------|-----------|
| `model`          | LLMModel       | ❌  | -              | LLM 模型配置  |
| `threshold`      | float          | ❌  | 0.7            | 相似度阈值     |
| `contextLength`  | int            | ❌  | 2000           | 上下文长度     |
| `topN`           | int            | ❌  | 5              | 返回前 N 个结果 |
| `polishPrompt`   | str            | ❌  | "请根据引文润色之后再返回" | 润色提示词     |
| `dialogueConfig` | DialogueConfig | ❌  | -              | 对话配置      |

**DialogueConfig（对话配置）**

| 参数                 | 类型        | 必填 | 默认值                 | 说明          |
|--------------------|-----------|----|---------------------|-------------|
| `enableFaq`        | bool      | ❌  | True                | 是否启用 FAQ 匹配 |
| `enableChitchat`   | bool      | ❌  | True                | 是否启用闲聊兜底    |
| `fallbackResponse` | list[str] | ❌  | ["抱歉，我暂时无法回答这个问题。"] | 兜底回复列表      |

**LLMModel（大语言模型配置）**

| 参数            | 类型    | 必填 | 默认值                                                 | 说明                  |
|---------------|-------|----|-----------------------------------------------------|---------------------|
| `modelId`     | str   | ❌  | None                                                | 模型 ID               |
| `maxTokens`   | int   | ❌  | None                                                | 最大 token 数          |
| `temperature` | float | ❌  | 0.7                                                 | 温度参数（0.0-2.0）       |
| `topP`        | float | ❌  | 0.9                                                 | Top-P 采样参数（0.0-1.0） |
| `extraConfig` | dict  | ❌  | {"frequency_penalty": 0.0, "presence_penalty": 0.0} | 额外配置                |

**TTSConfig（语音合成配置）**

> 详见接口文档 §7 TTS（语音合成）配置，含发音人列表（§13.2.1）、文本过滤器（§7.1.2）、Markup 标签（§7.1.5）等完整说明。

| 参数            | 类型                     | 必填 | 默认值                                 | 说明                                                            |
|---------------|------------------------|----|-------------------------------------|---------------------------------------------------------------|
| `enable`      | bool                   | ❌  | False                               | 是否启用 TTS                                                      |
| `textFilters` | list[str]              | ❌  | ["filter_markdown", "filter_emoji"] | 文本过滤器列表                                                       |
| `voices`      | dict[str, VoiceConfig] | ❌  | {}                                  | 语音配置字典（语言 -> VoiceConfig）                                     |
| `markupTags`  | TTSMarkupTagsConfig    | ❌  | None                                | 讯飞语音合成标签配置（None 表示未配置，默认关闭）                                   |
| `rateKbps`    | int                    | ❌  | None                                | TTS 音频下发速率（Kbps）。`None` 表示不限制，有效范围 [8, 4096]。对 raw-opus 编码不适用 |

**TTSMarkupTagsConfig（讯飞语音合成标签配置）**

仅讯飞系引擎（xfyun / xfyun-hyper / hyperrealistic / turing）解析标签，其他引擎自动跳过。标签须用英文小写字母输入才生效。

| 参数             | 类型                         | 必填 | 默认值   | 说明                     |
|----------------|----------------------------|----|-------|------------------------|
| `enable`       | bool                       | ❌  | False | 是否启用标签处理               |
| `globalPrefix` | str                        | ❌  | ""    | 全局前缀标签，拼接在每段送往引擎文本的最前面 |
| `replacements` | list[TTSMarkupReplacement] | ❌  | []    | 自定义替换规则列表，按顺序应用        |

**TTSMarkupReplacement（标签替换规则）**

| 参数            | 类型   | 必填 | 默认值   | 说明                                                      |
|---------------|------|----|-------|---------------------------------------------------------|
| `pattern`     | str  | ❌  | ""    | 匹配内容，为空时该条规则跳过                                          |
| `replacement` | str  | ❌  | ""    | 替换内容                                                    |
| `regex`       | bool | ❌  | False | `False` 字面量匹配；`True` 正则解析，replacement 支持 `$1`/`$2` 反向引用 |
| `ignoreCase`  | bool | ❌  | False | 是否忽略大小写                                                 |

> **使用示例**：通过 `markupTags` 注入朗读控制标签（数字读法、停顿、拼音矫正、强调等）。仅讯飞系引擎生效，默认关闭，向后兼容。
>

```python
await client.set_config({
    "tts": {
        "enable": True,
        "markupTags": {
            "enable": True,
            "globalPrefix": "[n2][b0]",  # 数字按数值读、不停顿
            "replacements": [
                {"pattern": "AI", "replacement": "[h1]AI"},  # 强调"AI"
                {"pattern": "重了", "replacement": "[=chong2]重了"}  # 拼音矫正
            ]
        }
    }
})
```

**VoiceConfig（语音配置）**

| 参数            | 类型          | 必填 | 默认值  | 说明                       |
|---------------|-------------|----|------|--------------------------|
| `voiceId`     | str         | ❌  | "4"  | 发音人 ID                   |
| `speed`       | int         | ❌  | 5    | 语速（0-10，5 为正常）           |
| `pitch`       | int         | ❌  | 5    | 音调（0-10，5 为正常）           |
| `volume`      | int         | ❌  | 5    | 音量（0-10，5 为正常）           |
| `audioConfig` | AudioConfig | ❌  | None | 音频输出配置（省略时使用该发音人引擎的默认配置） |

#### 完整配置示例

```python
# 使用字典配置
config = {
    "mode": "full_duplex",
    "simplifiedResponse": True,
    "multiTurnEnabled": True,
    "stt": {
        "enable": True,
        "language": "zh",
        "audioConfig": {
            "audioEncoding": "raw",
            "format": "plain",
            "sampleRate": 16000,
            "bitDepth": 16,
            "channels": 1
        },
        "vad": {
            "enable": True,
            "minSpeechDuration": 50,
            "minSilenceDuration": 600,
            "activationThreshold": 0.5
        },
        "turnDetection": {
            "enable": True,
            "minEndpointingDelay": 500,
            "maxEndpointingDelay": 3000
        },
        "interrupt": {
            "enable": True
        }
    },
    "nlu": {
        "enable": True,
        "streamingBufferLength": 10,
        "tools": {
            "model": {
                "modelId": "spark-v3.5",
                "maxTokens": 2000,
                "temperature": 0.7,
                "topP": 0.9
            }
        }
    },
    "tts": {
        "enable": True,
        "textFilters": ["filter_markdown", "filter_emoji"],
        "voices": {
            "zh": {
                "voiceId": "4",
                "speed": 5,
                "pitch": 5,
                "volume": 5,
                "audioConfig": {
                    "audioEncoding": "raw",
                    "sampleRate": 24000,
                    "bitDepth": 16,
                    "channels": 1
                }
            }
        }
    }
}

await client.set_config(config)
```

---

## API 参考

> **协议参考**：本章 SDK 方法对应的 WebSocket 协议细节见接口文档各章节（连接见 §1，首帧配置见 §2，消息发送见 §3，打断见
> §3.5）。

### 连接管理

#### `async connect() -> None`

建立 WebSocket 连接并等待会话创建。

**示例:**

```python
await client.connect()
print("连接成功")
```

#### `async disconnect() -> None`

关闭 WebSocket 连接。

**示例:**

```python
await client.disconnect()
```

#### `async reconnect() -> None`

手动重连到服务器。

> 协议参考：重连机制见接口文档 §1.5 重连机制，重连时需携带 `reconnect=true` 和原 `sid`，签名需用新的 `curtime` 重新计算。

**示例:**

```python
await client.reconnect()
```

#### `is_connected() -> bool`

检查当前是否已连接。

**返回值:**

+ `True`: 已连接
+ `False`: 未连接

**示例:**

```python
if client.is_connected():
    await client.send_text("Hello")
```

#### `get_connection_state() -> ConnectionState`

获取当前连接状态。

**返回值:**

+ `ConnectionState.DISCONNECTED`: 未连接
+ `ConnectionState.CONNECTING`: 连接中
+ `ConnectionState.CONNECTED`: 已连接
+ `ConnectionState.RECONNECTING`: 重连中
+ `ConnectionState.CLOSED`: 已关闭

**示例:**

```python
state = client.get_connection_state()
print(f"当前状态: {state}")
```

#### `get_connection_info() -> ConnectionInfo`

获取详细的连接信息。

**返回值:**

+ `ConnectionInfo`  
  对象，包含：
+ `state`: 连接状态
+ `sid`: 会话  
  ID
+ `connected_at`: 连接时间戳
+ `reconnect_attempts`: 重连次数
+ `last_error`: 最后一次错误

**示例:**

```python
info = client.get_connection_info()
print(f"会话ID: {info.sid}")
print(f"重连次数: {info.reconnect_attempts}")
```

### 配置管理

> **协议参考**：配置管理对应接口文档 §2 首帧请求（session.config 事件），配置响应见 §2.3 首帧响应，完整参数说明见 §4。

#### `async set_config(config: dict | SessionConfig) -> None`

设置会话配置。

**参数:**

+ `config`: 会话配置对象（字典或 SessionConfig 实例）

**示例:**

```python
config = {
    "mode": "full_duplex",
    "stt": {"enable": True},
    "nlu": {"enable": True},
    "tts": {"enable": True}
}

await client.set_config(config)
```

#### `get_config() -> SessionConfig | None`

获取当前会话配置。

**返回值:**

+ 当前配置对象，如果未设置则返回 `None`

**示例:**

```python
current_config = client.get_config()
if current_config:
    print(f"当前模式: {current_config.mode}")
```

### 消息发送

> **协议参考**：消息发送对应接口文档 §3 后续请求（纯数据），包括请求格式（§3.1）、参数说明（§3.2）、数据类型定义（§3.3）、能力路由规则（§3.4）。

#### `async begin_send() -> None`

生成新的对话 ID（CID），开始新一轮对话。

**示例:**

```python
await client.begin_send()
```

#### `async cancel() -> None`

取消当前所有活跃的服务端响应（向服务端发送 `response.cancel`），并立即生成新的 CID。

> 协议参考：打断机制见接口文档 §3.5 打断事件（response.cancel），打断策略配置见 §5.1.4。

**示例:**

```python
await client.cancel()
```

#### `async cancel_cid(cid: str) -> None`

取消指定的服务端响应 CID（向服务端发送 `response.cancel`）。与 `cancel()` 不同，仅针对单个 CID，适用于全双工等存在多个并发响应的场景。

**参数:**

+ `cid`: 要取消的响应 CID

**示例:**

```python
await client.cancel_cid(cid)
```

#### `async send_text(text: str, extend: dict | None = None) -> str`

发送文本消息（自动调用 begin_send）。

**参数:**

+ `text`: 文本内容
+ `extend`: 扩展信息（可选）

**返回值:**

+ `str`: 本次消息使用的对话 ID（CID）。可用于关联后续 `nlu.answer` / `tts.audio` / `event.cid_end` 等事件。

**示例:**

```python
cid = await client.send_text("你好，世界！")
print(f"本轮对话 cid={cid}")
```

#### `async send_text_stream(text: str, is_end: bool, extend: dict | None = None) -> str`

以流式方式发送文本。

**参数:**

+ `text`: 文本内容
+ `is_end`: 是否为最后一个分片
+ `extend`: 扩展信息（可选）

**返回值:**

+ `str`: 本次发送对应的 CID。流式发送过程中各分片返回同一个 CID（直到 `is_end=True` 时该 CID 会被切换为下一轮）。

**示例:**

```python
await client.begin_send()
cid = await client.send_text_stream("这是第一部分", False)
await client.send_text_stream("这是第二部分", False)
await client.send_text_stream("这是最后一部分", True)
print(f"本轮对话 cid={cid}")
```

#### `async send_audio(audio_data: bytes, extend: dict | None = None) -> str`

发送音频数据（单帧，自动调用 begin_send）。

**参数:**

+ `audio_data`: 音频字节数组
+ `extend`: 扩展信息（可选）

**返回值:**

+ `str`: 本次消息使用的 CID。

**示例:**

```python
audio_data = b'\x00\x01\x02...'
cid = await client.send_audio(audio_data)
```

#### `async send_audio_stream(audio_data: bytes, is_end: bool, extend: dict | None = None) -> str`

以流式方式发送音频。

**参数:**

+ `audio_data`: 音频字节数组
+ `is_end`: 是否为最后一帧
+ `extend`: 扩展信息（可选）

**返回值:**

+ `str`: 本次发送对应的 CID。流式过程中所有分片返回同一个 CID。

**示例:**

```python
await client.begin_send()

# 持续发送音频流
for chunk in audio_chunks:
    cid = await client.send_audio_stream(chunk, False)

# 发送最后一帧
cid = await client.send_audio_stream(last_chunk, True)
print(f"本轮对话 cid={cid}")
```

#### `async send_image(image_data: str, extend: dict | None = None) -> str`

发送图片（自动调用 begin_send）。

**参数:**

+ `image_data`: Base64 编码的图片数据或图片 URL
+ `extend`: 扩展信息（可选）

**返回值:**

+ `str`: 本次消息使用的 CID。

**示例:**

```python
base64_image = "data:image/png;base64,iVBORw0KGgo..."
cid = await client.send_image(base64_image, {"format": "png"})
```

#### `async send_msgs(items: list) -> str`

发送多模态消息（自动调用 begin_send）。

**参数:**

+ `items`: 消息项列表（TextItem, ImageItem, AudioItem）

**返回值:**

+ `str`: 本次消息使用的 CID。

**示例:**

```python
from aichain_sdk.events import TextItem, ImageItem

items = [
    TextItem(data="请看这张图片"),
    ImageItem(data="base64-image-data", extend={"format": "jpg"}),
    TextItem(data="这是什么？")
]
cid = await client.send_msgs(items)
```

### 事件监听

#### `on(event: str) -> Callable`

添加事件监听器（装饰器方式）。

**参数:**

+ `event`: 事件名称

**示例:**

```python
@client.on("nlu.answer")
def on_answer(cid: str, result, is_last: bool):
    print(result.answer, end="", flush=True)
    if is_last:
        print()
```

#### `off(event: str, listener: Callable | None = None) -> None`

移除事件监听器。

**参数:**

+ `event`: 事件名称
+ `listener`: 要移除的监听器，如果为 `None` 则移除该事件的所有监听器

**示例:**

```python
def my_listener(cid, result, is_last):
    print(result.answer)


client.on("nlu.answer")(my_listener)
# 移除特定监听器
client.off("nlu.answer", my_listener)
# 移除所有监听器
client.off("nlu.answer")
```

#### `remove_all_listeners(event: str | None = None) -> None`

移除所有事件监听器。

**参数:**

+ `event`: 事件名称，如果为 `None` 则移除所有事件的所有监听器

**示例:**

```python
# 移除特定事件的所有监听器
client.remove_all_listeners("nlu.answer")
# 移除所有监听器
client.remove_all_listeners()
```

---

## 事件系统

SDK 提供了完整的事件系统，所有事件回调都在异步上下文中执行。

> **协议参考**：事件系统对应的协议响应格式见接口文档 §9 响应格式（统一响应结构 §9.1、STT 响应 §9.2、NLU 响应 §9.3、TTS 响应
> §9.4、Welcome 响应 §9.5、系统消息 §9.6、End 帧 §9.7、STT 撤销 §9.8、错误消息 §9.9、组合能力响应流程 §9.10）。

### 连接事件

> **协议参考**：连接生命周期见接口文档 §1.4 连接响应 和 §1.5 重连机制。

#### `connecting`

连接开始时触发。

**回调参数:** 无

**示例:**

```python
@client.on("connecting")
def on_connecting():
    print("开始连接...")
```

#### `session.created`

会话创建成功时触发。

**回调参数:**

+ `sid` (str): 会话 ID

**示例:**

```python
@client.on("session.created")
def on_session_created(sid: str):
    print(f"会话已创建: {sid}")
```

#### `session.configed`

会话配置成功时触发。

**回调参数:**

+ `sid` (str): 会话 ID
+ `config` (SessionConfig): 配置对象

**示例:**

```python
@client.on("session.configed")
def on_session_configed(sid: str, config):
    print(f"会话已配置: {sid}")
```

#### `session.error`

会话错误时触发。

**回调参数:**

+ `error` (ErrorMessage): 错误对象

**示例:**

```python
@client.on("session.error")
def on_session_error(error):
    print(f"会话错误: {error.message} (code: {error.code})")
```

#### `reconnecting`

重连开始时触发。

**回调参数:**

+ `attempt` (int): 当前重连次数
+ `max_attempts` (int): 最大重连次数

**示例:**

```python
@client.on("reconnecting")
def on_reconnecting(attempt: int, max_attempts: int):
    print(f"重连中: {attempt}/{max_attempts}")
```

#### `close`

连接关闭时触发。

**回调参数:**

+ `code`(int): 关闭代码
+ `reason`(str): 关闭原因

**示例:**

```python
@client.on("close")
def on_close(code: int, reason: str):
    print(f"连接已关闭: {code} - {reason}")
```

#### `error`

WebSocket 错误时触发。

**回调参数:**

+ `code` (int): 错误代码
+ `message` (str): 错误消息

**示例:**

```python
@client.on("error")
def on_error(code: int, message: str):
    print(f"错误: {code} - {message}")
```

#### `pong`

收到心跳响应时触发。

**回调参数:** 无

**示例:**

```python
@client.on("pong")
def on_pong():
    print("心跳正常")
```

### STT 事件

> **协议参考**：STT 响应格式见接口文档 §9.2 STT 响应，含 action 机制（append/replace）、position 位置替换等详细说明。

#### `stt.result`

语音识别结果。

**回调参数:**

+ `cid` (str): 对话 ID
+ `result` (STTAnswerResult): 识别结果
+ `is_last` (bool): 是否为最后一个结果

**STTAnswerResult 字段:**

+ `text`: 识别的文本
+ `action`: 动作类型（"append" 追加，"replace" 替换）
+ `position`: 替换位置（仅当 action 为 "replace" 时存在，包含 `start`/`end` 字符索引）
+ `index`: 结果索引
+ `language`: 识别出的语种代码（如 "zh"、"en"）

**示例:**

```python
@client.on("stt.result")
def on_stt_result(cid: str, result, is_last: bool):
    if result.action == "append":
        print(f"追加文本: {result.text}")
    elif result.action == "replace":
        print(f"替换文本: {result.text}")

    if is_last:
        print("识别完成")
```

### NLU 事件

> **协议参考**：NLU 响应格式见接口文档 §9.3 NLU 响应，含直接问答（§9.3.1）、工具调用流程（§9.3.2）、RAG 检索（§9.3.3）、链路追踪（§9.3.4）。

#### `nlu.answer`

NLU 回答结果。

**回调参数:**

+ `cid` (str): 对话 ID
+ `result` (NLUAnswerResult): 回答结果
+ `is_last` (bool): 是否为最后一个分片

**NLUAnswerResult 字段:**

+ `answer`: 回答文本
+ `index`: 分片索引
+ `answerSource`: 回答来源（"llm", "rag", "faq" 等）
+ `replyLanguage`: 回复语种代码（如 "zh"、"en"），仅当配置了 `nlu.replyLanguage` 语种跟随时返回

**示例:**

```python
@client.on("nlu.answer")
def on_nlu_answer(cid: str, result, is_last: bool):
    print(result.answer, end="", flush=True)  # 流式输出

    if is_last:
        print(f"\n回答完成，来源: {result.answerSource}")
```

#### `nlu.tool_selection`

工具选择事件。

**回调参数:**

+ `cid` (str): 对话 ID
+ `tool` (str): 工具名称
+ `toolArgs` (dict): 工具参数

**NLUToolSelectionResult 字段:**

+ `tool`: 工具名称
+ `toolArgs`: 工具参数（dict）
+ `toolSelectedDuration`: 工具选择耗时（毫秒）

**示例:**

```python
@client.on("nlu.tool_selection")
def on_tool_selection(cid: str, tool: str, tool_args: dict):
    print(f"选择工具: {tool}")
    print(f"参数: {tool_args}")
```

#### `nlu.tool_execution`

工具执行结果。

**回调参数:**

+ `cid` (str): 对话 ID
+ `result` (NLUToolExecutionResult): 执行结果

**NLUToolExecutionResult 字段:**

+ `tool`: 工具名称
+ `result`: 执行结果
+ `customData`: 自定义数据
+ `executionDuration`: 执行耗时（毫秒）

**示例:**

```python
@client.on("nlu.tool_execution")
def on_tool_execution(cid: str, result):
    print(f"工具 {result.tool} 执行完成")
    print(f"结果: {result.result}")
    print(f"耗时: {result.executionDuration}ms")
```

#### `nlu.rag_retrieval`

RAG  
检索结果。

**回调参数:**

+ `cid`(str): 对话 ID
+ `result`(NLURagRetrievalResult): 检索结果

**NLURagRetrievalResult字段:**

+ `docSource`: 文档来源列表
+ `retrievalDuration`: 检索耗时（毫秒）

**DocSource字段:**

+ `docId`: 文档ID
+ `title`: 文档标题
+ `score`: 相似度分数
+ `content`: 文档内容

**示例:**

```python
@client.on("nlu.rag_retrieval")
def on_rag_retrieval(cid: str, result):
    print(f"检索到 {len(result.docSource)} 个文档")
    for doc in result.docSource:
        print(f"文档: {doc.title}, 分数: {doc.score}")
```

#### `nlu.trace`

NLU 追踪信息。

**回调参数:**

+ `cid` (str): 对话 ID
+ `result` (NLUTraceResult): 追踪信息

**NLUTraceResult 字段:**

+ `nluTimeConsuming`: NLU 总耗时（毫秒）
+ `extend`: 扩展信息
+ `otherParam`: 其他参数

**示例:**

```python
@client.on("nlu.trace")
def on_nlu_trace(cid: str, result):
    print(f"NLU 耗时: {result.nluTimeConsuming}ms")
```

#### `nlu.postprocess`

NLU 结果后处理事件，服务端在生成回答后对其做额外加工时下发。

**回调参数:**

+ `cid` (str): 对话 ID
+ `result` (NLUPostProcessResult): 后处理结果

**NLUPostProcessResult 字段:**

+ `content`: 处理后的内容
+ `originalAnswerSource`: 原始回答来源
+ `customData`: 自定义数据

**示例:**

```python
@client.on("nlu.postprocess")
def on_nlu_postprocess(cid: str, result):
    print(f"后处理结果: {result.content}")
```

### TTS 事件

> **协议参考**：TTS 响应格式见接口文档 §9.4 TTS 响应，音频数据为 Base64 编码，SDK 自动解码为 bytes。

#### `tts.audio`

TTS 音频数据。

**回调参数:**

+ `cid` (str): 对话 ID
+ `audio` (TTSAudioResult): 音频数据
+ `is_last` (bool): 是否为最后一帧

**TTSAudioResult 字段:**

+ `data`: 解码后的音频字节数据
+ `index`: 音频帧索引

**示例:**

```python
import wave

audio_chunks = []


@client.on("tts.audio")
def on_tts_audio(cid: str, audio, is_last: bool):
    audio_chunks.append(audio.data)

    if is_last:
        # 保存为 WAV 文件
        pcm_data = b"".join(audio_chunks)
        with wave.open("output.wav", "wb") as wf:
            wf.setnchannels(1)
            wf.setsampwidth(2)
            wf.setframerate(24000)
            wf.writeframes(pcm_data)
        audio_chunks.clear()
        print("音频已保存")
```

### 欢迎消息事件

> **协议参考**：欢迎消息响应格式见接口文档 §9.5 Welcome 响应，分为文本（welcome.answer）和音频（welcome.audio）两部分。

#### `welcome.answer`

欢迎消息文本。

**回调参数:**

+ `cid` (str): 对话 ID
+ `result` (WelcomeAnswerResult): 欢迎消息

**WelcomeAnswerResult 字段:**

+ `answer`: 欢迎文本
+ `answerSource`: 来源（通常为 "welcome"）

**示例:**

```python
@client.on("welcome.answer")
def on_welcome_answer(cid: str, result):
    print(f"欢迎消息: {result.answer}")
```

#### `welcome.audio`

欢迎消息音频。

**回调参数:**

+ `cid` (str): 对话 ID
+ `result` (WelcomeAudioResult): 音频数据

**WelcomeAudioResult 字段:**

+ `data`: 解码后的音频字节数据

**示例:**

```python
@client.on("welcome.audio")
def on_welcome_audio(cid: str, result):
    # 播放欢迎音频
    play_audio(result.data)
```

### 对话控制事件

> **协议参考**：系统消息响应格式见接口文档 §9.6 系统消息响应，End 帧见 §9.7，STT 撤销见 §9.8，空闲超时见 §9.6.1。

#### `event.interrupted`

对话被打断。

**回调参数:**

+ `cid` (str): 对话 ID

**示例:**

```python
@client.on("event.interrupted")
def on_interrupted(cid: str):
    print(f"对话 {cid} 被打断")
    # 停止音频播放
    audio_player.stop()
```

#### `event.user_speech_started`

用户开始说话。

**回调参数:**

+ `cid` (str): 对话 ID

**示例:**

```python
@client.on("event.user_speech_started")
def on_speech_started(cid: str):
    print("用户开始说话")
    # 显示监听指示器
```

#### `event.user_speech_stopped`

用户停止说话。

**回调参数:**

+ `cid` (str): 对话 ID

**示例:**

```python
@client.on("event.user_speech_stopped")
def on_speech_stopped(cid: str):
    print("用户停止说话")
    # 隐藏监听指示器
```

#### `event.cid_end`

对话结束。

**回调参数:**

+ `cid`(str): 对话 ID
+ `stats`(CidEndStats): 统计信息

**CidEndStats字段:**

+ `totalDuration`: 总耗时（毫秒）
+ `sttDuration`: STT 耗时（毫秒）
+ `nluDuration`: NLU 耗时（毫秒）
+ `ttsDuration`: TTS 耗时（毫秒）
+ `totalTimeConsuming`: 端到端总耗时（毫秒）

**示例**

```python
@client.on("event.cid_end")
def on_cid_end(cid: str, stats):
    print(f"对话结束")
    print(f"总耗时: {stats.totalDuration}ms")
    print(f"STT: {stats.sttDuration}ms, NLU: {stats.nluDuration}ms, TTS: {stats.ttsDuration}ms")
```

#### `event.stt_revocation`

STT 结果撤销事件，服务端通知客户端撤销上一轮的 STT 结果（例如打断时）。

**回调参数:**

+ `cid` (str): 对话 ID
+ `source` (str): 撤销来源（如 `"INTERRUPT"` 表示打断触发）

**示例:**

```python
@client.on("event.stt_revocation")
def on_stt_revocation(cid: str, source: str):
    print(f"撤销 cid={cid} 的 STT 结果, source={source}")
    # 清除上一轮在 UI 上展示的识别文本
```

#### `event.idle_timeout`

客户端空闲超时，服务端主动关闭连接。SDK 会在派发该事件前完成 WebSocket 关闭、心跳/重连任务取消等工作。

**回调参数:**

+ `sid` (str | None): 会话 ID
+ `reason` (str): 关闭原因

**示例:**

```python
@client.on("event.idle_timeout")
def on_idle_timeout(sid, reason: str):
    print(f"空闲超时关闭: sid={sid}, reason={reason}")
```

---

## 完整集成流程

> **协议参考**：完整的 WebSocket 交互流程示例见接口文档 §8 完整示例，含完整连接流程（§8.1）、纯语音识别（§8.2）、多模态输入（§8.3）等场景。

### 1. 基础文本对话集成

```python
"""
场景：纯文本对话（仅 NLU）
演示多轮文本对话的基本用法。
"""
import asyncio
from gc import enable
import os
import logging

from aichain_sdk import AIChainClient
from aichain_sdk.events import NLUAnswerResult


async def main():
    log_file = "../log.txt"
    if os.path.exists(log_file):
        os.remove(log_file)
    logger = logging.getLogger("aichain_sdk")
    client = AIChainClient(
        app_id="***",
        app_key="***",
        host="wss://aichain-sh.xfyun.cn/",
        sn="SN123456789",
        log_level="DEBUG",
        log_file=log_file
    )

    @client.on("nlu.answer")
    def on_answer(cid: str, result: NLUAnswerResult, is_last: bool):
        print(result.answer, end="", flush=True)
        if is_last:
            print(f"\n[done, source={result.answerSource}]")

    @client.on("error")
    def on_error(code, message):
        print(f"[error] {code}: {message}")

    @client.on("session.created")
    def on_session_created(sid: str):
        print(f"[session_created] {sid}")

    await client.connect()
    await client.set_config({
        "multiTurnEnabled": True,
        "stt": {
            "enable": False
        },
        "nlu": {
            "enable": True
        },
        "tts": {
            "enable": False
        }
    })

    questions = ["北京的天气怎么样？", "那上海呢？", "两个城市哪个更适合旅游？"]
    for q in questions:
        print(f"\n[Q] {q}")
        await client.send_text(q)
        await asyncio.sleep(15)

    await client.disconnect()


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

### 2. 半双工语音识别集成

```bash
# 安装依赖
uv add aichain-sdk==1.0.7
```

```python
"""
场景：半双工语音识别+语义处理+语音合成（仅 STT + NLU + TTS）
演示半双工模式下发送音频流并获取识别文本，进行语义处理，并合成语音输出。
"""
import asyncio
import wave
import os
import logging
import json

from aichain_sdk import AIChainClient
from aichain_sdk.events import STTAnswerResult, CidEndStats, NLUAnswerResult, TTSAudioResult


def read_wav_chunks(path: str, chunk_ms: int = 40, sample_rate: int = 16000) -> list[bytes]:
    """将 WAV 文件按 chunk_ms 毫秒切片，返回 PCM 字节块列表。"""
    with wave.open(path, "rb") as wf:
        assert wf.getnchannels() == 1, "仅支持单声道"
        assert wf.getsampwidth() == 2, "仅支持 16bit"
        chunk_frames = int(sample_rate * chunk_ms / 1000)
        chunks = []
        while True:
            data = wf.readframes(chunk_frames)
            if not data:
                break
            chunks.append(data)
    return chunks


def _save_wav(path: str, pcm: bytes, sample_rate: int = 16000):
    with wave.open(path, "wb") as wf:
        wf.setnchannels(1)
        wf.setsampwidth(2)
        wf.setframerate(sample_rate)
        wf.writeframes(pcm)


async def main():
    log_file = "../log.txt"
    if os.path.exists(log_file):
        os.remove(log_file)
    logger = logging.getLogger("aichain_sdk")
    client = AIChainClient(
        app_id="***",
        app_key="***",
        host="wss://aichain-sh.xfyun.cn/",
        sn="SN123456789",
        log_level="DEBUG",
        log_file=log_file,
    )

    stt_text = []

    @client.on("stt.result")
    def on_stt(cid: str, result: STTAnswerResult, is_last: bool):
        if result.action == "append":
            # 追加文本
            stt_text.append(result.text)

        elif result.action == "replace":
            if not result.position or result.position.start == 0:
                # 无位置信息或从头开始：直接替换整个文本
                # 服务端通常返回 start=0 表示"从头开始的完整识别结果"
                stt_text.clear()
                stt_text.append(result.text)
            else:
                # 替换中间部分（罕见情况）
                start = result.position.start
                end = result.position.end
                current_text = "".join(stt_text)

                if 0 < start <= end < len(current_text):
                    # 替换 [start, end] 区间的字符
                    new_text = current_text[:start] + result.text + current_text[end + 1:]
                    stt_text.clear()
                    stt_text.append(new_text)
                else:
                    # 索引越界，降级为全量替换
                    stt_text.clear()
                    stt_text.append(result.text)

        display_text = "".join(stt_text)
        # print(f"[stt] {display_text}", end="\r", flush=True)
        if is_last:
            print(f"\n[stt done] {display_text}")

    @client.on("event.cid_end")
    def on_cid_end(cid: str, stats: CidEndStats):
        print(f"[cid_end] stt={stats.sttDuration}ms total={stats.totalDuration}ms")

    @client.on("error")
    def on_error(code, message):
        print(f"[error] {code}: {message}")

    @client.on("session.created")
    def on_session_created(sid: str):
        print(f"[session_created] {sid}")

    acc_text = []

    @client.on("nlu.answer")
    def on_nlu_answer(cid: str, result: NLUAnswerResult, is_last: bool):
        acc_text.append(result.answer)
        display_text = "".join(acc_text)
        # print(f"[NLU] {display_text}", end="\r", flush=True)
        if is_last:
            print(f"[NLU done] 语义结果 ${display_text}")

    tts_chunks: list[bytes] = []

    @client.on("tts.audio")
    def on_tts(cid: str, audio: TTSAudioResult, is_last: bool):
        tts_chunks.append(audio.data)
        if is_last:
            # 将所有音频块合并保存为 WAV
            pcm = b"".join(tts_chunks)
            tts_chunks.clear()
            _save_wav("examples/outputs/tts_output.wav", pcm)
            print("[tts] 音频已保存到 tts_output.wav")

    await client.connect()
    await client.set_config({
        "mode": "half_duplex",
        "stt": {
            "enable": True,
            "language": "zh",
            "audioConfig": {
                "audioEncoding": "raw",
                "format": "plain",
                "sampleRate": 16000,
                "bitDepth": 16,
                "channels": 1,
            },
            "vad": {
                "enable": True,
                "minSpeechDuration": 50,
                "minSilenceDuration": 550,
                "activationThreshold": 0.5,
            },
            "turnDetection": {
                "enable": False
            },
            "interrupt": {
                "enable": False
            }
        },
        "nlu": {
            "enable": True
        },
        "tts": {
            "enable": True,
            "textFilters": ["filter_markdown", "filter_emoji"],
            "voices": {
                "zh": {
                    "voiceId": "4",
                    "audioConfig": {
                        "audioEncoding": "raw",
                        "format": "wav",
                        "sampleRate": 16000,
                        "bitDepth": 16,
                        "channels": 1,
                    },
                }
            },
        }
    })
    # print(json.dumps(client.get_config().to_dict(), ensure_ascii=False, indent=2))

    # 替换为实际 WAV 文件路径（16kHz, 16bit, 单声道）
    wav_path = "examples/inputs/test.wav"
    chunks = read_wav_chunks(wav_path)

    await client.begin_send()
    for i, chunk in enumerate(chunks):
        is_end = i == len(chunks) - 1
        await client.send_audio_stream(chunk, is_end)
        await asyncio.sleep(0.04)  # 模拟实时发送间隔 40ms

    # 等待识别结果返回
    await asyncio.sleep(10)
    await client.disconnect()


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

### 3. 全双工语音对话集成（麦克风输入）

```bash
# 安装依赖
uv add aichain-sdk==1.0.7 sounddevice numpy
```

```python
"""
场景：全双工语音对话（STT + NLU + TTS，支持打断）
演示持续发送音频、服务端 VAD 检测、打断处理的全双工模式。

使用麦克风作为音频输入

全双工特性：
1. 麦克风持续录音并发送音频流
2. 用户可随时打断AI回答（通过 VAD 检测）
3. 检测到用户说话时立即停止TTS播放
4. 支持多轮对话，无需等待上一轮结束
5. 使用音频缓冲区平滑播放TTS，避免断断续续

注意事项：
- 建议使用耳机避免回声（麦克风录到扬声器声音）
- 或使用带回声消除的音频设备
- VAD参数可根据环境调整灵敏度
"""
import asyncio
import queue
import signal
import sounddevice as sd
import os
import logging
import numpy as np

from aichain_sdk import AIChainClient
from aichain_sdk.events import STTAnswerResult, NLUAnswerResult, TTSAudioResult, CidEndStats


async def main():
    log_file = "./log.txt"
    if os.path.exists(log_file):
        os.remove(log_file)
    logger = logging.getLogger("aichain_sdk")
    # local环境
    client = AIChainClient(
        app_id="***",
        app_key="***",
        host="wss://aichain-sh.xfyun.cn/",
        sn="SN123456789",
        log_level="DEBUG",
        log_file=log_file,
    )

    tts_playing = False
    # TTS 播放队列（使用标准 queue 以支持跨线程）
    tts_queue: queue.Queue[bytes] = queue.Queue(maxsize=100)
    # TTS 音频缓冲区，用于平滑播放
    tts_buffer = bytearray()

    # 音频队列，用于线程安全的跨线程数据传递
    audio_queue: queue.Queue[bytes] = queue.Queue(maxsize=100)
    recording = True
    stop_event = asyncio.Event()

    def audio_callback(indata, frames, time, status):
        """麦克风录音回调，将音频数据放入队列"""
        if status:
            print(f"[audio callback] {status}")
        audio_data = indata.tobytes()
        try:
            audio_queue.put_nowait(audio_data)
        except queue.Full:
            pass  # 队列满则丢弃

    # 设置 Ctrl+C 信号处理
    def handle_interrupt():
        nonlocal recording
        recording = False
        stop_event.set()

    loop = asyncio.get_running_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        try:
            loop.add_signal_handler(sig, handle_interrupt)
        except NotImplementedError:
            # Windows 上不支持 add_signal_handler，使用替代方案
            pass

    stt_text = []
    stt_streaming = False
    MAX_DISPLAY_WIDTH = 100  # 最大显示宽度（字符数）

    @client.on("stt.result")
    def on_stt(cid: str, result: STTAnswerResult, is_last: bool):
        nonlocal stt_streaming

        if result.action == "append":
            # 追加文本
            stt_text.append(result.text)
        elif result.action == "replace":
            if not result.position or result.position.start == 0:
                # 无位置信息或从头开始：直接替换整个文本
                # 服务端通常返回 start=0 表示"从头开始的完整识别结果"
                stt_text.clear()
                stt_text.append(result.text)
            else:
                # 替换中间部分（罕见情况）
                start = result.position.start
                end = result.position.end
                current_text = "".join(stt_text)

                if 0 < start <= end < len(current_text):
                    # 替换 [start, end] 区间的字符
                    new_text = current_text[:start] + result.text + current_text[end + 1:]
                    stt_text.clear()
                    stt_text.append(new_text)
                else:
                    # 索引越界，降级为全量替换
                    stt_text.clear()
                    stt_text.append(result.text)

        # 流式输出：使用 \r 覆盖同一行，限制显示长度避免换行
        display_text = "".join(stt_text)
        if len(display_text) > MAX_DISPLAY_WIDTH:
            display_text = display_text[:MAX_DISPLAY_WIDTH - 3] + "..."

        if not stt_streaming:
            stt_streaming = True
        print(f"\r[STT] {display_text}", end="", flush=True)

        if is_last:
            # 最后一次输出完整文本（可能多行）
            full_text = "".join(stt_text)
            print(f"\r[STT] {full_text}")
            stt_text.clear()
            stt_streaming = False

    nlu_text = []
    nlu_streaming = False

    @client.on("nlu.answer")
    def on_nlu(cid: str, result: NLUAnswerResult, is_last: bool):
        nonlocal nlu_streaming

        nlu_text.append(result.answer)
        display_text = "".join(nlu_text)

        # 流式输出：使用 \r 覆盖同一行，限制显示长度避免换行
        if len(display_text) > MAX_DISPLAY_WIDTH:
            display_text = display_text[:MAX_DISPLAY_WIDTH - 3] + "..."

        if not nlu_streaming:
            nlu_streaming = True
        print(f"\r[NLU] {display_text}", end="", flush=True)

        if is_last:
            # 最后一次输出完整文本（可能多行）
            full_text = "".join(nlu_text)
            print(f"\r[NLU] {full_text}")
            nlu_text.clear()
            nlu_streaming = False

    def clear_tts_queue():
        """清空TTS播放队列和缓冲区"""
        nonlocal tts_playing
        tts_playing = False
        tts_buffer.clear()
        while not tts_queue.empty():
            try:
                tts_queue.get_nowait()
            except queue.Empty:
                break

    @client.on("tts.audio")
    def on_tts(cid: str, audio: TTSAudioResult, is_last: bool):
        nonlocal tts_playing
        if not tts_playing:
            tts_playing = True
            print("[TTS] 开始播放")
        # 实时放入播放队列
        try:
            tts_queue.put_nowait(audio.data)
        except queue.Full:
            pass
        if is_last:
            tts_playing = False
            print("[TTS] 播放结束")

    @client.on("event.interrupted")
    def on_interrupted(cid: str):
        # 如果正在流式输出，先换行
        if stt_streaming or nlu_streaming:
            print()
        clear_tts_queue()
        print(f"[INTERRUPTED] 停止 TTS 播放")

    @client.on("event.user_speech_started")
    def on_speech_started(cid: str):
        # 如果正在流式输出，先换行
        if stt_streaming or nlu_streaming:
            print()
        print(f"[SPEECH_STARTED] 用户开始说话")

    @client.on("event.user_speech_stopped")
    def on_speech_stopped(cid: str):
        # 如果正在流式输出，先换行
        if stt_streaming or nlu_streaming:
            print()
        print(f"[SPEECH_STOPPED] 用户停止说话")

    @client.on("event.cid_end")
    def on_cid_end(cid: str, stats: CidEndStats):
        # 如果正在流式输出，先换行
        if stt_streaming or nlu_streaming:
            print()
        print(f"[CID_END] 本轮对话结束，耗时 {stats.totalDuration}ms\n")

    @client.on("error")
    def on_error(code, message):
        print(f"[error] {code}: {message}")

    await client.connect()
    await client.set_config({
        "mode": "full_duplex",
        "simplifiedResponse": True,
        "multiTurnEnabled": True,
        "stt": {
            "enable": True,
            "language": "zh",
            "audioConfig": {
                "audioEncoding": "raw",
                "format": "plain",
                "sampleRate": 16000,
                "bitDepth": 16,
                "channels": 1,
            },
            "vad": {
                "enable": True,
                "minSpeechDuration": 50,
                "minSilenceDuration": 550,
                "activationThreshold": 0.5,
            },
            "turnDetection": {
                "enable": True,
                "minEndpointingDelay": 500,
                "maxEndpointingDelay": 3000,
            },
            "interrupt": {"enable": True},
        },
        "nlu": {"enable": True},
        "tts": {
            "enable": True,
            "textFilters": ["filter_markdown", "filter_emoji"],
            "voices": {
                "zh": {
                    "voiceId": "4",
                    "audioConfig": {"audioEncoding": "raw"},  # 使用 raw 编码以便直接播放
                }
            },
        },
    })

    # 全双工：使用麦克风持续发送音频
    stt_sample_rate = 16000  # STT 采样率
    tts_sample_rate = 16000  # TTS 采样率（通常为 24000Hz）
    chunk_size = int(stt_sample_rate * 0.04)  # 40ms chunk

    # 启动麦克风录音流
    stream = sd.InputStream(
        samplerate=stt_sample_rate,
        channels=1,
        dtype='int16',
        blocksize=chunk_size,
        callback=audio_callback,
    )
    stream.start()

    # TTS 播放回调
    def tts_play_callback(outdata, frames, time, status):
        """TTS 播放回调，从队列获取音频数据并使用缓冲区平滑播放"""
        if status:
            print(f"[tts play] {status}")

        # 需要的字节数（每帧2字节，int16）
        bytes_needed = frames * 2

        # 从队列补充缓冲区
        while len(tts_buffer) < bytes_needed:
            try:
                audio_data = tts_queue.get_nowait()
                tts_buffer.extend(audio_data)
            except queue.Empty:
                break

        # 从缓冲区取出数据填充输出
        if len(tts_buffer) >= bytes_needed:
            # 缓冲区有足够数据
            audio_bytes = bytes(tts_buffer[:bytes_needed])
            del tts_buffer[:bytes_needed]
            audio_np = np.frombuffer(audio_bytes, dtype=np.int16)
            outdata[:, 0] = audio_np
        elif len(tts_buffer) > 0:
            # 缓冲区数据不足，部分填充
            audio_bytes = bytes(tts_buffer)
            tts_buffer.clear()
            audio_np = np.frombuffer(audio_bytes, dtype=np.int16)
            outdata[:len(audio_np), 0] = audio_np
            outdata[len(audio_np):, 0] = 0
        else:
            # 缓冲区为空，输出静音
            outdata[:] = 0

    # 启动 TTS 音频播放流（使用 TTS 采样率）
    tts_chunk_size = int(tts_sample_rate * 0.04)
    tts_output_stream = sd.OutputStream(
        samplerate=tts_sample_rate,
        channels=1,
        dtype='int16',
        blocksize=tts_chunk_size,
        callback=tts_play_callback,
    )
    tts_output_stream.start()

    print("开始录音，按 Ctrl+C 结束...")

    # 从队列读取音频并发送
    loop = asyncio.get_running_loop()

    def get_audio_from_queue():
        """从队列获取音频数据（同步函数）"""
        try:
            return audio_queue.get_nowait()
        except queue.Empty:
            raise queue.Empty()

    try:
        while recording and not stop_event.is_set():
            try:
                # 使用 run_in_executor 在线程池中从 queue.Queue 获取数据
                audio_data = await loop.run_in_executor(None, get_audio_from_queue)
                await client.send_audio_stream(audio_data, False)
            except queue.Empty:
                await asyncio.sleep(0.01)
                continue
    except asyncio.CancelledError:
        pass
    finally:
        stream.stop()
        stream.close()
        tts_output_stream.stop()
        tts_output_stream.close()
        print("录音已停止")
        await client.disconnect()


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




```

---

## 最佳实践

> **协议参考**：服务端相关的最佳实践建议见接口文档 §12
> 最佳实践，含音频发送建议（§12.1）、连接管理（§12.2）、性能优化（§12.3）、错误处理（§12.4）、安全建议（§12.5）。

### 1. 异步编程最佳实践

```python
import asyncio
from aichain_sdk import AIChainClient


# 使用 async with 管理客户端生命周期
class AIChainClientManager:
    def __init__(self, **kwargs):
        self.client = AIChainClient(**kwargs)

    async def __aenter__(self):
        await self.client.connect()
        return self.client

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.client.disconnect()


# 使用示例
async def main():
    async with AIChainClientManager(
            app_id="your-app-id",
            app_key="your-app-key",
            host="your-host",
            sn="device-12345"
    ) as client:
        await client.set_config({"nlu": {"enable": True}})
        await client.send_text("你好")
        await asyncio.sleep(5)
```

### 2. 错误处理和重试

```python
import asyncio
from aichain_sdk import AIChainClient


async def connect_with_retry(client, max_retries=3):
    """带重试的连接"""
    for attempt in range(max_retries):
        try:
            await client.connect()
            print("连接成功")
            return True
        except Exception as e:
            print(f"连接失败 (尝试 {attempt + 1}/{max_retries}): {e}")
            if attempt < max_retries - 1:
                await asyncio.sleep(2 ** attempt)  # 指数退避
    return False


async def main():
    client = AIChainClient(
        app_id="your-app-id",
        app_key="your-app-key",
        host="your-host",
        sn="device-12345"
    )

    if await connect_with_retry(client):
        # 继续执行
        pass
```

### 3. 日志管理

```python
import logging
from aichain_sdk import AIChainClient

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(name)s - %(message)s',
    handlers=[
        logging.FileHandler('aichain.log', encoding='utf-8'),
        logging.StreamHandler()
    ]
)

client = AIChainClient(
    app_id="your-app-id",
    app_key="your-app-key",
    host="your-host",
    sn="device-12345",
    log_level="DEBUG",
    log_file="aichain_sdk.log"
)
```

### 4. 内存管理

```python
import asyncio
from collections import deque


class AudioBufferPool:
    """音频缓冲区对象池"""

    def __init__(self, buffer_size=4096, max_pool_size=10):
        self.buffer_size = buffer_size
        self.pool = deque(maxlen=max_pool_size)

    def get_buffer(self):
        """获取缓冲区"""
        if self.pool:
            return self.pool.pop()
        return bytearray(self.buffer_size)

    def return_buffer(self, buffer):
        """归还缓冲区"""
        if len(self.pool) < self.pool.maxlen:
            self.pool.append(buffer)


# 使用示例
buffer_pool = AudioBufferPool()


async def process_audio():
    buffer = buffer_pool.get_buffer()
    try:
        # 使用缓冲区
        pass
    finally:
        buffer_pool.return_buffer(buffer)
```

### 5. 并发任务管理

```python
import asyncio
from aichain_sdk import AIChainClient


async def handle_multiple_conversations():
    """处理多个并发对话"""
    client = AIChainClient(
        app_id="your-app-id",
        app_key="your-app-key",
        host="your-host",
        sn="device-12345"
    )

    await client.connect()
    await client.set_config({"nlu": {"enable": True}})

    # 并发发送多个问题
    questions = ["问题1", "问题2", "问题3"]
    tasks = []

    for q in questions:
        async def send_question(question):
            await client.send_text(question)
            await asyncio.sleep(5)

        tasks.append(send_question(q))

    # 等待所有任务完成
    await asyncio.gather(*tasks)
    await client.disconnect()
```

### 6. 优雅关闭

```python
import asyncio
import signal
from aichain_sdk import AIChainClient


async def main():
    client = AIChainClient(
        app_id="your-app-id",
        app_key="your-app-key",
        host="your-host",
        sn="device-12345"
    )

    # 设置信号处理
    stop_event = asyncio.Event()

    def signal_handler():
        print("\n收到退出信号，正在关闭...")
        stop_event.set()

    loop = asyncio.get_running_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        try:
            loop.add_signal_handler(sig, signal_handler)
        except NotImplementedError:
            # Windows 不支持 add_signal_handler
            pass

    await client.connect()
    await client.set_config({"nlu": {"enable": True}})

    try:
        # 主循环
        while not stop_event.is_set():
            await asyncio.sleep(1)
    finally:
        # 优雅关闭
        await client.disconnect()
        print("已断开连接")


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

---

## 错误处理

> **协议参考**：完整的错误码体系见接口文档 §10 错误码，含系统级错误（§10.1）、业务级错误（§10.2）、STT 错误（§10.3）、NLU
> 错误（§10.4）、TTS 错误（§10.5）。

### 错误码列表

下表为常见错误码（具体含义以服务端协议为准）：

| 错误码   | 说明                          | 处理建议                                                 |
|-------|-----------------------------|------------------------------------------------------|
| 10010 | WebSocket 连接失败              | 检查网络、host 与端口、是否需要 wss                               |
| 其它    | 业务错误（认证、配置、STT/NLU/TTS 异常等） | 通过 `session.error` / `error` 事件获取 `code` 与 `message` |

> 当收到 `session.error` 时，SDK 会自动禁用后续 `auto_reconnect`、终止当前重连任务，并将连接置为 `DISCONNECTED`  
> ，由上层根据错误自行决定是否手动 `reconnect()`。
>

### 错误处理示例

```python
import asyncio
from aichain_sdk import AIChainClient
from aichain_sdk.errors import AIChainError


async def main():
    client = AIChainClient(
        app_id="your-app-id",
        app_key="your-app-key",
        host="your-host",
        sn="device-12345"
    )

    @client.on("error")
    def on_error(code: int, message: str):
        if code == 10010:
            # WebSocket 连接失败
            print("网络连接失败，请检查 host / 网络")
        else:
            print(f"发生错误: code={code}, message={message}")

    @client.on("session.error")
    def on_session_error(error):
        # error 是 AIChainError，包含 sid/cid/code/message
        print(f"业务错误: code={error.code}, message={error.message}")
        # 注意：收到 session.error 后 SDK 会关闭自动重连，需要手动调用 reconnect()

    try:
        await client.connect()
        await client.set_config({"nlu": {"enable": True}})
        await client.send_text("你好")
        await asyncio.sleep(5)
    except Exception as e:
        print(f"异常: {e}")
    finally:
        await client.disconnect()


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

---

## 常见问题

> **更多 FAQ**：服务端协议层面的常见问题见接口文档 §14 FAQ，涵盖能力配置、session 配置、交互模式选择、并发限流、模型配置等。

### Q1: 如何处理音频格式转换？

**A:** SDK 接受 PCM 格式的音频数据。如果你的音频是其他格式，需要先转换：

```python
import wave
import numpy as np


def convert_to_pcm(input_file: str) -> bytes:
    """将音频文件转换为 PCM 格式"""
    with wave.open(input_file, "rb") as wf:
        # 读取音频参数
        channels = wf.getnchannels()
        sample_width = wf.getsampwidth()
        framerate = wf.getframerate()

        # 读取所有帧
        frames = wf.readframes(wf.getnframes())

        # 如果需要重采样到 16kHz
        if framerate != 16000:
            # 使用 scipy 或 librosa 进行重采样
            pass

        return frames
```

### Q2: 如何实现语音打断？

**A:** 启用 interrupt 配置，并监听 `event.interrupted` 事件：

```python
config = {
    "stt": {
        "interrupt": {"enable": True}
    }
}


@client.on("event.interrupted")
def on_interrupted(cid: str):
    # 停止当前播放
    audio_player.stop()
    # 开始新的对话
    asyncio.create_task(client.begin_send())
```

### Q3: 如何实现多轮对话？

**A:** 启用 `multiTurnEnabled` 并使用 `begin_send()` 管理对话轮次：

```python
config = {
    "mode": "half_duplex",
    "multiTurnEnabled": True
}

await client.set_config(config)

# 第一轮对话
await client.send_text("今天天气怎么样？")
await asyncio.sleep(10)

# 第二轮对话（上下文保持）
await client.send_text("明天呢？")
```

### Q4: 如何优化内存使用？

**A:** 使用对象池和及时清理资源：

```python
from collections import deque


class AudioBufferPool:
    def __init__(self, buffer_size=4096, max_pool_size=10):
        self.buffer_size = buffer_size
        self.pool = deque(maxlen=max_pool_size)

    def get_buffer(self):
        return self.pool.pop() if self.pool else bytearray(self.buffer_size)

    def return_buffer(self, buffer):
        if len(self.pool) < self.pool.maxlen:
            self.pool.append(buffer)


# 使用示例
buffer_pool = AudioBufferPool()


async def process_audio():
    buffer = buffer_pool.get_buffer()
    try:
        # 使用缓冲区
        pass
    finally:
        buffer_pool.return_buffer(buffer)
```

### Q5: 如何处理网络切换？

**A:** 监听连接状态并自动重连：

```python
@client.on("close")
def on_close(code: int, reason: str):
    print(f"连接关闭: {code} - {reason}")
    if client.auto_reconnect:
        print("自动重连中...")


@client.on("reconnecting")
def on_reconnecting(attempt: int, max_attempts: int):
    print(f"重连尝试: {attempt}/{max_attempts}")
```

### Q6: 如何实现流式文本输入？

**A:** 使用 `send_text_stream()` 方法：

```python
await client.begin_send()

text_chunks = ["这是", "一段", "流式", "文本"]
for i, chunk in enumerate(text_chunks):
    is_last = i == len(text_chunks) - 1
    await client.send_text_stream(chunk, is_last)
```

### Q7: 如何调试 WebSocket 连接？

**A:** 启用 DEBUG 日志级别：

```python
client = AIChainClient(
    app_id="your-app-id",
    app_key="your-app-key",
    host="your-host",
    sn="your-sn",
    log_level="DEBUG",  # 启用详细日志
    log_file="debug.log"
)

# 查看日志文件
# tail -f debug.log
```

### Q8: 如何处理大文件上传（如图片）？

**A:** 将图片转换为 Base64 并发送：

```python
import base64


def image_to_base64(image_path: str) -> str:
    """将图片文件转换为 Base64"""
    with open(image_path, "rb") as f:
        image_data = f.read()
    return base64.b64encode(image_data).decode("utf-8")


# 发送图片
image_base64 = image_to_base64("photo.jpg")
await client.send_image(
    f"data:image/jpeg;base64,{image_base64}",
    {"format": "jpeg"}
)
```

### Q9: 如何在 Windows 上使用信号处理？

**A:** Windows 不支持 `add_signal_handler`，使用替代方案：

```python
import asyncio
import signal
import sys


async def main():
    client = AIChainClient(...)
    stop_event = asyncio.Event()

    def signal_handler(sig, frame):
        print("\n收到退出信号")
        stop_event.set()

    # Windows 兼容的信号处理
    signal.signal(signal.SIGINT, signal_handler)
    if sys.platform != "win32":
        signal.signal(signal.SIGTERM, signal_handler)

    await client.connect()

    try:
        while not stop_event.is_set():
            await asyncio.sleep(1)
    finally:
        await client.disconnect()
```

### Q10: 如何实现音频实时播放？

**A:** 使用 sounddevice 进行实时播放：

```python
import queue
import sounddevice as sd
import numpy as np

tts_queue = queue.Queue(maxsize=100)


def tts_play_callback(outdata, frames, time, status):
    """TTS 播放回调"""
    try:
        audio_data = tts_queue.get_nowait()
        audio_np = np.frombuffer(audio_data, dtype=np.int16)
        if len(audio_np) < frames:
            outdata[:len(audio_np), 0] = audio_np
            outdata[len(audio_np):, 0] = 0
        else:
            outdata[:, 0] = audio_np[:frames]
    except queue.Empty:
        outdata[:] = 0


# 启动播放流
tts_stream = sd.OutputStream(
    samplerate=24000,
    channels=1,
    dtype='int16',
    blocksize=int(24000 * 0.04),
    callback=tts_play_callback
)
tts_stream.start()


@client.on("tts.audio")
def on_tts(cid: str, audio, is_last: bool):
    try:
        tts_queue.put_nowait(audio.data)
    except queue.Full:
        pass
```

---

## 附录

> **协议参考**：接口文档的附录含 ASR 引擎列表（§13.1）、TTS 发音人完整列表（§13.2.1）、音频格式说明（§13.2.2），SDK
> 的类型定义均基于这些协议定义构建。

### A. 完整类型定义

SDK 使用 Pydantic 提供完整的类型定义，支持 IDE 自动补全和类型检查。

```python
from aichain_sdk import (
    AIChainClient,
    SessionConfig,
    VoiceConfig,
    AudioConfig,
    AIChainError,
    get_version,
)

from aichain_sdk.config import (
    SessionConfig,
    STTConfig,
    NLUConfig,
    TTSConfig,
    TTSMarkupTagsConfig,
    TTSMarkupReplacement,
    AudioConfig,
    VadConfig,
    FarFieldFilterConfig,
    TurnDetectionConfig,
    InterruptConfig,
    FillerConfig,
    LLMModel,
    ToolsConfig,
    RagConfig,
    DialogueConfig,
    VoiceConfig,
)

from aichain_sdk.events import (
    STTAnswerResult,
    NLUAnswerResult,
    NLUToolSelectionResult,
    NLUToolExecutionResult,
    NLUPostProcessResult,
    NLURAGRetrievalDoc,
    NLURAGRetrievalResult,
    NLUTraceResult,
    TTSAudioResult,
    WelcomeAnswerResult,
    WelcomeAudioResult,
    CidEndStats,
    TextItem,
    ImageItem,
    AudioItem,
)

from aichain_sdk.types import (
    ConnectionState,
    ConnectionInfo,
)
```

### B. 环境变量配置

可以使用环境变量配置客户端：

```python
import os
from aichain_sdk import AIChainClient

client = AIChainClient(
    app_id=os.getenv("AICHAIN_APP_ID"),
    app_key=os.getenv("AICHAIN_APP_KEY"),
    host=os.getenv("AICHAIN_HOST"),
    sn=os.getenv("AICHAIN_SN"),
    log_level=os.getenv("AICHAIN_LOG_LEVEL", "INFO")
)
```

### C. 性能优化建议

1. **使用连接池**: 对于高并发场景，考虑使用连接池管理多个客户端实例
2. **音频缓冲**: 使用适当的缓冲区大小（推荐 40ms）
3. **异步处理**: 充分利用 asyncio 的并发能力
4. **内存管理**: 使用对象池减少内存分配
5. **日志级别**: 生产环境使用 INFO 或 WARN 级别
