Python Gemini Live API ConnectionClosedError при попытке загрузить предыдущий разговор в новый ⇐ Python

[Anonymous]

Я работаю с Gemini Live API и хочу загрузить контекст предыдущего разговора в текущий сеанс в реальном времени, что позволит мне возобновить любой сеанс из локального сохранения.
На данный момент я придумал следующий код:

Код: Выделить всё

import asyncio
from pathlib import Path
import sys
import json
from google import genai
from google.genai import types

key_path = Path("./api-key.txt")
if not key_path.exists():
key_path.touch()

with open(key_path, "r") as f:
key = f.readline().strip()
if not key:
print(f"Paste your API key into {key_path.resolve()}")
sys.exit(1)

client = genai.Client(
api_key=key,
http_options={'api_version': 'v1beta'}
)

MODEL = "gemini-2.5-flash-native-audio-preview-12-2025"

get_weather_tool = {
"name": "getWeather",
"description": "gets the weather (temperature in Celsius) for a requested city",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string"}
},
"required": ["city"]
}
}
tools_config = [types.Tool(function_declarations=[get_weather_tool])]

CONFIG = types.LiveConnectConfig(
response_modalities=["AUDIO"],
speech_config=types.SpeechConfig(
voice_config=types.VoiceConfig(
prebuilt_voice_config=types.PrebuiltVoiceConfig(
voice_name="Zephyr")
)
),
tools=tools_config,
)

async def run():
try:
async with client.aio.live.connect(model=MODEL, config=CONFIG) as session:
print("Connected! Loading context...")

func_call_id = "call_12345"

history_turns = [
# 1. User
types.Content(
role="user",
parts=[types.Part(
text="Hello, my name is Sasha. My hobby is skiing")]
),
# 2. Model
types.Content(
role="model",
parts=[types.Part(text="Hello Sasha! Nice to meet you.")]
),
# 3. User
types.Content(
role="user",
parts=[types.Part(text="What is the weather in Kyiv?")]
),
# 4. Function call
types.Content(
role="model",
parts=[types.Part(
function_call=types.FunctionCall(
name="getWeather",
args={"city": "Kyiv"},
id=func_call_id
)
)]
),
# 5. Tool Response
types.Content(
role="user",
parts=[types.Part(
function_response=types.FunctionResponse(
name="getWeather",
response={"result": "25"},
id=func_call_id
)
)]
),
# 6.  Model
types.Content(
role="model",
parts=[types.Part(text="It's 25 degrees celsius in Kyiv.")]
),
]

await session.send_client_content(
turns=history_turns,
turn_complete=False
)

print("Context loaded.")

while True:
text = await asyncio.to_thread(input, "\nYou: ")
if text.lower() == "q":
break

await session.send_client_content(
turns=[types.Content(
role="user",
parts=[types.Part(text=text)]
)],
turn_complete=True
)

print("Gemini: ", end="")
async for response in session.receive():
if response.server_content:
if response.server_content.output_transcription:
print(
response.server_content.output_transcription.text, end="", flush=True)

if response.server_content.model_turn:
for part in response.server_content.model_turn.parts:
if part.text:
print(part.text, end="", flush=True)

if response.server_content and response.server_content.turn_complete:
break
print()

except Exception as e:
print("\nConnection failed/closed:", repr(e))

if __name__ == "__main__":
asyncio.run(run())

Когда я запускаю его и ввожу любое приглашение, я получаю следующую ошибку:

Код: Выделить всё

Connected! Loading context...
Context loaded.

You: What was the weather like?

Connection failed/closed: ConnectionClosedError(Close(code=1007, reason='Request contains an invalid argument.'), Close(code=1007, reason='Request contains an invalid argument.'), True)

Не работает асинхронная строка ответа в session.receive():,
Но это явно связано с этой частью:

Код: Выделить всё

                # 4. Function call
types.Content(
role="model",
parts=[types.Part(
function_call=types.FunctionCall(
name="getWeather",
args={"city": "Kyiv"},
id=func_call_id
)
)]
),
# 5. Tool Response
types.Content(
role="user",
parts=[types.Part(
function_response=types.FunctionResponse(
name="getWeather",
response={"result": "25"},
id=func_call_id
)
)]
),

потому что если я удалю его, выполнение пройдет и модель сможет сказать, что мое хобби - это, например, катание на лыжах.
Я также пробовал обернуть {"city": "Kyiv"} и {"result": "25" в json.dumps(), но получаю еще одну ошибку:

Код: Выделить всё

Connection failed/closed: 1 validation error for FunctionCall
args
Input should be a valid dictionary [type=dict_type, input_value='{"city": "Kyiv"}', input_type=str]
For further information visit https://errors.pydantic.dev/2.12/v/dict_type

Есть ли способ передать вызовы инструментов во вновь созданный диалог? Или как предполагается, что сеансы будут сохраняться навсегда в контексте Gemini Live API?


Подробнее здесь: https://stackoverflow.com/questions/798 ... ous-conver