图像生成大模型
图像生成大模型
图像生成大模型
文档指南
请输入
图像生成大模型
实时互动数字人大模型
视频生成
智能绘图（文生图）
智能绘图（图生图）
智能绘图（图像特效）
图像识别与理解
身份认证
身份认证（增强版）
面部比对
图片静默活体检测
图像识别
通用图像分割
智能图像处理
SaaS应用
- 文档首页
图像生成大模型图像生成大模型实时互动数字人大模型实时互动数字人API实时互动数字人API-接口文档
复制全文
下载 pdf实时互动数字人API-接口文档
文档反馈
问问助手接口地址
协议名称 | 地址 |
协议名称 | 地址 |
WebSocket | wss://openspeech.bytedance.com/virtual_human/avatar_live/live |
协议帧汇总说明
客户端 → 服务端
Header | Message Type | Body | 用途 |
Header | Message Type | Body | 用途 |
|CTL|00| | Text | JSON | 开播或重连初始化信息 |
|CTL|01| | Text | N/A | 正常结束直播 |
|CTL|12| | Text | N/A | 流式音频驱动结束命令 |
|DAT|02| | Binary | Binary PCM Data | 流式音频驱动 |
|DAT|04| | Text | JSON | 结构化流式音频驱动 |
|DAT|05| | Text | JSON | Prompt 驱动动作/表情 |
服务端 → 客户端
Header | Message Type | Body | 用途 |
Header | Message Type | Body | 用途 |
|MSG|00| | Text | JSON | 确认建立连接，状态码详见下述状态码说明 |
|MSG|01| | Text | JSON | 服务异常通知，状态码详见下述状态码说明 |
|MSG|02| | Text | N/A | 心跳 |
|DAT|02| | Text | JSON | 返回合成状态 |
交互阶段说明
开播或重连初始化
客户端请求
- Message Type：Text
- Header：|CTL|00|
- Body：参数如下
参数名称 | 类型 | 层级 | 必需 | 说明 |
参数名称 | 类型 | 层级 | 必需 | 说明 |
live | object | 1 | 必需 | 直播会话相关配置 |
live_id | string | 2 | 必需 | 直播 ID 限制说明：
|
keep_alive_duration | int | 2 | 非必需 | 保活时长（单位：秒），websocket 异常断连后服务对此路并发的保活时长，保活期间端侧可进行重连，保活到期后并发会释放
默认值：0 |
auth | object | 1 | 必需 | 鉴权相关配置 |
type | int | 2 | 必需 | 鉴权类型 固定值：1 |
appid | string | 2 | 必需 | 火山引擎账号ID（可在火山引擎官网右上角查看） |
token | string | 2 | 必需 | token 可设置过期时间，详见代码示例中的 expires 参数 |
avatar | object | 1 | 必需 | 数字人相关信息 |
avatar_type | string | 2 | 必需 | 数字人类型 固定值：flowact-r1 |
input_mode | string | 2 | 必需 | 驱动模式 固定值：audio |
role | string | 2 | 必需 | 角色形象图URL（需公网可访问） |
flowact_r1 | object | 2 | 非必需 | flowact-r1参数 |
text_list | array of string | 3 | 非必需 | 动作指令列表 开播后，数字人将按列表顺序循环执行预设动作。当未设置该参数时，将启用默认值，默认值请参考表格下方的 动作指令默认列表。 支持用户自定义该动作指令列表，动作指令描述可参考默认值中的描述。 |
streaming | object | 1 | 必需 | 推流相关配置 |
type | string | 2 | 必需 | 推流方式 固定值：bytertc |
rtc_app_id | string | 2 | 必需 | ByteRTC 应用 ID（AppId） |
rtc_room_id | string | 2 | 必需 | |
rtc_uid | string | 2 | 必需 | |
rtc_token | string | 2 | 必需 |
*动作指令默认列表
*
*
动作指令默认列表
{
"text_list": [
"固定镜头，背景固定。人物自然站立，随呼吸节奏轻轻起伏身体，同时目光专注地看向前方，面带微笑。接着，她抬起一只手，指尖轻柔地梳理耳边的发丝，动作轻微自然，仿佛不经意间整理仪表，全程保持专注聆听的状态。",
"固定镜头，背景固定。人物自然站立，随呼吸节奏轻轻起伏身体，姿态松弛，保持自然，手轻垂身侧指尖微放松，眼神自然看向前方，面部微表情自然柔和。",
"固定镜头，背景固定。人物自然站立，随呼吸节奏轻轻起伏身体，接着右手食指和拇指轻柔摩擦，其他三指放松微曲，同时目光温柔地扫视前方，面带微笑。",
"固定镜头，背景固定。人物自然站立，随呼吸节奏轻轻起伏身体，姿态松弛，保持自然，手轻垂身侧指尖微放松，眼神自然看向前方，面部微表情自然柔和。",
"固定镜头，背景固定。人物自然站立，随呼吸节奏轻轻起伏身体，接着右手肘支撑于左手心，将右手手掌轻柔地托住下巴，身体重心稍稍前倾，眼神专注地望向前方，嘴角微扬，似在细致聆听。",
"固定镜头，背景固定。人物自然站立，随呼吸节奏轻轻起伏身体，姿态松弛，保持自然，手轻垂身侧指尖微放松，眼神自然看向前方，面部微表情自然柔和。",
"固定镜头，背景固定。人物自然站立，随呼吸节奏轻轻起伏身体，接着面带微笑地缓慢向下点头，同时保持眼神专注。",
"固定镜头，背景固定。人物自然站立，随呼吸节奏轻轻起伏身体，姿态松弛，保持自然，手轻垂身侧指尖微放松，眼神自然看向前方，面部微表情自然柔和。",
"固定镜头，背景固定。人物自然站立，随呼吸节奏轻轻起伏身体，接着面部表情平静放松，双眼视线柔和地聚焦在正前方，身体重心保持稳定，全程动作流畅且幅度适中，不带动肩部或身体晃动。",
"固定镜头，背景固定。人物自然站立，随呼吸节奏轻轻起伏身体，姿态松弛，保持自然，手轻垂身侧指尖微放松，眼神自然看向前方，面部微表情自然柔和。",
"固定镜头，背景固定。人物自然站立，随呼吸节奏轻轻起伏身体，接着身体微微前倾，双眉紧锁，嘴角向下轻微拉动，头部随疑惑轻轻左右摇摆，眼神聚焦镜头，面露沉思和不解。",
"固定镜头，背景固定。人物自然站立，随呼吸节奏轻轻起伏身体，姿态松弛，保持自然，手轻垂身侧指尖微放松，眼神自然看向前方，面部微表情自然柔和。",
"固定镜头，背景固定。人物自然站立，随呼吸节奏轻轻起伏身体，接着身体微微前倾，双臂交叠在胸前，眼神专注地望向前方，瞳孔随话语内容在眼眶内缓缓左右转动，眉头微蹙又舒展，露出若有所思的表情。",
"固定镜头，背景固定。人物自然站立，随呼吸节奏轻轻起伏身体，姿态松弛，保持自然，手轻垂身侧指尖微放松，眼神自然看向前方，面部微表情自然柔和。",
"固定镜头，背景固定。人物自然站立，随呼吸节奏轻轻起伏身体，接着目光缓慢地从镜头中心向左侧移动，短暂停留后，又缓慢地向右侧移动，仿佛在思考或搜寻，嘴唇轻微抿起，神情平和，保持呼吸节奏的轻微起伏。",
"固定镜头，背景固定。人物自然站立，随呼吸节奏轻轻起伏身体，姿态松弛，保持自然，手轻垂身侧指尖微放松，眼神自然看向前方，面部微表情自然柔和。",
"固定镜头，背景固定。人物自然站立，随呼吸节奏轻轻起伏身体，头微侧，眼神专注地看着观众，面带理解的微笑，身体随节奏轻柔摇摆，双手自然交叠于身前或背后，偶尔轻抚下巴，表现出沉思与认同，全程保持放松且友善的姿态。",
"固定镜头，背景固定。人物自然站立，随呼吸节奏轻轻起伏身体，姿态松弛，保持自然，手轻垂身侧指尖微放松，眼神自然看向前方，面部微表情自然柔和。"
]
}
Body参数示例：
{
"live": {
"live_id": "company_livename",
"keep_alive_duration": 0
},
"auth": {
"type": 1,
"appid": "填写火山引擎账号ID",
"token": "https://billing.volcengineapi.com/?X-Algorithm=..."
},
"avatar": {
"avatar_type": "flowact-r1",
"input_mode": "audio",
"role": "https://xxxx"
},
"streaming": {
"type": "bytertc",
"rtc_app_id": "xxx",
"rtc_room_id": "xxx",
"rtc_uid": "xxx",
"rtc_token": "xxx"
}
}
服务端响应
- Message Type：Text
- Header：|MSG|00|
- Body：参数如下
参数名称 | 类型 | 层级 | 必需 | 说明 |
参数名称 | 类型 | 层级 | 必需 | 说明 |
code | int | 1 | 必需 | 状态码 1000：表示成功 其它为错误码，详见下述状态码说明 |
message | string | 1 | 必需 | 状态信息 code=1000 时为 success 其它时为错误信息 |
data | object | 1 | 非必需 | 事件数据 |
Body参数示例：
{
"code": 1000,
"message": "success",
"data": {
"rtc": {
"app_id": "xxx",
"room_id": "xxx",
"inner_uid": "xxx",
"remote_uid": "",
"token": ""
}
}
}
播放状态反馈
返回合成状态
- Message Type：Text
- Header：|DAT|02|
- Body：参数如下
参数名称 | 类型 | 层级 | 必需 | 说明 |
参数名称 | 类型 | 层级 | 必需 | 说明 |
type | string | 1 | 必需 | 事件类型，常用值：
|
data | object | 1 | 非必需 | 事件数据 |
extra_data | string | 2 | 非必需
| 自定义数据 若驱动音频带有 extra_data，会在合成并推流到对应帧时将 extra_data 放入某个事件内返回 |
Body参数示例：
{
"type": "voice_start",
"data": {},
"extra_data": ""
}
实时驱动
音频驱动
说明
- 流式音频驱动、结构化流式音频驱动结束时需发送结束指令；
- 在进行流式音频驱动 或 结构化流式音频驱动的同时，可以进行Prompt动作指令驱动；
流式音频驱动
- Message Type：Binary
- Header：|DAT|02|
- Body：二进制音频数据（PCM），16kHz 单通道 2字节深度，长度不小于 40ms（1280 字节，一视频帧对应的音频长度）
结构化流式音频驱动
- Message Type：Text
- Header：|DAT|04|
- Body：参数如下
参数名称 | 类型 | 层级 | 必需 | 备注 |
参数名称 | 类型 | 层级 | 必需 | 备注 |
audio | string | 1 | 必需 | base64 编码的音频数据 音频应为 16kHz 单通道 2字节深度，长度不小于 40ms（1280 字节，一视频帧对应的音频长度） |
流式音频驱动结束命令
- Message Type：Text
- Header：|CTL|12|
Prompt驱动
- Message Type：Text
- Header：|DAT|05|
- Body：参数如下
参数名称 | 类型 | 层级 | 必需 | 备注 |
参数名称 | 类型 | 层级 | 必需 | 备注 |
prompt | string | 1 | 必需 | 驱动提示词 通过 prompt 文本驱动数字人展示高表现力的动作、表情 提示词详见下表 |
提示词场景类型 | 提示词 |
提示词场景类型 | 提示词 |
聆听（不说话） | 聆听（不说话）- 自然呼吸 |
| 聆听（不说话）- 小动作 - 整理头发 |
| 聆听（不说话）- 开心 |
| 聆听（不说话）- 惊讶 |
开场 | 开场 - 招手 |
| 开场 - 双手合十 |
| 开场 - 整理仪态 |
感谢礼物与打赏 | 感谢礼物与打赏 - 双手并拢比心 |
| 感谢礼物与打赏 - 起立致敬 |
| 感谢礼物与打赏 - 双手捂嘴感动 |
| 感谢礼物与打赏 - 单手竖起大拇指 |
| 感谢礼物与打赏 - 双手竖起大拇指 |
引导关注与点赞 | 引导关注与点赞 - 手指上方引导 |
| 引导关注与点赞 - 手指下方引导 |
| 引导关注与点赞 - 比 “1” |
情绪调节与氛围 | 情绪调节与氛围 - 比耶（V 手势） |
| 情绪调节与氛围 - 单手托下巴 |
| 情绪调节与氛围 - 捂嘴笑 |
| 情绪调节与氛围 - 嘟嘟嘴 |
| 情绪调节与氛围 - 凑近镜头 |
| 情绪调节与氛围 - 摸摸头 |
| 情绪调节与氛围 - 比比嘘 |
| 情绪调节与氛围 - 隔空 “飞吻” 气流 |
| 情绪调节与氛围 - 开心笑 |
| 情绪调节与氛围 - 整理衣服 |
| 情绪调节与氛围 - 双手欢呼 |
Body参数示例：
{
"prompt": "开场 - 招手"
}
关播
客户端请求
- Message Type：Text
- Header：|CTL|01|
服务收到结束直播消息后会以 1000 作为 code 关闭 websocket 连接。
异常通知
服务异常通知
- Message Type：Text
- Header：|MSG|01|
- Body：参数如下
参数名称 | 类型 | 层级 | 必需 | 备注 |
参数名称 | 类型 | 层级 | 必需 | 备注 |
code | int | 1 | 必需 | 错误码 详见下述状态码说明 |
message | string | 1 | 必需 | 错误信息 |
服务发送异常通知后会以 4999 作为 code 关闭 websocket 连接。
Body参数示例：
{
"code": 4002,
"message": "failed to load or create engine: concurrency has no quota"
}
心跳
服务端维持心跳
通过 websocket 自带的 Ping/Pong 机制实现心跳，服务端每5s发送一次 Ping 包，客户端收到 Ping 包后应立即返回 Pong 包进行确认。服务端和客户端应自行对心跳消息进行超时处理（如超过3倍的心跳间隔判定为连接异常）。
鉴于前端无法收发 websocket 自带的 Ping/Pong 消息，服务端会同时每5s发一包以下格式的消息：
- Message Type：Text
- Header：|MSG|02|
状态码说明
错误方 | 状态码 | 含义 | 原因列举 |
错误方 | 状态码 | 含义 | 原因列举 |
无 | 1000 | 成功 |
|
Client | 4000 | 请求错误 |
|
| 4001 | 鉴权错误 |
|
| 4002 | 并发超限 |
|
| 4003 | 连接过多 |
|
| 4004 | Live ID 重复 |
|
| 4006 | 直播不存在 |
|
Server | 5000 | Live 服务内部错误 |
|
| 5001 | Avatar 服务内部错误 |
|
| 5002 | 服务器忙 |
|
| 5100 | 重连 Avatar 原子服务 |
|
代码示例
*获取火山预签名URL
*
*
获取火山预签名URL
import datetime as dt
import hashlib
import hmac
from typing import Mapping, Optional
from urllib.parse import quote
ALGORITHM = "HMAC-SHA256"
EMPTY_SHA256 = hashlib.sha256(b"").hexdigest()
def _sha256_hex(data: str) -> str:
return hashlib.sha256(data.encode("utf-8")).hexdigest()
def _hmac_sha256(key: bytes, data: str) -> bytes:
return hmac.HMAC(key, data.encode("utf-8"), hashlib.sha256).digest()
def _url_encode(value: str) -> str:
return quote(str(value), safe="-_.~")
def _canonical_query(params: Mapping[str, str]) -> str:
items = sorted((str(k), str(v)) for k, v in params.items())
return "&".join(f"{_url_encode(k)}={_url_encode(v)}" for k, v in items)
def _signing_key(secret_access_key: str, short_date: str, region: str, service: str) -> bytes:
k_date = _hmac_sha256(secret_access_key.encode("utf-8"), short_date)
k_region = _hmac_sha256(k_date, region)
k_service = _hmac_sha256(k_region, service)
return _hmac_sha256(k_service, "request")
def generate_volcengine_presigned_url(
access_key_id: str,
secret_access_key: str,
region: str,
service: str,
host: str,
expires: int = 86400,
path: str = "/",
request_time: Optional[dt.datetime] = None,
) -> str:
"""
生成一个火山引擎 V4 预签名 URL
生成结果示例：
https://billing.volcengineapi.com/?X-Algorithm=HMAC-SHA256&...
参数:
access_key_id: 火山引擎 Access Key ID（AK）。
secret_access_key: 火山引擎 Secret Access Key（SK）。
region: 凭证作用域中使用的地域，例如 "cn-beijing"。
service: 凭证作用域中使用的服务名。
host: API 域名，例如 "billing.volcengineapi.com"。
expires: 过期时间，单位为秒，默认 900。
path: 请求路径。大多数火山引擎 OpenAPI 接口使用 "/"。
request_time: 可选的 UTC 时间；如果不传，则使用当前 UTC 时间。
返回:
完整签名后的 HTTPS URL 字符串。
"""
if not access_key_id:
raise ValueError("access_key_id is required")
if not secret_access_key:
raise ValueError("secret_access_key is required")
if not region:
raise ValueError("region is required")
if not service:
raise ValueError("service is required")
if not host:
raise ValueError("host is required")
if expires <= 0:
raise ValueError("expires must be a positive integer")
if request_time is None:
request_time = dt.datetime.now(dt.timezone.utc)
elif request_time.tzinfo is None:
request_time = request_time.replace(tzinfo=dt.timezone.utc)
else:
request_time = request_time.astimezone(dt.timezone.utc)
canonical_path = path or "/"
amz_date = request_time.strftime("%Y%m%dT%H%M%SZ")
short_date = amz_date[:8]
credential_scope = f"{short_date}/{region}/{service}/request"
signed_headers = "host;x-date"
canonical_query_params = {
"X-Algorithm": ALGORITHM,
"X-Credential": f"{access_key_id}/{credential_scope}",
"X-Date": amz_date,
"X-Expires": str(expires),
"X-SignedHeaders": signed_headers,
"X-SignedQueries": "",
}
canonical_headers = f"host:{host}\n" f"x-date:{amz_date}\n"
canonical_request = "\n".join(
[
"GET",
canonical_path,
_canonical_query(canonical_query_params),
canonical_headers,
signed_headers,
EMPTY_SHA256,
]
)
string_to_sign = "\n".join(
[
ALGORITHM,
amz_date,
credential_scope,
_sha256_hex(canonical_request),
]
)
signing_key = _signing_key(secret_access_key, short_date, region, service)
signature = hmac.HMAC(signing_key, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest()
final_query = "&".join(
[
f"X-Algorithm={_url_encode(ALGORITHM)}",
f"X-Credential={_url_encode(f'{access_key_id}/{credential_scope}')}",
f"X-Date={_url_encode(amz_date)}",
f"X-Expires={_url_encode(str(expires))}",
f"X-Signature={_url_encode(signature)}",
f"X-SignedHeaders={_url_encode(signed_headers)}",
"X-SignedQueries=",
]
)
return f"https://{host}{canonical_path}?{final_query}"
if __name__ == "__main__":
url = generate_volcengine_presigned_url(
# TODO 请修改 access_key_id、secret_access_key 为您的火山账号AK、SK
# AK、SK可以从火山引擎官网的API访问密钥页面获取：https://console.volcengine.com/iam/keymanage
access_key_id="your_ak",
secret_access_key="your_sk",
region="cn-beijing",
service="avatar_streaming_auth",
host="billing.volcengineapi.com",
expires=86400,
)
print(url)
API交互示例
import base64
import json
import logging
import traceback
import uuid
import websocket
import threading
import time
import wave
import os
from typing import Iterator, Optional
logging.basicConfig(
format="%(asctime)s %(levelname)s %(filename)s:%(funcName)s:%(lineno)s: %(message)s",
level=logging.INFO,
)
# client side header
StartLiveHeader = "|CTL|00|" # 开播或重连初始化信息
CloseLiveHeader = "|CTL|01|" # 正常结束直播
EndStreamingAudioHeader = "|CTL|12|" # 流式音频驱动结束命令
StreamingAudioHeader = "|DAT|02|" # 流式音频驱动
StreamingstructurationAudioHeader = "|DAT|04|" # 结构化流式音频驱动
PromptHeader = "|DAT|05|" # Prompt 驱动
# server side header
StartLiveAckHeader = "|MSG|00|" # 确认建立连接
PlayStatusHeader = "|DAT|02|" # 返回合成状态
PCM_BYTES_PER_SECOND = 16000 * 2 # 1s PCM数据 (16kHz, 1 channel, 16 bits)
MIN_PCM_CHUNK_BYTES = 1280 # 40ms PCM，要求单包不小于 1280 字节
FIXED_ACTION_INTERVAL_SECONDS = 2.5 # 固定动作间隔，单位秒
MANUAL_PROMPT_PAUSE_SECONDS = 5 # 手动暂停固定动作时间，单位秒
SEND_LOCK = threading.Lock()
AUDIO_TASK_LOCK = threading.Lock()
FIXED_ACTION_LOCK = threading.Lock()
ACTIVE_AUDIO_THREAD = None
ACTIVE_FIXED_ACTION_THREAD = None
FIXED_ACTION_ENABLED = False
FIXED_ACTION_PROMPT = ""
FIXED_ACTION_PAUSED_UNTIL = 0.0
# websocket.enableTrace(True)
def init(startLiveInfo: dict) -> Optional[websocket.WebSocket]:
"""
初始化WebSocket连接
:return: WebSocket连接对象
"""
logging.info("[init] 开始初始化WebSocket连接")
ws = None
try:
ws = websocket.create_connection(
"wss://openspeech.bytedance.com/virtual_human/avatar_live/live",
timeout=10, # 连接建立的超时时间（秒）
)
# 设置recv方法的超时时间（全局生效）
ws.settimeout(10)
logging.info(f"[init] 准备发送开播信息: {startLiveInfo}")
ws.send(StartLiveHeader + json.dumps(startLiveInfo))
ack = ws.recv()
logging.info(f"[init] 收到开播响应: {ack}")
ackInfo = json.loads(ack[8:])
if ackInfo["code"] != 1000:
raise Exception(f"[init] [{ackInfo['code']}] {ackInfo['message']}")
t = threading.Thread(target=heartbeat, args=(ws,), daemon=True)
t.start()
return ws
except Exception as e:
# 初始化失败时主动关闭连接，避免配额泄漏
logging.error(f"[init] 初始化WebSocket连接失败: {e}")
if ws:
try:
ws.close()
logging.info("[init] 初始化失败，已关闭WebSocket连接")
except Exception as close_e:
logging.error(f"[init] 初始化失败，关闭连接异常: {close_e}")
raise e
def heartbeat(ws):
"""
发送心跳线程
:param ws: WebSocket连接对象
"""
while ws.connected:
try:
with SEND_LOCK:
ws.ping()
logging.info("[heartbeat] 发送心跳包成功")
time.sleep(5) # 每5秒发一次心跳
except Exception as e:
logging.error(f"[heartbeat] Heartbeat failed: {e}")
break
def safeSend(ws: websocket.WebSocket, payload, opcode=websocket.ABNF.OPCODE_TEXT):
"""
安全发送消息，确保线程安全
:param ws: WebSocket连接对象
:param payload: 要发送的消息内容
:param opcode: 消息类型，默认文本消息
"""
with SEND_LOCK:
ws.send(payload, opcode=opcode)
def startAsyncAudioTask(ws: websocket.WebSocket, send_func, audio_file_path: str, task_name: str):
"""
启动异步音频任务，确保线程安全
:param ws: WebSocket连接对象
:param send_func: 发送音频文件的函数，参数为 (ws, audio_file_path)
:param audio_file_path: 音频文件路径（WAV格式）
:param task_name: 任务名称，用于日志记录和异常处理
"""
global ACTIVE_AUDIO_THREAD
with AUDIO_TASK_LOCK:
if ACTIVE_AUDIO_THREAD and ACTIVE_AUDIO_THREAD.is_alive():
logging.info("[startAsyncAudioTask] 当前已有音频任务在发送，请等待完成后再试")
return
def runner():
global ACTIVE_AUDIO_THREAD
try:
send_func(ws, audio_file_path)
except Exception as e:
logging.error(f"[{task_name}] 后台发送失败: {e}")
finally:
with AUDIO_TASK_LOCK:
ACTIVE_AUDIO_THREAD = None
ACTIVE_AUDIO_THREAD = threading.Thread(target=runner, name=task_name, daemon=True)
ACTIVE_AUDIO_THREAD.start()
def setFixedActionPrompt(prompt: str):
global FIXED_ACTION_PROMPT, FIXED_ACTION_ENABLED
normalized_prompt = prompt.strip()
with FIXED_ACTION_LOCK:
FIXED_ACTION_PROMPT = normalized_prompt
if not normalized_prompt:
FIXED_ACTION_ENABLED = False
if normalized_prompt:
logging.info(f"[setFixedActionPrompt] 已更新固定动作提示词: {normalized_prompt}")
else:
logging.info("[setFixedActionPrompt] 已关闭固定动作")
def startFixedActions(ws: websocket.WebSocket):
global ACTIVE_FIXED_ACTION_THREAD, FIXED_ACTION_ENABLED
with FIXED_ACTION_LOCK:
if not FIXED_ACTION_PROMPT:
print("未设置固定动作提示词，无法启动固定动作")
return
if ACTIVE_FIXED_ACTION_THREAD and ACTIVE_FIXED_ACTION_THREAD.is_alive():
logging.info("[startFixedActions] 固定动作已在运行中")
return
FIXED_ACTION_ENABLED = True
ACTIVE_FIXED_ACTION_THREAD = threading.Thread(target=fixedActions, args=(ws,), daemon=True)
ACTIVE_FIXED_ACTION_THREAD.start()
def stopFixedActions():
global FIXED_ACTION_ENABLED
with FIXED_ACTION_LOCK:
FIXED_ACTION_ENABLED = False
def pauseFixedActions(duration_seconds: int = MANUAL_PROMPT_PAUSE_SECONDS):
global FIXED_ACTION_PAUSED_UNTIL
with FIXED_ACTION_LOCK:
FIXED_ACTION_PAUSED_UNTIL = max(FIXED_ACTION_PAUSED_UNTIL, time.time() + duration_seconds)
def fixedActions(ws):
"""
发送固定动作线程
:param ws: WebSocket连接对象
"""
global ACTIVE_FIXED_ACTION_THREAD, FIXED_ACTION_ENABLED
try:
while ws.connected:
with FIXED_ACTION_LOCK:
enabled = FIXED_ACTION_ENABLED
prompt = FIXED_ACTION_PROMPT
paused_until = FIXED_ACTION_PAUSED_UNTIL
if not enabled:
break
if not prompt:
time.sleep(0.5)
continue
if time.time() < paused_until:
time.sleep(0.2)
continue
sendPrompt(ws, prompt, pause_fixed_action=False)
time.sleep(FIXED_ACTION_INTERVAL_SECONDS)
except Exception as e:
logging.error(f"[fixedActions] fixedActions failed: {e}")
finally:
with FIXED_ACTION_LOCK:
FIXED_ACTION_ENABLED = False
ACTIVE_FIXED_ACTION_THREAD = None
def sendPrompt(ws: websocket.WebSocket, prompt: str, pause_fixed_action: bool = True):
"""
发送提示词指令
:param ws: WebSocket连接对象
:param prompt: 提示词内容
:param pause_fixed_action: 是否暂停固定动作线程
"""
prompt = prompt.strip()
if not prompt:
raise ValueError("[sendPrompt] prompt 不能为空")
payload = json.dumps({"prompt": prompt}, ensure_ascii=False)
if pause_fixed_action:
pauseFixedActions()
logging.info(f"[sendPrompt] 准备发送指令内容: {payload}")
safeSend(ws, PromptHeader + payload)
return ws
def sleepUntilNextAudioTick(next_send_at: float):
sleep_seconds = next_send_at - time.monotonic()
if sleep_seconds > 0:
time.sleep(sleep_seconds)
def sendStreamingAudio(ws: websocket.WebSocket, audio_file_path: str):
"""
发送流式音频文件到数字人直播服务（DAT|02| + Binary PCM）
:param ws: WebSocket连接对象
:param audio_file_path: 音频文件路径（WAV格式）
"""
try:
logging.info(f"[sendStreamingAudio] 开始发送音频文件: {audio_file_path}")
start_at = time.monotonic()
total_bytes = 0
for pkg_index, pcm_chunk in enumerate(iterPcmChunks(audio_file_path), start=1):
message = StreamingAudioHeader.encode("utf-8") + pcm_chunk
safeSend(ws, message, opcode=websocket.ABNF.OPCODE_BINARY)
total_bytes += len(pcm_chunk)
sleepUntilNextAudioTick(start_at + total_bytes / PCM_BYTES_PER_SECOND)
safeSend(ws, EndStreamingAudioHeader)
logging.info("[sendStreamingAudio] 音频发送完成，已发送结束指令")
return ws
except wave.Error as e:
raise Exception(f"[sendStreamingAudio] 音频文件解析失败: {e}")
except Exception as e:
raise Exception(f"[sendStreamingAudio] 发送音频失败: {e}")
def sendStructurationAudio(ws: websocket.WebSocket, audio_file_path: str):
"""
发送结构化流式音频文件到数字人直播服务（DAT|04| + JSON）
:param ws: WebSocket连接对象
:param audio_file_path: 音频文件路径（WAV格式）
"""
try:
logging.info(f"[sendStructurationAudio] 开始发送音频文件: {audio_file_path}")
start_at = time.monotonic()
total_bytes = 0
for pkg_index, pcm_chunk in enumerate(iterPcmChunks(audio_file_path), start=1):
payload = json.dumps(
{"audio": base64.b64encode(pcm_chunk).decode("utf-8")},
ensure_ascii=False,
)
safeSend(ws, StreamingstructurationAudioHeader + payload)
total_bytes += len(pcm_chunk)
sleepUntilNextAudioTick(start_at + total_bytes / PCM_BYTES_PER_SECOND)
safeSend(ws, EndStreamingAudioHeader)
logging.info("[sendStructurationAudio] 音频发送完成，已发送结束指令")
return ws
except wave.Error as e:
raise Exception(f"[sendStructurationAudio] 音频文件解析失败: {e}")
except Exception as e:
raise Exception(f"[sendStructurationAudio] 发送音频失败: {e}")
def iterPcmChunks(audio_file_path: str, chunk_bytes: int = MIN_PCM_CHUNK_BYTES) -> Iterator[bytes]:
"""按协议要求读取 PCM 分片。"""
if not os.path.exists(audio_file_path):
raise FileNotFoundError(f"音频文件不存在: {audio_file_path}")
with wave.open(audio_file_path, "rb") as wf:
if wf.getframerate() != 16000:
raise ValueError(f"音频采样率必须为16000Hz，当前为{wf.getframerate()}Hz")
if wf.getnchannels() != 1:
raise ValueError(f"音频必须为单声道，当前为{wf.getnchannels()}声道")
if wf.getsampwidth() != 2:
raise ValueError(f"音频位深必须为16bit，当前为{wf.getsampwidth() * 8}bit")
frame_count = max(1, chunk_bytes // wf.getsampwidth())
while True:
pcm_chunk = wf.readframes(frame_count)
if not pcm_chunk:
break
if len(pcm_chunk) < MIN_PCM_CHUNK_BYTES:
pcm_chunk = pcm_chunk.ljust(MIN_PCM_CHUNK_BYTES, b"\x00")
yield pcm_chunk
def close(ws: websocket.WebSocket):
"""
安全关闭WebSocket连接
:param ws: WebSocket连接对象
"""
if not ws or not ws.connected:
logging.info("[close] WebSocket连接已关闭，无需重复操作")
return
try:
# 发送关播指令
safeSend(ws, CloseLiveHeader)
time.sleep(0.5) # 给服务端处理指令的时间
logging.info("[close] 已发送关播指令")
except Exception as send_e:
logging.warning(f"[close] 发送关播指令失败: {send_e}")
finally:
try:
# 关闭WebSocket连接
ws.close()
logging.info("[close] WebSocket连接已正常关闭")
except Exception as close_e:
logging.error(f"[close] 关闭WebSocket异常: {close_e}")
if __name__ == "__main__":
ws = None
# TODO 如需测试音频驱动，需要补充音频文件路径
audio_path = "/.../xxx.wav"
# TODO 填写你的开播body参数
startLiveInfo = {
"live": {"live_id": "xxx_" + str(uuid.uuid4())},
"auth": {
"type": 1,
"appid": "",
"token": "",
},
"avatar": {
"avatar_type": "flowact-r1",
"input_mode": "audio",
"role": "https://xxx.png",
},
"streaming": {
"type": "bytertc",
"rtc_app_id": "",
"rtc_room_id": "",
"rtc_uid": "",
"rtc_token": "",
},
}
try:
print("正在连接 WebSocket 服务器...")
ws = init(startLiveInfo)
print("连接成功！")
while True:
print("\n请选择操作:")
print("1. 发送提示词指令")
print("2. 发送流式音频")
print("3. 发送结构化流式音频")
print("4. 设置固定动作提示词 (设置后数字人会一直保持固定动作)")
print("5. 停止固定动作")
print("0. 关播退出 (close)")
choice = input("请输入选项 (0/1/2/3/4/5): ").strip()
if choice == "1":
content = input("请输入指令内容: ").strip()
if content:
sendPrompt(ws, content)
elif choice == "2":
if audio_path:
startAsyncAudioTask(ws, sendStreamingAudio, audio_path, "sendStreamingAudio")
print("流式音频已开始后台发送")
elif choice == "3":
if audio_path:
startAsyncAudioTask(ws, sendStructurationAudio, audio_path, "sendStructurationAudio")
print("结构化流式音频已开始后台发送")
elif choice == "4":
# 可设置固定动作提示词，设置之后，数字人在直播过程中会保持固定动作
fixed_prompt = input("请输入固定动作提示词（留空则关闭）: ").strip()
setFixedActionPrompt(fixed_prompt)
if fixed_prompt:
print(f"固定动作提示词已更新为: {fixed_prompt}")
startFixedActions(ws)
print("固定动作已启动")
else:
print("固定动作已关闭")
elif choice == "5":
stopFixedActions()
print("固定动作已停止")
elif choice == "0":
stopFixedActions()
close(ws)
print("连接已关闭")
break
else:
print("无效选项，请重试")
except Exception as e:
logging.error(f"程序执行异常: {e}")
logging.error(f"异常堆栈: {traceback.format_exc()}")
finally:
# 最终兜底关闭连接，避免配额泄漏
if ws:
close(ws)
最近更新时间：2026.08.06 10:38:26
这个页面对您有帮助吗？
有用
有用
无用
无用