Голосовой диалог (Realtime)

import asyncio
import json

import websockets

URL = "wss://api.zveno.ai/v1/realtime?model=qwen/qwen-audio-3.1-realtime-plus"


async def main():
    async with websockets.connect(
        URL,
        subprotocols=["realtime"],
        additional_headers={"Authorization": "Bearer <ZVENOAI_API_KEY>"},
    ) as ws:
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {"type": "realtime", "output_modalities": ["text"]},
        }))
        await ws.send(json.dumps({
            "type": "conversation.item.create",
            "item": {
                "type": "message",
                "role": "user",
                "content": [{"type": "input_text", "text": "Привет! Расскажи о себе."}],
            },
        }))
        await ws.send(json.dumps({"type": "response.create"}))

        async for message in ws:
            event = json.loads(message)
            if event["type"] == "response.output_text.delta":
                print(event["delta"], end="", flush=True)
            if event["type"] in ("response.done", "error"):
                break


asyncio.run(main())

import WebSocket from 'ws'

const url = 'wss://api.zveno.ai/v1/realtime?model=qwen/qwen-audio-3.1-realtime-plus'
const ws = new WebSocket(url, 'realtime', {
  headers: { Authorization: 'Bearer <ZVENOAI_API_KEY>' },
})

ws.on('open', () => {
  ws.send(JSON.stringify({
    type: 'session.update',
    session: { type: 'realtime', output_modalities: ['text'] },
  }))
  ws.send(JSON.stringify({
    type: 'conversation.item.create',
    item: {
      type: 'message',
      role: 'user',
      content: [{ type: 'input_text', text: 'Привет! Расскажи о себе.' }],
    },
  }))
  ws.send(JSON.stringify({ type: 'response.create' }))
})

ws.on('message', (data) => {
  const event = JSON.parse(data.toString())
  if (event.type === 'response.output_text.delta') process.stdout.write(event.delta)
  if (event.type === 'response.done' || event.type === 'error') ws.close()
})