# Gemini Live WebSocket

Gemini 原生双向 WebSocket 会话。支持文本、音频、图片和视频输入；模型以 AUDIO 返回，并可通过 outputAudioTranscription 获取同步文本。

## 接口与鉴权

`GET https://api.tu-zi.com/v1beta/live?model={model}`

传输方式: websocket

使用模型调用 API Key。示例凭证为占位符，请从环境变量读取真实密钥。

```http
Authorization: Bearer YOUR_API_KEY
```

## 查询参数

| 字段 | 类型 | 必填 | 说明与约束 |
| --- | --- | --- | --- |
| model | string | 是 | 当前模型标识。 示例: YOUR_MODEL |

## 最小请求示例

```json
{
  "setup": {
    "model": "models/YOUR_MODEL",
    "generationConfig": {
      "responseModalities": [
        "AUDIO"
      ]
    }
  }
}
```

## 运行前准备

将 https://api.tu-zi.com 替换为本站地址（不要重复追加 /v1），YOUR_MODEL 替换为模型广场中的完整模型名。Python 示例需要先安装 requests；JavaScript 示例使用 Node.js 20+ 并保存为 .mjs 文件。WebSocket 示例按代码注释安装专用依赖。

先在本站令牌页面创建模型调用密钥，并设置 API_KEY 环境变量；cURL 中的 YOUR_API_KEY 需要替换。上传示例中的 input.png/input.wav 是本地文件，必须提前准备。

首次运行使用最小请求；成功后对照下方响应字段读取结果。异步接口返回任务 ID 后仍需查询，任务提交成功不代表生成已完成。

## 代码示例

### curl

```bash
# Gemini Live requires a WebSocket client.
# See the Python or JavaScript example and /docs/api/gemini.
```

### javascript

```javascript
// Node.js: npm install ws
import WebSocket from 'ws';

const ws = new WebSocket("wss://api.tu-zi.com/v1beta/live?model=YOUR_MODEL", {
  headers: { Authorization: "Bearer " + process.env.API_KEY },
  perMessageDeflate: false,
  maxPayload: 2 * 1024 * 1024,
});
let completed = false, sent = false, receivedOutput = false, failed = false;
function fail(error) {
  if (failed) return;
  failed = true;
  console.error(error);
  process.exitCode = 1;
  ws.terminate();
}
const timeout = setTimeout(() => fail(new Error('Timed out before a complete turn')), 120000);
ws.on('close', () => {
  clearTimeout(timeout);
  if (!completed && !failed) fail(new Error('Connection closed before a complete turn'));
});
ws.on('error', fail);
ws.on('open', () => ws.send(JSON.stringify({
  "setup": {
    "model": "models/YOUR_MODEL",
    "generationConfig": {
      "responseModalities": [
        "AUDIO"
      ]
    }
  }
})));
ws.on('message', (message) => {
  if (failed || completed) return;
  try {
    const event = JSON.parse(message.toString());
    if (event.error) throw new Error(JSON.stringify(event.error));
    if (event.setupComplete && !sent) {
      sent = true;
      ws.send(JSON.stringify({"clientContent":{"turns":[{"role":"user","parts":[{"text":"Say hello briefly."}]}],"turnComplete":true}}));
    }
    const content = event.serverContent || {};
    if (content.interrupted) throw new Error('Turn was interrupted; no complete result');
    const text = content.outputTranscription?.text;
    if (text) { receivedOutput = true; process.stdout.write(text); }
    // Consume content.modelTurn.parts[].inlineData for PCM audio.
    if (content.modelTurn?.parts?.some(part => part.inlineData?.data || part.text)) receivedOutput = true;
    if (content.turnComplete) {
      if (!sent || !receivedOutput) throw new Error('Turn ended without model output');
      completed = true;
      console.log(event.usageMetadata);
      ws.close();
    }
  } catch (error) { fail(error); }
});
```

### python

```python
# Python 3.11+: pip install 'websockets>=14,<16'
import asyncio, json, os
from websockets.asyncio.client import connect

async def main():
    async with asyncio.timeout(120):
        async with connect("wss://api.tu-zi.com/v1beta/live?model=YOUR_MODEL", additional_headers={
            'Authorization': "Bearer " + os.environ["API_KEY"]
        }, max_size=2*1024*1024, max_queue=4, compression=None) as ws:
            completed = sent = received_output = False
            await ws.send("{\"setup\":{\"model\":\"models/YOUR_MODEL\",\"generationConfig\":{\"responseModalities\":[\"AUDIO\"]}}}")
            async for message in ws:
                event = json.loads(message)
                if 'error' in event:
                    raise RuntimeError(event['error'])
                if 'setupComplete' in event and not sent:
                    sent = True
                    await ws.send("{\"clientContent\":{\"turns\":[{\"role\":\"user\",\"parts\":[{\"text\":\"Say hello briefly.\"}]}],\"turnComplete\":true}}")
                content = event.get('serverContent', {})
                if content.get('interrupted'):
                    raise RuntimeError('Turn was interrupted; no complete result')
                text = content.get('outputTranscription', {}).get('text', '')
                if text:
                    received_output = True
                    print(text, end='', flush=True)
                # Consume content.modelTurn.parts[].inlineData for PCM audio.
                if any(part.get('inlineData', {}).get('data') or part.get('text') for part in content.get('modelTurn', {}).get('parts', [])):
                    received_output = True
                if content.get('turnComplete'):
                    if not sent or not received_output:
                        raise RuntimeError('Turn ended without model output')
                    completed = True
                    print(event.get('usageMetadata', {}))
                    break
            if not completed:
                raise RuntimeError('Connection closed before a complete turn')

asyncio.run(main())
```

## 响应字段

| 字段 | 类型 | 说明与约束 |
| --- | --- | --- |
| setupComplete | object | 握手完成事件。 |
| serverContent.modelTurn.parts[] | object | 模型音频或其他内容块。 |
| serverContent.turnComplete | boolean | 本轮结束。 |

以下为示例响应，部分接口仅展示主要字段；可选字段不保证出现。

```json
{
  "setupComplete": {}
}
```

## 流式响应

连接使用 wss，先发 setup，再发送 realtimeInput 或 clientContent。按服务端事件接收音频与转写，处理 turnComplete、interrupted 和错误。 示例等待 setupComplete 后只发送一次请求，以收到模型输出并出现 turnComplete 为完成；错误、超时、提前断开或 interrupted 均视为未完成，不能自动重提。

## 错误处理与重试

检查 HTTP 状态与该协议的错误对象；不要将不同协议的错误字段混用。401/403 检查凭证与权限，429 退避等待，5xx 检查网关或上游状态。错误正文可能含 error/message/code 等字段，以实际协议为准。

查询可采用有上限的指数退避；创建任务、生成内容、创建或删除密钥等操作不要盲目重试，以免重复计费或重复执行。

## 适用限制

本页是协议基础结构。模型可用参数、枚举、上限和异步行为以该模型“快速上手”为准；示例中的 YOUR_MODEL 必须替换。表中表单初始值不代表上游默认值。

## 文档依据

定义版本: 1

审核日期: 2026-11-02

来源: https://api.tu-zi.com/docs/api/reference-relay-gemini-live

关联协议说明: https://api.tu-zi.com/docs/api/gemini

本文只包含接口文档和示例。真实密钥、账号数据及用户请求内容不包含在复制内容中。