"""Debate Room Handler — AG 인스턴스 간 토론 중재. #variet-debate 채널에서 !debate-start로 시작. 사회자(이 봇)가 #ag-debate_gemini와 #ag-debate_opus 사이에서 턴을 관리. """ import asyncio import logging import random from dataclasses import dataclass, field from typing import Optional import discord logger = logging.getLogger("variet.debate") # ── 채널 ID ── DEBATE_CONTROL_CH = 1484167157620146227 # #variet-debate DEBATE_AGENTS = { "gemini": 1484156194187771905, # #ag-debate_gemini "opus": 1484156521209401476, # #ag-debate_opus } # Embed 색상 AGENT_COLORS = { "gemini": 0x2ECC71, # 초록 "opus": 0x3498DB, # 파랑 } AGENT_EMOJI = { "gemini": "🟢", "opus": "🔵", } MAX_ROUNDS = 5 RESPONSE_TIMEOUT = 180 # AG 응답 대기 (초) @dataclass class DebateSession: topic: str = "" round: int = 0 max_rounds: int = MAX_ROUNDS active: bool = False current_speaker: str = "" history: list = field(default_factory=list) _response_event: Optional[asyncio.Event] = field(default=None, repr=False) _last_response: str = "" class DebateHandler: def __init__(self, bot: discord.ext.commands.Bot): self.bot = bot self.session = DebateSession() # ── 공개 API ── async def start(self, ctx, topic: str): """토론 시작.""" if self.session.active: await ctx.reply("⚠️ 이미 진행 중인 토론이 있습니다. `!debate-stop`으로 먼저 종료하세요.") return self.session = DebateSession( topic=topic, active=True, max_rounds=MAX_ROUNDS, ) # 시작 알림 embed = discord.Embed( title="🏛️ Debate Room 시작", description=f"**주제**: {topic}\n**최대 라운드**: {MAX_ROUNDS}\n**참여자**: 🟢 Gemini · 🔵 Opus", color=0x9B59B6, ) control_ch = self.bot.get_channel(DEBATE_CONTROL_CH) if control_ch: await control_ch.send(embed=embed) # 첫 턴 시작 await self._next_turn() async def stop(self, ctx): """토론 중단.""" if not self.session.active: await ctx.reply("진행 중인 토론이 없습니다.") return self.session.active = False await self._post_conclusion(ctx.channel) async def inject(self, ctx, opinion: str): """사용자 의견 삽입.""" if not self.session.active: await ctx.reply("진행 중인 토론이 없습니다.") return self.session.history.append({ "speaker": "user", "content": opinion, }) embed = discord.Embed( title="👤 사용자 의견", description=opinion, color=0xF39C12, ) control_ch = self.bot.get_channel(DEBATE_CONTROL_CH) if control_ch: await control_ch.send(embed=embed) # 다음 턴에 반영 await self._next_turn(user_injected=opinion) async def on_agent_message(self, message: discord.Message): """AG 채널에서 응답 수신 시 호출.""" if not self.session.active: return if not self.session._response_event: return # 어떤 에이전트가 응답했는지 확인 agent_name = None for name, ch_id in DEBATE_AGENTS.items(): if message.channel.id == ch_id: agent_name = name break if not agent_name: return # 현재 대기 중인 에이전트의 응답만 처리 if agent_name != self.session.current_speaker: return # 응답 저장 (여러 메시지 합치기 - 마지막 메시지로) self.session._last_response = message.content self.session._response_event.set() # ── 내부 로직 ── async def _next_turn(self, user_injected: str = ""): """다음 턴 실행.""" if not self.session.active: return self.session.round += 1 if self.session.round > self.session.max_rounds: control_ch = self.bot.get_channel(DEBATE_CONTROL_CH) if control_ch: await control_ch.send("⏹️ 최대 라운드 도달. 토론을 종료합니다.") await self._post_conclusion(control_ch) return # 발언자 선택 (랜덤, 단 직전과 다른 에이전트) agents = list(DEBATE_AGENTS.keys()) if self.session.current_speaker: agents.remove(self.session.current_speaker) speaker = random.choice(agents) self.session.current_speaker = speaker # 프롬프트 구성 prompt = await self._build_prompt(speaker, user_injected) # 진행 상태 표시 control_ch = self.bot.get_channel(DEBATE_CONTROL_CH) if control_ch: await control_ch.send( f"{AGENT_EMOJI[speaker]} **Round {self.session.round}** — " f"`{speaker}` 발언 대기 중..." ) # AG 채널에 프롬프트 전송 agent_ch = self.bot.get_channel(DEBATE_AGENTS[speaker]) if not agent_ch: logger.error(f"토론 채널 접근 불가: {speaker}") return await agent_ch.send(prompt) # 응답 대기 self.session._response_event = asyncio.Event() self.session._last_response = "" try: # 첫 응답 후 추가 메시지가 있을 수 있으므로 약간의 딜레이 await asyncio.wait_for( self.session._response_event.wait(), timeout=RESPONSE_TIMEOUT, ) # AG가 여러 메시지를 보낼 수 있으므로 잠시 대기 await asyncio.sleep(5) except asyncio.TimeoutError: if control_ch: await control_ch.send(f"⏰ `{speaker}` 응답 시간 초과 ({RESPONSE_TIMEOUT}초)") return response = self.session._last_response if not response: return # 응답 기록 self.session.history.append({ "speaker": speaker, "content": response, }) # #variet-debate에 게시 if control_ch: embed = discord.Embed( title=f"{AGENT_EMOJI[speaker]} Round {self.session.round} — {speaker}", description=response[:4000], color=AGENT_COLORS[speaker], ) if len(response) > 4000: embed.add_field( name="(계속)", value=response[4000:4000+1024] + "...", inline=False, ) await control_ch.send(embed=embed) # 사용자에게 계속 진행 여부 물어보기 await control_ch.send( f"**다음 턴으로 진행할까요?**\n" f"✅ `!debate-next` — 계속\n" f"✏️ `!debate-inject 의견` — 의견 추가 후 계속\n" f"⏹️ `!debate-stop` — 종료" ) async def _build_prompt(self, speaker: str, user_injected: str = "") -> str: """사회자(Flash)가 에이전트에게 보낼 프롬프트를 생성.""" other = "opus" if speaker == "gemini" else "gemini" # 직전 상대 발언 가져오기 prev_opinions = [h for h in self.session.history if h["speaker"] != "user"] last_opponent = prev_opinions[-1] if prev_opinions else None # 사용자 발언 가져오기 user_opinions = [h for h in self.session.history if h["speaker"] == "user"] last_user = user_opinions[-1] if user_opinions else None # 사회자에게 보내는 메타 프롬프트 moderator_prompt = f"""당신은 AI 토론 사회자입니다. 두 AI 토론자({', '.join(DEBATE_AGENTS.keys())}) 사이에서 턴을 관리합니다. ## 현재 상황 - 토론 주제: {self.session.topic} - 현재 라운드: {self.session.round}/{self.session.max_rounds} - 다음 발언자: {speaker} - 상대: {other} """ if last_opponent: moderator_prompt += f""" ## 상대방({last_opponent['speaker']})의 직전 발언 (전문): {last_opponent['content']} """ if last_user: moderator_prompt += f""" ## 사용자의 의견/판단: {last_user['content']} """ if user_injected: moderator_prompt += f""" ## 사용자가 방금 추가한 의견: {user_injected} """ moderator_prompt += f""" ## 지시 다음 발언자({speaker})에게 보낼 메시지를 작성하세요. 규칙: 1. 상대방의 발언이 있으면 **전문을 그대로 포함**하세요 (요약하지 마세요) 2. 이 발언이 **사용자의 어떤 판단이나 의견에 기반**하는지 설명하세요 3. 상대 발언의 **오류나 논리적 허점을 확인**하고, 개선된 방향이나 잘못된 부분을 지적하라고 지시하세요 4. 현재 토론의 **핵심 쟁점**이 무엇인지 방향을 제시하세요 5. 첫 발언이면, 주제를 잘 설명하고 자유롭게 의견을 제시하라고 안내하세요 (사회자의 메시지만 출력하세요. 다른 설명은 불필요합니다.) """ try: from core.gemini_caller import GeminiCaller caller = GeminiCaller() moderated = await caller.call_simple(moderator_prompt, timeout=60) return moderated except Exception as e: logger.error(f"사회자 Flash 호출 실패: {e}, 기본 프롬프트 사용") # fallback: 기본 프롬프트 return self._build_fallback_prompt(speaker, user_injected) def _build_fallback_prompt(self, speaker: str, user_injected: str = "") -> str: """Flash 호출 실패 시 기본 프롬프트.""" parts = [f"## 토론 주제: {self.session.topic}\n"] parts.append(f"현재 라운드: {self.session.round}/{self.session.max_rounds}\n") prev = [h for h in self.session.history if h["speaker"] != "user"] if prev: last = prev[-1] parts.append(f"### 상대방({last['speaker']})의 직전 발언:\n") parts.append(last["content"]) parts.append("\n\n---\n") parts.append( "위 발언에 오류가 없는지 확인하고, " "더 개선된 방향이나 잘못된 부분을 지적하세요.\n" ) if user_injected: parts.append(f"\n### 사용자 의견:\n{user_injected}\n") if not prev and not user_injected: parts.append("당신이 첫 번째 발언자입니다. 주제에 대해 의견을 제시하세요.\n") return "\n".join(parts) async def _post_conclusion(self, channel): """토론 종료 — 결론 정리.""" self.session.active = False summary_parts = [] for h in self.session.history: emoji = AGENT_EMOJI.get(h["speaker"], "👤") content = h["content"][:300] summary_parts.append(f"{emoji} **{h['speaker']}**: {content}...") embed = discord.Embed( title="🏛️ 토론 종료", description=f"**주제**: {self.session.topic}\n" f"**라운드**: {self.session.round}\n\n" + "\n\n".join(summary_parts[-6:]), # 최근 6개만 color=0x9B59B6, ) await channel.send(embed=embed)