Google Gemini 2.0 Flash API实时对话应用开发全攻略

这是关于 Google DeepMind Gemini 2.0 Flash API 最实用的攻略，没有之一。

核心内容：

• Gemini 2.0 Flash API 的强大功能
• 有效使用的前提条件
• 两个脚本的主要特征及共享组件

Google DeepMind 的 Gemini 2.0 Flash API，确实是实时 AI 驱动对话领域的一大突破。它让开发者能构建出无缝处理实时音频交互的应用程序，语音输入输出的集成度称得上“无与伦比”。不管是做客户服务聊天机器人、增强无障碍工具，还是开发互动 AI 导师，这个 API 都是个相当扎实的基础。下面，我们就来拆解 Gemini 2.0 Flash API 的能力，并演示如何用 Python 高效上手。我们会通过两个不同的脚本 —— live_test.py 和 no_queue_version.py —— 一步步带你落地，帮你快速搭建动态对话 AI 方案。

前提条件

1. API 密钥： 先在 Google AI Platform 上注册获取访问权限，拿到你的 Google Gemini API 密钥。
2. Python 环境： 确保装了 Python 3.8 或更高版本。
3. 依赖项： 安装必需的库：

pip install websockets pyaudio asyncio

4. 环境变量： 把 API 密钥设成环境变量：

export GEMINI_API_KEY="your_api_key_here"

脚本的主要特征

两个脚本都实现了实时语音交互，但在处理方式上略有不同：

• live_test.py：使用音频队列来管理和播放接收到的音频数据，适合需要缓冲控制的场景。
• no_queue_version.py：直接播放接收到的音频，没有队列，过程更简单直接。

共享组件

• WebSocket 连接： 负责与 Gemini API 通信。
• 音频捕获： 用 PyAudio 录制用户输入。
• 音频播放： 把 AI 生成的响应流式回放给用户。

live_test.py 的逐步解释

初始化

class GeminiVoice:
    def __init__(self):
        self.audio_queue = asyncio.Queue()
        self.api_key = os.environ.get("GEMINI_API_KEY")
        self.model = "gemini-2.0-flash-exp"
        self.uri = f"wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1alpha.GenerativeService.BidiGenerateContent?key={self.api_key}"
        self.FORMAT = pyaudio.paInt16
        self.CHANNELS = 1
        self.CHUNK = 512
        self.RATE = 16000

• 音频队列： 临时存储接收到的音频块，等播放就绪再取。
• API 密钥和模型： 配置对 Gemini API 的访问。
• 音频设置： 定义输入输出参数（格式、声道、块大小、采样率）。

开始连接

async def start(self):
    self.ws = await connect(
        self.uri, additional_headers={"Content-Type": "application/json"}
    )
    await self.ws.send(json.dumps({"setup": {"model": f"models/{self.model}"}}))
    await self.ws.recv(decode=False)
    print("Connected to Gemini, You can start talking now")
    async with asyncio.TaskGroup() as tg:
        tg.create_task(self.capture_audio())
        tg.create_task(self.stream_audio())
        tg.create_task(self.play_response())

• WebSocket 连接： 建立与 Gemini 服务的链接。
• 任务组： 并发管理音频捕获、流式传输和播放三个任务。

捕获音频

async def capture_audio(self):
    audio = pyaudio.PyAudio()
    stream = audio.open(
        format=self.FORMAT,
        channels=self.CHANNELS,
        rate=self.RATE,
        input=True,
        frames_per_buffer=self.CHUNK,
    )
    while True:
        data = await asyncio.to_thread(stream.read, self.CHUNK)
        await self.ws.send(
            json.dumps(
                {
                    "realtime_input": {
                        "media_chunks": [
                            {
                                "data": base64.b64encode(data).decode(),
                                "mime_type": "audio/pcm",
                            }
                        ]
                    }
                }
            )
        )

• 音频录制： 持续从麦克风读取数据，base64 编码后实时发送给 API。

流音频

async def stream_audio(self):
    async for msg in self.ws:
        response = json.loads(msg)
        try:
            audio_data = response["serverContent"]["modelTurn"]["parts"][0]["inlineData"]["data"]
            self.audio_queue.put_nowait(base64.b64decode(audio_data))
        except KeyError:
            pass

• 响应处理： 从 API 返回的数据中提取音频块，解码后放入播放队列。

播放响应

async def play_response(self):
    audio = pyaudio.PyAudio()
    stream = audio.open(
        format=self.FORMAT, channels=self.CHANNELS, rate=24000, output=True
    )
    while True:
        data = await self.audio_queue.get()
        await asyncio.to_thread(stream.write, data)

• 音频播放： 从队列中取出音频数据，通过扬声器输出（采样率 24000 Hz）。

no_queue_version.py 的逐步解释

no_queue_version.py 脚本通过直接流式传输和播放接收到的音频，简化了流程，省去了中间队列。

主要区别

• 简化的响应处理： 不再单独用队列，而是把接收和播放合并在一个异步迭代里。

async def recv_model_audio(self):
    audio = pyaudio.PyAudio()
    stream = audio.open(
        format=self.FORMAT, channels=self.CHANNELS, rate=24000, output=True
    )
    async for msg in self.ws:
        response = json.loads(msg)
        try:
            audio_data = response["serverContent"]["modelTurn"]["parts"][0]["inlineData"]["data"]
            await asyncio.to_thread(stream.write, base64.b64decode(audio_data))
        except KeyError:
            pass

运行脚本

1. 设置环境变量：

export GEMINI_API_KEY="your_api_key_here"

2. 运行脚本：

python live_test.py

3. 对着麦克风说话： 脚本捕获你的输入，发送给 Gemini 服务，然后播放 AI 的实时回应。

结论

用 Google DeepMind Gemini 2.0 Flash API，你可以构建出支持动态实时对话的创新应用。这项技术让音频交互变得丝滑流畅，客户支持、互动教程、语言学习等场景都能轻松驾驭。

选择 live_test.py 获得高级排队功能，或者 no_queue_version.py 追求简洁直接，开发者可以根据具体需求灵活选用。API 的能力搭配 Python 的灵活性，足以创造出高度互动且反应迅速的应用程序。花点时间研究这两个脚本，试着调整自己的配置，把 AI 驱动通信的潜力真正释放出来。可能性很多，有了 Gemini 2.0 Flash，那些最野心勃勃的对话 AI 想法，都能变成现实。

你是一名 AI 行业编辑，请围绕下面这条热点输出一份资讯解读：
热点：Google Gemini 2.0 Flash API实时对话应用开发全攻略要求：
1. 先用一句话解释这条热点在讲什么
2. 再总结它为什么重要
3. 说明会影响哪些 AI 产品或内容方向
4. 最后给出 3 个适合资讯站使用的标题