ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;color: rgb(63, 63, 63);">ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;letter-spacing: 0.1em;color: rgb(63, 63, 63);">Google DeepMind 的 Gemini 2.0 Flash API 代表了实时 AI 驱动对话领域的重大突破。它使开发者能够构建能够无缝处理实时音频交互的应用程序，提供了语音输入和输出的无与伦比的集成。无论您是在创建客户服务聊天机器人、增强无障碍工具，还是开发互动 AI 导师，这个 API 都是一个强大的基础。在本博客中，我们将探讨 Gemini 2.0 Flash API 的功能，并演示如何有效地使用 Python。我们将通过两个不同的脚本
ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;color: rgb(63, 63, 63);" class="list-paddingleft-1">ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 14px;text-indent: -1em;display: block;margin: 0.5em 8px;color: rgb(63, 63, 63);">1.ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: inherit;color: rgb(15, 76, 129);">API 密钥：通过在 Google AI Platform 上注册以获取访问权限来获取您的 Google Gemini API 密钥。2.Python 环境：确保安装了 Python 3.8+。pipinstallwebsocketspyaudioasyncio 4.环境变量：将 API 密钥设置为环境变量： exportGEMINI_API_KEY="your_api_key_here" 脚本的主要特征两个脚本都实现了实时语音交互，但在方法上略有不同： - •
live_test.py:使用音频队列来管理和播放接收到的音频数据。 - •
no_queue_version.py:直接播放接收到的音频，没有队列，简化了过程。
共享组件- •WebSocket 连接:促进与 Gemini API 的通信。
live_test.py 的逐步解释初始化classGeminiVoice:
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 的访问。
开始连接asyncdefstart(self):
self.ws =awaitconnect(
self.uri, additional_headers={"Content-Type":"application/json"}
)
awaitself.ws.send(json.dumps({"setup": {"model":f"models/{self.model}"}}))
awaitself.ws.recv(decode=False)
print("Connected to Gemini, You can start talking now")
asyncwithasyncio.TaskGroup()astg:
tg.create_task(self.capture_audio())
tg.create_task(self.stream_audio())
tg.create_task(self.play_response())
- •WebSocket 连接:建立与 Gemini 服务的链接。
捕获音频asyncdefcapture_audio(self):
audio = pyaudio.PyAudio()
stream = audio.open(
format=self.FORMAT,
channels=self.CHANNELS,
rate=self.RATE,
input=True,
frames_per_buffer=self.CHUNK,
)
whileTrue:
data =awaitasyncio.to_thread(stream.read,self.CHUNK)
awaitself.ws.send(
json.dumps(
{
"realtime_input": {
"media_chunks": [
{
"data": base64.b64encode(data).decode(),
"mime_type":"audio/pcm",
}
]
}
}
)
)
- •音频录制：捕获来自麦克风的输入并将其发送到API。
流音频asyncdefstream_audio(self):
asyncformsginself.ws:
response = json.loads(msg)
try:
audio_data = response["serverContent"]["modelTurn"]["parts"][0]["inlineData"]["data"]
self.audio_queue.put_nowait(base64.b64decode(audio_data))
exceptKeyError:
pass
- •响应处理：解码来自API的音频响应并将其存储在队列中。
播放响应asyncdefplay_response(self):
audio = pyaudio.PyAudio()
stream = audio.open(
format=self.FORMAT, channels=self.CHANNELS, rate=24000, output=True
)
whileTrue:
data =awaitself.audio_queue.get()
awaitasyncio.to_thread(stream.write, data)
no_queue_version.py 的逐步解释no_queue_version.py脚本通过直接流式传输和播放接收到的音频，简化了这一过程，而无需中间队列。
主要区别asyncdefrecv_model_audio(self):
audio = pyaudio.PyAudio()
stream = audio.open(
format=self.FORMAT, channels=self.CHANNELS, rate=24000, output=True
)
asyncformsginself.ws:
response = json.loads(msg)
try:
audio_data = response["serverContent"]["modelTurn"]["parts"][0]["inlineData"]["data"]
awaitasyncio.to_thread(stream.write, base64.b64decode(audio_data))
exceptKeyError:
pass
运行脚本
exportGEMINI_API_KEY="your_api_key_here" 2.运行脚本： pythonlive_test.py 3.对着麦克风说话：脚本捕获您的输入，将其发送到Gemini服务，并播放AI的响应。 结论使用 Google DeepMind Gemini 2.0 Flash API，您可以构建支持动态和实时对话的创新应用程序。这项尖端技术实现了无缝的音频交互，非常适合客户支持、互动教程和语言学习等多种用例。 通过选择live_test.py以获得高级排队功能或no_queue_version.py以实现简单性，开发人员可以根据具体需求定制 API 集成。 API 的强大功能结合 Python 的灵活性，使得创建高度互动和响应迅速的应用程序成为可能。花时间探索脚本，尝试自定义选项，释放 AI 驱动通信的全部潜力。可能性是巨大的，借助 Gemini 2.0 Flash，您可以将最雄心勃勃的对话 AI 创意变为现实 |
