Gemini 3.8 Live Avatar 正式可用：如何构建能看、能说、还能办事的实时智能体

浏览器 / App / 交互终端
        │ 音频、视频、屏幕共享、控制事件
        ▼
实时会话网关
        │ 身份认证、限流、区域路由、会话恢复
        ▼
Gemini Live 会话 ──────► Live Avatar 输出
        │
        ├──► 订单 / 保单 / 库存 API
        ├──► Agent Development Kit 工作流
        └──► 审计、指标与人工接管系统

mkdir live-agent-demo && cd live-agent-demo
cat > requirements.txt <<'EOF'
fastapi==0.115.12
uvicorn[standard]==0.34.0
websockets==15.0.1
EOF

cat > app.py <<'PY'
import asyncio
import uuid
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

async def check_policy(policy_id: str) -> dict:
    # 替换为真实的保单、订单或库存 API。
    await asyncio.sleep(2)
    return {
        "policy_id": policy_id,
        "active": True,
        "coverage": "collision"
    }

@app.websocket("/live")
async def live_session(ws: WebSocket):
    await ws.accept()
    send_lock = asyncio.Lock()
    background_tasks = set()

    async def send(event: dict):
        async with send_lock:
            await ws.send_json(event)

    async def run_policy_tool(task_id: str, policy_id: str):
        result = await check_policy(policy_id)
        await send({
            "type": "tool.result",
            "task_id": task_id,
            "tool": "check_policy",
            "result": result,
            "assistant_text": "保单有效，已确认包含碰撞保障。"
        })

    try:
        while True:
            event = await ws.receive_json()

            if event.get("type") == "claim.start":
                task_id = str(uuid.uuid4())

                # 立即响应，不等待后台工具完成。
                await send({
                    "type": "assistant.message",
                    "text": "我正在核对保单。您可以继续描述事故经过。",
                    "task_id": task_id
                })

                task = asyncio.create_task(
                    run_policy_tool(task_id, event["policy_id"])
                )
                background_tasks.add(task)
                task.add_done_callback(background_tasks.discard)
            else:
                await send({
                    "type": "assistant.message",
                    "text": "已收到当前会话事件。"
                })
    except WebSocketDisconnect:
        for task in background_tasks:
            task.cancel()

PY

cat > client.py <<'PY'
import asyncio
import json
import websockets

async def main():
    async with websockets.connect("ws://127.0.0.1:8000/live") as ws:
        await ws.send(json.dumps({
            "type": "claim.start",
            "policy_id": "POLICY-2026-001"
        }))

        # 第一条是即时确认，第二条是稍后返回的工具结果。
        for _ in range(2):
            print(json.loads(await ws.recv()))

asyncio.run(main())
PY

python -m pip install -r requirements.txt
uvicorn app:app --reload

python client.py