Viaim AI Open Python SDK

Viaim AI Open（Viaim AI 开放平台）官方 Python 服务端 SDK。当前版本支持通过 text-stream 使用实时语音识别，并提供同步、异步两套 API。

功能

• App Key HMAC-SHA256 鉴权
• 同步和异步客户端
• text-stream Session 生命周期管理
• WebSocket PCM 实时推流
• ASR partial/final、句级 utteranceId 与 input 收尾事件
• 翻译、语种、说话人、情绪和性别强类型事件
• 统一异常、超时和资源清理

环境要求

• Python 3.10 或更高版本
• Viaim AI Open 平台的 App Key 和 App Secret
• 输入音频：16 kHz、单声道、signed 16-bit little-endian PCM

安装

python -m pip install viaim-ai-open

升级到最新版：

python -m pip install --upgrade viaim-ai-open

凭证

可以在创建客户端时传入凭证：

from viaim_ai_open import AsyncViaimAIOpen

client = AsyncViaimAIOpen(
    app_key="app_xxx",
    app_secret="sk_xxx",
)

也可以通过环境变量提供：

VIAIM_AI_OPEN_APP_KEY=app_xxx
VIAIM_AI_OPEN_APP_SECRET=sk_xxx

凭证只能用于可信服务端，不能放入网页、桌面客户端、移动 App 或公开仓库。

快速开始

异步识别 PCM 文件

import asyncio

from viaim_ai_open import AsyncViaimAIOpen


async def main() -> None:
    pcm = open("sample.pcm", "rb").read()

    async with AsyncViaimAIOpen(
        app_key="app_xxx",
        app_secret="sk_xxx",
    ) as client:
        result = await client.text_stream.transcribe(pcm)
        print(result.text)


asyncio.run(main())

同步识别 PCM 文件

from viaim_ai_open import ViaimAIOpen

with open("sample.pcm", "rb") as source:
    pcm = source.read()

with ViaimAIOpen(app_key="app_xxx", app_secret="sk_xxx") as client:
    result = client.text_stream.transcribe(pcm)
    print(result.text)

实时推流

每帧推荐发送 6400 字节，对应 16 kHz mono s16le 的 200 ms 音频：

import asyncio

from viaim_ai_open import AsyncViaimAIOpen
from viaim_ai_open.types.text_stream import (
    AsrFinalEvent,
    AsrPartialEvent,
    EmotionDetectedEvent,
    GenderDetectedEvent,
    SpeakerChangedEvent,
    SpeakerDetectedEvent,
    TextStreamAbility,
)


async def main() -> None:
    async with AsyncViaimAIOpen() as client:
        async with client.text_stream.connect(
            abilities=[
                TextStreamAbility.SPEAKER_ID,
                TextStreamAbility.EMOTION_ID,
                TextStreamAbility.GENDER_ID,
            ],
        ) as stream:
            await stream.wait_ready()

            async def receive_events() -> None:
                async for event in stream.events():
                    if isinstance(event, AsrPartialEvent):
                        print("PARTIAL:", event.utterance_id, event.text)
                    elif isinstance(event, AsrFinalEvent):
                        if not event.session_final:
                            print("FINAL:", event.utterance_id, event.text)
                    elif isinstance(event, (SpeakerDetectedEvent, SpeakerChangedEvent)):
                        print(
                            "SPEAKER:",
                            event.payload.utterance_id,
                            event.payload.speaker_id,
                        )
                    elif isinstance(event, EmotionDetectedEvent):
                        print(
                            "EMOTION:",
                            event.payload.utterance_id,
                            event.payload.emotion,
                        )
                    elif isinstance(event, GenderDetectedEvent):
                        print(
                            "GENDER:",
                            event.payload.utterance_id,
                            event.payload.gender,
                        )

            receiver = asyncio.create_task(receive_events())

            with open("sample.pcm", "rb") as source:
                while chunk := source.read(6400):
                    await stream.send_audio(chunk)
                    await asyncio.sleep(0.2)

            await stream.end_input()
            await stream.wait_input_final(timeout=30)
            await stream.close()
            await receiver


asyncio.run(main())

events() 只能有一个消费者。等待方法可与事件消费者同时使用，不会争抢 WebSocket 消息：

• wait_final() 等待下一条句级 asr.final。
• wait_input_final() 等待当前 input.end 对应的 sessionFinal=true 收尾事件。

同一轮输入可能产生多条句级 final。字幕以及说话人、情绪、性别信息都应使用 payload.utteranceId 关联，不能使用外层 segmentId 作为句子 ID。

启用扩展能力

调用 App 必须已在开放平台开通对应能力：

from viaim_ai_open.types.text_stream import (
    TextStreamAbility,
    TextStreamOptions,
    TranslationOptions,
)

async with client.text_stream.connect(
    abilities=[
        TextStreamAbility.TRANSLATION,
        TextStreamAbility.SPEAKER_ID,
        TextStreamAbility.EMOTION_ID,
        TextStreamAbility.GENDER_ID,
    ],
    options=TextStreamOptions(
        translation=TranslationOptions(target_lang="en"),
    ),
) as stream:
    ...

当前事件类型包括：

• session.ready
• asr.partial / asr.final
• translation.partial / translation.final
• language.detected
• speaker.detected / speaker.changed
• emotion.detected
• gender.detected
• pong、error、session.ended

句子级信息聚合

增强事件会在对应的句级 asr.final 之后到达。可以用 utteranceId 将它们聚合到同一条记录：

utterances: dict[str, dict[str, object]] = {}

async for event in stream.events():
    if isinstance(event, AsrFinalEvent) and not event.session_final:
        if event.utterance_id is not None:
            utterances[event.utterance_id] = {"text": event.text}
    elif isinstance(event, (SpeakerDetectedEvent, SpeakerChangedEvent)):
        uid = event.payload.utterance_id
        if uid in utterances:
            utterances[uid]["speaker_id"] = event.payload.speaker_id
    elif isinstance(event, EmotionDetectedEvent):
        uid = event.payload.utterance_id
        if uid in utterances:
            utterances[uid]["emotion"] = event.payload.emotion
    elif isinstance(event, GenderDetectedEvent):
        uid = event.payload.utterance_id
        if uid in utterances:
            utterances[uid]["gender"] = event.payload.gender

speaker.detected / speaker.changed 是说话人状态事件：相同说话人连续说话时，不保证每句话都重复发送。业务侧可保存当前说话人，并在收到新事件后按 utteranceId 修正对应句子。

验证凭证

async with AsyncViaimAIOpen() as client:
    result = await client.credentials.verify()
    print(result)

异常处理

from viaim_ai_open import (
    AuthenticationError,
    PermissionDeniedError,
    StreamConnectionError,
    ViaimAIOpenError,
)

try:
    result = await client.text_stream.transcribe(pcm)
except AuthenticationError:
    print("App Key 或 App Secret 无效")
except PermissionDeniedError:
    print("应用尚未开通该服务或能力")
except StreamConnectionError as exc:
    print("流式连接失败:", exc)
except ViaimAIOpenError as exc:
    print("调用失败:", exc.code, exc.message)

超时配置

from viaim_ai_open import AsyncViaimAIOpen, ClientOptions

client = AsyncViaimAIOpen(
    options=ClientOptions(
        connect_timeout=5,
        read_timeout=30,
        session_ready_timeout=10,
        final_timeout=30,
    )
)

本地开发

git clone https://gitee.com/qimijiu/openplatform.git
cd openplatform
python -m pip install -e "sdk/backend/python[test]"
python -m pytest -q -c sdk/backend/python/pyproject.toml sdk/backend/python/tests

运行生产环境冒烟测试：

python sdk/backend/python/scripts/test_text_stream.py sample.pcm

版本兼容

SDK 遵循语义化版本。0.x 版本仍处于早期阶段，公开接口可能在次版本中调整；正式稳定接口将从 1.0.0 开始。

0.2.0

• 新增说话人、情绪和性别强类型事件及 ability 枚举。
• 使用 utteranceId 关联 ASR 文本与句级增强信息。
• 新增 wait_input_final()，正确区分句级 final 与 sessionFinal 输入收尾。
• 补齐 ASR 的 confidence、words 等新协议字段。

安全说明

• 不要记录或提交 App Secret。
• 不要把凭证发送给浏览器或移动端。
• SDK 不会主动记录原始音频、识别文本或完整认证信息。

License

本项目基于 Apache License 2.0 发布。