Spaces:
Configuration error
Configuration error
| import telebot | |
| import requests | |
| import os | |
| from flask import Flask | |
| import threading | |
| import time | |
| app = Flask(__name__) | |
| def health_check(): | |
| return "Hermes AI Agent is Active and Thinking!" | |
| TOKEN = os.getenv("TELEGRAM_BOT_TOKEN") | |
| HF_TOKEN = os.getenv("HF_TOKEN") | |
| bot = telebot.TeleBot(TOKEN, threaded=False) | |
| API_URL = "https://api-inference.huggingface.co/models/meta-llama/Meta-Llama-3-70B-Instruct" | |
| headers = {"Authorization": f"Bearer {HF_TOKEN}"} | |
| def query_llm(prompt): | |
| payload = { | |
| "inputs": f"<|begin_of_text|><|start_header_id|>system<|end_header_id|>\n\nYou are Hermes, a highly intelligent AI worker. You have access to 30+ free APIs from GitHub. You should think before you act and provide helpful, smart responses in Arabic or English as requested.<|eot_id|><|start_header_id|>user<|end_header_id|>\n\n{prompt}<|eot_id|><|start_header_id|>assistant<|end_header_id|>\n\n", | |
| "parameters": {"max_new_tokens": 500, "return_full_text": False} | |
| } | |
| try: | |
| response = requests.post(API_URL, headers=headers, json=payload, timeout=40) | |
| res_json = response.json() | |
| if isinstance(res_json, list): | |
| return res_json[0]['generated_text'] | |
| elif isinstance(res_json, dict) and "error" in res_json: | |
| return f"عذراً، المحرك يخبرني: {res_json['error']}" | |
| else: | |
| return "عذراً، أنا أواجه مشكلة في معالجة طلبك." | |
| except Exception as e: | |
| return f"حدث خطأ في الاتصال: {str(e)}" | |
| def handle_message(message): | |
| print(f"Received: {message.text}") | |
| # 1. Start typing indicator | |
| bot.send_chat_action(message.chat.id, 'typing') | |
| # 2. Query LLM (with typing indicator logic) | |
| def process_and_reply(): | |
| try: | |
| # We keep sending typing action every 5 seconds if it's slow | |
| ai_response = query_llm(message.text) | |
| bot.reply_to(message, ai_response) | |
| except Exception as e: | |
| bot.reply_to(message, f"Error: {e}") | |
| threading.Thread(target=process_and_reply).start() | |
| def run_bot(): | |
| print("Hermes Smart Agent is starting with Typing indicator...") | |
| while True: | |
| try: | |
| bot.polling(none_stop=True, interval=1, timeout=30) | |
| except Exception as e: | |
| print(f"Polling error: {e}") | |
| time.sleep(5) | |
| if __name__ == "__main__": | |
| t = threading.Thread(target=run_bot) | |
| t.daemon = True | |
| t.start() | |
| port = int(os.environ.get("PORT", 7860)) | |
| app.run(host='0.0.0.0', port=port) | |