From 3b5ef010df1137591c9e7037f6476c97405c4701 Mon Sep 17 00:00:00 2001 From: Dita Aji Pratama Date: Tue, 16 Jun 2026 22:44:10 +0700 Subject: [PATCH] Telegram Bot Client --- .env.example | 6 +- config.py | 8 ++ config.yaml | 5 + hendrik.py | 56 +++++++- requirements.txt | 1 + services/agent_loop.py | 68 +++++++++ services/telegram_client.py | 275 ++++++++++++++++++++++++++++++++++++ 7 files changed, 414 insertions(+), 5 deletions(-) create mode 100644 services/agent_loop.py create mode 100644 services/telegram_client.py diff --git a/.env.example b/.env.example index 910a3d7..554330c 100644 --- a/.env.example +++ b/.env.example @@ -1,3 +1,5 @@ -# XMPP_USERNAME= -# XMPP_PASSWORD= +XMPP_USERNAME= +XMPP_PASSWORD= + +TELEGRAM_TOKEN= diff --git a/config.py b/config.py index a53ef35..d984d07 100644 --- a/config.py +++ b/config.py @@ -101,6 +101,14 @@ XMPP_NICKNAME = os.getenv("XMPP_NICKNAME", default=_yaml_get("xmpp", "nickname XMPP_SELECTIVE_RESPONSE = os.getenv("XMPP_SELECTIVE_RESPONSE", default=str(_yaml_get("xmpp", "selective_response", default="true"))).strip().lower() in ("true", "1", "yes") +# ─── Telegram (non-credential dari YAML, credential dari .env) ───────────────── + +TELEGRAM_ENABLED = os.getenv("TELEGRAM_ENABLED", default=str(_yaml_get("telegram", "enabled", default="false"))).strip().lower() in ("true", "1", "yes") +TELEGRAM_TOKEN = os.getenv("TELEGRAM_TOKEN", default="") +TELEGRAM_ALLOWED_GROUP_IDS = os.getenv("TELEGRAM_ALLOWED_GROUP_IDS", default=_yaml_get("telegram", "allowed_group_ids", default="")).strip() +TELEGRAM_SELECTIVE_RESPONSE = os.getenv("TELEGRAM_SELECTIVE_RESPONSE", default=str(_yaml_get("telegram", "selective_response", default="true"))).strip().lower() in ("true", "1", "yes") + + # ─── RAG (YAML) ───────────────────────────────────────────────────────────────── RAG_PERSIST_DIR = os.getenv("RAG_PERSIST_DIR", default=_yaml_get("rag", "persist_dir", default="chroma_db")) diff --git a/config.yaml b/config.yaml index 9b71bdc..3e0fe0c 100644 --- a/config.yaml +++ b/config.yaml @@ -12,6 +12,11 @@ xmpp: nickname: "" # custom MUC nickname (empty = use username) selective_response: true # true = only response if mentioned/relevant +telegram: + enabled: false + allowed_group_ids: "" # comma-separated, kosong = semua grup + selective_response: true + # Humanize Delay (anti-bot detection) delay: read_min: 1.0 # second diff --git a/hendrik.py b/hendrik.py index 382ae24..594cfb5 100644 --- a/hendrik.py +++ b/hendrik.py @@ -1,4 +1,6 @@ -import os, sys +import os, sys, threading, time +import signal + import config from services.xmpp_client import XMPPClient @@ -32,6 +34,7 @@ tools_definition = [ TOOLS = gadget.tool_schemas (tools_definition) TOOL_HANDLERS = gadget.tool_handlers (tools_definition) + def main(): llm_client = LLMClient(config.llm_baseurl, config.llm_model, config.llm_api_key, config.llm_timeout) @@ -51,6 +54,8 @@ def main(): sys.exit(1) os.chdir(resolved) + services = [] + if config.XMPP_ENABLED: muc_rooms = [] if config.XMPP_MUC_ROOMS.strip(): @@ -66,7 +71,52 @@ def main(): agent_max_iterations = config.AGENT_MAX_ITERATIONS, muc_rooms = muc_rooms, ) - client.start() + services.append(client) + + if config.TELEGRAM_ENABLED: + from services.telegram_client import TelegramClient + + allowed_ids = [] + if config.TELEGRAM_ALLOWED_GROUP_IDS.strip(): + allowed_ids = [r.strip() for r in config.TELEGRAM_ALLOWED_GROUP_IDS.split(',') if r.strip()] + + tg = TelegramClient( + token = config.TELEGRAM_TOKEN, + llm_client = llm_client, + tools_definition = tools_definition, + TOOLS = TOOLS, + TOOL_HANDLERS = TOOL_HANDLERS, + build_system_prompt = build_system_prompt, + agent_max_iterations = config.AGENT_MAX_ITERATIONS, + allowed_group_ids = allowed_ids, + ) + services.append(tg) + + if services: + threads = [] + for svc in services: + t = threading.Thread(target=svc.start, daemon=True) + t.start() + threads.append(t) + + _shutdown = False + + def _handle_sig(signum, frame): + nonlocal _shutdown + print("\nShutting down...") + for svc in services: + svc.stop() + _shutdown = True + + signal.signal(signal.SIGTERM, _handle_sig) + signal.signal(signal.SIGINT, _handle_sig) + + try: + while not _shutdown: + time.sleep(1) + except KeyboardInterrupt: + _shutdown = True + print("Exiting.") else: from tui import HendrikTUI HendrikTUI( @@ -78,6 +128,6 @@ def main(): agent_max_iterations = config.AGENT_MAX_ITERATIONS, ).run() + if __name__ == "__main__": main() - diff --git a/requirements.txt b/requirements.txt index 82f051a..ab857cd 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,3 +3,4 @@ PyYAML>=6.0 chromadb>=0.5.0 openpyxl>=3.1.0 slixmpp +python-telegram-bot>=20.0 diff --git a/services/agent_loop.py b/services/agent_loop.py new file mode 100644 index 0000000..87801ef --- /dev/null +++ b/services/agent_loop.py @@ -0,0 +1,68 @@ +import json +from datetime import datetime + + +def _ts(): + return datetime.now().strftime('%H:%M:%S') + + +def execute_tool(tool_call, TOOL_HANDLERS): + tname = tool_call['function']['name'] + targs = json.loads(tool_call['function']['arguments']) + handler = TOOL_HANDLERS.get(tname) + if not handler: + return f'Tool {tname} not found' + try: + if tname == 'search_code': + return handler( + pattern=targs['pattern'], + search_type=targs['search_type'], + path=targs.get('path', '.'), + ) + elif tname == 'git_operation': + return handler(args=targs['args']) + else: + return handler(**targs) + except Exception as e: + return f'Error executing tool: {str(e)}' + + +def run_agent_loop(session, llm_client, TOOLS, TOOL_HANDLERS, max_iterations, on_tool_calls=None): + for step in range(max_iterations): + print(f'[{_ts()}] Step {step + 1} — calling LLM...') + response = llm_client.chat(session.messages, tools=TOOLS) + + if response.tool_calls: + amsg = { + 'role': 'assistant', + 'content': response.content, + 'tool_calls': response.tool_calls, + } + session.messages.append(amsg) + + tnames = [tc['function']['name'] for tc in response.tool_calls] + print(f'[{_ts()}] Using tools: {", ".join(tnames)}') + + if on_tool_calls: + on_tool_calls(tnames) + + for tc in response.tool_calls: + result = execute_tool(tc, TOOL_HANDLERS) + session.messages.append({ + 'role': 'tool', + 'tool_call_id': tc['id'], + 'content': str(result), + }) + else: + if response.content: + print(f'[{_ts()}] Response generated ({len(response.content)} chars)') + session.messages.append({'role': 'assistant', 'content': response.content}) + return response.content + return None + + print(f'[{_ts()}] Max iterations ({max_iterations}) reached') + session.messages.append({ + 'role': 'assistant', + 'content': 'Max iterations reached without final answer.', + }) + return None diff --git a/services/telegram_client.py b/services/telegram_client.py new file mode 100644 index 0000000..14d76a6 --- /dev/null +++ b/services/telegram_client.py @@ -0,0 +1,275 @@ +import asyncio +import random +import re +import threading +import time +from datetime import datetime + +import config +from services.session_manager import SessionManager +from services.agent_loop import run_agent_loop +from scripts.persona import PERSONALITY +from tools.roleplayer import _name_mentioned + + +def _ts(): + return datetime.now().strftime('%H:%M:%S') + + +def _strip_bot_mention(text: str, bot_username: str) -> str: + if not bot_username: + return text + pattern = re.compile(r'@' + re.escape(bot_username), re.IGNORECASE) + return pattern.sub('', text).strip() + + +class TelegramClient: + def __init__(self, token, llm_client, tools_definition, TOOLS, + TOOL_HANDLERS, build_system_prompt, agent_max_iterations, + allowed_group_ids=None): + self._token = token + self._llm = llm_client + self._tools_def = tools_definition + self._TOOLS = TOOLS + self._TOOL_HANDLERS = TOOL_HANDLERS + self._build_system_prompt = build_system_prompt + self._max_iterations = agent_max_iterations + self._allowed_group_ids = allowed_group_ids or [] + self._skill = config.AGENT_SKILL + self._bot_username = "" + + self._session_mgr = SessionManager() + self._loop = None + self._stopped = None + + from telegram.ext import Application + self._app = Application.builder().token(token).build() + + def start(self): + print(f'[{_ts()}] Starting Telegram service...') + asyncio.run(self._async_run()) + + def stop(self): + if self._loop and not self._loop.is_closed(): + asyncio.run_coroutine_threadsafe(self._async_stop(), self._loop) + + async def _async_stop(self): + self._stopped.set() + + async def _async_run(self): + self._loop = asyncio.get_running_loop() + self._stopped = asyncio.Event() + + bot_user = await self._app.bot.get_me() + self._bot_username = bot_user.username or "" + print(f'[{_ts()}] Telegram bot: @{self._bot_username}') + + self._register_handlers() + + await self._app.initialize() + await self._app.start() + await self._app.updater.start_polling() + print(f'[{_ts()}] Telegram bot is polling') + + try: + await self._stopped.wait() + except (asyncio.CancelledError, KeyboardInterrupt): + pass + + await self._app.updater.stop() + await self._app.stop() + await self._app.shutdown() + print(f'[{_ts()}] Telegram service stopped') + + def _register_handlers(self): + from telegram.ext import MessageHandler, filters, CommandHandler + + self._app.add_handler( + MessageHandler( + filters.TEXT & ~filters.COMMAND & filters.ChatType.PRIVATE, + self._on_private_message + ) + ) + self._app.add_handler( + MessageHandler( + filters.TEXT & ~filters.COMMAND & filters.ChatType.GROUPS, + self._on_group_message + ) + ) + self._app.add_handler(CommandHandler('start', self._on_start_command)) + self._app.add_handler(CommandHandler('new', self._on_new_command)) + + async def _on_start_command(self, update, context): + await update.message.reply_text( + 'Halo! Aku adalah asisten AI. Kirim pesan untuk memulai percakapan.' + ) + + async def _on_new_command(self, update, context): + chat_id = str(update.effective_chat.id) + self._session_mgr.reset(chat_id) + await update.message.reply_text('Memulai sesi baru. Ada yang bisa dibantu?') + + async def _on_private_message(self, update, context): + chat_id = update.effective_chat.id + text = update.message.text.strip() + if not text: + return + msg_id = update.message.message_id + print(f'[{_ts()}] Telegram DM from {chat_id}: {text[:60]}') + threading.Thread( + target=self._process_message, + args=(chat_id, text, 'private', '', msg_id), + daemon=True + ).start() + + async def _on_group_message(self, update, context): + chat_id = update.effective_chat.id + chat_id_str = str(chat_id) + + if self._allowed_group_ids and chat_id_str not in self._allowed_group_ids: + return + + text = update.message.text.strip() + if not text: + return + + replied_to_bot = ( + update.message.reply_to_message + and update.message.reply_to_message.from_user + and update.message.reply_to_message.from_user.id == context.bot.id + ) + + has_mention = False + if update.message.entities: + for ent in update.message.entities: + if ent.type == 'mention' and self._bot_username: + start = ent.offset + end = ent.offset + ent.length + mention_text = update.message.text[start:end] + if mention_text.lower() == f'@{self._bot_username.lower()}': + has_mention = True + break + elif ent.type == 'text_mention' and ent.user.id == context.bot.id: + has_mention = True + break + + name_mentioned = _name_mentioned(PERSONALITY.name, text) + + if not replied_to_bot and not has_mention and not name_mentioned: + print(f'[{_ts()}] Telegram Group [{chat_id}] NO-REPLY: {text[:60]}') + return + + if has_mention and self._bot_username: + text = _strip_bot_mention(text, self._bot_username) + if not text: + return + + msg_id = update.message.message_id + sender = update.effective_user + sender_name = sender.full_name or sender.username or str(sender.id) if sender else str(chat_id) + print(f'[{_ts()}] Telegram Group [{chat_id}] <{sender_name}>: {text[:60]}') + threading.Thread( + target=self._process_message, + args=(chat_id, text, 'group', sender_name, msg_id), + daemon=True + ).start() + + def _process_message(self, chat_id, body, chat_type, sender_name, reply_to_msg_id): + session = self._session_mgr.get_or_create( + str(chat_id), self._build_system_prompt( + tools_definition=self._tools_def, + character=config.AGENT_CHARACTER or None, + skills=config.AGENT_SKILLS.split(",") if config.AGENT_SKILLS else None, + ) + ) + session.cancel_timer() + + if body in (':new', '/new'): + self._session_mgr.reset(str(chat_id)) + print(f'[{_ts()}] Session reset for {chat_id}') + self._schedule_send(chat_id, 'Memulai sesi baru. Ada yang bisa dibantu?', reply_to_msg_id) + return + + session.add_message('user', body) + is_roleplay = self._skill == 'roleplayer' + + asyncio.run_coroutine_threadsafe( + self._app.bot.send_chat_action(chat_id=chat_id, action='typing'), + self._loop + ) + + delay = random.uniform(config.READ_DELAY_MIN, config.READ_DELAY_MAX) + time.sleep(delay) + + def on_tool_calls(tnames): + info = f'Using: {", ".join(tnames)}' + print(f'[{_ts()}] {info}') + if not is_roleplay: + self._schedule_send(chat_id, info, reply_to_msg_id) + + final_content = run_agent_loop( + session, self._llm, self._TOOLS, self._TOOL_HANDLERS, + self._max_iterations, on_tool_calls=on_tool_calls + ) + + if final_content is not None: + if is_roleplay: + my_name = PERSONALITY.name + if config.TELEGRAM_SELECTIVE_RESPONSE: + recent_msgs = [] + for msg in session.messages[-6:]: + if msg.get('role') == 'user': + recent_msgs.append(f"User: {msg.get('content', '')}") + elif msg.get('role') == 'assistant' and msg.get('content'): + recent_msgs.append(f"{my_name}: {msg.get('content', '')}") + recent_history = "\n".join(recent_msgs) + + from tools.roleplayer import should_respond + if should_respond( + message=body, + sender_nickname=sender_name or str(chat_id), + recent_history=recent_history, + my_name=my_name, + ): + print(f'[{_ts()}] need_response=True → sending response') + self._schedule_send(chat_id, final_content, reply_to_msg_id) + else: + print(f'[{_ts()}] need_response=False → staying silent') + else: + from tools.roleplayer import _name_mentioned + if _name_mentioned(my_name, body): + print(f'[{_ts()}] Name mentioned → sending response') + self._schedule_send(chat_id, final_content, reply_to_msg_id) + else: + print(f'[{_ts()}] Name not mentioned → staying silent') + else: + self._schedule_send(chat_id, final_content, reply_to_msg_id) + else: + msg = 'Max iterations reached without final answer.' + self._schedule_send(chat_id, msg, reply_to_msg_id) + + timeout = 86400 if chat_type == 'private' else 300 + session.start_timer(timeout, self._timeout_session, chat_id) + + def _schedule_send(self, chat_id, text, reply_to_msg_id=None): + if self._loop and not self._loop.is_closed(): + char_count = len(text) if text else 0 + sleep_delay = max(1.0, min(char_count / config.TYPING_SPEED, config.TYPING_MAX)) + print(f'[{_ts()}] Typing delay: {sleep_delay:.1f}s ({char_count} chars)') + time.sleep(sleep_delay) + + asyncio.run_coroutine_threadsafe( + self._app.bot.send_message( + chat_id=chat_id, + text=text, + reply_to_message_id=reply_to_msg_id + ), + self._loop + ) + else: + print(f'[{_ts()}] WARNING: cannot send to {chat_id} — loop unavailable') + + def _timeout_session(self, chat_id): + print(f'[{_ts()}] Session timeout: {chat_id}') + self._schedule_send(chat_id, 'Sesi ditutup. Sampai jumpa') + self._session_mgr.reset(str(chat_id))