Spaces:
Sleeping
Sleeping
| """ | |
| Personal AI Agent - Telegram Bot (Webhook Mode) | |
| Deployed on Hugging Face Spaces | |
| Uses webhook mode: Telegram pushes updates to us via HTTP POST. | |
| Replies are sent via the webhook response body (no outgoing connections to Telegram needed). | |
| This avoids DNS/firewall issues on HF Spaces. | |
| """ | |
| import os | |
| import logging | |
| import json | |
| import threading | |
| import queue | |
| from collections import defaultdict | |
| from datetime import datetime | |
| import urllib.request | |
| import urllib.parse | |
| import gradio as gr | |
| from fastapi import FastAPI, Request | |
| from fastapi.responses import JSONResponse | |
| logging.basicConfig( | |
| format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", | |
| level=logging.INFO, | |
| ) | |
| logger = logging.getLogger(__name__) | |
| # ── Config ──────────────────────────────────────────────────── | |
| TELEGRAM_TOKEN = os.environ.get("TELEGRAM_TOKEN", "") | |
| HF_TOKEN = os.environ.get("HF_TOKEN", "") | |
| ALLOWED_USERS = os.environ.get("ALLOWED_USERS", "") | |
| MODEL = os.environ.get("MODEL", "Qwen/Qwen2.5-Coder-32B-Instruct") | |
| MAX_HISTORY = 30 | |
| SPACE_HOST = os.environ.get("SPACE_HOST", "") # e.g. "mlnjsh-milan-ai-agent.hf.space" | |
| HF_API_URL = "https://router.huggingface.co/v1/chat/completions" | |
| SYSTEM_PROMPT = """You are Dr. Milan Joshi's personal AI assistant running on Telegram. You are smart, concise, and genuinely helpful. | |
| Your capabilities: | |
| - Answer questions on any topic (especially Math, ML, AI, Data Science) | |
| - Summarize text and articles | |
| - Help with writing, editing, brainstorming | |
| - Explain complex concepts simply | |
| - Remember conversation context | |
| Rules: | |
| - Be conversational and to the point. No fluff. | |
| - Use short paragraphs. Telegram messages should be easy to read on mobile. | |
| - For code, use monospace formatting with backticks. | |
| - For math, write equations clearly (no LaTeX rendering on Telegram). | |
| - If you don't know something, say so honestly. | |
| - Keep responses under 300 words unless the user asks for detail.""" | |
| # ── Memory ──────────────────────────────────────────────────── | |
| conversations = defaultdict(list) | |
| user_stats = defaultdict(lambda: {"messages": 0, "first_seen": None}) | |
| pending_replies = defaultdict(queue.Queue) | |
| def get_history(user_id): | |
| return conversations[str(user_id)] | |
| def add_message(user_id, role, content): | |
| uid = str(user_id) | |
| conversations[uid].append({"role": role, "content": content}) | |
| if len(conversations[uid]) > MAX_HISTORY * 2: | |
| conversations[uid] = conversations[uid][-MAX_HISTORY * 2:] | |
| if role == "user": | |
| user_stats[uid]["messages"] += 1 | |
| if not user_stats[uid]["first_seen"]: | |
| user_stats[uid]["first_seen"] = datetime.now().isoformat() | |
| def is_allowed(user_id): | |
| if not ALLOWED_USERS: | |
| return True | |
| allowed = [u.strip() for u in ALLOWED_USERS.split(",") if u.strip()] | |
| return str(user_id) in allowed | |
| # ── LLM Call ────────────────────────────────────────────────── | |
| def call_llm(messages, max_tokens=600, temperature=0.7): | |
| try: | |
| payload = json.dumps({ | |
| "model": MODEL, | |
| "messages": messages, | |
| "max_tokens": max_tokens, | |
| "temperature": temperature, | |
| }).encode("utf-8") | |
| req = urllib.request.Request( | |
| HF_API_URL, | |
| data=payload, | |
| headers={ | |
| "Authorization": f"Bearer {HF_TOKEN}", | |
| "Content-Type": "application/json", | |
| }, | |
| ) | |
| with urllib.request.urlopen(req, timeout=60) as resp: | |
| data = json.loads(resp.read().decode("utf-8")) | |
| return data["choices"][0]["message"]["content"] | |
| except Exception as e: | |
| logger.error(f"LLM error: {e}") | |
| return f"Sorry, I hit an error: {str(e)[:200]}" | |
| # ── Message Processing ─────────────────────────────────────── | |
| MODES = { | |
| "casual": "You speak casually, like a smart friend texting. Use short sentences, occasional humor.", | |
| "formal": "You speak formally and professionally. Structured responses, precise language.", | |
| "technical": "You are highly technical. Use precise terminology, include details, assume expertise.", | |
| "creative": "You are creative and expressive. Use analogies, metaphors, and vivid language.", | |
| } | |
| def make_reply(chat_id, text): | |
| """Create a Telegram sendMessage webhook response.""" | |
| return { | |
| "method": "sendMessage", | |
| "chat_id": chat_id, | |
| "text": text[:4096], | |
| } | |
| def process_update(update): | |
| """Process a Telegram update and return a webhook response dict (or None).""" | |
| msg = update.get("message") | |
| if not msg: | |
| return None | |
| chat_id = msg["chat"]["id"] | |
| user_id = msg.get("from", {}).get("id", chat_id) | |
| text = msg.get("text", "") | |
| first_name = msg.get("from", {}).get("first_name", "there") | |
| if not text or not is_allowed(user_id): | |
| return None | |
| uid = str(user_id) | |
| # /start | |
| if text.startswith("/start"): | |
| return make_reply(chat_id, | |
| f"Hey {first_name}! I'm your personal AI assistant.\n\n" | |
| "Just send me a message and I'll respond. I remember our conversation.\n\n" | |
| "Commands:\n" | |
| "/update - Latest AI tools & model news\n" | |
| "/movies - Kids movies in Bangalore theaters\n" | |
| "/kids - Kids events & activities in Bangalore\n" | |
| "/summary - Summarize text\n" | |
| "/mode <style> - Change personality\n" | |
| "/clear - Fresh start\n" | |
| "/stats - Your usage stats\n" | |
| "/help - Show all commands" | |
| ) | |
| # /help | |
| if text.startswith("/help"): | |
| return make_reply(chat_id, | |
| "I'm your personal AI assistant. Here's what I can do:\n\n" | |
| "AI & Tech:\n" | |
| "/update - Latest AI tools news (Claude, GPT, Gemini, DeepSeek, Grok...)\n" | |
| "/update <topic> - Focus on specific AI topic\n\n" | |
| "Bangalore Life:\n" | |
| "/movies - Kids movies in Bangalore theaters with prices\n" | |
| "/kids - Kids events & activities in Bangalore\n\n" | |
| "Utilities:\n" | |
| "/summary <text> - Summarize text (or reply to a message)\n" | |
| "/mode <style> - Change personality (casual/formal/technical/creative)\n" | |
| "/clear - Wipe conversation memory\n" | |
| "/stats - Your usage stats\n\n" | |
| "Tips:\n" | |
| "- Just chat normally for anything else\n" | |
| "- I remember our conversation (~30 messages)\n" | |
| "- Forward me articles/text to summarize" | |
| ) | |
| # /clear | |
| if text.startswith("/clear"): | |
| conversations[uid] = [] | |
| return make_reply(chat_id, "Memory cleared. Fresh start!") | |
| # /stats | |
| if text.startswith("/stats"): | |
| stats = user_stats[uid] | |
| history_len = len(conversations.get(uid, [])) | |
| return make_reply(chat_id, | |
| f"Your Stats:\n" | |
| f"- Messages sent: {stats['messages']}\n" | |
| f"- Memory: {history_len} messages in context\n" | |
| f"- First seen: {stats.get('first_seen', 'just now')}\n" | |
| f"- Model: {MODEL}" | |
| ) | |
| # /summary | |
| if text.startswith("/summary"): | |
| summary_text = text[len("/summary"):].strip() | |
| reply = msg.get("reply_to_message", {}) | |
| if reply and reply.get("text"): | |
| summary_text = reply["text"] | |
| if not summary_text: | |
| return make_reply(chat_id, "Send /summary followed by text, or reply to a message with /summary") | |
| messages = [ | |
| {"role": "system", "content": "Summarize the following text in 2-4 concise bullet points. Be brief and capture the key points."}, | |
| {"role": "user", "content": summary_text[:3000]}, | |
| ] | |
| result = call_llm(messages, max_tokens=300, temperature=0.3) | |
| return make_reply(chat_id, result) | |
| # /update - AI tools news | |
| if text.startswith("/update"): | |
| topic = text[len("/update"):].strip() | |
| today = datetime.now().strftime("%B %d, %Y") | |
| prompt = ( | |
| f"Today is {today}. Give a concise update on the LATEST news, releases, and updates from these AI tools and companies. " | |
| "Cover each one briefly (1-2 lines max per tool):\n\n" | |
| "1. Anthropic Claude - any new model, feature, or API update\n" | |
| "2. OpenAI ChatGPT - new GPT models, features, plugins\n" | |
| "3. Google Gemini - new versions, capabilities\n" | |
| "4. DeepSeek - new models, math/reasoning breakthroughs\n" | |
| "5. xAI Grok - updates, new features\n" | |
| "6. Perplexity AI - search, new features\n" | |
| "7. Kimi (Moonshot AI) - updates\n" | |
| "8. Notable agentic AI / coding tool updates (Cursor, Windsurf, Claude Code, Copilot, etc.)\n\n" | |
| "Focus on the MOST RECENT updates you know about. " | |
| "Use bullet points. Be specific with version numbers and dates where possible. " | |
| "If you're unsure about very recent news, say so and mention the last update you know of." | |
| ) | |
| if topic: | |
| prompt += f"\n\nThe user is specifically interested in: {topic}" | |
| messages = [ | |
| {"role": "system", "content": "You are an AI news reporter. Provide the latest AI tool updates concisely. Use bullet points and emojis for readability on Telegram."}, | |
| {"role": "user", "content": prompt}, | |
| ] | |
| result = call_llm(messages, max_tokens=1200, temperature=0.3) | |
| return make_reply(chat_id, result) | |
| # /movies - Kids movies in Bangalore theaters | |
| if text.startswith("/movies"): | |
| extra = text[len("/movies"):].strip() | |
| today = datetime.now().strftime("%A, %B %d, %Y") | |
| prompt = ( | |
| f"Today is {today}. I'm looking for movies suitable for kids under 9 years old " | |
| "currently showing or recently released in Bangalore theaters.\n\n" | |
| "Please provide:\n" | |
| "- Movie name (Hindi, English, Marathi, Kannada - any kid-friendly ones)\n" | |
| "- Which malls/theaters in Bangalore are showing them (PVR, INOX, Cinepolis etc.)\n" | |
| "- Approximate ticket price range\n" | |
| "- Age suitability and brief description\n\n" | |
| "Cover major malls like: Phoenix Marketcity, Orion Mall, Mantri Square, " | |
| "Forum Mall, VR Bengaluru, Nexus Shantiniketan, Lulu Mall, GT World Mall.\n\n" | |
| "If you don't have exact current listings, provide the most recent kid-friendly movies " | |
| "you know about and suggest checking BookMyShow or the theater websites for exact showtimes and prices. " | |
| "Format nicely for Telegram with emojis." | |
| ) | |
| if extra: | |
| prompt += f"\n\nAdditional preference: {extra}" | |
| messages = [ | |
| {"role": "system", "content": "You are a helpful movie guide for parents in Bangalore. Focus on kid-friendly content for children under 9."}, | |
| {"role": "user", "content": prompt}, | |
| ] | |
| result = call_llm(messages, max_tokens=1200, temperature=0.4) | |
| return make_reply(chat_id, result) | |
| # /kids - Kids events in Bangalore | |
| if text.startswith("/kids"): | |
| extra = text[len("/kids"):].strip() | |
| today = datetime.now().strftime("%A, %B %d, %Y") | |
| prompt = ( | |
| f"Today is {today}. I'm looking for upcoming kids events, activities, and things to do " | |
| "for children (under 9 years old) in Bangalore.\n\n" | |
| "Please cover:\n" | |
| "- Weekend events and workshops for kids\n" | |
| "- Fun zones, play areas, and amusement parks\n" | |
| "- Educational workshops (science, art, coding for kids)\n" | |
| "- Outdoor activities and nature events\n" | |
| "- Festival or seasonal events happening now\n" | |
| "- Popular venues: Cubbon Park, Lumbini Gardens, Innovative Film City, " | |
| "Wonderla, Fun World, art galleries with kids programs\n\n" | |
| "For each, mention: what it is, where, approximate cost, and age suitability.\n" | |
| "If you don't have exact current listings, suggest popular recurring events " | |
| "and recommend checking Bookmyshow, Eventbrite, or LBB Bangalore for latest listings. " | |
| "Format nicely for Telegram with emojis." | |
| ) | |
| if extra: | |
| prompt += f"\n\nSpecific interest: {extra}" | |
| messages = [ | |
| {"role": "system", "content": "You are a helpful Bangalore kids activity guide. Focus on fun, safe, age-appropriate activities for children under 9."}, | |
| {"role": "user", "content": prompt}, | |
| ] | |
| result = call_llm(messages, max_tokens=1200, temperature=0.4) | |
| return make_reply(chat_id, result) | |
| # /mode | |
| mode = text[len("/mode"):].strip().lower() | |
| if not mode: | |
| return make_reply(chat_id, f"Available modes: {', '.join(MODES.keys())}\nUsage: /mode casual") | |
| if mode not in MODES: | |
| return make_reply(chat_id, f"Unknown mode. Choose from: {', '.join(MODES.keys())}") | |
| add_message(uid, "system", f"Style change: {MODES[mode]}") | |
| return make_reply(chat_id, f"Mode switched to: {mode}") | |
| # Regular message - chat with LLM | |
| add_message(uid, "user", text) | |
| messages = [{"role": "system", "content": SYSTEM_PROMPT}] | |
| messages.extend(get_history(uid)) | |
| result = call_llm(messages) | |
| add_message(uid, "assistant", result) | |
| return make_reply(chat_id, result) | |
| # ── Gradio App + Webhook ───────────────────────────────────── | |
| bot_status = "Not configured" | |
| webhook_url = "" | |
| if TELEGRAM_TOKEN: | |
| if SPACE_HOST: | |
| webhook_url = f"https://{SPACE_HOST}/webhook" | |
| bot_status = "Webhook mode (waiting for setup)" | |
| else: | |
| bot_status = "SPACE_HOST not available yet" | |
| else: | |
| bot_status = "TELEGRAM_TOKEN not set" | |
| def get_status(): | |
| active_users = len([u for u in conversations if conversations[u]]) | |
| total_messages = sum(s["messages"] for s in user_stats.values()) | |
| wh_display = webhook_url or "Not configured" | |
| return ( | |
| f"## Bot Status: {bot_status}\n\n" | |
| f"- **Webhook URL:** `{wh_display}`\n" | |
| f"- **Model:** {MODEL}\n" | |
| f"- **Active conversations:** {active_users}\n" | |
| f"- **Total messages processed:** {total_messages}\n" | |
| f"- **Memory per user:** {MAX_HISTORY * 2} messages max\n" | |
| f"- **Allowed users:** {'Everyone' if not ALLOWED_USERS else ALLOWED_USERS}\n" | |
| ) | |
| def setup_webhook_instructions(): | |
| """Return instructions for setting up the webhook.""" | |
| if not webhook_url: | |
| return "Webhook URL not available. Make sure TELEGRAM_TOKEN is set and the Space is running." | |
| return ( | |
| f"**Run this command in your terminal to activate the bot:**\n\n" | |
| f"```\n" | |
| f"curl -X POST \"https://api.telegram.org/bot{TELEGRAM_TOKEN}/setWebhook\"" | |
| f" -d \"url={webhook_url}\"\n" | |
| f"```\n\n" | |
| f"Or open this URL in your browser:\n\n" | |
| f"`https://api.telegram.org/bot{TELEGRAM_TOKEN}/setWebhook?url={webhook_url}`" | |
| ) | |
| with gr.Blocks(title="Personal AI Agent", theme=gr.themes.Soft()) as demo: | |
| gr.Markdown( | |
| """ | |
| # Personal AI Agent - Telegram Bot | |
| **Dr. Milan Joshi's AI Assistant** running on Telegram. | |
| This Space runs a Telegram bot that acts as your personal AI assistant | |
| with persistent memory, multiple conversation modes, and text summarization. | |
| """ | |
| ) | |
| status_md = gr.Markdown(get_status()) | |
| refresh_btn = gr.Button("Refresh Status") | |
| refresh_btn.click(fn=get_status, outputs=status_md) | |
| with gr.Accordion("Webhook Setup", open=True): | |
| gr.Markdown(setup_webhook_instructions()) | |
| gr.Markdown( | |
| """ | |
| --- | |
| ### Commands | |
| | Command | Description | | |
| |---------|-------------| | |
| | Just chat | AI responds with context memory | | |
| | `/update` | Latest AI tools & model news | | |
| | `/movies` | Kids movies in Bangalore theaters | | |
| | `/kids` | Kids events & activities in Bangalore | | |
| | `/summary` | Summarize text (reply to msg or type after) | | |
| | `/mode` | Change style: casual, formal, technical, creative | | |
| | `/clear` | Wipe conversation memory | | |
| | `/stats` | Your usage stats | | |
| | `/help` | Show all commands | | |
| """ | |
| ) | |
| # Mount webhook endpoint on the Gradio app's FastAPI | |
| app = FastAPI() | |
| async def telegram_webhook(request: Request): | |
| """Handle incoming Telegram webhook updates.""" | |
| try: | |
| update = await request.json() | |
| logger.info(f"Webhook received update: {update.get('update_id', '?')}") | |
| reply = process_update(update) | |
| if reply: | |
| return JSONResponse(content=reply) | |
| return JSONResponse(content={"ok": True}) | |
| except Exception as e: | |
| logger.error(f"Webhook error: {e}") | |
| return JSONResponse(content={"ok": True}) | |
| # Mount Gradio inside FastAPI | |
| app = gr.mount_gradio_app(app, demo, path="/") | |
| # For HF Spaces, we need to use the app directly | |
| if __name__ == "__main__": | |
| import uvicorn | |
| uvicorn.run(app, host="0.0.0.0", port=7860) | |