viaim-ai-open 0.1.0

Official Python server SDK for Viaim AI Open services

Viaim AI Open Python SDK

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

功能

• 官方服务地址内置，无需配置 Base URL
• App Key HMAC-SHA256 鉴权
• 同步和异步客户端
• text-stream Session 生命周期管理
• WebSocket PCM 实时推流
• ASR partial/final 强类型事件
• 翻译、语种和说话人事件模型
• 统一异常、超时和资源清理

环境要求

• 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


async def main() -> None:
    async with AsyncViaimAIOpen() as client:
        async with client.text_stream.connect() as stream:
            await stream.wait_ready()

            async def receive_events() -> None:
                async for event in stream.events():
                    if isinstance(event, AsrPartialEvent):
                        print("PARTIAL:", event.text)
                    elif isinstance(event, AsrFinalEvent):
                        print("FINAL:", event.text)

            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_final(timeout=30)
            await stream.close()
            await receiver


asyncio.run(main())

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

启用扩展能力

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

from viaim_ai_open.types.text_stream import TextStreamOptions, TranslationOptions

async with client.text_stream.connect(
    abilities=["translation"],
    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
• pong、error、session.ended

验证凭证

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 开始。

安全说明

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

License

本项目基于 Apache License 2.0 发布。