"""다중 모델 통합 봇 (Party Bot) Gemini, Codex, Claude 봇을 하나의 Python 프로세스에서 구동하며, 대화 기록(chat history)을 공유합니다. 봇들이 서로를 멘션하면(예: "코덱스야, 네 생각은 어때?"), 해당 봇이 자동으로 대화를 이어받아 토론(의견 조율)을 할 수 있습니다. """ import asyncio import logging from typing import Any, Dict, List, Set from bot_utils import send_response, should_respond, split_message from engine_v2.bot_api import call_claude, call_codex, call_gemini from telegram import Update from telegram.constants import ChatAction from telegram.ext import Application, ContextTypes, MessageHandler, filters from config import CLAUDE_BOT_TOKEN, CODEX_BOT_TOKEN, GEMINI_BOT_TOKEN, MAX_MESSAGE_LENGTH, OWNER_USER_ID logging.basicConfig( format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO, ) logger = logging.getLogger(__name__) chat_history: Dict[int, List[Dict[str, str]]] = {} seen_message_ids: Set[int] = set() MAX_HISTORY = 10 BOT_CONFIGS = { "gemini": { "token": GEMINI_BOT_TOKEN, "name": "잼민이(Gemini)", "keywords": ["잼민", "제미나이", "gemini", "모두", "다같이", "너희들", "얘들아"], "call_func": call_gemini, "app": None, }, "codex": { "token": CODEX_BOT_TOKEN, "name": "코덱스(Codex)", "keywords": ["코덱스", "codex", "모두", "다같이", "너희들", "얘들아"], "call_func": call_codex, "app": None, }, "claude": { "token": CLAUDE_BOT_TOKEN, "name": "클로디(Claude)", "keywords": ["클로드", "클로디", "claude", "모두", "다같이", "너희들", "얘들아"], "call_func": call_claude, "app": None, }, } def build_prompt(chat_id: int, bot_name: str, new_user_text: str | None = None) -> str: history = chat_history.get(chat_id, []) lines = [ f"너는 지금 텔레그램 그룹 채팅방에서 사용자 및 다른 AI들과 함께 대화하고 있어. 너의 이름(역할)은 '{bot_name}'야.", "자연스러운 말투로 대화에 참여하고, 필요한 경우 다른 AI(잼민이, 코덱스, 클로디)를 이름으로 불러서 의견을 물어보거나 조율할 수 있어.", "", ] if history: lines.append("[이전 대화 기록]") for msg in history[-MAX_HISTORY:]: lines.append(f"{msg['role']}: {msg['content']}") lines.append("") if new_user_text: lines.append(f"[현재 입력 - User]: {new_user_text}\n") else: lines.append("[시스템]: 위 대화 흐름에 이어서 네 의견을 말해줘.\n") lines.append(f"자 이제 {bot_name}의 답변을 말해줘:") return "\n".join(lines) def check_triggers(text: str) -> List[str]: triggered = [] text_lower = text.lower() for bot_id, config in BOT_CONFIGS.items(): unique_kws = [kw for kw in config["keywords"] if kw not in ["모두", "다같이", "너희들", "얘들아"]] if any(kw in text_lower for kw in unique_kws): triggered.append(bot_id) return triggered async def trigger_bot_internally(chat_id: int, target_bot_id: str, trigger_text: str): await asyncio.sleep(2) config = BOT_CONFIGS[target_bot_id] app = config["app"] if not app: return logger.info(f"내부 트리거 발동: {target_bot_id} in chat {chat_id}") try: await app.bot.send_chat_action(chat_id=chat_id, action=ChatAction.TYPING) prompt = build_prompt(chat_id, config["name"], new_user_text=None) result = await config["call_func"](prompt) if chat_id not in chat_history: chat_history[chat_id] = [] chat_history[chat_id].append({"role": config["name"], "content": result}) chunks = split_message(result, MAX_MESSAGE_LENGTH) for chunk in chunks: await app.bot.send_message(chat_id=chat_id, text=chunk) next_targets = check_triggers(result) for next_id in next_targets: if next_id != target_bot_id: asyncio.create_task(trigger_bot_internally(chat_id, next_id, result)) except Exception as e: logger.error(f"내부 트리거 오류 ({target_bot_id}): {e}") def create_handler(bot_id: str): async def handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not update.message or not update.message.from_user: return if update.message.from_user.id != OWNER_USER_ID: return bot_username = context.bot.username or "" if update.effective_chat is None: return chat_id = update.effective_chat.id msg_id = update.message.message_id user_text = (update.message.text or "").strip() if not user_text: return config = BOT_CONFIGS[bot_id] is_called = False text_lower = user_text.lower() if update.effective_chat.type == "private": is_called = True else: if should_respond(update, bot_username, OWNER_USER_ID): is_called = True elif any(kw in text_lower for kw in config["keywords"]): is_called = True if msg_id not in seen_message_ids: seen_message_ids.add(msg_id) if chat_id not in chat_history: chat_history[chat_id] = [] chat_history[chat_id].append({"role": "User", "content": user_text}) if not is_called: return try: await context.bot.send_chat_action(chat_id=chat_id, action=ChatAction.TYPING) prompt = build_prompt(chat_id, config["name"], new_user_text=user_text) result = await config["call_func"](prompt) chat_history[chat_id].append({"role": config["name"], "content": result}) await send_response(update, context, result, max_len=MAX_MESSAGE_LENGTH) next_targets = check_triggers(result) for next_id in next_targets: if next_id != bot_id: asyncio.create_task(trigger_bot_internally(chat_id, next_id, result)) except Exception as e: logger.error(f"{bot_id} handle_message 오류: {e}") return handler async def main_async(): apps = [] for bot_id, config in BOT_CONFIGS.items(): if not config["token"]: logger.warning(f"{bot_id} 토큰이 없습니다. 건너뜁니다.") continue app = Application.builder().token(config["token"]).build() app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, create_handler(bot_id))) config["app"] = app apps.append(app) if not apps: logger.error("실행할 봇이 없습니다.") return logger.info("다중 모델 통합 봇 시작 중...") for app in apps: await app.initialize() await app.start() await app.updater.start_polling() logger.info("모든 봇이 폴링을 시작했습니다.") stop_event = asyncio.Event() try: await stop_event.wait() except KeyboardInterrupt: pass finally: for app in apps: await app.updater.stop() await app.stop() await app.shutdown() def main(): asyncio.run(main_async()) if __name__ == "__main__": main()