#!/usr/bin/env python3
"""
group_chat.py — 팀 단톡 데몬 스크립트
트리거 파일 폴링으로 멀티 페르소나 단톡을 구동합니다.
"""
import json
import logging
import os
import signal
import subprocess
import sys
import time
from datetime import datetime
from pathlib import Path
import requests
# ── 경로 상수 ──────────────────────────────────────────────────
WORKSPACE = "/home/jay/workspace"
TRIGGER_FILE = f"{WORKSPACE}/memory/events/group_chat_trigger.json"
SESSION_FILE = f"{WORKSPACE}/memory/events/group_chat_session.json"
DAILY_DIR = f"{WORKSPACE}/memory/daily"
PERSONAS_FILE = f"{WORKSPACE}/config/personas.json"
ORG_STRUCTURE_FILE = f"{WORKSPACE}/memory/organization-structure.json"
ENV_KEYS_FILE = f"{WORKSPACE}/.env.keys"
DEFAULT_CHAT_ID = "6937032012"
POLL_INTERVAL = 1 # 초
SESSION_TIMEOUT = 300 # 5분
MAX_AUTO_TURNS = 6
# ── 자연어 트리거 키워드 ──────────────────────────────────────
START_KEYWORDS = ["팀 모여", "회의하자", "단톡 시작", "모여봐", "전원 집합", "소집"]
END_KEYWORDS = ["잘래", "빠이", "해산", "끝", "그만", "여기까지"]
CONTROL_TYPES = {
"limit_personas": "인원 수 제한",
"add_persona": "인원 추가",
"remove_persona": "인원 퇴장",
"filter_by_role": "역할별 필터",
"filter_by_team": "팀별 필터",
"set_auto_turns": "자동 발화 수 변경",
}
ROLE_FILTERS = {
"백엔드": ["vulcan", "thor", "anubis"],
"프론트": ["iris", "freya", "isis"],
"프론트엔드": ["iris", "freya", "isis"],
"UX": ["athena", "mimir", "sobek"],
"테스터": ["argos", "heimdall", "horus"],
"팀장": ["hermes", "odin", "ra"],
}
TEAM_FILTERS = {
"1팀": ["hermes", "vulcan", "iris", "athena", "argos"],
"개발1팀": ["hermes", "vulcan", "iris", "athena", "argos"],
"2팀": ["odin", "thor", "freya", "mimir", "heimdall"],
"개발2팀": ["odin", "thor", "freya", "mimir", "heimdall"],
"3팀": ["ra", "anubis", "isis", "sobek", "horus"],
"개발3팀": ["ra", "anubis", "isis", "sobek", "horus"],
}
# ── 전체 페르소나 ID 목록 (detect_intent에서 사용) ────────────
ALL_PERSONA_IDS = [
"hermes",
"vulcan",
"iris",
"athena",
"argos",
"odin",
"thor",
"freya",
"mimir",
"heimdall",
"ra",
"anubis",
"isis",
"sobek",
"horus",
"loki",
"maat",
"janus",
"venus",
]
# ── 한국어 이름 → 페르소나 ID 매핑 ──────────────────────────
_PERSONA_NAMES = {
"헤르메스": "hermes",
"불칸": "vulcan",
"이리스": "iris",
"아테나": "athena",
"아르고스": "argos",
"오딘": "odin",
"토르": "thor",
"프레이야": "freya",
"미미르": "mimir",
"헤임달": "heimdall",
"라": "ra",
"아누비스": "anubis",
"이시스": "isis",
"소베크": "sobek",
"호루스": "horus",
"로키": "loki",
"마아트": "maat",
"야누스": "janus",
"비너스": "venus",
}
# ── 이모지 매핑 (코드 내 상수) ────────────────────────────────
EMOJI_MAP = {
"hermes": "⚡",
"odin": "👁️",
"ra": "☀️",
"vulcan": "🔥",
"iris": "🌈",
"athena": "🏛️",
"argos": "🔍",
"thor": "🔨",
"freya": "✨",
"mimir": "📚",
"heimdall": "🛡️",
"anubis": "⚱️",
"isis": "🌙",
"sobek": "🐊",
"horus": "🦅",
"loki": "🎭",
"venus": "🎨",
"janus": "🚪",
"maat": "⚖️",
}
# ── 로깅 설정 ──────────────────────────────────────────────────
logging.basicConfig(
level=logging.INFO,
format="[%(asctime)s] [%(levelname)s] %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger(__name__)
# ── 기본 페르소나 정의 (조직도 + personas.json 모두 실패 시 폴백) ──
DEFAULT_PERSONAS = {
"hermes": {
"name": "헤르메스",
"team": "개발1팀",
"role": "개발1팀장",
"expertise": "백엔드 아키텍처, 시스템 설계, 코드 리뷰",
"personality": "냉철하고 실용적, 효율을 중시하며 직설적으로 말함",
"persona_desc": "",
},
"athena": {
"name": "아테나",
"team": "개발1팀",
"role": "UX/UI 설계자",
"expertise": "UX/UI 설계, 사용자 경험, 프로토타이핑",
"personality": "분석적이고 꼼꼼함, 사용자 관점을 항상 강조함",
"persona_desc": "",
},
"thor": {
"name": "토르",
"team": "개발2팀",
"role": "백엔드 개발자",
"expertise": "백엔드 개발, 성능 최적화, 데이터베이스",
"personality": "열정적이고 추진력 강함, 문제 해결에 적극적",
"persona_desc": "",
},
"vulcan": {
"name": "불칸",
"team": "개발1팀",
"role": "백엔드 개발자",
"expertise": "백엔드 개발, API 설계, 인프라",
"personality": "실용주의적, 코드 품질을 중시하며 간결하게 말함",
"persona_desc": "",
},
"iris": {
"name": "이리스",
"team": "개발1팀",
"role": "프론트엔드 개발자",
"expertise": "프론트엔드 개발, React, UI 구현",
"personality": "창의적이고 활발함, 새로운 기술 도입에 적극적",
"persona_desc": "",
},
"odin": {
"name": "오딘",
"team": "개발2팀",
"role": "개발2팀장",
"expertise": "시스템 아키텍처, 기술 전략, 팀 리딩",
"personality": "지혜롭고 신중함, 큰 그림을 보며 전략적으로 접근함",
"persona_desc": "",
},
"loki": {
"name": "로키",
"team": "보안팀",
"role": "보안팀 리더",
"expertise": "보안 검토, 취약점 분석, 리스크 평가",
"personality": "비판적이고 도전적, 허점을 찾아내는 데 탁월함",
"persona_desc": "",
},
}
# ── Claude CLI 호출 ──────────────────────────────────────────────
CLAUDE_CLI_PATH = "/home/jay/.local/bin/claude"
def call_claude(
system_prompt: str,
user_message: str,
max_tokens: int = 300,
model: str = "claude-sonnet-4-6",
) -> str:
"""Claude CLI를 subprocess로 호출하여 응답 생성."""
cmd = [
CLAUDE_CLI_PATH,
"--print",
"--model",
model,
]
if system_prompt:
cmd.extend(["--system-prompt", system_prompt])
result = subprocess.run(
cmd,
input=user_message,
capture_output=True,
text=True,
timeout=30,
)
if result.returncode != 0:
raise RuntimeError(f"Claude CLI 에러: {result.stderr[:200]}")
return result.stdout.strip()
# ── Telegram getUpdates 폴링 ─────────────────────────────────
class TelegramPoller:
"""Telegram Bot API getUpdates 폴링으로 유저 메시지 수신."""
def __init__(self, token: str, chat_id: str):
self.token = token
self.chat_id = chat_id
self.base_url = f"https://api.telegram.org/bot{token}"
self.last_update_id = 0
def get_updates(self) -> list:
"""새 메시지 가져오기."""
try:
resp = requests.get(
f"{self.base_url}/getUpdates",
params={"offset": self.last_update_id + 1, "timeout": 1},
timeout=5,
)
data = resp.json()
if not data.get("ok"):
return []
messages = []
for update in data.get("result", []):
self.last_update_id = update["update_id"]
msg = update.get("message", {})
# 지정된 chat_id에서 온 메시지만 처리
if str(msg.get("chat", {}).get("id")) == self.chat_id:
text = msg.get("text", "")
if text:
messages.append(text)
return messages
except Exception as e:
logger.warning(f"getUpdates 오류: {e}")
return []
def format_history_for_cli(history: list, last_n: int = 20) -> str:
"""대화 히스토리를 CLI 입력용 텍스트로 변환."""
lines = []
for msg in history[-last_n:]:
speaker = msg.get("speaker", "")
speaker_name = msg.get("speaker_name", "")
content = msg.get("content", "")
if speaker_name:
lines.append(f"[{speaker_name}]: {content}")
elif speaker == "user":
lines.append(f"[제이회장님]: {content}")
else:
lines.append(content)
return "\n".join(lines)
# ── 자연어 의도 감지 (하이브리드: 키워드 + Claude) ──────────
def detect_intent(user_message: str, session_active: bool) -> dict:
"""유저 메시지의 의도를 감지한다.
Returns:
{"intent": "start_chat", "topic": "...", "personas": [...]}
{"intent": "user_input", "message": "..."}
{"intent": "end_chat"}
{"intent": "control", "type": "...", ...}
{"intent": "none"}
"""
import re as _re
# 1차: 키워드 매칭 (즉시)
if not session_active:
if any(kw in user_message for kw in START_KEYWORDS):
# 자연어 인원 지정
personas = None
if "전원" in user_message or "다 모여" in user_message:
personas = list(ALL_PERSONA_IDS)
else:
for role_name, pids in ROLE_FILTERS.items():
if role_name in user_message:
personas = list(pids)
break
if personas is None:
for team_name, pids in TEAM_FILTERS.items():
if team_name in user_message:
personas = list(pids)
break
if personas is None:
personas = ["hermes", "athena", "thor"]
# 인원 수 제한
limit_match = _re.search(r"(\d+)명", user_message)
if limit_match:
count = int(limit_match.group(1))
if count > 0:
personas = personas[:count]
return {
"intent": "start_chat",
"topic": "자유 대화",
"personas": personas,
}
else:
if any(kw in user_message for kw in END_KEYWORDS):
return {"intent": "end_chat"}
# 제어 명령 키워드 매칭 (즉시 반응, Claude 호출 없음)
# 인원 제한: "N명만", "N명까지만", "N명까지"
limit_match = _re.search(r"(\d+)명(?:만|까지만|까지)", user_message)
if limit_match:
return {"intent": "control", "type": "limit_personas", "count": int(limit_match.group(1))}
# 인원 퇴장/추가: "{이름} 빠져/불러" 등
for name, pid in _PERSONA_NAMES.items():
if name in user_message:
if any(kw in user_message for kw in ["빠져", "나가", "퇴장", "빠지", "나가라"]):
return {"intent": "control", "type": "remove_persona", "persona": pid}
if any(kw in user_message for kw in ["불러", "합류", "들어와", "참여"]):
return {"intent": "control", "type": "add_persona", "persona": pid}
# 역할/팀 필터
for role_name, persona_ids in ROLE_FILTERS.items():
if role_name in user_message and any(kw in user_message for kw in ["만", "만 남아", "만 얘기", "만 모여"]):
return {"intent": "control", "type": "filter_by_role", "personas": persona_ids, "role": role_name}
for team_name, persona_ids in TEAM_FILTERS.items():
if team_name in user_message and any(kw in user_message for kw in ["만", "만 남아", "만 얘기", "만 모여"]):
return {"intent": "control", "type": "filter_by_team", "personas": persona_ids, "team": team_name}
# 자동 발화 수 변경: "N턴까지만", "계속 얘기해"
turn_match = _re.search(r"(\d+)턴", user_message)
if turn_match:
return {"intent": "control", "type": "set_auto_turns", "count": int(turn_match.group(1))}
if any(kw in user_message for kw in ["계속 얘기", "계속 대화", "멈추지 마"]):
return {"intent": "control", "type": "set_auto_turns", "count": 99}
# 2차: 모호한 경우 Claude 호출
if session_active:
prompt = f"""유저 메시지를 분석하세요.
현재 팀 단톡 세션이 진행 중입니다.
유저 메시지: "{user_message}"
아래 중 하나를 JSON으로 응답:
1. 종료 의도 ("잘래", "빠이", "해산", "끝", "그만" 등): {{"intent": "end_chat"}}
2. 대화 참여 (질문, 의견, 지시 등): {{"intent": "user_input", "message": "{user_message}"}}
주의: "너 잘래?" 같은 질문은 종료가 아님. "나 잘래", "오늘은 여기까지" 같은 1인칭 표현만 종료.
JSON만 응답:"""
else:
persona_list = ", ".join(ALL_PERSONA_IDS)
prompt = f"""유저 메시지를 분석하세요.
현재 팀 단톡 세션이 없습니다.
유저 메시지: "{user_message}"
아래 중 하나를 JSON으로 응답:
1. 단톡 시작 의도 ("팀 모여", "회의하자", "단톡 시작", "모여봐", "전원 집합" 등):
{{"intent": "start_chat", "topic": "추정 주제", "personas": ["hermes", "athena", "thor"]}}
- personas: 주제에 맞는 페르소나 3~5명 선택 (전체 소집이면 전원)
2. 단톡과 무관한 메시지 ("안녕", "하이", 일반 대화 등):
{{"intent": "none"}}
전체 페르소나 목록: {persona_list}
JSON만 응답:"""
try:
result = call_claude("", prompt, model="claude-haiku-4-5-20251001")
json_match = _re.search(r"\{.*\}", result, _re.DOTALL)
if json_match:
parsed = json.loads(json_match.group())
# 유효성 검사
intent = parsed.get("intent", "none")
if intent in ("start_chat", "user_input", "end_chat", "none"):
return parsed
except Exception as e:
logger.warning(f"의도 감지 오류: {e}")
# 세션 활성 중이면 user_input으로 폴백, 아니면 none
if session_active:
return {"intent": "user_input", "message": user_message}
return {"intent": "none"}
# ── 환경변수 로드 ──────────────────────────────────────────────
def load_env_keys():
"""GROUP_CHAT_BOT_TOKEN이 없으면 .env.keys에서 로드."""
if os.environ.get("GROUP_CHAT_BOT_TOKEN"):
return
env_path = Path(ENV_KEYS_FILE)
if not env_path.exists():
logger.warning(f".env.keys 파일 없음: {ENV_KEYS_FILE}")
return
try:
result = subprocess.run(
["bash", "-c", f"source {ENV_KEYS_FILE} && env"],
capture_output=True,
text=True,
)
for line in result.stdout.splitlines():
if "=" in line:
k, _, v = line.partition("=")
if k.strip() == "GROUP_CHAT_BOT_TOKEN":
os.environ["GROUP_CHAT_BOT_TOKEN"] = v.strip()
logger.info("GROUP_CHAT_BOT_TOKEN을 .env.keys에서 로드했습니다.")
return
except Exception as e:
logger.error(f"env.keys 로드 실패: {e}")
# ── 봇 토큰 로드 ──────────────────────────────────────────────
def load_bot_token() -> str:
"""GROUP_CHAT_BOT_TOKEN 환경변수에서 봇 토큰 로드. 폴백: .env.keys 파싱."""
token = os.environ.get("GROUP_CHAT_BOT_TOKEN")
if token:
logger.info("봇 토큰을 환경변수에서 로드했습니다.")
return token
# 폴백: .env.keys에서 파싱
env_path = Path(ENV_KEYS_FILE)
if env_path.exists():
try:
result = subprocess.run(
["bash", "-c", f"source {ENV_KEYS_FILE} && env"],
capture_output=True,
text=True,
)
for line in result.stdout.splitlines():
if "=" in line:
k, _, v = line.partition("=")
if k.strip() == "GROUP_CHAT_BOT_TOKEN":
logger.info("봇 토큰을 .env.keys에서 로드했습니다.")
return v.strip()
except Exception as e:
logger.error(f"봇 토큰 .env.keys 로드 실패: {e}")
raise RuntimeError("GROUP_CHAT_BOT_TOKEN을 환경변수 또는 .env.keys에서 찾을 수 없습니다.")
# ── 조직도 기반 페르소나 로드 ─────────────────────────────────
def load_personas_from_org() -> dict:
"""organization-structure.json에서 페르소나 동적 로드."""
org_path = Path(ORG_STRUCTURE_FILE)
with open(org_path, "r", encoding="utf-8") as f:
org = json.load(f)
personas = {}
# 1. 개발실 하위 팀들 (sub_teams) + 레드팀
for team in org["structure"]["columns"]["teams"]:
if team.get("status") != "active":
continue
if team.get("team_id") == "development-office":
for sub_team in team.get("sub_teams", []):
if sub_team.get("status") != "active":
continue
team_name = sub_team["sub_team_name"]
# 팀장
lead = sub_team.get("lead")
if lead and lead.get("id") != "anu" and lead.get("status") == "active":
personas[lead["id"]] = {
"name": lead["name"].split(" (")[0],
"team": team_name,
"role": lead["role"],
"expertise": lead.get("expertise", {}).get("primary", ""),
"personality": lead.get("expertise", {}).get("style", ""),
"persona_desc": lead.get("persona", ""),
}
# 멤버
for member in sub_team.get("members", []):
if member.get("id") == "anu":
continue
if member.get("status") in ("active", "available"):
personas[member["id"]] = {
"name": member["name"].split(" (")[0],
"team": team_name,
"role": member["role"],
"expertise": member.get("expertise", {}).get("primary", ""),
"personality": member.get("expertise", {}).get("style", ""),
"persona_desc": member.get("persona", ""),
}
else:
# 2. 모든 다른 active 팀 (security-team, marketing-team, consulting-team, publishing-team 등)
team_name = team.get("team_name", team.get("team_id", ""))
# lead 처리 (status가 없으면 active로 간주 - 팀 자체가 active이므로)
lead = team.get("lead")
if lead and lead.get("id") != "anu" and lead.get("name"):
lead_status = lead.get("status", "active")
if lead_status in ("active", "available"):
personas[lead["id"]] = {
"name": lead["name"].split(" (")[0],
"team": team_name,
"role": lead["role"],
"expertise": lead.get("expertise", {}).get("primary", ""),
"personality": lead.get("expertise", {}).get("style", ""),
"persona_desc": lead.get("persona", ""),
}
# members 처리
for member in team.get("members", []):
if member.get("id") == "anu":
continue
member_status = member.get("status", "active")
if member_status in ("active", "available"):
personas[member["id"]] = {
"name": member["name"].split(" (")[0],
"team": team_name,
"role": member["role"],
"expertise": member.get("expertise", {}).get("primary", ""),
"personality": member.get("expertise", {}).get("style", ""),
"persona_desc": member.get("persona", ""),
}
# 3. 횡단조직 (QC, DevOps, 디자인 등)
for center in org["structure"]["rows"]["centers"]:
if center.get("status") != "active":
continue
lead = center.get("lead")
if lead and lead.get("id") != "anu" and lead.get("status") in ("active", "available"):
personas[lead["id"]] = {
"name": lead["name"].split(" (")[0],
"team": center["center_name"],
"role": lead["role"],
"expertise": lead.get("expertise", {}).get("primary", ""),
"personality": lead.get("expertise", {}).get("style", ""),
"persona_desc": lead.get("persona", ""),
}
# center members 처리
for member in center.get("members", []):
if member.get("id") == "anu":
continue
member_status = member.get("status", "active")
if member_status in ("active", "available"):
personas[member["id"]] = {
"name": member["name"].split(" (")[0],
"team": center["center_name"],
"role": member["role"],
"expertise": member.get("expertise", {}).get("primary", ""),
"personality": member.get("expertise", {}).get("style", ""),
"persona_desc": member.get("persona", ""),
}
return personas
# ── 페르소나 로드 (조직도 우선, personas.json 폴백) ───────────
def load_personas() -> dict:
"""조직도에서 페르소나를 로드하고, 실패 시 personas.json → DEFAULT_PERSONAS 순으로 폴백."""
# 1차: 조직도에서 동적 로드
try:
personas = load_personas_from_org()
if personas:
logger.info(f"페르소나 {len(personas)}명을 조직도에서 로드했습니다.")
return personas
except Exception as e:
logger.warning(f"조직도 로드 실패, personas.json으로 폴백: {e}")
# 2차: config/personas.json 폴백
path = Path(PERSONAS_FILE)
if path.exists():
try:
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
logger.info(f"페르소나 설정 로드 (폴백): {PERSONAS_FILE}")
return data
except Exception as e:
logger.warning(f"personas.json 로드 실패, 기본값 사용: {e}")
# 3차: 하드코딩 기본값
return DEFAULT_PERSONAS
# ── 프로필 태그 포맷 ─────────────────────────────────────────
def format_persona_tag(persona_key: str, personas_data: dict) -> str:
"""페르소나 키로부터 '⚡ 헤르메스(개발1팀/개발1팀장)' 형태의 태그 생성."""
p = personas_data.get(persona_key, {"name": persona_key, "team": "", "role": ""})
emoji = EMOJI_MAP.get(persona_key, "💬")
team = p.get("team", "")
role = p.get("role", "")
name = p.get("name", persona_key)
if team and role:
return f"{emoji} {name}({team}/{role})"
elif role:
return f"{emoji} {name}({role})"
else:
return f"{emoji} {name}"
# ── Telegram 전송 ─────────────────────────────────────────────
def send_telegram(token: str, chat_id: str, text: str):
url = f"https://api.telegram.org/bot{token}/sendMessage"
payload = {"chat_id": chat_id, "text": text, "parse_mode": "HTML"}
try:
resp = requests.post(url, json=payload, timeout=10)
if not resp.ok:
logger.error(f"Telegram 전송 실패: {resp.status_code} {resp.text[:200]}")
except Exception as e:
logger.error(f"Telegram 요청 오류: {e}")
time.sleep(0.5)
# ── 트리거 파일 읽기/삭제 ─────────────────────────────────────
def read_trigger() -> dict | None:
path = Path(TRIGGER_FILE)
if not path.exists():
return None
try:
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
path.unlink()
return data
except Exception as e:
logger.error(f"트리거 파일 읽기 오류: {e}")
return None
# ── 세션 상태 파일 덤프/로드 ─────────────────────────────────
def dump_session(session: dict):
try:
with open(SESSION_FILE, "w", encoding="utf-8") as f:
json.dump(session, f, ensure_ascii=False, indent=2)
except Exception as e:
logger.error(f"세션 덤프 오류: {e}")
def load_session() -> dict | None:
path = Path(SESSION_FILE)
if not path.exists():
return None
try:
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
if data.get("active"):
logger.info("기존 세션 복구됨.")
return data
except Exception as e:
logger.error(f"세션 로드 오류: {e}")
return None
# ── 세션 로그 저장 ────────────────────────────────────────────
def save_session_log(session: dict):
daily_dir = Path(DAILY_DIR)
daily_dir.mkdir(parents=True, exist_ok=True)
today = datetime.now().strftime("%Y-%m-%d")
log_path = daily_dir / f"{today}.md"
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
n_personas = len(session.get("personas", []))
total_turns = sum(session.get("speak_counts", {}).values())
entry = (
f"\n## 단톡 세션 [{timestamp}]\n"
f"- 주제: {session.get('topic', '없음')}\n"
f"- 참여: {n_personas}인\n"
f"- 총 발화: {total_turns}회\n"
f"- 참여자: {', '.join(session.get('personas', []))}\n\n"
)
for msg in session.get("history", []):
speaker = msg.get("speaker", "unknown")
content = msg.get("content", "")
entry += f"**{speaker}**: {content}\n\n"
try:
with open(log_path, "a", encoding="utf-8") as f:
f.write(entry)
logger.info(f"세션 로그 저장: {log_path}")
except Exception as e:
logger.error(f"세션 로그 저장 실패: {e}")
# ── Claude CLI: 페르소나 응답 생성 ───────────────────────────
def generate_persona_response(
persona_key: str,
persona: dict,
history: list,
topic: str,
) -> str:
name = persona.get("name", persona_key)
team = persona.get("team", "")
role = persona.get("role", "")
persona_desc = persona.get("persona_desc", "")
expertise = persona.get("expertise", "")
personality = persona.get("personality", "")
system_prompt = f"당신은 {name}입니다."
if team and role:
system_prompt += f" {team}의 {role}입니다."
elif role:
system_prompt += f" 역할: {role}."
system_prompt += "\n"
if persona_desc:
system_prompt += f"{persona_desc}\n"
if expertise:
system_prompt += f"전문 분야: {expertise}\n"
if personality:
system_prompt += f"성격: {personality}\n"
system_prompt += (
f"\n현재 팀 단톡 주제: {topic}\n\n"
"규칙:\n"
"- 2-4문장 이내 짧게 답하세요.\n"
"- 한국어로 대화하세요.\n"
"- 유저(제이회장님)를 '제이회장님'이라 호칭하세요.\n"
"- 다른 팀원은 이름으로 직접 호칭하세요.\n"
"- 자연스러운 대화체를 사용하세요.\n"
"- 자신의 전문성을 살린 실질적인 의견을 말하세요."
)
history_text = format_history_for_cli(history, last_n=10)
if not history_text:
history_text = f"단톡이 시작되었습니다. 주제: {topic}"
try:
return call_claude(system_prompt, history_text, max_tokens=300)
except (RuntimeError, subprocess.TimeoutExpired):
time.sleep(2)
return call_claude(system_prompt, history_text, max_tokens=300)
# ── Claude CLI: 다음 발화자 선택 ────────────────────────────
def select_next_speaker(
persona_names: list,
history: list,
last_speaker: str,
speak_counts: dict,
) -> str:
last_5 = history[-5:] if len(history) >= 5 else history
history_text = "\n".join(f"[{m.get('speaker_name', m.get('speaker', ''))}]: {m.get('content', '')}" for m in last_5)
speaker_prompt = (
f"현재 대화 참여자: {', '.join(persona_names)}\n"
f"대화 히스토리:\n{history_text}\n"
f"직전 발화자: {last_speaker}\n"
f"각 발화 횟수: {speak_counts}\n\n"
"다음에 가장 자연스럽게 말할 참여자 1명을 선택하세요.\n"
f'JSON으로 응답: {{"next_speaker": "persona_key", "reason": "한줄이유"}}\n'
f"persona_key는 반드시 다음 중 하나여야 합니다: {list(speak_counts.keys())}"
)
try:
text = call_claude("", speaker_prompt, max_tokens=50, model="claude-haiku-4-5-20251001")
start = text.find("{")
end = text.rfind("}") + 1
if start >= 0 and end > start:
data = json.loads(text[start:end])
speaker_key = data.get("next_speaker", "")
if speaker_key in speak_counts:
return speaker_key
except Exception as e:
logger.error(f"발화자 선택 오류: {e}")
# 폴백: 발화 횟수 적은 순 (직전 발화자 제외)
candidates = [k for k in speak_counts if k != last_speaker]
if not candidates:
candidates = list(speak_counts.keys())
return min(candidates, key=lambda k: speak_counts.get(k, 0))
# ── 단톡 세션 클래스 ─────────────────────────────────────────
class GroupChatSession:
def __init__(self, token: str, personas_data: dict):
self.token = token
self.personas_data = personas_data
self.active = False
self.topic = ""
self.chat_id = DEFAULT_CHAT_ID
self.personas: list = []
self.history: list = []
self.last_activity = 0.0
self.auto_turns = 0
self.speak_counts: dict = {}
self.last_speaker = ""
def send(self, text: str):
send_telegram(self.token, self.chat_id, text)
def _persona(self, key: str) -> dict:
default = {"name": key, "team": "", "role": "", "expertise": "", "personality": "", "persona_desc": ""}
p = dict(self.personas_data.get(key, default))
p.setdefault("team", "")
p.setdefault("role", "")
p.setdefault("persona_desc", "")
return p
def _tag(self, key: str) -> str:
return format_persona_tag(key, self.personas_data)
def to_dict(self) -> dict:
return {
"active": self.active,
"topic": self.topic,
"personas": self.personas,
"chat_id": self.chat_id,
"history": self.history,
"last_activity": self.last_activity,
"auto_turns": self.auto_turns,
"speak_counts": self.speak_counts,
}
def from_dict(self, data: dict):
self.active = data.get("active", False)
self.topic = data.get("topic", "")
self.personas = data.get("personas", [])
self.chat_id = data.get("chat_id", DEFAULT_CHAT_ID)
self.history = data.get("history", [])
self.last_activity = data.get("last_activity", time.time())
self.auto_turns = data.get("auto_turns", 0)
self.speak_counts = data.get("speak_counts", {})
if self.history:
self.last_speaker = self.history[-1].get("speaker", "")
def start(self, trigger: dict):
self.topic = trigger.get("topic", "일반 대화")
self.personas = trigger.get("personas", ["hermes", "athena", "thor"])
self.chat_id = trigger.get("chat_id", DEFAULT_CHAT_ID)
self.history = []
self.last_activity = time.time()
self.auto_turns = 0
self.speak_counts = {p: 0 for p in self.personas}
self.active = True
user_message = trigger.get("user_message", "")
if user_message:
self.history.append({"speaker": "user", "speaker_name": "제이회장님", "content": user_message})
logger.info(f"세션 시작 | 주제: {self.topic} | 참여자: {self.personas}")
# 입장 시퀀스
self.send("잠깐요, 관련 팀원들 모을게요.")
for p_key in self.personas:
p = self._persona(p_key)
tag = self._tag(p_key)
name = p.get("name", p_key)
team = p.get("team", "")
try:
sys_text = f"당신은 {name}입니다."
if team:
sys_text += f" {team}의 {p.get('role', '')}입니다."
else:
sys_text += f" 역할: {p.get('role', '')}."
sys_text += f" 성격: {p.get('personality', '')}. 짧고 자연스러운 한국어 대화체로 한 문장만 말하세요."
one_liner = call_claude(
sys_text,
f"단톡에 참여하며 한 마디 짧게 인사해주세요. 주제는 '{self.topic}'입니다.",
max_tokens=80,
)
except Exception as e:
logger.error(f"입장 인사 생성 오류 ({p_key}): {e}")
one_liner = "잘 부탁드립니다."
self.send(f"{tag} 참여합니다. {one_liner}")
n = len(self.personas)
self.send(f"{n}명 모였습니다. 시작하시죠.")
dump_session(self.to_dict())
def add_user_input(self, message: str):
self.history.append({"speaker": "user", "speaker_name": "제이회장님", "content": message})
self.auto_turns = 0
self.last_activity = time.time()
logger.info(f"유저 입력 추가: {message[:50]}")
def speak(self, p_key: str) -> bool:
p = self._persona(p_key)
tag = self._tag(p_key)
name = p.get("name", p_key)
try:
text = generate_persona_response(p_key, p, self.history, self.topic)
except subprocess.TimeoutExpired:
self.send(f"{tag} (잠시 생각 중...)")
logger.warning(f"TimeoutExpired: {p_key}")
return False
except Exception as e:
logger.error(f"응답 생성 오류 ({p_key}): {e}")
raise
msg_text = f"{tag} {text}"
self.send(msg_text)
self.history.append({"speaker": p_key, "speaker_name": name, "content": text})
self.speak_counts[p_key] = self.speak_counts.get(p_key, 0) + 1
self.auto_turns += 1
self.last_speaker = p_key
self.last_activity = time.time()
logger.info(f"발화: [{name}] {text[:60]}")
dump_session(self.to_dict())
return True
def end(self, reason: str = "user_exit"):
logger.info(f"세션 종료 | 이유: {reason}")
# 퇴장 시퀀스
for p_key in self.personas:
p = self._persona(p_key)
tag = self._tag(p_key)
name = p.get("name", p_key)
team = p.get("team", "")
try:
sys_text = f"당신은 {name}입니다."
if team:
sys_text += f" {team}의 {p.get('role', '')}입니다."
else:
sys_text += f" 역할: {p.get('role', '')}."
sys_text += " 짧고 자연스러운 한국어 대화체로 제이회장님께 한 문장만 말하세요."
farewell = call_claude(
sys_text,
"단톡을 마치며 짧게 작별 인사 한 문장만 해주세요.",
max_tokens=80,
)
except Exception:
farewell = "수고하셨습니다, 제이회장님."
self.send(f"{tag} {farewell}")
n = len(self.personas)
total_turns = sum(self.speak_counts.values())
self.send(f"단톡 종료합니다. (주제: {self.topic}, 참여: {n}인, 발화: {total_turns}회)")
self.active = False
save_session_log(self.to_dict())
# 세션 파일 정리
path = Path(SESSION_FILE)
if path.exists():
path.unlink()
logger.info("세션 파일 정리 완료.")
def is_active(self) -> bool:
"""세션 활성 여부 반환."""
return self.active
def _all_available_personas(self) -> list:
"""로드된 전체 페르소나 ID 목록 반환."""
return list(self.personas_data.keys())
def send_system_message(self, text: str):
"""시스템 안내 메시지 전송 (페르소나 포맷 없이)."""
send_telegram(self.token, self.chat_id, f"⚙️ {text}")
def get_persona(self, key: str) -> dict:
"""페르소나 데이터 반환 (_persona와 동일, 공개 인터페이스)."""
return self._persona(key)
def handle_control(self, control: dict):
"""세션 중 제어 명령 처리."""
ctype = control.get("type")
if ctype == "limit_personas":
count = control["count"]
if count < 1:
return # 최소 1명은 남아야 함
if count < len(self.personas):
removed = self.personas[count:]
self.personas = self.personas[:count]
removed_names = [self.get_persona(p).get("name", p) for p in removed]
# speak_counts에서도 제거
for p in removed:
self.speak_counts.pop(p, None)
self.send_system_message(f"인원 조정: {', '.join(removed_names)} 퇴장. 현재 {count}명.")
elif ctype == "remove_persona":
pid = control["persona"]
if pid in self.personas and len(self.personas) > 1:
name = self.get_persona(pid).get("name", pid)
self.personas.remove(pid)
self.speak_counts.pop(pid, None)
self.send_system_message(f"{name} 퇴장. 현재 {len(self.personas)}명.")
elif ctype == "add_persona":
pid = control["persona"]
if pid not in self.personas:
name = self.get_persona(pid).get("name", pid)
self.personas.append(pid)
self.speak_counts[pid] = 0
self.send_system_message(f"{name} 합류. 현재 {len(self.personas)}명.")
elif ctype == "filter_by_role":
new_personas = [p for p in control["personas"] if p in self._all_available_personas()]
if new_personas:
# 기존 참여자 중 제거되는 인원의 speak_counts 정리
for p in self.personas:
if p not in new_personas:
self.speak_counts.pop(p, None)
# 새로 추가되는 인원의 speak_counts 초기화
for p in new_personas:
if p not in self.speak_counts:
self.speak_counts[p] = 0
self.personas = new_personas
role = control.get("role", "")
self.send_system_message(f"{role} 담당만 남았습니다. 현재 {len(self.personas)}명.")
elif ctype == "filter_by_team":
new_personas = [p for p in control["personas"] if p in self._all_available_personas()]
if new_personas:
for p in self.personas:
if p not in new_personas:
self.speak_counts.pop(p, None)
for p in new_personas:
if p not in self.speak_counts:
self.speak_counts[p] = 0
self.personas = new_personas
team = control.get("team", "")
self.send_system_message(f"{team}만 남았습니다. 현재 {len(self.personas)}명.")
elif ctype == "set_auto_turns":
count = control["count"]
self.max_auto_turns = count
self.auto_turns = 0
if count >= 99:
self.send_system_message("자동 대화 모드: 계속 진행합니다.")
else:
self.send_system_message(f"자동 발화 {count}턴으로 설정.")
def run_one_turn(self):
"""대화 1턴 실행 (발화자 선택 → 응답 → 전송).
메인 루프에서 매 반복마다 호출된다.
기존 run_loop()의 단일 턴 로직을 분리한 것.
"""
if not self.active:
return
# 타임아웃 체크
if time.time() - self.last_activity > SESSION_TIMEOUT:
logger.info("5분 타임아웃으로 세션 자동 종료.")
self.end("timeout")
return
# MAX_AUTO_TURNS 체크 → 유저 입력 대기
max_turns = getattr(self, "max_auto_turns", MAX_AUTO_TURNS)
if self.auto_turns >= max_turns:
return
# 다음 발화자 선택
persona_names = [self._persona(k).get("name", k) for k in self.personas]
try:
next_key = select_next_speaker(
persona_names,
self.history,
self.last_speaker,
self.speak_counts,
)
except Exception as e:
logger.error(f"발화자 선택 실패: {e}")
next_key = self.personas[0]
# 발화
try:
self.speak(next_key)
except Exception as e:
self.send(f"⚠️ 시스템 오류로 단톡을 종료합니다: {str(e)[:100]}")
self.end("error")
return
def run_loop(self):
"""메인 대화 루프 (하위 호환용, 트리거 파일만 사용하는 레거시 모드)."""
while self.active:
trigger = read_trigger()
if trigger:
action = trigger.get("action")
if action == "user_input":
msg = trigger.get("message", "")
if msg:
self.add_user_input(msg)
elif action == "end":
self.end(trigger.get("reason", "user_exit"))
return
elif action == "start":
logger.warning("활성 세션 중 start 트리거 무시됨.")
self.run_one_turn()
if not self.active:
return
time.sleep(POLL_INTERVAL)
# ── 트리거 파일 체크 (하위 호환) ──────────────────────────────
def check_trigger_file(session: GroupChatSession, token: str):
"""트리거 파일을 확인하고 세션에 반영한다."""
trigger = read_trigger()
if not trigger:
return
action = trigger.get("action")
logger.info(f"트리거 감지: action={action}")
if action == "start":
if session.is_active():
logger.warning("이미 활성 세션 있음. 기존 세션 종료 후 재시작.")
session.end("restart")
try:
session.start(trigger)
except Exception as e:
logger.error(f"세션 시작 오류: {e}")
try:
send_telegram(token, trigger.get("chat_id", DEFAULT_CHAT_ID), f"⚠️ 세션 시작 오류: {str(e)[:100]}")
except Exception:
pass
elif action == "user_input":
if session.is_active():
msg = trigger.get("message", "")
if msg:
session.add_user_input(msg)
else:
logger.warning("활성 세션 없음. user_input 무시.")
elif action == "end":
if session.is_active():
session.end(trigger.get("reason", "user_exit"))
else:
logger.warning("활성 세션 없음. end 무시.")
# ── 데몬 메인 루프 ────────────────────────────────────────────
def main():
load_env_keys()
# Claude CLI 존재 확인
if not os.path.exists(CLAUDE_CLI_PATH):
logger.error(f"Claude CLI를 찾을 수 없습니다: {CLAUDE_CLI_PATH}")
sys.exit(1)
token = load_bot_token()
personas_data = load_personas()
session = GroupChatSession(token, personas_data)
poller = TelegramPoller(token, chat_id=DEFAULT_CHAT_ID)
# 재시작 시 기존 세션 복구
saved = load_session()
if saved:
session.from_dict(saved)
logger.info(f"세션 복구: 주제={session.topic}, 참여자={session.personas}")
# 시그널 핸들링
def _shutdown(signum, frame):
logger.info(f"시그널 {signum} 수신. Graceful shutdown...")
if session.is_active():
session.end("daemon_shutdown")
sys.exit(0)
signal.signal(signal.SIGTERM, _shutdown)
signal.signal(signal.SIGINT, _shutdown)
logger.info("그룹챗 데몬 시작. Telegram 폴링 + 트리거 파일 감시 중...")
while True:
try:
# 1. Telegram 메시지 수신
messages = poller.get_updates()
for msg in messages:
intent = detect_intent(msg, session.is_active())
if intent["intent"] == "start_chat" and not session.is_active():
session.start(
{
"action": "start",
"topic": intent.get("topic", "자유 대화"),
"personas": intent.get("personas", ["hermes", "athena", "thor"]),
"chat_id": DEFAULT_CHAT_ID,
"user_message": msg,
}
)
elif intent["intent"] == "user_input" and session.is_active():
session.add_user_input(intent.get("message", msg))
elif intent["intent"] == "end_chat" and session.is_active():
session.end(reason="user_exit")
elif intent["intent"] == "control" and session.is_active():
session.handle_control(intent)
elif intent["intent"] == "none" and not session.is_active():
send_telegram(
token,
DEFAULT_CHAT_ID,
'💬 안녕하세요! 팀 단톡을 시작하려면 "팀 모여" 또는 "회의하자"라고 말씀해주세요.',
)
# 2. 트리거 파일 체크 (기존 호환)
check_trigger_file(session, token)
# 3. 활성 세션이면 대화 루프 1턴 실행
if session.is_active():
session.run_one_turn()
time.sleep(POLL_INTERVAL)
except KeyboardInterrupt:
break
except Exception as e:
logger.error(f"메인 루프 오류: {e}")
time.sleep(5)
if __name__ == "__main__":
main()