7장: 메모리 시스템 - 에이전트의 기억과 학습

class ConversationMemory:
    def __init__(self):
        self.messages: list[dict] = []
 
    def add_user_message(self, content: str):
        self.messages.append({"role": "user", "content": content})
 
    def add_assistant_message(self, content: str):
        self.messages.append({"role": "assistant", "content": content})
 
    def get_messages(self) -> list[dict]:
        return self.messages.copy()
 
    def clear(self):
        self.messages = []

class SlidingWindowMemory:
    def __init__(self, max_messages: int = 20):
        self.messages: list[dict] = []
        self.max_messages = max_messages
 
    def add_message(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        if len(self.messages) > self.max_messages:
            self.messages = self.messages[-self.max_messages:]
 
    def get_messages(self) -> list[dict]:
        return self.messages.copy()

import anthropic
 
client = anthropic.Anthropic()
 
class SummaryMemory:
    def __init__(self, max_recent: int = 10):
        self.summary: str = ""
        self.recent_messages: list[dict] = []
        self.max_recent = max_recent
 
    def add_message(self, role: str, content: str):
        self.recent_messages.append({"role": role, "content": content})
 
        if len(self.recent_messages) > self.max_recent * 2:
            self._compress()
 
    def _compress(self):
        """오래된 메시지를 요약하여 압축합니다."""
        old_messages = self.recent_messages[:self.max_recent]
        self.recent_messages = self.recent_messages[self.max_recent:]
 
        conversation_text = "\n".join(
            f"{m['role']}: {m['content']}" for m in old_messages
        )
 
        prompt = f"""다음 대화 내용을 핵심 정보 위주로 요약하십시오.
 
기존 요약: {self.summary if self.summary else '없음'}
 
새로운 대화:
{conversation_text}
 
다음 정보를 반드시 보존하십시오:
1. 사용자의 요청과 선호사항
2. 중요한 결정 사항과 그 근거
3. 발견된 핵심 정보와 데이터
4. 아직 완료되지 않은 작업
 
간결하게 요약하십시오."""
 
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}],
        )
        self.summary = response.content[0].text
 
    def get_context(self) -> list[dict]:
        """요약 + 최근 메시지를 반환합니다."""
        context = []
        if self.summary:
            context.append({
                "role": "user",
                "content": f"[이전 대화 요약]\n{self.summary}"
            })
            context.append({
                "role": "assistant",
                "content": "이전 대화 내용을 확인했습니다."
            })
        context.extend(self.recent_messages)
        return context

from dataclasses import dataclass
from datetime import datetime
import numpy as np
 
@dataclass
class MemoryEntry:
    content: str
    embedding: list[float]
    metadata: dict
    timestamp: str
    memory_type: str  # episodic, semantic, procedural
 
class VectorMemory:
    def __init__(self):
        self.entries: list[MemoryEntry] = []
        self.client = anthropic.Anthropic()
 
    def store(self, content: str, memory_type: str, metadata: dict = None):
        """정보를 장기 메모리에 저장합니다."""
        embedding = self._get_embedding(content)
        entry = MemoryEntry(
            content=content,
            embedding=embedding,
            metadata=metadata or {},
            timestamp=datetime.now().isoformat(),
            memory_type=memory_type,
        )
        self.entries.append(entry)
 
    def retrieve(self, query: str, top_k: int = 5,
                 memory_type: str = None) -> list[MemoryEntry]:
        """쿼리와 관련된 메모리를 검색합니다."""
        query_embedding = self._get_embedding(query)
 
        # 코사인 유사도 계산
        scored = []
        for entry in self.entries:
            if memory_type and entry.memory_type != memory_type:
                continue
            similarity = self._cosine_similarity(
                query_embedding, entry.embedding
            )
            scored.append((similarity, entry))
 
        scored.sort(key=lambda x: x[0], reverse=True)
        return [entry for _, entry in scored[:top_k]]
 
    def _get_embedding(self, text: str) -> list[float]:
        """텍스트의 임베딩 벡터를 생성합니다."""
        # 실제로는 임베딩 API를 호출합니다
        # 예: OpenAI text-embedding-3-small 또는 Voyage AI
        import hashlib
        hash_val = hashlib.md5(text.encode()).hexdigest()
        return [int(c, 16) / 15.0 for c in hash_val]  # 시연용
 
    def _cosine_similarity(self, a: list[float], b: list[float]) -> float:
        a_arr = np.array(a)
        b_arr = np.array(b)
        return float(np.dot(a_arr, b_arr) / (
            np.linalg.norm(a_arr) * np.linalg.norm(b_arr) + 1e-8
        ))

class AgentMemorySystem:
    def __init__(self):
        self.vector_memory = VectorMemory()
 
    def remember_episode(self, interaction: str, outcome: str):
        """에피소드 기억: 과거 상호작용과 결과를 저장합니다."""
        self.vector_memory.store(
            content=f"상호작용: {interaction}\n결과: {outcome}",
            memory_type="episodic",
            metadata={"interaction": interaction, "outcome": outcome},
        )
 
    def remember_fact(self, fact: str, source: str = ""):
        """의미 기억: 학습한 사실과 지식을 저장합니다."""
        self.vector_memory.store(
            content=fact,
            memory_type="semantic",
            metadata={"source": source},
        )
 
    def remember_procedure(self, task: str, procedure: str):
        """절차 기억: 작업 수행 방법을 저장합니다."""
        self.vector_memory.store(
            content=f"작업: {task}\n절차: {procedure}",
            memory_type="procedural",
            metadata={"task": task},
        )
 
    def recall(self, query: str, memory_type: str = None) -> str:
        """관련 기억을 검색하여 반환합니다."""
        entries = self.vector_memory.retrieve(query, top_k=5, memory_type=memory_type)
        if not entries:
            return "관련 기억이 없습니다."
 
        memories = []
        for entry in entries:
            memories.append(
                f"[{entry.memory_type}] {entry.content} "
                f"({entry.timestamp[:10]})"
            )
        return "\n".join(memories)

class UserProfileMemory:
    def __init__(self):
        self.profiles: dict[str, dict] = {}
        self.client = anthropic.Anthropic()
 
    def extract_preferences(self, user_id: str, conversation: str):
        """대화에서 사용자 선호를 추출합니다."""
        prompt = f"""다음 대화에서 사용자의 선호, 습관, 특성을 추출하십시오.
 
대화:
{conversation}
 
기존 프로필: {json.dumps(self.profiles.get(user_id, {}), ensure_ascii=False)}
 
JSON으로 업데이트된 프로필을 반환하십시오:
{{
  "preferences": {{"key": "value", ...}},
  "expertise_level": "beginner|intermediate|advanced",
  "communication_style": "formal|casual",
  "interests": ["topic1", "topic2"],
  "constraints": ["constraint1"]
}}"""
 
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}],
        )
        try:
            profile = json.loads(response.content[0].text)
            self.profiles[user_id] = profile
        except json.JSONDecodeError:
            pass
 
    def get_context(self, user_id: str) -> str:
        """사용자 프로필을 컨텍스트 문자열로 반환합니다."""
        profile = self.profiles.get(user_id, {})
        if not profile:
            return ""
        return f"[사용자 프로필]\n{json.dumps(profile, ensure_ascii=False, indent=2)}"

class MemoryAgent:
    def __init__(self, system_prompt: str, tools: list[dict]):
        self.client = anthropic.Anthropic()
        self.system_prompt = system_prompt
        self.tools = tools
        self.short_term = SummaryMemory(max_recent=10)
        self.long_term = AgentMemorySystem()
        self.user_profiles = UserProfileMemory()
 
    def run(self, user_id: str, user_message: str) -> str:
        """메모리를 활용한 에이전트를 실행합니다."""
 
        # 1. 관련 장기 기억 검색
        relevant_memories = self.long_term.recall(user_message)
 
        # 2. 사용자 프로필 로드
        user_context = self.user_profiles.get_context(user_id)
 
        # 3. 시스템 프롬프트에 메모리 주입
        enhanced_system = f"""{self.system_prompt}
 
{f'[관련 기억]{relevant_memories}' if relevant_memories != '관련 기억이 없습니다.' else ''}
{user_context}"""
 
        # 4. 단기 메모리에 메시지 추가
        self.short_term.add_message("user", user_message)
 
        # 5. 대화 히스토리 구성
        messages = self.short_term.get_context()
 
        # 6. LLM 호출
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            system=enhanced_system,
            tools=self.tools,
            messages=messages,
        )
 
        assistant_message = response.content[0].text
 
        # 7. 단기 메모리에 응답 추가
        self.short_term.add_message("assistant", assistant_message)
 
        # 8. 장기 메모리에 에피소드 저장
        self.long_term.remember_episode(user_message, assistant_message)
 
        # 9. 사용자 프로필 업데이트
        conversation = f"사용자: {user_message}\n에이전트: {assistant_message}"
        self.user_profiles.extract_preferences(user_id, conversation)
 
        return assistant_message

import math
from datetime import datetime, timedelta
 
class MemoryWithForgetting(VectorMemory):
    def retrieve(self, query: str, top_k: int = 5,
                 memory_type: str = None) -> list[MemoryEntry]:
        """시간 감쇠를 적용한 검색입니다."""
        query_embedding = self._get_embedding(query)
        now = datetime.now()
 
        scored = []
        for entry in self.entries:
            if memory_type and entry.memory_type != memory_type:
                continue
 
            # 의미적 유사도
            similarity = self._cosine_similarity(
                query_embedding, entry.embedding
            )
 
            # 시간 감쇠 (지수 함수)
            entry_time = datetime.fromisoformat(entry.timestamp)
            days_ago = (now - entry_time).days
            time_decay = math.exp(-0.01 * days_ago)
 
            # 접근 빈도 가중치
            access_count = entry.metadata.get("access_count", 0)
            frequency_boost = math.log(access_count + 1) * 0.1
 
            # 최종 점수
            final_score = similarity * 0.6 + time_decay * 0.3 + frequency_boost * 0.1
            scored.append((final_score, entry))
 
        scored.sort(key=lambda x: x[0], reverse=True)
        return [entry for _, entry in scored[:top_k]]
 
    def cleanup(self, max_age_days: int = 90, min_score: float = 0.1):
        """오래되고 사용되지 않는 메모리를 정리합니다."""
        cutoff = datetime.now() - timedelta(days=max_age_days)
        self.entries = [
            entry for entry in self.entries
            if (
                datetime.fromisoformat(entry.timestamp) > cutoff
                or entry.metadata.get("pinned", False)
                or entry.metadata.get("access_count", 0) > 5
            )
        ]

class MemoryConsolidator:
    def __init__(self, memory: VectorMemory):
        self.memory = memory
        self.client = anthropic.Anthropic()
 
    def consolidate(self, similarity_threshold: float = 0.85):
        """유사한 메모리를 통합합니다."""
        to_merge = []
 
        for i, entry_a in enumerate(self.memory.entries):
            for j, entry_b in enumerate(self.memory.entries[i + 1:], i + 1):
                sim = self.memory._cosine_similarity(
                    entry_a.embedding, entry_b.embedding
                )
                if sim > similarity_threshold:
                    to_merge.append((i, j, sim))
 
        # 유사도 높은 순으로 병합
        to_merge.sort(key=lambda x: x[2], reverse=True)
        merged_indices = set()
 
        for i, j, _ in to_merge:
            if i in merged_indices or j in merged_indices:
                continue
 
            merged_content = self._merge_entries(
                self.memory.entries[i],
                self.memory.entries[j],
            )
            self.memory.entries[i].content = merged_content
            self.memory.entries[i].embedding = self.memory._get_embedding(merged_content)
            merged_indices.add(j)
 
        # 병합된 항목 제거
        self.memory.entries = [
            e for idx, e in enumerate(self.memory.entries)
            if idx not in merged_indices
        ]
 
    def _merge_entries(self, a: MemoryEntry, b: MemoryEntry) -> str:
        prompt = f"""다음 두 메모리 항목을 하나로 통합하십시오.
중복을 제거하고 핵심 정보를 보존하십시오.
 
항목 1: {a.content}
항목 2: {b.content}
 
통합된 하나의 문장으로 작성하십시오."""
 
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=256,
            messages=[{"role": "user", "content": prompt}],
        )
        return response.content[0].text