讯飞星火（xhuo）音频两条核心链路（素材目录 D:/wks/study/xhuo，含官方 demo 与 API 文档 PDF）： | 能力 | 服务名 | 协议 | 端点 | |------|--------|------|------| | ASR 语音识别 | 实时语音转写大模型（rtasr_llm） | WebSocket 双向流 | wss://office-api-ast-dx.iflyaisol.com/ast/communicate/v1 | | TTS 语音合成 | 超拟人语音合成（super smart-tts） | WebSocket 双向流 | wss://cbm01.cn-huabei-1.xf-yun.com/v1/private/mcd9m97e6 | 两条都是 **wss 双向流式**：ASR 流式推音频、流式出文字；TTS 流式推文本、流式出音频（TTS 天然适配大模型逐字输出场景）。 音频硬性约束（ASR）：采样率 16k、位长 16bit、单声道 PCM，**每 40ms 发送 1280 字节**。

两套鉴权完全不同（最容易搞混的点）： | | ASR（rtasr_llm） | TTS（超拟人） | |--|------------------|---------------| | 密钥 | AppID + AccessKeyId + AccessKeySecret | 方式一：APIPassword；方式二：APPID/APIKey/APISecret | | 签名算法 | **HMAC-SHA1 + Base64** | **HMAC-SHA256 + Base64**（方式二） | | 签名内容 | 全部请求参数排序拼接 | `host: ...\ndate: ...\nGET {path} HTTP/1.1` | | 放置位置 | URL 查询参数 signature | URL 查询参数 authorization / 请求头 x-api-key | | 时间要求 | utc=北京时间 +0800 | date=GMT 标准时间 | 计费口径：TTS 按字符量（汉字/字母/数字/标点/空格/回车各计 1 字符），高调用量场景可买并发套餐；ASR 按转写时长套餐。用前都要在控制台领免费额度。

握手 URL 带全部参数： ```text wss://office-api-ast-dx.iflyaisol.com/ast/communicate/v1 ?accessKeyId=xxx&appId=xxx&uuid=xxx &utc=2026-09-16T19%3A39%3A47%2B0800 &audio_encode=pcm_s16le&lang=autodialect&samplerate=16000 &signature=... ``` 关键业务参数（PDF 文档提炼）： | 参数 | 说明 | |------|------| | audio_encode | pcm_s16le / speex-7、speex-10 / opus-wb（推荐 opus） | | lang | autodialect=中英+202 种方言免切；autominor=37 语种（需工单开通） | | role_type | 0 关闭说话人分离；2 开启实时角色分离（盲分），可配声纹 | | feature_ids | 声纹 Id 列表（需 role_type=2），eng_spk_match=1 角色全来自声纹库 | | pd | 领域个性化：court/finance/medical/tech/edu/isp/gov/game/ecom 等 16 域 | | eng_punc / eng_vad_mdn | 标点过滤（punc=0 去标点）/ vad 远近场（1 远 2 近） |

签名生成（demo 已按文档实现，实测可复现）： ```python # 1. 除 signature 外全部参数：过滤空值 → 字典序升序 # 2. 键值分别 urlencode 后按 "k=v&" 拼接，去尾 & 得 baseString # 3. HmacSHA1(accessKeySecret, baseString) → Base64 base_str = "&".join(f"{quote(k,safe='')}={quote(v,safe='')}" for k, v in sorted(auth_params.items())) signature = base64.b64encode( hmac.new(secret.encode(), base_str.encode(), hashlib.sha1).digest() ).decode() ``` utc 格式必须是北京时间：`yyyy-MM-dd'T'HH:mm:ss+0800`（uuid 每次随机生成即可）。

发送节奏——demo 的精髓（帧率自洽公式：16000Hz × 2字节 × 1声道 × 40ms = **1280B/帧**）： ```python AUDIO_FRAME_SIZE = 1280 # 每帧字节数 FRAME_INTERVAL_MS = 40 # 发送间隔 # 关键：基于起始时间算"理论发送时刻"，抵消累计误差（而非简单 sleep(0.04)） expected = start_time_ms + frame_index * FRAME_INTERVAL_MS diff = expected - now_ms if diff > 0.1: time.sleep(diff / 1000) ws.send_binary(chunk) ``` 发完后送**结束标记**（JSON 文本帧，带 sessionId 关联会话）： ```json {"end": true, "sessionId": "d49ddd6f-c451-35ea-b8c4-2c75af837caa"} ``` sessionId 来自服务端 `msg_type=action` 消息的 data.sessionId，收到后要存下来。

结果 JSON 的嵌套解析路径（层级很深，注意取法）： ```text msg_type=result / res_type=asr data.seg_id 消息号（从0开始） data.cn.st.type 0=确定性结果；1=中间结果 data.ls true=最终结果 data.cn.st.rt[].ws[].cw[].w 逐词内容（lg=语言，wb/we=起止帧） ``` 即：外层 `rt` 数组 → `ws` 词数组 → `cw` 候选 → `w` 取词文本。实践口径：**中间结果实时上屏，type=0 且 ls=true 的定稿落库**。异常时 `data.normal=false`，读 `desc/detail.domain` 定位。 提准确率：控制台上传**个性化热词**（业务专有名词提高权重）。

调用地址固定：`wss://cbm01.cn-huabei-1.xf-yun.com/v1/private/mcd9m97e6` **鉴权方式一（最简，推荐）**：控制台拿 APIPassword，直接放请求头： ```python ws = websocket.WebSocketApp(requrl, on_message=on_message, on_error=on_error, on_close=on_close, header={"x-api-key": "你的APIPassword"}) ``` **鉴权方式二**：HMAC-SHA256 签名拼 URL（demo 的 assemble_ws_auth_url）： ```python signature_origin = f"host: {host}\ndate: {date}\nGET {path} HTTP/1.1" signature = b64(hmac_sha256(api_secret, signature_origin)) authorization = b64(f'api_key="{api_key}", algorithm="hmac-sha256", ' f'headers="host date request-line", signature="{signature}"') url = requrl + "?" + urlencode({"host": host, "date": date, "authorization": authorization}) ``` 注意 date 用 **GMT 格式**（`Wed, 16 Sep 2026 11:39:47 GMT`），与 ASR 的北京时间完全不同。

一次请求 = 三段式协议体（header / parameter / payload）： ```json { "header": {"app_id": "xxx", "status": 2}, "parameter": {"tts": { "vcn": "x6_lingxiaoxuan_pro", "volume": 50, "speed": 50, "pitch": 50, "rhy": 0, "bgs": 0, "reg": 0, "rdn": 0, "audio": {"encoding": "lame", "sample_rate": 24000, "channels": 1, "bit_depth": 16, "frame_size": 0}}}, "payload": {"text": { "encoding": "utf8", "compress": "raw", "format": "plain", "status": 2, "seq": 0, "text": "base64(待合成文本)"}} } ``` 响应同样是三段式：`header.code==0` 正常（sid 供排障）、`payload.audio.audio` 为 **base64 音频分片**、`payload.audio.status==2` 表示合成结束。客户端逐片 decode 后 append 写文件即可（demo.mp3 即此产物，实测 70KB 有效 MP3）。 x4 系列发音人还可加 `parameter.oral`（口语化配置 oral_level 等）。

发音人（vcn）与音频参数： | 参数 | 取值 | 说明 | |------|------|------| | vcn | x6_lingxiaoxuan_pro / x5_lingfeiyi_flow / ... | 发音人列表；**正式调用前需在控制台开通对应发音人权限** | | volume | 0~100 | 0 静音，默认 50 | | speed | 0~100 | 越大越快 | | pitch | 0~100 | 振幅/语调 | | rhy | 0/1 | 是否返回拼音标注 | | bgs | 0/1/2 | 背景音 | | rdn / reg | 0~3 / 0~2 | 数字读法 / 英文读法 | | audio.encoding | **lame(mp3)/raw(pcm)**/speex/opus | 推荐 lame 或 raw | | audio.sample_rate | 8000/16000/**24000** | 推荐 24000 | 另支持：静音停顿与多音字读法占位符（如 `#=2#=`）、Latex 公式朗读。

**双向流式**是超拟人的招牌：同一会话内可**分段多次发文本**、流式收音频，延迟低——专为"大模型逐字输出边合成边播放"设计。 单次上传（demo 用法）：text.status=2 一次发完。流式上传（生产推荐）： ```text 第一帧: text.status=0, seq=0 中间帧: text.status=1, seq=1,2,...（LLM 每产出一小段就发） 最后一帧: text.status=2, seq=n ``` 每段文本同样 base64；音频侧 status==2 后再关连接。

验证环境：Windows + Python + websocket-client 1.9.0（依赖安装：`pip install websocket-client`）。两个 demo 均 `py_compile` 语法通过。 **签名与参数验证**（假密钥只验算法，全部通过）： | 项 | 结果 | |----|------| | ASR `_generate_auth_params` | 8 字段齐全；utc 匹配 `yyyy-MM-dd'T'HH:mm:ss+0800` | | ASR 签名复算 | 按文档规则（排序→urlencode→拼接→HmacSHA1→Base64）独立重算**与 demo 输出一致** | | PCM 帧率自洽 | 16000×2byte×1ch×40ms = 1280B 与常量一致 | | test_1.pcm | 153998B → 121 帧 → 预估时长 4.84s（帧数计算含尾帧） | | TTS 鉴权 URL | host/date/authorization 三参数；authorization 解码含 api_key/algorithm/headers/signature | | TTS 请求体 | header.status=2 / tts.vcn+lame/24000 / text base64 可解码还原 | | demo.mp3 样本 | 70272B，帧同步头 `fff3`，**有效 MP3**（此前真实密钥成功运行的产物） | **真实密钥端到端复测**（2026-09-22，密钥配置于 `D:/wks/jupyter/yinpin/env.txt`，驱动 `yinpin_demo.ipynb` 全格重跑）： | 链路 | 结果 | |------|------| | ASR rtasr_llm 转写 test_1.pcm | 连接成功（action/started + sessionId）→ 发送 121 帧 → 收到 11 条消息 → **定稿全文：床前明月光，疑是地上霜。举头望明月，低头思故乡。** | | TTS 超拟人合成（HMAC-SHA256 方式二鉴权） | `code=0`，**59 个音频分片 → tts_out.mp3 58608B**，帧同步头 `fff3` 有效 MP3 | | 复测结论 | 双链路真实密钥端到端**全通过**；复测中修复 notebook 精简版一处解析 bug（见踩坑 6） |

**真实网络握手验证**（无真实密钥，预期走到"服务端鉴权拒绝"即证明链路正确——实际完全符合预期）： ```text TTS：WebSocketBadStatusException Handshake status 401 Unauthorized（kong 网关） → 端点可达、TLS 正常、签名请求格式正确，假签名被正确拒绝 ASR：首次 [SSL: CERTIFICATE_VERIFY_FAILED]（本地证书问题，见踩坑） → 加 sslopt={"cert_reqs": ssl.CERT_NONE} 后重试： Handshake status 35010，响应体 {'error': 'AccessKeyId Not Exists'} → 请求已被业务层受理，假 AccessKeyId 被标准错误码拒绝 ``` 结论：两个 demo 的 URL 构造、签名算法、协议封装**均与线上服务实际吻合**，换上控制台真实密钥即可端到端跑通（TTS 侧另有 demo.mp3 成功产物佐证）。 **更新（2026-09-22）**：已用控制台真实密钥完成端到端复测，上方"静态验证"表中新增复测结果——ASR 定稿全文输出、TTS 58608B 有效 MP3，链路验证期的推断全部得到实证。

验证中发现的坑（按文档模式记录）： 1. **Windows SSL 证书验证失败**：ASR demo 的 `create_connection` 未配 sslopt，本机报 `unable to get local issuer certificate`；TTS demo 有 `sslopt={"cert_reqs": ssl.CERT_NONE}` 所以没事。修复：ASR 的 connect 加同样的 sslopt（或 `pip install certifi` 后配 ca_certs） 2. **demo 音频路径**：`AUDIO_FILE_PATH = r"python\\test_1.pcm"` 是相对写法，注释里也提醒"建议使用绝对路径"——实际运行要改成绝对路径 3. **两套鉴权别混用**：ASR 用 AccessKey 三元组 + HMAC-SHA1 + 北京时间；TTS 用 APIPassword 或 HMAC-SHA256 + GMT 时间。混淆必 401/35010 4. **发音人权限**：vcn 填了没开通的音色会报错，先去控制台开通 5. ASR 接收线程要判断 `is_connected` 再 recv，否则 Windows 下抛"非套接字操作"（demo 已修复此点） 6. **notebook 精简版解析变量错写（2026-09-22 复测发现并修复）**：`yinpin_demo.ipynb` 的 RTASRClient 结果解析生成器误写 `w.get("w", "")`——循环变量实际是 `cw`（`for cw in ws_.get("cw", [])`），`w` 未定义抛 NameError，**接收线程静默崩溃**，表现为"收到消息数=2、定稿为空"（链路与握手其实都正常）。修复为 `cw.get("w", "")` 后重跑：11 条消息、定稿全文输出。教训：**守护线程内的异常不会中断主流程，只表现为结果缺失**——收发分离架构下，接收线程必须 try/except 记录错误，且"消息数异常偏少"要第一时间怀疑解析侧而非链路侧

把两个 demo 收敛成可复用组件的建议： 1. **密钥进 env 文件**（如 xfyun_app_id / asr_access_key_id / asr_access_key_secret / tts_api_password），杜绝硬编码 2. **收发分离双线程**：发送线程管节奏，接收线程管解析（demo 已如此）；生产再加重连与指数退避 3. **ASR 结束三要素**：音频发完 → 发 `{"end":true,"sessionId":...}` → 继续收直到 `ls=true` 4. **TTS 文件后缀跟 encoding 走**：lame→.mp3，raw→.pcm；收包循环里 append，`status==2` 才关连接 5. 面向 LLM 场景直接用 TTS 流式模式（status 0/1/2），与大模型 token 流天然对齐

星火 ASR/TTS 在链路家族中的位置： | 方案 | 优势 | 适用 | |------|------|------| | 星火 rtasr_llm | 中英+202 方言免切、说话人分离/声纹、16 领域个性化、不限时长 | 实时字幕/会议转写/客服质检 | | 星火超拟人 TTS | 超拟人情感音色、双向流式低延迟、发音人丰富 | 有声内容/数字人/LLM 语音输出 | | whisper 类本地 ASR | 免费/离线/隐私 | 无网环境、数据不出域 | | GPT-SoVITS（本 tab 首页） | 音色克隆/本地部署 | 个性化定制音色 | 组合范式：rtasr_llm（听）→ LLM（想）→ 超拟人 TTS（说）= 完整语音对话链路。

**ASR demo 第一段：全局配置与签名**（rtasr_llm_demo.py，逐行注释） ```python FIXED_PARAMS = { "audio_encode": "pcm_s16le", # 音频格式：16k/16bit 原始 PCM 流（speex/opus 需换值） "lang": "autodialect", # 中英 + 202 种方言免切识别（37 语种用 autominor） "samplerate": "16000" # 采样率：仅 pcm 格式必须传，支持 16000/8000 } AUDIO_FRAME_SIZE = 1280 # 每帧字节数 = 16000Hz × 2byte × 1声道 × 0.04s（自洽公式，勿乱改） FRAME_INTERVAL_MS = 40 # 发送间隔毫秒，与帧大小一一对应 def __init__(self, app_id, access_key_id, access_key_secret, audio_path): self.app_id = app_id # 控制台应用 ID self.access_key_id = access_key_id # 控制台 APIKey self.access_key_secret = access_key_secret # 控制台 APISecret（只参与签名，不上URL） self.audio_path = audio_path # 建议绝对路径（demo 原注释同款提醒） self.base_ws_url = "wss://office-api-ast-dx.iflyaisol.com/ast/communicate/v1" self.ws = None # WebSocket 连接对象 self.is_connected = False # 连接状态标志（收发线程都看它） self.recv_thread = None # 接收线程句柄 self.session_id = None # 服务端会话ID，结束标记要用 self.is_sending_audio = False # 防重入锁：阻止并发发送 self.audio_file_size = 0 # 提前缓存的文件大小（算时长用） ``` ```python def _get_audio_file_size(self): with open(self.audio_path, "rb") as f: # 二进制只读打开 f.seek(0, 2) # 偏移 0 从文件末尾(whence=2)定位 → 指针移到结尾 return f.tell() # 指针位置即文件字节数（等价 os.path.getsize 但自带开闭容错） def _get_utc_time(self): beijing_tz = datetime.timezone(datetime.timedelta(hours=8)) # 东八区 now = datetime.datetime.now(beijing_tz) # 带时区的当前时间 return now.strftime("%Y-%m-%dT%H:%M:%S%z") # 文档要求格式: 2026-09-16T19:39:47+0800 ``` ```python def _generate_auth_params(self): auth_params = { "accessKeyId": self.access_key_id, # 三个身份参数 + "appId": self.app_id, # 一个随机 uuid（标识业务侧用户） "uuid": uuid.uuid4().hex, # 一个北京时间戳 "utc": self._get_utc_time(), # 都要参与签名 **FIXED_PARAMS # 展开 3 个音频参数，一并签名 } sorted_params = dict(sorted([ # 核心：过滤空值 → 按参数名升序排序 (k, v) for k, v in auth_params.items() #（与官方 Java TreeMap 行为对齐） if v is not None and str(v).strip() != "" ])) base_str = "&".join([ # 键、值分别 urlencode 后按 k=v& 拼接 f"{urllib.parse.quote(k, safe='')}={urllib.parse.quote(v, safe='')}" for k, v in sorted_params.items() ]) signature = hmac.new( # HmacSHA1(密钥=accessKeySecret, 内容=base_str) self.access_key_secret.encode("utf-8"), base_str.encode("utf-8"), hashlib.sha1 ).digest() # 得 20 字节二进制摘要 auth_params["signature"] = base64.b64encode(signature).decode("utf-8") return auth_params # signature 是事后追加，不参与自身签名 ```

**ASR demo 第二段：连接、接收、发送、收尾** ```python def connect(self): auth_params = self._generate_auth_params() # 生成含签名的全部参数 params_str = urllib.parse.urlencode(auth_params) full_ws_url = f"{self.base_ws_url}?{params_str}" # 参数全部拼在 URL 查询串 self.ws = create_connection( full_ws_url, timeout=15, # 握手+IO 超时 15s enable_multithread=True # 允许收发双线程并发操作同一连接 ) self.is_connected = True time.sleep(1.5) # 等服务端完全就绪（实测经验值） self.recv_thread = threading.Thread(target=self._recv_msg, daemon=True) self.recv_thread.start() # daemon=True：主进程退出时接收线程自动结束 return True # 异常分支：WebSocketException 打印状态码；其余 Exception 兜底，返回 False ``` ```python def _recv_msg(self): # 独立接收线程：只管收与解析 while True: if not self.is_connected or not self.ws: # 每轮先验状态，防操作已关连接 break #（修复 Windows"非套接字操作"的关键） msg = self.ws.recv() # 阻塞等待一条服务端消息 if not msg: # 空消息 = 服务端关闭 → self.close() 后退出 break if isinstance(msg, str): # 服务端 JSON 一律是文本帧 msg_json = json.loads(msg) if (msg_json.get('msg_type') == 'action' # action 消息里取 and 'sessionId' in msg_json.get('data', {})): # sessionId 存下来 self.session_id = msg_json['data']['sessionId'] # msg_type=result 的转写结果也在本线程打印（结构见"结果解析"节） else: pass # 二进制消息按 demo 约定忽略 # except WebSocketException / OSError / Exception → 统一 close() 退出循环 ``` ```python def send_audio(self): if not self.is_connected: return False # 双保险①：未连接不发包 if self.is_sending_audio: return False # 双保险②：已有发送任务时不重入 self.is_sending_audio = True frame_index = 0; start_time = None # 帧号 & 首帧时间基准（毫秒） self.audio_file_size = self._get_audio_file_size() with open(self.audio_path, "rb") as f: total_frames = self.audio_file_size // AUDIO_FRAME_SIZE # 整帧数 if self.audio_file_size % AUDIO_FRAME_SIZE: # 尾部不足一帧 total_frames += 1 # 也算一帧 while True: chunk = f.read(AUDIO_FRAME_SIZE) # 每次读 1280B if not chunk: break # 读空 = 发送完毕 if start_time is None: # 首帧记录基准时间 start_time = time.time() * 1000 expected = start_time + frame_index * FRAME_INTERVAL_MS # 本帧理论时刻 diff = expected - time.time() * 1000 # 与当前时刻的差值 if diff > 0.1: # 只在 diff 为正时休眠， time.sleep(diff / 1000) # 抵消读取/网络造成的累计误差 self.ws.send_binary(chunk) # 明确二进制帧发送 frame_index += 1 end_msg = {"end": True} # 结束标记：JSON 文本帧 if self.session_id: end_msg["sessionId"] = self.session_id # 关联会话，服务端好收尾 self.ws.send(json.dumps(end_msg, ensure_ascii=False)) # finally: self.is_sending_audio = False 释放重入锁 ``` ```python def close(self): if self.is_connected and self.ws: self.is_connected = False # 先置 False：让接收线程先退出 if self.ws.connected: # 连接仍开才发关闭帧 self.ws.close(status=1000, reason="客户端正常关闭") # main 流程：connect() → send_audio() → 等待(预估时长+5s，每秒查 is_connected) → close() ```

**TTS demo 第一段：参数组装与鉴权 URL**（super smart-tts.py） ```python def __init__(self, APPID, APIKey, APISecret, Text): self.CommonArgs = {"app_id": self.APPID, "status": 2} # status=2：单次上传模式一次发完全部文本（流式时首帧0/中间1/末帧2） self.BusinessArgs = {"tts": { "vcn": "x6_lingxiaoxuan_pro", # 发音人（需控制台开通权限），换音色改这里 "volume": 50, # 音量 0~100（0 静音，默认音量的缩放） "rhy": 0, # 0 不返回拼音；1 返回（utf8 纯文本标注） "speed": 50, # 语速 0~100，越大越快 "pitch": 50, # 音调/振幅 0~100 "bgs": 0, # 背景音：0 无 / 1 内置1 / 2 内置2 "reg": 0, # 英文读法：0 自动 / 1 全按字母 / 2 自动+不确定按字母 "rdn": 0, # 数字读法：0 自动 / 1 数值 / 2 字符串 / 3 字符串优先 "audio": { "encoding": "lame", # lame=mp3（raw=pcm）；文件后缀必须跟着改 "sample_rate": 24000, # 推荐 24000（可选 8000/16000） "channels": 1, # 单声道 "bit_depth": 16, # 位深 16/8 "frame_size": 0 # 分帧大小，0 默认 }}} self.Data = {"text": { # payload：待合成文本 "encoding": "utf8", "compress": "raw", "format": "plain", "status": 2, "seq": 0, # 单次模式：status=2、序号0 "text": str(base64.b64encode(self.Text.encode('utf-8')), "UTF8") }} # 文本必须 base64 后再入 JSON ``` ```python def sha256base64(data): # 工具：SHA256 摘要 → Base64 文本 sha256 = hashlib.sha256() sha256.update(data) return base64.b64encode(sha256.digest()).decode(encoding='utf-8') def parse_url(requset_url): # 工具：拆 URL 成 host/path/schema 三件 stidx = requset_url.index("://") # wss://cbm01.../v1/private/mcd9m97e6 host = requset_url[stidx + 3:] # → host=cbm01.cn-huabei-1.xf-yun.com edidx = host.index("/") # → path=/v1/private/mcd9m97e6 ... def assemble_ws_auth_url(requset_url, method="GET", api_key="", api_secret=""): u = parse_url(requset_url) now = datetime.now() date = format_date_time(mktime(now.timetuple())) # GMT 时间: Wed, 16 Sep 2026 .. GMT signature_origin = "host: {}\ndate: {}\n{} {} HTTP/1.1".format(host, date, method, path) # 签名原文 = host行+date行+请求行，三行按序拼接 signature_sha = hmac.new(api_secret.encode('utf-8'), signature_origin.encode('utf-8'), digestmod=hashlib.sha256).digest() # HMAC-SHA256 signature_sha = base64.b64encode(signature_sha).decode() # → Base64 authorization_origin = 'api_key="%s", algorithm="%s", headers="%s", signature="%s"' % ( api_key, "hmac-sha256", "host date request-line", signature_sha) authorization = base64.b64encode(authorization_origin.encode()).decode() # authorization 是"再包一层 Base64"的整体串 values = {"host": host, "date": date, "authorization": authorization} return requset_url + "?" + urlencode(values) # 三参数 urlencode 拼回 URL ```

**TTS demo 第二段：回调收音与两种鉴权入口** ```python def on_open(ws): # 连接建立后触发 def run(*args): # 注意：在子线程里 send，避免阻塞握手线程 d = {"header": wsParam.CommonArgs, # 三段式请求体一次发齐： "parameter": wsParam.BusinessArgs, # 身份 / 合成参数 / "payload": wsParam.Data} # base64 文本 ws.send(json.dumps(d)) if os.path.exists('./demo.mp3'): os.remove('./demo.mp3') # 删旧文件：收音是 append 模式，防脏数据 thread.start_new_thread(run, ()) def on_message(ws, message): # 每收到一条服务端消息触发一次 message = json.loads(message) code = message["header"]["code"] # 0=正常；非0看 message 错误说明 sid = message["header"]["sid"] # 会话ID，报障时提供给讯飞排查 if "payload" in message: audio = message["payload"]["audio"]['audio'] audio = base64.b64decode(audio) # 音频分片是 base64，先解码成二进制 status = message["payload']['audio"]["status"] # 嵌套取 status if status == 2: # 2=最后一帧 → 主动关连接结束会话 ws.close() if code != 0: print("sid:%s call error code:%s" % (sid, code)) else: with open('./demo.mp3', 'ab') as f: # 追加写：分片到达顺序即播放顺序 f.write(audio) # 文件后缀必须对应 audio.encoding # on_error(ws, error): 打印错误；on_close: 空实现（demo 保持简单） ``` ```python if __name__ == "__main__": AUTH_METHOD = 2 # 1=APIPassword 最简；2=三元组签名 wsParam = Ws_Param(APPID=appid, APISecret=apisecret, APIKey=apikey, Text="全红婵，2007年3月28日出生于广东省湛江市……") requrl = 'wss://cbm01.cn-huabei-1.xf-yun.com/v1/private/mcd9m97e6' if AUTH_METHOD == 1: # 鉴权方式一：URL 用原始地址，密钥放请求头 auth_header = {"x-api-key": apipassword} ws = websocket.WebSocketApp(requrl, on_message=..., on_error=..., on_close=..., header=auth_header) else: # 鉴权方式二：签名后拼 URL（见上一 tab） wsUrl = assemble_ws_auth_url(requrl, "GET", apikey, apisecret) ws = websocket.WebSocketApp(wsUrl, on_message=..., on_error=..., on_close=...) ws.on_open = on_open ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE}) # run_forever 阻塞驱动事件循环；sslopt 跳过证书校验（Windows 免 SSL 坑） ```

**最小复用改造**（拿走即用） ASR 复用三步（类不动，只换配置）： ```python from rtasr_llm_demo import RTASRClient client = RTASRClient( app_id=os.getenv("XF_APP_ID"), # ① 密钥进环境变量 access_key_id=os.getenv("XF_ASR_KEY_ID"), access_key_secret=os.getenv("XF_ASR_KEY_SECRET"), audio_path=r"D:/abs/path/test_1.pcm") # ② 绝对路径 client.connect() # ③ connect → send → wait → close client.send_audio() time.sleep((client.audio_file_size // 1280) * 40 // 1000 + 5) client.close() # 生产补丁：connect 里加 sslopt={"cert_reqs": ssl.CERT_NONE} # 结果落库：在 _recv_msg 的 msg_type=result 分支挂回调，按 type==0 且 ls==true 取定稿 ``` TTS 方式一 20 行精简版（最省事的接入形态）： ```python import websocket, base64, json, ssl, os TEXT, VCN, OUT = "你好，星火超拟人合成。", "x6_lingxiaoxuan_pro", "out.mp3" body = json.dumps({ "header": {"app_id": os.getenv("XF_APP_ID"), "status": 2}, "parameter": {"tts": {"vcn": VCN, "volume": 50, "speed": 50, "pitch": 50, "audio": {"encoding": "lame", "sample_rate": 24000, "channels": 1, "bit_depth": 16, "frame_size": 0}}}, "payload": {"text": {"encoding": "utf8", "compress": "raw", "format": "plain", "status": 2, "seq": 0, "text": base64.b64encode(TEXT.encode()).decode()}}}) def on_open(ws): if os.path.exists(OUT): os.remove(OUT) ws.send(body) def on_message(ws, m): m = json.loads(m) if "payload" in m and m["header"]["code"] == 0: a = m["payload"]["audio"] with open(OUT, "ab") as f: f.write(base64.b64decode(a["audio"])) # 分片追加 if a["status"] == 2: ws.close() # 末帧关连接 ws = websocket.WebSocketApp( "wss://cbm01.cn-huabei-1.xf-yun.com/v1/private/mcd9m97e6", on_message=on_message, on_open=on_open, header={"x-api-key": os.getenv("XF_TTS_PASSWORD")}) # 方式一鉴权 ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE}) ``` 改造清单：密钥全部进 env；ASR 补 sslopt；音频路径绝对化；输出文件后缀跟 encoding（lame→.mp3 / raw→.pcm）；LLM 逐字输出场景把 text.status 改 0/1/2 流式分帧发送。