Makes a request to the gemini-2.5-pro-preview-03-25 model with retry functionality. Args: prompt (str): The text prompt to send to the model log_id (str, optional): The log ID for tracking requests, defaults to tkb+timestamp max_tokens (int, optional): Maximum token
(prompt, log_id=None, max_tokens=8000, max_retries=3)
| 366 | |
| 367 | |
| 368 | def request_gemini(prompt, log_id=None, max_tokens=8000, max_retries=3): |
| 369 | """ |
| 370 | Makes a request to the gemini-2.5-pro-preview-03-25 model with retry functionality. |
| 371 | |
| 372 | Args: |
| 373 | prompt (str): The text prompt to send to the model |
| 374 | log_id (str, optional): The log ID for tracking requests, defaults to tkb+timestamp |
| 375 | max_tokens (int, optional): Maximum tokens for response, default 8000 |
| 376 | max_retries (int, optional): Maximum number of retry attempts, default 3 |
| 377 | |
| 378 | Returns: |
| 379 | dict: The model's response |
| 380 | """ |
| 381 | base_url = cfg("gemini", "base_url") |
| 382 | api_version = cfg("gemini", "api_version") |
| 383 | api_key = cfg("gemini", "api_key") |
| 384 | model_name = cfg("gemini", "model") |
| 385 | |
| 386 | client = openai.AzureOpenAI( |
| 387 | azure_endpoint=base_url, |
| 388 | api_version=api_version, |
| 389 | api_key=api_key, |
| 390 | ) |
| 391 | |
| 392 | if log_id is None: |
| 393 | log_id = generate_log_id() |
| 394 | |
| 395 | extra_headers = {"X-TT-LOGID": log_id} |
| 396 | |
| 397 | retry_count = 0 |
| 398 | while retry_count < max_retries: |
| 399 | try: |
| 400 | completion = client.chat.completions.create( |
| 401 | model=model_name, |
| 402 | messages=[{"role": "user", "content": prompt}], |
| 403 | max_tokens=max_tokens, |
| 404 | extra_headers=extra_headers, |
| 405 | ) |
| 406 | return completion |
| 407 | except Exception as e: |
| 408 | retry_count += 1 |
| 409 | if retry_count >= max_retries: |
| 410 | raise Exception(f"Failed after {max_retries} attempts. Last error: {str(e)}") |
| 411 | |
| 412 | # Exponential backoff with jitter |
| 413 | delay = (2**retry_count) * 0.1 + (random.random() * 0.1) |
| 414 | print( |
| 415 | f"Request failed with error: {str(e)}. Retrying in {delay:.2f} seconds... (Attempt {retry_count}/{max_retries})" |
| 416 | ) |
| 417 | time.sleep(delay) |
| 418 | |
| 419 | |
| 420 | def request_gemini_token(prompt, log_id=None, max_tokens=8000, max_retries=3): |
no test coverage detected