Closed as not planned
Description
Why is 1011 timeout error occurs when Gemini response is long?
Code (can contain Ukrainian strings and comments)
import asyncio
import pyaudio
import speech_recognition as sr
from google import genai
from google.genai import types
import sys
client = genai.Client(api_key="I WILL DIDN'T PUT HERE MY API KEY ON PUBLIC ISSUE")
model = "gemini-2.0-flash-live-001"
tools = [
{"code_execution": {}},
{"google_search": {}}
]
config = {"response_modalities": ["AUDIO"], "tools": tools}
def speech_to_text():
"""Перетворює мовлення з мікрофона на текст."""
r = sr.Recognizer()
r.dynamic_energy_threshold = True
with sr.Microphone() as source:
print("Скажіть щось...")
try:
r.adjust_for_ambient_noise(source, duration=1)
audio = r.listen(source, timeout=5, phrase_time_limit=None)
print("Розпізнавання...")
text = r.recognize_google(audio, language="uk-UA")
print(f"Ви сказали: {text}")
return text
except sr.UnknownValueError:
print("Не вдалося розпізнати мову.")
return None
except sr.RequestError as e:
print(f"Помилка сервісу розпізнавання: {e}")
return None
except sr.WaitTimeoutError:
return None
except Exception as e:
print(f"Неочікувана помилка розпізнавання: {e}")
return None
async def main():
p = pyaudio.PyAudio()
stream = p.open(format=p.get_format_from_width(2),
channels=1,
rate=24000,
output=True)
print("Підключення до Gemini Live...")
try:
async with client.aio.live.connect(model=model, config=config) as session:
print("Підключено. Розпочинаємо діалог.")
loop = asyncio.get_running_loop()
while True:
user_text = speech_to_text()
if user_text is None:
continue
if user_text.lower() in ["до побачення", "вихід", "стоп"]:
print("Завершення діалогу.")
break
print("Надсилання повідомлення ШІ...")
mfai = f"""
Ти CraftAssistant (CraftAI), і ти призначений для того, щоб просто говорити з користувачем. Відповідай на такій самій мові як користувач.
Якщо користувач попросить щось зробити таке що вимагає виконання коду то ти маєш виконати код за допомогою code_execution на Python на Windows 10/11.
Повідомлення користувача:
{user_text}
"""
await session.send_client_content(
turns={"role": "user", "parts": [{"text": mfai}]},
)
print("Очікування відповіді ШІ...")
received_text = ""
async for response in session.receive():
if response.data is not None:
await loop.run_in_executor(None, lambda: stream.write(response.data))
if response.text is not None:
received_text += response.text
sys.stdout.flush()
if received_text:
print(f"\nAI: {received_text}")
else:
print("\nAI: Не отримано текстової відповіді.")
except Exception as e:
print(f"Виникла помилка під час сесії з Gemini: {e}")
finally:
print("Завершення роботи з аудіо.")
stream.stop_stream()
stream.close()
p.terminate()
print("Аудіо пристрої звільнено.")
if __name__ == "__main__":
asyncio.run(main())Environment details
- Programming language: Python
- OS: Windows 10
- Language runtime version: 3.13.2
- Package version: 1.15.0
Thanks!