"""대화 메모리 모듈.
인메모리 deque + JSONL 파일 영속화 + 자동 요약 생성.
"""
from __future__ import annotations
import asyncio
import fcntl
import json
import logging
import os
import re
from collections import deque
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# 봇별 페르소나 설정
# ---------------------------------------------------------------------------
BOT_PERSONAS: dict[str, dict[str, str]] = {
"gemini_view_bot": {
"name": "잼민이",
"persona": "너는 잼민이다. Google의 시각을 가지고 있다. 실용적이고 데이터 중심적으로 답변해라.",
},
"codex_view_bot": {
"name": "코덱스",
"persona": "너는 코덱스다. OpenAI의 시각을 가지고 있다. 코드 중심적이고 구현 지향적으로 답변해라. 반드시 한국어로 답변해라.",
},
"claude_view_bot": {
"name": "클로디",
"persona": "너는 클로디다. Anthropic의 시각을 가지고 있다. 신중하고 균형 잡힌 분석을 해라.",
},
}
# ---------------------------------------------------------------------------
# 기본 저장 경로
# ---------------------------------------------------------------------------
DEFAULT_STORAGE_BASE = "/home/jay/workspace/memory/groupchat"
# ---------------------------------------------------------------------------
# 무활동 타임아웃 (초)
# ---------------------------------------------------------------------------
INACTIVITY_TIMEOUT = 1800 # 30분
# ---------------------------------------------------------------------------
# PII 마스킹 패턴 (모듈 상수)
# ---------------------------------------------------------------------------
_PII_PATTERNS = [
(re.compile(r"\d{6}-[1-4]\d{6}"), "[주민번호]"), # 주민등록번호
(re.compile(r"01[016789]-?\d{3,4}-?\d{4}"), "[전화번호]"), # 휴대폰
(re.compile(r"\d{3,4}-\d{2,4}-\d{4,6}"), "[계좌번호]"), # 계좌번호
]
# ---------------------------------------------------------------------------
# 주제 전환 감지 상수 (모듈 상수)
# ---------------------------------------------------------------------------
_TOPIC_CHANGE_KEYWORDS = ["다른 주제", "그건 그렇고", "화제를 바꿔", "다음 안건", "본론으로"]
_SILENCE_GAP_SECONDS = 300 # 5분
# ---------------------------------------------------------------------------
# call_claude import (요약/insight 생성에 사용)
# ---------------------------------------------------------------------------
try:
from engine_v2.bot_api import call_claude # type: ignore[import-not-found]
except ImportError: # pragma: no cover – 테스트 환경에서 engine_v2 없을 때 대비
async def call_claude(prompt: str, timeout: int = 120) -> str: # type: ignore[misc]
raise ImportError("engine_v2 모듈을 찾을 수 없습니다.")
# ---------------------------------------------------------------------------
# 모듈 레벨 함수: PII 마스킹
# ---------------------------------------------------------------------------
def _mask_pii(text: str) -> str:
"""텍스트에서 PII(개인식별정보)를 마스킹한다."""
for pattern, replacement in _PII_PATTERNS:
text = pattern.sub(replacement, text)
return text
# ---------------------------------------------------------------------------
# 데이터클래스
# ---------------------------------------------------------------------------
@dataclass
class ChatMessage:
sender: str # 봇이름 or "제이회장님"
text: str
timestamp: datetime
is_bot: bool
# ---------------------------------------------------------------------------
# ConversationMemory
# ---------------------------------------------------------------------------
_SUMMARY_TRIGGER = 50 # 메시지 N개마다 요약 트리거
_JSONL_ROTATION_THRESHOLD = 50000 # 5만줄
class ConversationMemory:
def __init__(
self,
max_messages: int = 50,
storage_base: str | None = None,
) -> None:
self._messages: dict[int, deque[ChatMessage]] = {}
self._max_messages = max_messages
self._storage_base: str | None = storage_base
# chat_id별 메시지 카운트 (요약 트리거용)
self._message_count: dict[int, int] = {}
# chat_id별 요약 파일 번호
self._summary_counter: dict[int, int] = {}
# 이미 load_today를 호출한 chat_id 추적 (lazy load)
self._loaded_chats: set[int] = set()
# JSONL 파일경로 → 줄 수 캐시 (rotation 판단용)
self._jsonl_line_count: dict[str, int] = {}
# chat_id별 마지막 활동 시간 (무활동 타이머용)
self._last_activity: dict[int, datetime] = {}
# chat_id별 무활동 타이머 태스크
self._inactivity_tasks: dict[int, asyncio.Task[None]] = {}
# chat_id별 현재 주제 (topic tracking)
self._current_topic: dict[int, str] = {}
# chat_id별 요약 중복 실행 방지 플래그
self._summary_lock: dict[int, bool] = {}
# 날짜 문자열별 LLM 호출 횟수 (비용 DoS 방지)
self._llm_call_count: dict[str, int] = {}
# 일일 LLM 호출 상한
self._daily_llm_budget: int = 50
# chat_id별 롤링 서머리 (현재 토론 중 자동 생성)
self._rolling_summaries: dict[int, str] = {}
# ------------------------------------------------------------------
# 내부 헬퍼: 경로
# ------------------------------------------------------------------
def _today_jsonl_path(self, ts: datetime | None = None) -> Path | None:
"""오늘 날짜 JSONL 파일 경로. storage_base가 None이면 None 반환.
ts가 주어지면 해당 timestamp 기준 날짜를 사용한다 (날짜 경계 역전 방지).
"""
if self._storage_base is None:
return None
today = (ts or datetime.now()).strftime("%Y-%m-%d")
return Path(self._storage_base) / f"{today}.jsonl"
def _summaries_dir(self) -> Path | None:
if self._storage_base is None:
return None
return Path(self._storage_base) / "summaries"
def _insights_dir(self) -> Path | None:
if self._storage_base is None:
return None
return Path(self._storage_base) / "insights"
# ------------------------------------------------------------------
# 내부 헬퍼: 파일 저장
# ------------------------------------------------------------------
def _get_jsonl_path_with_rotation(self, base_path: Path) -> Path:
"""현재 JSONL 파일의 줄 수를 확인하고 rotation이 필요하면 새 파일 경로 반환."""
key = str(base_path)
if key not in self._jsonl_line_count:
if base_path.exists():
try:
with open(base_path, "r", encoding="utf-8") as fh:
count = sum(1 for _ in fh)
except Exception:
count = 0
else:
count = 0
self._jsonl_line_count[key] = count
if self._jsonl_line_count[key] < _JSONL_ROTATION_THRESHOLD:
self._jsonl_line_count[key] += 1
return base_path
# rotation 필요: _part2, _part3, ... 순서로 빈 자리 또는 threshold 미만인 파일 찾기
stem = base_path.stem # e.g. "2026-03-15"
suffix = base_path.suffix # ".jsonl"
parent = base_path.parent
part = 2
while True:
candidate = parent / f"{stem}_part{part}{suffix}"
cand_key = str(candidate)
if cand_key not in self._jsonl_line_count:
if candidate.exists():
try:
with open(candidate, "r", encoding="utf-8") as fh:
count = sum(1 for _ in fh)
except Exception:
count = 0
else:
count = 0
self._jsonl_line_count[cand_key] = count
if self._jsonl_line_count[cand_key] < _JSONL_ROTATION_THRESHOLD:
self._jsonl_line_count[cand_key] += 1
return candidate
part += 1
def _append_to_jsonl(self, chat_id: int, msg: ChatMessage) -> None:
"""JSONL 파일에 메시지를 append. 실패 시 로깅만 하고 예외 없음.
인메모리 deque에는 원본 저장, JSONL 파일에만 PII 마스킹 적용.
"""
base_path = self._today_jsonl_path(ts=msg.timestamp)
if base_path is None:
return
# rotation 적용
jsonl_path = self._get_jsonl_path_with_rotation(base_path)
# PII 마스킹 (JSONL 파일에만 적용, 인메모리 원본은 유지)
masked_text = _mask_pii(msg.text)
record: dict[str, Any] = {
"sender": msg.sender,
"text": masked_text,
"timestamp": msg.timestamp.isoformat(),
"is_bot": msg.is_bot,
"chat_id": chat_id,
"topic_tag": self._current_topic.get(chat_id, "general"),
}
line = json.dumps(record, ensure_ascii=False)
try:
os.makedirs(jsonl_path.parent, exist_ok=True)
os.chmod(jsonl_path.parent, 0o700)
with open(jsonl_path, "a", encoding="utf-8") as fh:
fcntl.flock(fh, fcntl.LOCK_EX)
try:
fh.write(line + "\n")
finally:
fcntl.flock(fh, fcntl.LOCK_UN)
os.chmod(jsonl_path, 0o600)
except Exception as exc:
logger.warning("JSONL 파일 저장 실패 (인메모리만 동작): %s", exc)
# ------------------------------------------------------------------
# 주제 전환 감지
# ------------------------------------------------------------------
def _detect_topic_change(self, chat_id: int, msg: ChatMessage) -> bool:
"""침묵 갭 5분 초과 또는 전환 키워드 감지 시 True."""
last = self._last_activity.get(chat_id)
if last and (msg.timestamp - last).total_seconds() > _SILENCE_GAP_SECONDS:
return True
for kw in _TOPIC_CHANGE_KEYWORDS:
if kw in msg.text:
return True
return False
# ------------------------------------------------------------------
# 핵심 public 메서드
# ------------------------------------------------------------------
def add_message(self, chat_id: int, sender: str, text: str, is_bot: bool) -> None:
"""메시지를 추가한다. 링 버퍼 초과 시 오래된 메시지 제거."""
# 첫 메시지 수신 시 오늘 JSONL에서 대화 복원 (lazy load)
if chat_id not in self._loaded_chats:
self._loaded_chats.add(chat_id)
self.load_today(chat_id)
if chat_id not in self._messages:
self._messages[chat_id] = deque(maxlen=self._max_messages)
msg = ChatMessage(
sender=sender,
text=text,
timestamp=datetime.now(),
is_bot=is_bot,
)
# 주제 전환 감지 (JSONL 저장 전에 호출)
if self._detect_topic_change(chat_id, msg):
self._current_topic[chat_id] = "pending"
self._messages[chat_id].append(msg)
# JSONL 파일 저장 (graceful degradation)
self._append_to_jsonl(chat_id, msg)
# 메시지 카운트 증가 + 요약 트리거 확인
self._message_count[chat_id] = self._message_count.get(chat_id, 0) + 1
if self._message_count[chat_id] % _SUMMARY_TRIGGER == 0:
self._schedule_summary(chat_id)
# 무활동 타이머 갱신
self._last_activity[chat_id] = datetime.now()
self._schedule_inactivity_check(chat_id)
def get_context(self, chat_id: int, limit: int = 30) -> list[ChatMessage]:
"""최근 limit개 메시지를 시간순으로 반환."""
if chat_id not in self._messages:
return []
msgs = list(self._messages[chat_id])
return msgs[-limit:]
# ------------------------------------------------------------------
# load_today
# ------------------------------------------------------------------
def load_today(self, chat_id: int) -> None:
"""오늘 날짜 JSONL 파일에서 해당 chat_id의 최근 50개 메시지를 로드하여 deque 복원.
rotation 파일(_partN)도 포함하여 로드한다.
"""
base_path = self._today_jsonl_path()
if base_path is None:
return
# 기본 파일 + _partN 파일을 날짜 순서대로 수집
today = datetime.now().strftime("%Y-%m-%d")
parent = base_path.parent
all_paths: list[Path] = sorted(parent.glob(f"{today}*.jsonl"))
if not all_paths:
return
loaded: list[ChatMessage] = []
for jsonl_path in all_paths:
if not jsonl_path.exists():
continue
try:
with open(jsonl_path, "r", encoding="utf-8") as fh:
for raw_line in fh:
raw_line = raw_line.strip()
if not raw_line:
continue
try:
data = json.loads(raw_line)
except json.JSONDecodeError:
logger.warning("손상된 JSONL 라인 건너뜀: %r", raw_line)
continue
if data.get("chat_id") != chat_id:
continue
try:
ts = datetime.fromisoformat(data["timestamp"])
except (KeyError, ValueError):
ts = datetime.now()
loaded.append(
ChatMessage(
sender=data.get("sender", ""),
text=data.get("text", ""),
timestamp=ts,
is_bot=bool(data.get("is_bot", False)),
)
)
except Exception as exc:
logger.warning("load_today 파일 읽기 실패 (%s): %s", jsonl_path, exc)
# 최근 50개만 유지
recent = loaded[-50:]
dq: deque[ChatMessage] = deque(maxlen=self._max_messages)
dq.extend(recent)
self._messages[chat_id] = dq
# ------------------------------------------------------------------
# 무활동 타이머
# ------------------------------------------------------------------
def _schedule_inactivity_check(self, chat_id: int) -> None:
"""무활동 타이머를 (재)시작한다. 기존 타이머가 있으면 취소 후 새로 생성."""
existing = self._inactivity_tasks.get(chat_id)
if existing is not None and not existing.done():
existing.cancel()
try:
loop = asyncio.get_running_loop()
task = loop.create_task(self._check_inactivity(chat_id))
self._inactivity_tasks[chat_id] = task
except RuntimeError:
pass # 동기 환경 (테스트 등)
async def _check_inactivity(self, chat_id: int) -> None:
"""INACTIVITY_TIMEOUT 동안 새 메시지가 없으면 요약을 트리거한다."""
activity_at_start = self._last_activity.get(chat_id)
await asyncio.sleep(INACTIVITY_TIMEOUT)
# sleep 후 활동 시간이 달라졌으면 (새 메시지가 왔으면) skip
if self._last_activity.get(chat_id) != activity_at_start:
return
# 최소 5개 이상 메시지가 있을 때만 요약 트리거
messages = self.get_context(chat_id, limit=50)
if len(messages) < 5:
return
self._schedule_summary(chat_id)
# ------------------------------------------------------------------
# 요약 생성
# ------------------------------------------------------------------
def _schedule_summary(self, chat_id: int) -> None:
"""비동기 이벤트 루프가 있으면 create_task로 요약 스케줄, 없으면 skip."""
result = self._generate_summary(chat_id)
# mock 또는 일반 호출 반환값이 코루틴인지 확인
if asyncio.iscoroutine(result):
try:
loop = asyncio.get_running_loop()
loop.create_task(result)
except RuntimeError:
# 동기 환경: 코루틴을 닫고 skip
result.close()
async def _generate_summary(self, chat_id: int) -> None:
"""call_claude로 요약을 생성하고 JSON 파일에 저장한다."""
# --- 항목 1: 요약 중복 실행 방지 세마포어 ---
if self._summary_lock.get(chat_id):
logger.debug("요약 이미 진행 중 (chat_id=%d), 스킵", chat_id)
return
self._summary_lock[chat_id] = True
try:
await self._do_generate_summary(chat_id)
finally:
self._summary_lock[chat_id] = False
async def _do_generate_summary(self, chat_id: int) -> None:
"""요약 생성 본체 (_generate_summary에서 세마포어 보호 하에 호출)."""
summaries_dir = self._summaries_dir()
messages = self.get_context(chat_id, limit=50)
if not messages:
return
# --- 항목 2: 일일 LLM 호출 예산 확인 ---
today = datetime.now().strftime("%Y-%m-%d")
current_count = self._llm_call_count.get(today, 0)
if current_count >= self._daily_llm_budget:
logger.warning(
"일일 LLM 호출 예산 초과 (%d/%d), 스킵",
current_count,
self._daily_llm_budget,
)
return
self._llm_call_count[today] = current_count + 1
# 텍스트 포맷
conversation_text = "\n".join(f"{m.sender}: {m.text}" for m in messages)
prompt = (
"아래 대화를 분석해줘. 반드시 JSON 형식으로만 응답해.\n"
"아래 대화의 핵심 논점과 결론을 3~5줄로 요약해줘. 주제 키워드 3~5개도 추출해줘.\n\n"
"한국어 대화 특성:\n"
"- 한국어 대화에서 주어는 자주 생략됩니다. 문맥으로 주어를 추론하세요.\n"
"- 감탄사(ㅋㅋ, ㅎㅎ, 와, 헐, 대박)는 주제 분류에 포함하지 마세요.\n\n"
"\n"
f"{conversation_text}\n"
"\n\n"
'{"summary": "3~5줄 요약", "key_topics": ["주제1", "주제2"], '
'"topic_tag": "주요 주제 한 단어(영문 snake_case)", '
'"key_decisions": ["결정사항1"], '
'"action_items": ["액션1"], '
'"consensus_level": "exploratory|tentative|agreed|decided"}'
)
try:
raw_response = await call_claude(prompt)
except Exception as exc:
logger.warning("요약 생성 실패, 다음 트리거 대기: %s", exc)
return
# 마크다운 코드펜스 제거
stripped = raw_response.strip()
if stripped.startswith("```"):
# 첫 줄(```json 등) 제거
stripped = re.sub(r"^```(?:\w+)?\s*\n?", "", stripped, count=1)
# 마지막 ``` 제거
stripped = re.sub(r"\n?```\s*$", "", stripped)
raw_response = stripped.strip()
# JSON 파싱 시도
summary_text: str
key_topics: list[str]
parsed: dict[str, Any]
try:
parsed = json.loads(raw_response)
summary_text = parsed["summary"]
key_topics = parsed.get("key_topics", [])
except (json.JSONDecodeError, KeyError, TypeError):
# 하위호환: 파싱 실패 시 전체 텍스트를 summary로, key_topics는 빈 배열
summary_text = raw_response
key_topics = []
parsed = {}
# 참여자 추출
participants = list({m.sender for m in messages})
count = self._message_count.get(chat_id, len(messages))
msg_from = max(1, count - len(messages) + 1)
msg_to = count
# topic_slug 생성
topic_slug = re.sub(r"[^a-zA-Z0-9가-힣_-]", "", parsed.get("topic_tag", "general"))[:20]
if not topic_slug:
topic_slug = "general"
summary_data: dict[str, Any] = {
"timestamp": datetime.now().isoformat(),
"date": datetime.now().strftime("%Y-%m-%d"),
"message_range": {"from": msg_from, "to": msg_to},
"summary": summary_text,
"key_topics": key_topics,
"topic_tag": topic_slug,
"key_decisions": parsed.get("key_decisions", []),
"action_items": parsed.get("action_items", []),
"consensus_level": parsed.get("consensus_level", "exploratory"),
"participants": participants,
}
# --- 항목 5: pending 상태의 topic을 LLM 응답 기반으로 확정 ---
if self._current_topic.get(chat_id) == "pending":
self._current_topic[chat_id] = topic_slug
if summaries_dir is None:
return
try:
os.makedirs(summaries_dir, exist_ok=True)
os.chmod(summaries_dir, 0o700)
today = datetime.now().strftime("%Y-%m-%d")
# _summary_counter 복구: 재시작 시 기존 파일 수 확인
if chat_id not in self._summary_counter:
existing = list(summaries_dir.glob(f"{today}_*.json"))
self._summary_counter[chat_id] = len(existing)
num = self._summary_counter.get(chat_id, 0) + 1
self._summary_counter[chat_id] = num
file_path = summaries_dir / f"{today}_{topic_slug}_{num:03d}.json"
file_path.write_text(
json.dumps(summary_data, ensure_ascii=False, indent=2),
encoding="utf-8",
)
os.chmod(file_path, 0o600)
# 지연 인덱싱: 요약 파일 저장 후 인덱스 갱신
new_index_entry = {
"filename": file_path.stem,
"date": summary_data.get("date", ""),
"topic_tag": summary_data.get("topic_tag", ""),
"summary": summary_data.get("summary", "")[:50],
}
self._update_summary_index(summaries_dir, new_entry=new_index_entry)
except Exception as exc:
logger.warning("요약 파일 저장 실패: %s", exc)
# ------------------------------------------------------------------
# 지연 인덱싱
# ------------------------------------------------------------------
def _update_summary_index(self, summaries_dir: Path, new_entry: dict[str, Any] | None = None) -> None:
"""summaries/_index.json 캐시 인덱스를 갱신한다.
_index.json이 이미 존재하면 new_entry만 append하여 갱신.
_index.json이 없고 파일 수 200개 초과면 전체 스캔하여 생성.
_index.json이 없고 파일 수 200개 이하면 아무것도 하지 않음.
"""
index_path = summaries_dir / "_index.json"
try:
if index_path.exists():
# 기존 인덱스에 new_entry만 append
if new_entry is not None:
try:
existing: list[dict[str, Any]] = json.loads(index_path.read_text(encoding="utf-8"))
except Exception:
existing = []
existing.append(new_entry)
index_path.write_text(
json.dumps(existing, ensure_ascii=False, indent=2),
encoding="utf-8",
)
return
# _index.json 미존재: 파일 수 200개 초과 시에만 생성
all_json_files = [f for f in summaries_dir.glob("*.json") if f.name != "_index.json"]
if len(all_json_files) <= 200:
return
# 전체 스캔하여 인덱스 생성
index_entries: list[dict[str, Any]] = []
for fp in sorted(all_json_files):
try:
data = json.loads(fp.read_text(encoding="utf-8"))
index_entries.append(
{
"filename": fp.stem,
"date": data.get("date", ""),
"topic_tag": data.get("topic_tag", ""),
"summary": data.get("summary", "")[:50],
}
)
except Exception as exc:
logger.warning("인덱스 생성 중 파일 읽기 실패 %s: %s", fp, exc)
index_path.write_text(
json.dumps(index_entries, ensure_ascii=False, indent=2),
encoding="utf-8",
)
except Exception as exc:
logger.warning("_update_summary_index 실패: %s", exc)
# ------------------------------------------------------------------
# generate_rolling_summary
# ------------------------------------------------------------------
async def generate_rolling_summary(self, chat_id: int) -> str:
"""현재 세션의 전체 메시지를 요약하여 롤링 서머리 생성.
Claude CLI를 호출하여 요약 생성.
결과를 self._rolling_summaries[chat_id]에 저장.
"""
messages = self.get_context(chat_id, limit=50) # 가능한 많은 메시지
if not messages:
return ""
# 일일 LLM 예산 확인
today = datetime.now().strftime("%Y-%m-%d")
current_count = self._llm_call_count.get(today, 0)
if current_count >= self._daily_llm_budget:
logger.warning("일일 LLM 예산 초과, 롤링 서머리 스킵")
return self._rolling_summaries.get(chat_id, "")
self._llm_call_count[today] = current_count + 1
conversation_text = "\n".join(f"{m.sender}: {m.text}" for m in messages)
prompt = (
"아래 토론 내용을 5줄 이내로 압축 요약해줘. "
"각 참여자의 핵심 주장과 합의/이견 사항을 포함해.\n\n"
"\n"
f"{conversation_text}\n"
""
)
try:
summary = await call_claude(prompt, timeout=30)
self._rolling_summaries[chat_id] = summary
logger.info("롤링 서머리 생성 완료 (chat_id=%d, %d자)", chat_id, len(summary))
return summary
except Exception as exc:
logger.warning("롤링 서머리 생성 실패: %s", exc)
return self._rolling_summaries.get(chat_id, "")
# ------------------------------------------------------------------
# get_recent_summaries
# ------------------------------------------------------------------
def get_recent_summaries(self, limit: int = 3) -> list[dict[str, Any]]:
"""summaries/ 디렉토리에서 오늘 파일들을 읽어 최근 N개 반환."""
summaries_dir = self._summaries_dir()
if summaries_dir is None or not summaries_dir.exists():
return []
today = datetime.now().strftime("%Y-%m-%d")
today_files = sorted(summaries_dir.glob(f"{today}_*.json"))
results: list[dict[str, Any]] = []
for fp in today_files[-limit:]:
try:
data = json.loads(fp.read_text(encoding="utf-8"))
results.append(data)
except Exception as exc:
logger.warning("요약 파일 읽기 실패 %s: %s", fp, exc)
return results
def get_all_summary_files(self, limit: int = 50) -> list[dict[str, Any]]:
"""summaries/ 디렉토리의 모든 .json 파일을 읽어 메타데이터 목록 반환.
/메모리 InlineKeyboard 지원용. 최신순 정렬 후 limit만큼 반환.
각 항목: {"filename": stem, "date": ..., "topic_tag": ..., "summary": ...(50자)}
summaries/_index.json 존재 시 그것을 사용, 없으면 기존 glob 방식.
"""
summaries_dir = self._summaries_dir()
if summaries_dir is None or not summaries_dir.exists():
return []
index_path = summaries_dir / "_index.json"
if index_path.exists():
try:
index_data: list[dict[str, Any]] = json.loads(index_path.read_text(encoding="utf-8"))
# 최신순 정렬 (filename 기준 내림차순)
index_data_sorted = sorted(index_data, key=lambda x: x.get("filename", ""), reverse=True)
return index_data_sorted[:limit]
except Exception as exc:
logger.warning("_index.json 읽기 실패, glob 방식으로 fallback: %s", exc)
# glob 방식 (기존 동작, _index.json 제외)
all_files = sorted(
(f for f in summaries_dir.glob("*.json") if f.name != "_index.json"),
reverse=True,
)
# 지연 인덱싱: 200개 초과 시 인덱스 자동 생성
if len(all_files) > 200:
self._update_summary_index(summaries_dir)
results: list[dict[str, Any]] = []
for fp in all_files[:limit]:
try:
data = json.loads(fp.read_text(encoding="utf-8"))
results.append(
{
"filename": fp.stem,
"date": data.get("date", ""),
"topic_tag": data.get("topic_tag", ""),
"summary": data.get("summary", "")[:50],
}
)
except Exception as exc:
logger.warning("요약 파일 읽기 실패 %s: %s", fp, exc)
return results
# ------------------------------------------------------------------
# smart_search (/스마트검색 명령어)
# ------------------------------------------------------------------
async def smart_search(self, query: str, chat_id: int) -> str:
"""요약 메타데이터에서 키워드 매칭 후 LLM 자연어 답변 생성."""
# 레이트 리밋 예산 확인
today = datetime.now().strftime("%Y-%m-%d")
current_count = self._llm_call_count.get(today, 0)
if current_count >= self._daily_llm_budget:
logger.warning(
"일일 LLM 호출 예산 초과 (%d/%d), 스마트검색 스킵",
current_count,
self._daily_llm_budget,
)
return "일일 LLM 호출 예산이 초과되었습니다."
summaries_dir = self._summaries_dir()
if summaries_dir is None or not summaries_dir.exists():
return "검색 결과가 없습니다."
# summaries/ 디렉토리의 모든 .json 파일에서 키워드 매칭
# (_index.json 제외)
query_lower = query.lower()
matched: list[dict[str, Any]] = []
for fp in sorted(summaries_dir.glob("*.json")):
if fp.name == "_index.json":
continue
try:
data = json.loads(fp.read_text(encoding="utf-8"))
# topic_tags, key_decisions, summary, key_topics 필드 대상 검색
searchable_parts = [
data.get("summary", ""),
data.get("topic_tag", ""),
" ".join(data.get("key_topics", [])),
" ".join(data.get("key_decisions", [])),
]
searchable = " ".join(searchable_parts)
if query_lower in searchable.lower():
matched.append(data)
except Exception:
continue
if not matched:
return "검색 결과가 없습니다."
# 상위 5건 추출 (최신순)
top5 = matched[-5:]
# LLM 예산 차감
self._llm_call_count[today] = current_count + 1
# LLM 프롬프트 구성
summary_lines = []
for i, item in enumerate(top5, start=1):
date = item.get("date", "")
topic_tag = item.get("topic_tag", "")
summary_text = item.get("summary", "")
summary_lines.append(f"{i}. [{date}] {topic_tag}: {summary_text}")
summaries_text = "\n".join(summary_lines)
prompt = (
"아래 대화 요약들을 참고하여 질문에 답변해줘.\n\n"
"\n"
f"질문: {query}\n\n"
f"관련 요약:\n{summaries_text}\n"
"\n\n"
"한국어로 간결하게 답변해줘."
)
try:
answer = await call_claude(prompt)
except Exception as exc:
logger.warning("smart_search LLM 호출 실패: %s", exc)
answer = f"검색 결과 처리 중 오류가 발생했습니다: {exc}"
return answer
# ------------------------------------------------------------------
# generate_insight (/정리 명령어)
# ------------------------------------------------------------------
async def generate_insight(self, chat_id: int) -> str:
"""현재까지 대화 요약 (핵심 논점, 결론, 액션 아이템) 생성 후 .md 파일 저장."""
# --- 항목 2: 일일 LLM 호출 예산 확인 ---
today = datetime.now().strftime("%Y-%m-%d")
current_count = self._llm_call_count.get(today, 0)
if current_count >= self._daily_llm_budget:
logger.warning(
"일일 LLM 호출 예산 초과 (%d/%d), 스킵",
current_count,
self._daily_llm_budget,
)
return "일일 LLM 호출 예산이 초과되었습니다."
self._llm_call_count[today] = current_count + 1
messages = self.get_context(chat_id, limit=50)
conversation_text = "\n".join(f"{m.sender}: {m.text}" for m in messages)
prompt = (
"아래 대화를 정리해줘.\n"
"핵심 논점, 결론, 액션 아이템을 마크다운 형식으로 요약해줘.\n\n"
"\n"
f"{conversation_text}\n"
""
)
try:
insight_text = await call_claude(prompt)
except Exception as exc:
logger.warning("insight 생성 실패: %s", exc)
insight_text = f"insight 생성 실패: {exc}"
# .md 파일 저장
insights_dir = self._insights_dir()
if insights_dir is not None:
try:
os.makedirs(insights_dir, exist_ok=True)
os.chmod(insights_dir, 0o700)
today = datetime.now().strftime("%Y-%m-%d")
# 파일 번호: 오늘 파일 개수 + 1
existing = list(insights_dir.glob(f"{today}_insight_*.md"))
num = len(existing) + 1
file_path = insights_dir / f"{today}_insight_{num:03d}.md"
file_path.write_text(insight_text, encoding="utf-8")
os.chmod(file_path, 0o600)
except Exception as exc:
logger.warning("insight 파일 저장 실패: %s", exc)
# 이벤트 파일 생성 (외부 시스템 연동)
self._create_insight_event()
return insight_text
def _create_insight_event(self, events_dir: Path | None = None) -> None:
"""새 insight 생성 시 이벤트 파일을 생성한다. 아누가 대화 시작 시 참조."""
if events_dir is None:
events_dir = Path("/home/jay/workspace/memory/events")
try:
events_dir.mkdir(parents=True, exist_ok=True)
today = datetime.now().strftime("%Y-%m-%d")
event_path = events_dir / f"groupchat-insight-{today}.event"
event_data: dict[str, Any] = {
"type": "groupchat-insight",
"timestamp": datetime.now().isoformat(),
"insights_dir": str(self._insights_dir()) if self._insights_dir() else None,
}
event_path.write_text(
json.dumps(event_data, ensure_ascii=False, indent=2),
encoding="utf-8",
)
except Exception as exc:
logger.warning("insight 이벤트 파일 생성 실패: %s", exc)
# ------------------------------------------------------------------
# format_context
# ------------------------------------------------------------------
def format_context(
self, chat_id: int, bot_username: str, current_message: str = "", phase: str | None = None
) -> str:
"""맥락 기반 프롬프트 문자열을 생성한다."""
persona_info = BOT_PERSONAS.get(bot_username, {"name": bot_username, "persona": ""})
persona = persona_info["persona"]
messages = self.get_context(chat_id) # 이제 기본값 30
summaries = self.get_recent_summaries(limit=3)
parts: list[str] = []
# 1. 이전 세션 요약
if summaries:
parts.append("[이전 대화 요약]")
for s in summaries:
summary_text = s.get("summary", "")
parts.append(f"- {summary_text}")
parts.append("")
# 2. 롤링 서머리 (현재 토론 압축)
rolling = self._rolling_summaries.get(chat_id)
if rolling:
parts.append("[현재 토론 요약]")
parts.append(rolling)
parts.append("")
# 3. 최근 메시지
if messages:
# summaries 있거나 rolling 있으면 [최근 대화], 둘 다 없으면 [이전 대화] (기존 호환)
if summaries or rolling:
parts.append("[최근 대화]")
else:
parts.append("[이전 대화]")
for msg in messages:
parts.append(f"{msg.sender}: {msg.text}")
parts.append("")
# 4. 현재 질문/발언
if current_message:
parts.append("[현재 질문/발언]")
parts.append(f"제이회장님: {current_message}")
parts.append("")
# 5. 페르소나
parts.append(persona)
# 6. Phase별 지시 (추가, 기존 "새로운 관점" 지시 대체)
PHASE_PROMPTS = {
"diverge": "새로운 관점과 아이디어를 자유롭게 제시하라. 다른 참여자의 의견에 동의할 필요 없다.",
"converge": "지금까지 논의에서 공통점을 먼저 정리하라. 이견이 있는 부분만 간결하게 언급하라. 새로운 주제를 꺼내지 마라.",
"consensus": "최종 합의문을 작성하라. 형식: [합의사항] 모두가 동의한 내용 / [미합의] 이견이 남은 부분 / [다음 단계] 추가 논의가 필요한 항목. 새로운 의견 제시 금지.",
}
if phase and phase in PHASE_PROMPTS:
parts.append(PHASE_PROMPTS[phase])
else:
parts.append("이전 발언들을 참고하되 중복되지 않는 새로운 관점을 제시해라.")
return "\n".join(parts)