| | import requests |
| | import json |
| | import time |
| |
|
| | def thinking_animation(): |
| | """ |
| | دالة بسيطة للاستخدام أثناء الانتظار (ليست ضرورية). |
| | """ |
| | for _ in range(3): |
| | print(".", end="", flush=True) |
| | time.sleep(0.5) |
| |
|
| | def call_o1_ai_api(formatted_chat_history): |
| | """ |
| | يُرسِل المحادثة الحالية إلى واجهة الذكاء الاصطناعي o1 |
| | ويعيد الرد مع تحديث المحادثة. |
| | """ |
| | url = "https://corvo-ai-xx-o1.hf.space/chat" |
| | headers = {"Content-Type": "application/json"} |
| | payload = { |
| | "chat_history": formatted_chat_history |
| | } |
| | max_retries = 5 |
| | retry_delay = 10 |
| | timeout = 600 |
| |
|
| | for attempt in range(max_retries): |
| | try: |
| | print("AI THINKING", end="", flush=True) |
| | thinking_animation() |
| | response = requests.post(url, headers=headers, data=json.dumps(payload), timeout=timeout) |
| | response.raise_for_status() |
| | assistant_response = response.json().get("assistant_response", "No response received.") |
| | |
| | formatted_chat_history.append({"role": "assistant", "content": assistant_response}) |
| | return assistant_response, formatted_chat_history |
| | except requests.exceptions.Timeout: |
| | print(f"Timeout on attempt {attempt + 1}, retrying...") |
| | time.sleep(retry_delay) |
| | except Exception as e: |
| | print(f"Error on attempt {attempt + 1}: {e}, retrying...") |
| | time.sleep(retry_delay) |
| |
|
| | return "Error processing request. Please try again.", formatted_chat_history |
| |
|
| |
|
