import asyncio
import base64
import json
import os
import uuid
import websockets
API_KEY = os.environ["MODELHUB_API_KEY"]
WS_URL = "wss://model-api.skyengine.com.cn/v1/stream_asr?model=aigc-cpp-funasr-sysj"
async def recognize(path: str) -> None:
headers = {"Authorization": f"Bearer {API_KEY}"}
trace_id = f"segment_{uuid.uuid4().hex}"
async with websockets.connect(WS_URL, additional_headers=headers) as ws:
with open(path, "rb") as audio_file:
while chunk := audio_file.read(3200):
await ws.send(json.dumps({
"type": "input_audio_buffer.append",
"trace_id": trace_id,
"audio": base64.b64encode(chunk).decode(),
}))
await asyncio.sleep(0.1)
# 发送静音,让服务端完成当前语音段
await ws.send(json.dumps({
"type": "input_audio_buffer.append",
"trace_id": trace_id,
"audio": base64.b64encode(bytes(32000)).decode(),
}))
completed = []
while True:
try:
message = await asyncio.wait_for(ws.recv(), timeout=3)
except asyncio.TimeoutError:
break
event = json.loads(message)
if event.get("type") == "error":
raise RuntimeError(event["error"]["message"])
if event.get("type") == "conversation.item.input_audio_transcription.delta":
print("实时结果:", event["transcript"])
if event.get("type") == "conversation.item.input_audio_transcription.completed":
completed.append(event["transcript"])
print("最终结果:", "".join(completed))
asyncio.run(recognize("input.pcm"))