"""Debate Room Handler — AG 인스턴스 간 토론 중재. 흐름: 1. AG가 로컬 response.md에 전문 작성 2. Discord에 "작성 완료" 시그널 3. 사회자(봇)가 response.md 읽기 → NC transcript에 append → response.md 비우기 4. Discord #debate에 요약 게시 5. 상대 AG 채널에 지시 전달 """ import asyncio import logging import random from dataclasses import dataclass, field from pathlib import Path 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 } # ── 로컬 프로젝트 경로 ── AGENT_PATHS = { "gemini": Path(r"C:\Users\Variet-Worker\Desktop\debate_gemini"), "opus": Path(r"C:\Users\Variet-Worker\Desktop\debate_opus"), } # Embed 색상 / 이모지 AGENT_COLORS = {"gemini": 0x2ECC71, "opus": 0x3498DB} AGENT_EMOJI = {"gemini": "🟢", "opus": "🔵"} MAX_ROUNDS = 5 RESPONSE_TIMEOUT = 300 # 긴 답변 대기 (5분) FILE_POLL_INTERVAL = 5 # 파일 확인 간격 (초) @dataclass class DebateSession: topic: str = "" topic_short: str = "" # 파일명용 약어 round: int = 0 max_rounds: int = MAX_ROUNDS active: bool = False current_speaker: str = "" history: list = field(default_factory=list) 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 # 약어 생성 (첫 10자, 공백은 _) short = topic[:10].replace(" ", "_").replace("/", "_") self.session = DebateSession( topic=topic, topic_short=short, active=True, max_rounds=MAX_ROUNDS, ) # 각 AG 프로젝트에 response.md 초기화 for name, path in AGENT_PATHS.items(): resp_file = path / "response.md" resp_file.write_text("", encoding="utf-8") # 시작 알림 embed = discord.Embed( title="🏛️ Debate Room 시작", description=( f"**주제**: {topic}\n" f"**최대 라운드**: {MAX_ROUNDS}\n" f"**참여자**: 🟢 Gemini · 🔵 Opus\n\n" f"AG는 `response.md`에 답변을 작성합니다." ), 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 채널에서 메시지 수신 — response.md 파일 완성 시그널로 사용.""" if not self.session.active: 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 or agent_name != self.session.current_speaker: return # response.md에 내용이 있는지 확인 resp_file = AGENT_PATHS[agent_name] / "response.md" if resp_file.exists(): content = resp_file.read_text(encoding="utf-8").strip() if content: await self._process_response(agent_name, content) # ── 내부 로직 ── 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 # response.md 비우기 resp_file = AGENT_PATHS[speaker] / "response.md" resp_file.write_text("", encoding="utf-8") # 사회자(Flash)가 프롬프트 생성 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}` 답변 대기 중...\n" f"📝 `response.md`에 작성 후 Discord에 알려주세요." ) # AG 채널에 프롬프트 전송 (분할) agent_ch = self.bot.get_channel(DEBATE_AGENTS[speaker]) if not agent_ch: logger.error(f"토론 채널 접근 불가: {speaker}") return # 2000자씩 분할 전송 for i in range(0, len(prompt), 1900): await agent_ch.send(prompt[i:i+1900]) async def _process_response(self, speaker: str, content: str): """AG 응답 처리 — 기록 + 게시 + 파일 비우기.""" # 히스토리에 추가 self.session.history.append({ "speaker": speaker, "content": content, }) # response.md 비우기 resp_file = AGENT_PATHS[speaker] / "response.md" resp_file.write_text("", encoding="utf-8") # NC에 transcript 저장 (TODO: nc_client 연동) # await self._append_to_nc_transcript(speaker, content) # #variet-debate에 요약 게시 control_ch = self.bot.get_channel(DEBATE_CONTROL_CH) if control_ch: # 요약 (첫 500자) summary = content[:500] if len(content) > 500: summary += f"\n\n... (전문 {len(content)}자)" embed = discord.Embed( title=f"{AGENT_EMOJI[speaker]} Round {self.session.round} — {speaker}", description=summary, color=AGENT_COLORS[speaker], ) 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 토론 사회자입니다. ## 현재 상황 - 토론 주제: {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. 첫 발언이면 주제 설명 + 자유 의견 안내 6. **중요: 답변을 response.md 파일에 작성하라고 안내** (사회자 메시지만 출력) """ try: from core.gemini_caller import GeminiCaller caller = GeminiCaller() return await caller.call_simple(moderator_prompt, timeout=60) except Exception as e: logger.error(f"Flash 호출 실패: {e}") 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}"] 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{last['content']}") parts.append("\n---\n오류를 확인하고 개선점을 지적하세요.") if user_injected: parts.append(f"\n### 사용자 의견:\n{user_injected}") if not prev and not user_injected: parts.append("첫 번째 발언자입니다. 주제에 대해 의견을 제시하세요.") parts.append("\n\n**답변은 response.md 파일에 작성하세요.**") 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:]) ), color=0x9B59B6, ) await channel.send(embed=embed)