Create telegram_bot.py

telegram_bot.py

@@ -0,0 +1,145 @@
+import logging
+from telegram import Update
+from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
+
+# Enable logging
+logging.basicConfig(
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+    level=logging.INFO
+)
+logger = logging.getLogger(__name__)
+
+# Global agent instance (will be set by start_telegram_bot)
+agent = None
+
+# Command Handlers
+async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    """Send a message when the command /start is issued."""
+    user = update.effective_user
+    await update.message.reply_text(
+        f'Hi {user.first_name}! I am your AI coding assistant. '
+        f'Ask me anything and I will solve it step by step!'
+    )
+
+async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    """Send a message when the command /help is issued."""
+    help_text = """
+    I can help you with:
+    - Coding questions
+    - Web searches
+    - Mathematical calculations
+    - Data analysis
+    - And much more!
+    
+    Just send me your question and I'll work on it!
+    """
+    await update.message.reply_text(help_text)
+
+async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    """Handle user messages and run agent"""
+    user_message = update.message.text
+    user_id = update.effective_user.id
+    
+    logger.info(f"User {user_id} asked: {user_message}")
+    
+    # Send "thinking" message
+    thinking_msg = await update.message.reply_text("🤔 Thinking...")
+    
+    try:
+        # Run agent
+        result = agent.run(user_message, reset=False)
+        
+        # Extract final answer
+        final_answer = str(result)
+        
+        # Send result
+        await thinking_msg.edit_text(f"✅ Answer:\n\n{final_answer}")
+        
+    except Exception as e:
+        logger.error(f"Error processing message: {e}")
+        await thinking_msg.edit_text(
+            f"❌ Sorry, I encountered an error: {str(e)}\n\n"
+            f"Please try rephrasing your question."
+        )
+
+async def handle_streaming_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    """Handle messages with streaming updates (advanced)"""
+    user_message = update.message.text
+    
+    # Send initial message
+    status_msg = await update.message.reply_text("🤔 Starting...")
+    
+    try:
+        step_count = 0
+        for step_log in agent.run(user_message, stream=True, reset=False):
+            step_count += 1
+            
+            # Update status for each step
+            if hasattr(step_log, 'model_output'):
+                await status_msg.edit_text(
+                    f"📝 Step {step_count}:\n{step_log.model_output[:500]}..."
+                )
+        
+        # Send final answer
+        final_answer = str(step_log)
+        await status_msg.edit_text(f"✅ Final Answer:\n\n{final_answer}")
+        
+    except Exception as e:
+        await status_msg.edit_text(f"❌ Error: {str(e)}")
+
+def start_telegram_bot(shared_agent, token="8634464564:AAGL7FzFkMN-Uktf97NKtDs1RFPGxec-HFI"):
+    """Start the Telegram bot with a shared agent instance
+    
+    Args:
+        shared_agent: The CodeAgent instance to use for processing messages
+        token: Telegram bot token (default provided, can be overridden)
+    """
+    global agent
+    agent = shared_agent
+    
+    # Create application
+    application = Application.builder().token(token).build()
+    
+    # Register handlers
+    application.add_handler(CommandHandler("start", start))
+    application.add_handler(CommandHandler("help", help_command))
+    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
+    
+    # Start bot
+    logger.info("Telegram Bot started!")
+    application.run_polling(allowed_updates=Update.ALL_TYPES)
+
+def main():
+    """Standalone entry point (for backward compatibility)"""
+    from smolagents import CodeAgent, LiteLLMModel
+    import yaml
+    from tools.final_answer import FinalAnswerTool
+    
+    # Use local Ollama model
+    model = LiteLLMModel(
+        model_id="ollama/qwen2.5-coder:7b",
+        api_base="http://localhost:11434",
+        max_tokens=2096,
+        temperature=0.5
+    )
+    
+    # Load prompts
+    with open("prompts.yaml", 'r') as stream:
+        prompt_templates = yaml.safe_load(stream)
+    
+    # Initialize tools
+    final_answer = FinalAnswerTool()
+    
+    # Create agent
+    standalone_agent = CodeAgent(
+        model=model,
+        tools=[final_answer],
+        max_steps=6,
+        verbosity_level=1,
+        prompt_templates=prompt_templates
+    )
+    
+    start_telegram_bot(standalone_agent)
+
+if __name__ == '__main__':
+    main()