aichain-sdk 1.0.7

AI Chain Python Client SDK for IIP Dispatch WebSocket protocol v2.0

AIChain Dispatch SDK Python 集成指南

目录

• 简介
• 快速开始
• 详细配置
• API 参考
• 事件系统
• 完整集成流程
• 最佳实践
• 错误处理
• 常见问题

---

简介

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

协议参考：本 SDK 基于 AIChain WebSocket 接口协议 v2.1

实现。完整的协议规范（连接鉴权、首帧配置、数据交互、响应格式等）见 API 接口文档。

主要特性

• ✅ 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：

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

或者使用uv管理依赖：

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

2. 初始化客户端

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. 连接服务器

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


asyncio.run(main())

4. 发送消息

@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
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

完整配置示例

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 注入朗读控制标签（数字读法、停顿、拼音矫正、强调等）。仅讯飞系引擎生效，默认关闭，向后兼容。

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	音频输出配置（省略时使用该发音人引擎的默认配置）

完整配置示例

# 使用字典配置
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 连接并等待会话创建。

示例:

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

async disconnect() -> None

关闭 WebSocket 连接。

示例:

await client.disconnect()

async reconnect() -> None

手动重连到服务器。

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

示例:

await client.reconnect()

is_connected() -> bool

检查当前是否已连接。

返回值:

• True: 已连接
• False: 未连接

示例:

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

get_connection_state() -> ConnectionState

获取当前连接状态。

返回值:

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

示例:

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

get_connection_info() -> ConnectionInfo

获取详细的连接信息。

返回值:

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

示例:

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 实例）

示例:

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

await client.set_config(config)

get_config() -> SessionConfig | None

获取当前会话配置。

返回值:

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

示例:

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），开始新一轮对话。

示例:

await client.begin_send()

async cancel() -> None

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

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

示例:

await client.cancel()

async cancel_cid(cid: str) -> None

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

参数:

• cid: 要取消的响应 CID

示例:

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 等事件。

示例:

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 会被切换为下一轮）。

示例:

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。

示例:

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。

示例:

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。

示例:

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。

示例:

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: 事件名称

示例:

@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 则移除该事件的所有监听器

示例:

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 则移除所有事件的所有监听器

示例:

# 移除特定事件的所有监听器
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

连接开始时触发。

回调参数: 无

示例:

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

session.created

会话创建成功时触发。

回调参数:

• sid (str): 会话 ID

示例:

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

session.configed

会话配置成功时触发。

回调参数:

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

示例:

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

session.error

会话错误时触发。

回调参数:

• error (ErrorMessage): 错误对象

示例:

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

reconnecting

重连开始时触发。

回调参数:

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

示例:

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

close

连接关闭时触发。

回调参数:

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

示例:

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

error

WebSocket 错误时触发。

回调参数:

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

示例:

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

pong

收到心跳响应时触发。

回调参数: 无

示例:

@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"）

示例:

@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 语种跟随时返回

示例:

@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: 工具选择耗时（毫秒）

示例:

@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: 执行耗时（毫秒）

示例:

@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: 文档内容

示例:

@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: 其他参数

示例:

@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: 自定义数据

示例:

@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: 音频帧索引

示例:

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"）

示例:

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

welcome.audio

欢迎消息音频。

回调参数:

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

WelcomeAudioResult 字段:

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

示例:

@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

示例:

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

event.user_speech_started

用户开始说话。

回调参数:

• cid (str): 对话 ID

示例:

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

event.user_speech_stopped

用户停止说话。

回调参数:

• cid (str): 对话 ID

示例:

@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: 端到端总耗时（毫秒）

示例

@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" 表示打断触发）

示例:

@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): 关闭原因

示例:

@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. 基础文本对话集成

"""
场景：纯文本对话（仅 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. 半双工语音识别集成

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

"""
场景：半双工语音识别+语义处理+语音合成（仅 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. 全双工语音对话集成（麦克风输入）

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

"""
场景：全双工语音对话（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. 异步编程最佳实践

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. 错误处理和重试

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. 日志管理

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. 内存管理

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. 并发任务管理

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. 优雅关闭

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()。

错误处理示例

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 格式的音频数据。如果你的音频是其他格式，需要先转换：

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 事件：

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() 管理对话轮次：

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: 使用对象池和及时清理资源：

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: 监听连接状态并自动重连：

@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() 方法：

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 日志级别：

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 并发送：

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，使用替代方案：

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 进行实时播放：

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 自动补全和类型检查。

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. 环境变量配置

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

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 级别