pipinstallwebsocketspyaudioasyncio
4.环境变量：将 API 密钥设置为环境变量：
exportGEMINI_API_KEY="your_api_key_here"
两个脚本都实现了实时语音交互，但在方法上略有不同：
live_test.py:使用音频队列来管理和播放接收到的音频数据。no_queue_version.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 =16000asyncdefstart(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())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",
}
]
}
}
)
)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:
passasyncdefplay_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脚本通过直接流式传输和播放接收到的音频，简化了这一过程，而无需中间队列。
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:
passexportGEMINI_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 创意变为现实
| 欢迎光临 链载Ai (https://www.lianzai.com/) | Powered by Discuz! X3.5 |