import os import json import ast import math import time import asyncio import threading from collections import defaultdict, deque import wikipedia import torch from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup from telegram.constants import ChatAction from telegram.ext import ( ApplicationBuilder, MessageHandler, CommandHandler, CallbackQueryHandler, ContextTypes, filters, ) from transformers import AutoTokenizer, AutoModelForCausalLM MODEL_NAME = "Qwen/Qwen2-1.5B-Instruct" MAX_HISTORY = 12 MAX_STEPS = 4 MAX_NEW_TOKENS_JSON = 220 torch.set_num_threads(max(1, os.cpu_count() // 2)) tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME) model = AutoModelForCausalLM.from_pretrained( MODEL_NAME, torch_dtype=torch.float32, device_map="cpu", low_cpu_mem_usage=True ) model.eval() memory = defaultdict(lambda: deque(maxlen=MAX_HISTORY)) button_state = defaultdict(dict) PLANNER_SYSTEM_PROMPT = """ You are an advanced Telegram AI agent. You must never expose internal tool calls, JSON planning, scratch work, or control tokens to the user. You have exactly three response formats. 1) Final: {"type":"final","text":"your message to the user"} 2) Buttons: {"type":"buttons","text":"question for the user","buttons":[{"id":"choice_1","label":"Yes"},{"id":"choice_2","label":"No"}]} 3) Tool: {"type":"tool","name":"wiki_search","arguments":{"query":"Finland"}} {"type":"tool","name":"calculate","arguments":{"expression":"(25*17)/5"}} Rules: - Output exactly one JSON object. - No markdown fences. - No extra text before or after JSON. - The button type must be exactly "buttons". - Use buttons when the user should choose between a few short options. - Use wiki_search for factual topics, places, people, concepts, summaries. - Use calculate for arithmetic or formula evaluation. - If you are unsure, do not call a tool. Respond with a final answer instead. - After receiving a tool result, continue and respond with exactly one JSON object. - Prefer Finnish if the user speaks Finnish. - Never output tokens like <|end|>, <|im_start|>, <|im_end|>. """ def safe_calculate(expression: str) -> str: allowed_names = { "abs": abs, "round": round, "min": min, "max": max, "pow": pow, "sqrt": math.sqrt, "sin": math.sin, "cos": math.cos, "tan": math.tan, "pi": math.pi, "e": math.e, } allowed_nodes = ( ast.Expression, ast.BinOp, ast.UnaryOp, ast.Num, ast.Constant, ast.Add, ast.Sub, ast.Mult, ast.Div, ast.FloorDiv, ast.Mod, ast.Pow, ast.USub, ast.UAdd, ast.Load, ast.Call, ast.Name, ast.Tuple, ast.List, ) try: tree = ast.parse(expression, mode="eval") except Exception: return "Virhe: laskua ei voitu lukea." for node in ast.walk(tree): if not isinstance(node, allowed_nodes): return "Virhe: laskua ei voitu suorittaa turvallisesti." if isinstance(node, ast.Call): if not isinstance(node.func, ast.Name): return "Virhe: laskua ei voitu suorittaa turvallisesti." if node.func.id not in allowed_names: return "Virhe: laskua ei voitu suorittaa turvallisesti." if isinstance(node, ast.Name): if node.id not in allowed_names: return "Virhe: laskua ei voitu suorittaa turvallisesti." try: result = eval(compile(tree, "", "eval"), {"__builtins__": {}}, allowed_names) return str(result) except Exception: return "Virhe: laskua ei voitu suorittaa." def wiki_search(query: str) -> str: query = query.strip() if not query: return "Virhe: tyhjä hakukysely." try: wikipedia.set_lang("fi") try: return wikipedia.summary(query, sentences=3, auto_suggest=True) except Exception: page = wikipedia.page(query, auto_suggest=True) return page.summary[:1200] except Exception: try: wikipedia.set_lang("en") try: return wikipedia.summary(query, sentences=3, auto_suggest=True) except Exception: page = wikipedia.page(query, auto_suggest=True) return page.summary[:1200] except Exception: return f"En löytänyt hakutulosta haulle: {query}" TOOLS = { "wiki_search": wiki_search, "calculate": safe_calculate, } def clean_text(text: str) -> str: bad_tokens = [ "<|end|>", "<|im_start|>", "<|im_end|>", "<|assistant|>", "<|user|>", "<|system|>", "", ] for token in bad_tokens: text = text.replace(token, "") return text.strip() def extract_json(text: str): text = text.strip() decoder = json.JSONDecoder() for i, ch in enumerate(text): if ch == "{": try: obj, end = decoder.raw_decode(text[i:]) trailing = text[i + end:].strip() if trailing: continue return obj except Exception: continue return None def build_planner_messages(user_id: int, user_text: str): messages = [{"role": "system", "content": PLANNER_SYSTEM_PROMPT}] for item in memory[user_id]: messages.append(item) messages.append({"role": "user", "content": user_text}) return messages def generate_chat_text(messages, max_new_tokens=220): prompt = tokenizer.apply_chat_template( messages, tokenize=False, add_generation_prompt=True ) inputs = tokenizer(prompt, return_tensors="pt") with torch.no_grad(): outputs = model.generate( **inputs, max_new_tokens=max_new_tokens, do_sample=False, use_cache=True, pad_token_id=tokenizer.eos_token_id ) new_tokens = outputs[0][inputs["input_ids"].shape[1]:] text = tokenizer.decode(new_tokens, skip_special_tokens=False) return clean_text(text) def run_agent(user_id: int, user_text: str): messages = build_planner_messages(user_id, user_text) for _ in range(MAX_STEPS): raw = generate_chat_text(messages, max_new_tokens=MAX_NEW_TOKENS_JSON) data = extract_json(raw) if not isinstance(data, dict): messages.append({"role": "assistant", "content": raw}) messages.append({ "role": "user", "content": 'Your previous response was invalid. Output ONLY one valid JSON object.' }) continue response_type = data.get("type") if response_type == "final": text = clean_text(str(data.get("text", ""))) if not text: text = "En saanut muodostettua vastausta." return {"type": "final", "text": text} if response_type == "buttons": text = clean_text(str(data.get("text", ""))) buttons = data.get("buttons", []) normalized = [] if isinstance(buttons, list): for b in buttons[:6]: if isinstance(b, dict): bid = str(b.get("id", "")).strip() label = str(b.get("label", "")).strip() if bid and label: normalized.append({"id": bid[:32], "label": label[:40]}) if text and normalized: return {"type": "buttons", "text": text, "buttons": normalized} messages.append({"role": "assistant", "content": json.dumps(data, ensure_ascii=False)}) messages.append({ "role": "user", "content": 'That buttons response was invalid. Output ONLY one valid JSON object.' }) continue if response_type == "tool": name = data.get("name") arguments = data.get("arguments", {}) if name not in TOOLS: return {"type": "final", "text": f"Tuntematon työkalu: {name}"} if not isinstance(arguments, dict): return {"type": "final", "text": "Työkalun argumentit olivat virheelliset."} if name == "wiki_search": query = str(arguments.get("query", "")).strip() result = TOOLS[name](query) elif name == "calculate": expression = str(arguments.get("expression", "")).strip() result = TOOLS[name](expression) else: result = "Työkalua ei voitu suorittaa." messages.append({"role": "assistant", "content": json.dumps(data, ensure_ascii=False)}) messages.append({ "role": "user", "content": f"Tool result for {name}:\n{result}\nNow continue and respond with exactly one JSON object." }) continue messages.append({"role": "assistant", "content": json.dumps(data, ensure_ascii=False)}) messages.append({ "role": "user", "content": 'That response type was invalid. Output ONLY one valid JSON object.' }) return {"type": "final", "text": "Pyyntö vaati liikaa välivaiheita."} async def typing_loop(chat, stop_event: asyncio.Event): while not stop_event.is_set(): try: await chat.send_action(ChatAction.TYPING) except Exception: return try: await asyncio.wait_for(stop_event.wait(), timeout=2.0) except asyncio.TimeoutError: pass def chunk_text_for_stream(text: str): words = text.split() if not words: return [""] chunks = [] current = "" for word in words: candidate = f"{current} {word}".strip() if len(candidate) >= 35: chunks.append(candidate) current = "" else: current = candidate if current: chunks.append(current) return chunks async def stream_text_reply(message, text: str): text = clean_text(text) if not text: text = " " chunks = chunk_text_for_stream(text) sent = await message.reply_text("...") assembled = "" last_edit_time = 0.0 for i, chunk in enumerate(chunks): assembled = f"{assembled} {chunk}".strip() now = time.time() if i < len(chunks) - 1: if now - last_edit_time < 0.55: await asyncio.sleep(0.55 - (now - last_edit_time)) safe_text = assembled[:4096] try: await sent.edit_text(safe_text) last_edit_time = time.time() except Exception: pass return sent async def process_user_text(message, context: ContextTypes.DEFAULT_TYPE, user_id: int, text: str): stop_event = asyncio.Event() typing_task = asyncio.create_task(typing_loop(message.chat, stop_event)) try: result = await asyncio.to_thread(run_agent, user_id, text) finally: stop_event.set() await typing_task memory[user_id].append({"role": "user", "content": text}) if result["type"] == "buttons": keyboard = [] button_state[user_id] = {} for b in result["buttons"]: button_state[user_id][b["id"]] = b["label"] keyboard.append([InlineKeyboardButton(b["label"], callback_data=f"btn:{b['id']}")]) memory[user_id].append({ "role": "assistant", "content": json.dumps( {"type": "buttons", "text": result["text"], "buttons": result["buttons"]}, ensure_ascii=False ) }) await message.reply_text( result["text"], reply_markup=InlineKeyboardMarkup(keyboard) ) return reply = clean_text(result["text"]) memory[user_id].append({ "role": "assistant", "content": json.dumps({"type": "final", "text": reply}, ensure_ascii=False) }) await stream_text_reply(message, reply) async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE): if not update.message or not update.effective_user: return text = (update.message.text or "").strip() if not text: return await process_user_text(update.message, context, update.effective_user.id, text) async def handle_button(update: Update, context: ContextTypes.DEFAULT_TYPE): query = update.callback_query if not query or not update.effective_user: return await query.answer() data = query.data or "" if not data.startswith("btn:"): return button_id = data[4:] user_id = update.effective_user.id label = button_state[user_id].get(button_id) if not label: await query.message.reply_text("Tämä valinta ei ole enää voimassa.") return await process_user_text(query.message, context, user_id, label) async def start(update: Update, context: ContextTypes.DEFAULT_TYPE): if not update.message: return await update.message.reply_text("Moi. Olen AI-agentti. Laita viestiä.") async def reset_chat(update: Update, context: ContextTypes.DEFAULT_TYPE): if not update.message or not update.effective_user: return user_id = update.effective_user.id memory[user_id].clear() button_state[user_id].clear() await update.message.reply_text("Muisti nollattu.") def main(): token = os.environ["TELEGRAM_TOKEN"] app = ApplicationBuilder().token(token).build() app.add_handler(CommandHandler("start", start)) app.add_handler(CommandHandler("reset", reset_chat)) app.add_handler(CallbackQueryHandler(handle_button)) app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message)) app.run_polling(drop_pending_updates=True) if __name__ == "__main__": main()