这是关于 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 方案。
前提条件
- API 密钥： 先在 Google AI Platform 上注册获取访问权限，拿到你的 Google Gemini API 密钥。
- Python 环境： 确保装了 Python 3.8 或更高版本。
- 依赖项： 安装必需的库：
pip install websockets pyaudio asyncio- 环境变量： 把 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运行脚本
- 设置环境变量：
export GEMINI_API_KEY="your_api_key_here"- 运行脚本：
python live_test.py- 对着麦克风说话： 脚本捕获你的输入，发送给 Gemini 服务，然后播放 AI 的实时回应。
结论
用 Google DeepMind Gemini 2.0 Flash API，你可以构建出支持动态实时对话的创新应用。这项技术让音频交互变得丝滑流畅，客户支持、互动教程、语言学习等场景都能轻松驾驭。
选择 live_test.py 获得高级排队功能，或者 no_queue_version.py 追求简洁直接，开发者可以根据具体需求灵活选用。API 的能力搭配 Python 的灵活性，足以创造出高度互动且反应迅速的应用程序。花点时间研究这两个脚本，试着调整自己的配置，把 AI 驱动通信的潜力真正释放出来。可能性很多，有了 Gemini 2.0 Flash，那些最野心勃勃的对话 AI 想法，都能变成现实。