에이전트 메모리와 상태 관리: 단기/장기 메모리와 컨텍스트 압축

'AI 에이전트 엔지니어링(AI Agent Engineering)' 시리즈의 2편입니다. 다단계 작업에 필요한 상태(State) 스키마, 단기·장기 메모리, 컨텍스트 압축(Context Compression)을 다룹니다.

---

1. 에이전트 상태(State) 스키마 설계: Pydantic v2 & TypedDict

대형 언어 모델(LLM) 본체는 상태가 없는(Stateless) 함수에 불과합니다. 동일한 프롬프트와 옵션을 전달하면 매번 독립적인 토큰 출력을 생성할 뿐, 이전 호출에서 수행한 일이나 도구(Tool) 실행 결과를 스스로 기억하지 못합니다.

따라서 AI 에이전트 런타임의 핵심은 "상태(State)를 에이전트 그래프(Graph) 내부에서 어떻게 일관되게 정의하고 불변성(Immutability)을 유지하며 전달할 것인가?"에 있습니다.

LLM 자체는 Stateless하지만, 에이전트 런타임은 Stateful한 상태 전이 시스템(State Transition System)입니다. 각 상태 전이(State Transition) 시점마다 Reducer 함수를 통해 이전 상태와 노드의 출력 결과를 결정론적으로 병합(Merge)합니다.

Stateless 호출은 Prompt Input → LLM Engine → Token Output으로 끝납니다. Stateful 런타임은 현재 상태와 노드 출력을 함께 Reducer에 전달합니다.

상태 스키마(State Schema) 설계의 3가지 핵심 원칙

1. 단일 진실 출처 (Single Source of Truth): 에이전트 노드 간 이동하는 State 객체는 현재 진행 중인 과업의 모든 정보(대화 이력, 도구 실행 결과, 서브 타스크 성공 여부, 사용자 메타데이터)를 담는 유일한 객체여야 합니다.
2. 명확한 전이 규칙 (Reducer Pattern): 노드가 상태를 업데이트할 때 기존 값을 덮어쓸지(Overwrite), 리스트에 추가할지(Append), 혹은 병합할지(Merge) 명시적으로 정의해야 합니다.
3. 엄격한 런타임 타입 검증 (Strict Runtime Validation): 잘못된 데이터가 노드 간에 전달되는 것을 방지하기 위해 Pydantic v2 또는 TypedDict 기반의 스키마 검증이 필수적입니다.

TypedDict vs Pydantic v2 비교 및 실전 구현

TypedDict는 Python 내장 타입 힌팅 메커니즘으로 오버헤드가 거의 없습니다. 에이전트 그래프(예: LangGraph) 내에서 노드 간 고속 데이터 전달과 Reducer 타입 어노테이션에 유리합니다.

Pydantic v2는 C++ 엔진 기반(pydantic-core)의 강력한 런타임 데이터 검증, 커스텀 Validator, 데이터 직렬화/역직렬화(JSON Schema, Redis/RDBMS Persistence) 기능을 제공합니다. External Gateway나 API 경계에서 에이전트 상태를 검증할 때 필수적입니다.

다음은 TypedDict와 Pydantic v2를 조합한 상태 스키마 예시입니다.

Copy1from typing import Annotated, Any, Dict, List, Literal, Optional, TypedDict2from pydantic import BaseModel, Field, ConfigDict3 4 5class ChatMessage(BaseModel):6    """단일 메시지 스키마"""7    role: Literal["system", "user", "assistant", "tool"]8    content: str9    name: Optional[str] = None10    tool_call_id: Optional[str] = None11    metadata: Dict[str, Any] = Field(default_factory=dict)12 13 14def add_messages(left: List[ChatMessage], right: List[ChatMessage]) -> List[ChatMessage]:15    """메시지 리스트를 안전하게 병합하는 Reducer 함수"""16    return left + right17 18 19# 1. TypedDict 기반 그래프 상태 스키마 (LangGraph 등 그래프 런타임용)20class GraphAgentState(TypedDict):21    session_id: str22    user_id: str23    messages: Annotated[List[ChatMessage], add_messages]24    current_step: int25    max_steps: int26    scratchpad: Dict[str, Any]27    is_completed: bool28 29 30# 2. Pydantic v2 기반 런타임 API 경계 검증 모델31class AgentStateModel(BaseModel):32    model_config = ConfigDict(arbitrary_types_allowed=True)33 34    session_id: str = Field(description="대화 및 실행 세션 고유 식별자")35    user_id: str = Field(description="사용자 ID")36    messages: List[ChatMessage] = Field(37        default_factory=list,38        description="대화 이력 및 도구 실행 메시지 버퍼"39    )40    current_step: int = Field(default=0, ge=0, description="현재 진행 중인 루프 스텝 수")41    max_steps: int = Field(default=15, gt=0, description="최대 허용 루프 스텝 수")42    scratchpad: Dict[str, Any] = Field(43        default_factory=dict,44        description="노드 간 일시적 데이터 공유를 위한 스크래치패드"45    )46    is_completed: bool = Field(default=False, description="과업 종료 여부")

---

2. 단기 메모리 vs 장기 메모리 아키텍처

에이전트의 메모리는 사용하는 목적과 보존 기간에 따라 나눌 수 있습니다. 여기서는 작업 기억(Working Memory), 에피소드 기억(Episodic Memory), 의미 기억(Semantic Memory)을 구분합니다.

User Input은 Short-Term Memory(작업 기억 / 대화 버퍼)로 들어갑니다. Agent Runtime System 안의 실행 순환은 Short-Term Memory → Context Window Assembly → LLM Engine → Tool Execution → Short-Term Memory입니다.

메모리 생애주기 전이 모델 (Memory Hierarchy State Diagram)

단기 메모리 버퍼에서 임계치를 초과한 대화 맥락은 동적 요약기를 거쳐 에피소딕 메모리(Vector DB)와 세만틱 메모리(Fact Base/RDBMS)로 분기되어 영구 저장됩니다.

상태 전이는 내부 동작과 바깥 연결을 나누어 읽습니다. 각 진입점은 해당 범위의 시작을 뜻합니다.

메모리 계층별 상세 비교 매트릭스 (Working vs Episodic vs Semantic vs Procedural)

Experience Replay(에피소딕 메모리 회상) 기법을 적용하면 에이전트가 복잡한 데이터 변환이나 API 호출을 수행할 때, 과거에 유사한 에러를 극복했던 성공 트레이스를 검색해 Few-Shot 예시로 주입할 수 있습니다. 이는 제로샷 프롬프팅 대비 도구 호환 오류율을 40% 이상 낮춥니다.

장기 메모리 연동 메커니즘 (RAG & Experience Replay)

장기 메모리는 단순 RAG(Retrieval-Augmented Generation) 문서 검색에 그치지 않습니다. 고급 에이전트 엔지니어링에서는 "에피소딕 메모리 회상" 기술을 활용합니다.

1. 유사 과업 검색 (Similarity Search): 사용자의 입력이 들어오면 과거에 가장 유사했던 과업의 해결 트레이스(Execution Trace)를 Vector Store에서 검색합니다.
2. In-Context Few-Shot Injection: 검색된 과거 성공 사례를 시스템 프롬프트에 Few-Shot 예시로 삽입하여 에이전트의 환각(Hallucination)을 줄이고 도구 호출 정확도를 대폭 향상시킵니다.

---

3. 컨텍스트 윈도우 한계와 압축(Context Compression) 전략

LLM의 컨텍스트 윈도우가 128k, 1M 토큰으로 대폭 확장되었음에도 불구하고, 대량의 컨텍스트를 그대로 유지하는 것은 실무에서 큰 장애물이 됩니다.

무작정 거대한 컨텍스트를 유지하면 API 비용 상승뿐만 아니라 프롬프트 중앙의 핵심 정보 손실(Lost in the Middle)로 인해 에이전트의 의도 파악 성능이 대폭 저하됩니다.

거대 컨텍스트 유지 시 발생하는 3가지 문제점

1. 비용 및 지연시간 급증: 입력 토큰 수에 비례하여 API 비용과 첫 번째 토큰 생성 시간(TTFT, Time-to-First-Token)이 선형 또는 이차 함수적으로 증가합니다.
2. Lost in the Middle (중간 정보 실종): 프롬프트 중앙에 위치한 핵심 지시사항이나 도구 실행 결과를 LLM 주의력(Attention) 메커니즘이 놓치는 현상이 발생합니다.
3. Context Drift (컨텍스트 표류): 대화가 길어질수록 초기의 시스템 지시사항이나 유저 목적이 누적된 도구 호출 결과 메시지들에 파묻혀 왜곡됩니다.

컨텍스트 압축 파이프라인 시퀀스 (Context Compression Sequence Flow)

에이전트 런타임이 유저 쿼리를 수신하고 최종 LLM 프롬프트 윈도우를 구성하기까지 실행되는 메시지 정제, 토큰 계산, 동적 요약, 장기 메모리 플러시의 시퀀스 흐름입니다.

3대 컨텍스트 압축 기법 비교

---

4. 구현 실전: Python 기반 윈도우 메모리 및 상태 압축 파이프라인

다음 AgentMemoryManager 예시는 Pydantic v2와 Python asyncio를 사용합니다. 토큰 예산에 따른 슬라이딩 윈도우, 도구 결과 절삭, LLM 요약, 메모리 플러시를 연결합니다.

메모리 관리자 파이프라인 흐름도:

↳토큰 한도를 넘을 때만 요약하기

도식을 글로 읽기

AgentMemoryManager는 Raw Conversation Input에서 오래된 Tool Payload를 잘라내고 System + Summary + History의 토큰을 추정합니다. Total Tokens > Trigger Limit이면 Old/Recent 분리, 오래된 메시지의 비동기 LLM 요약, Summary Buffer 갱신과 LTM Flush를 거칩니다. 아니면 이 세 단계를 건너뜁니다. 두 경로는 Context Window 조립에서 합류하여 System + Persistent Summary + Recent Window의 Final Compressed Context Array를 반환합니다.

AgentMemoryManager는 Raw Conversation Input에서 오래된 Tool Payload를 잘라내고 System + Summary + History의 토큰을 추정합니다. Total Tokens > Trigger Limit이면 Old/Recent 분리, 오래된 메시지의 비동기 LLM 요약, Summary Buffer 갱신과 LTM Flush를 거칩니다. 아니면 이 세 단계를 건너뜁니다. 두 경로는 Context Window 조립에서 합류하여 System + Persistent Summary + Recent Window의 Final Compressed Context Array를 반환합니다.

크게 보기 ↗

실무 프로덕션 환경에서는 estimate_tokens 구현 시 단순 글자 수 비례 연산 대신 tiktoken 라이브러리를 바인딩하여 사용 중인 모델의 정확한 BPE 토큰 수 파싱을 권장합니다.

Copy1import asyncio2from typing import List, Dict, Any, Optional, Literal3from pydantic import BaseModel, Field4 5 6class ChatMessage(BaseModel):7    """메시지 데이터 모델"""8    role: Literal["system", "user", "assistant", "tool"] = Field(..., description="메시지 발신자 역할")9    content: str = Field(..., description="메시지 본문")10    name: Optional[str] = Field(None, description="도구 이름 또는 작성자")11    tool_call_id: Optional[str] = Field(None, description="도구 호출 식별자")12 13 14class MemoryConfig(BaseModel):15    """메모리 관리자 설정 모델"""16    max_token_budget: int = Field(default=4000, description="최대 허용 토큰 예산")17    sliding_window_size: int = Field(default=10, description="슬라이딩 윈도우로 보존할 최신 메시지 수")18    tool_pruning_threshold: int = Field(default=150, description="오래된 Tool 출력의 절삭 임계 문자 수")19    summary_trigger_ratio: float = Field(default=0.8, description="요약 트리거 토큰 예산 비율")20 21 22class AgentMemoryManager:23    """24    토큰 예산 초과 방지 및 컨텍스트 압축을 담당하는 메모리 관리자25    """26    def __init__(self, config: MemoryConfig) -> None:27        self.config: MemoryConfig = config28        self.summary_buffer: str = ""29 30    def estimate_tokens(self, text: str) -> int:31        """토큰 수 추정 (실무에서는 tiktoken 사용 권장)"""32        if not text:33            return 034        return max(1, len(text) // 3)35 36    def prune_tool_outputs(self, messages: List[ChatMessage]) -> List[ChatMessage]:37        """오래된 대용량 Tool 실행 결과를 절삭하여 토큰 공간 확보"""38        pruned_messages: List[ChatMessage] = []39        for i, msg in enumerate(messages):40            if msg.role == "tool" and len(messages) - i > 3:41                if len(msg.content) > self.config.tool_pruning_threshold:42                    truncated_content = (43                        msg.content[:self.config.tool_pruning_threshold]44                        + f"\n... [Tool Output Truncated: Total Length {len(msg.content)} chars]"45                    )46                    pruned_messages.append(47                        ChatMessage(48                            role=msg.role,49                            content=truncated_content,50                            name=msg.name,51                            tool_call_id=msg.tool_call_id52                        )53                    )54                    continue55            pruned_messages.append(msg)56        return pruned_messages57 58    async def summarize_old_messages(self, messages_to_summarize: List[ChatMessage]) -> str:59        """오래된 대화 이력을 LLM을 이용해 요약 (가상 비동기 LLM 호출)"""60        print(f"[MemoryManager] Summarizing {len(messages_to_summarize)} old messages...")61        await asyncio.sleep(0.05)  # LLM API Latency 흉내62 63        extracted_facts = [64            f"- {m.role}: {m.content[:50]}..."65            for m in messages_to_summarize66            if m.role in ["user", "assistant"]67        ]68        return "이전 대화 핵심 요약:\n" + "\n".join(extracted_facts)69 70    async def prepare_context_window(71        self,72        system_prompt: ChatMessage,73        conversation_history: List[ChatMessage]74    ) -> List[ChatMessage]:75        """LLM에 전달할 최종 컨텍스트 윈도우 생성 및 압축 파이프라인"""76        # 1단계: Tool Output Pruning77        pruned_history = self.prune_tool_outputs(conversation_history)78 79        # 2단계: 전체 토큰 계산80        total_tokens = self.estimate_tokens(system_prompt.content) + self.estimate_tokens(self.summary_buffer)81        for msg in pruned_history:82            total_tokens += self.estimate_tokens(msg.content)83 84        trigger_limit = int(self.config.max_token_budget * self.config.summary_trigger_ratio)85 86        # 3단계: 토큰 예산 초과 시 Summarization & Windowing87        if total_tokens > trigger_limit and len(pruned_history) > self.config.sliding_window_size:88            cutoff_index = len(pruned_history) - self.config.sliding_window_size89            old_messages = pruned_history[:cutoff_index]90            recent_messages = pruned_history[cutoff_index:]91 92            new_summary = await self.summarize_old_messages(old_messages)93            self.summary_buffer = f"{self.summary_buffer}\n{new_summary}".strip()94 95            pruned_history = recent_messages96 97        # 4단계: 최종 메시지 리스트 조립 (System Prompt -> Summary Message -> Recent Window)98        final_messages: List[ChatMessage] = [system_prompt]99 100        if self.summary_buffer:101            final_messages.append(102                ChatMessage(103                    role="system",104                    content=f"[지속 메모리 요약 정보]\n{self.summary_buffer}"105                )106            )107 108        final_messages.extend(pruned_history)109        return final_messages110 111 112# --- 사용 예시 및 테스트 실행 ---113async def main() -> None:114    config = MemoryConfig(max_token_budget=100, sliding_window_size=3, tool_pruning_threshold=150)115    memory_mgr = AgentMemoryManager(config)116 117    sys_prompt = ChatMessage(role="system", content="당신은 유능한 파이썬 에이전트입니다.")118 119    history = [120        ChatMessage(role="user", content="데이터베이스 접속 설정 방법을 알려줘."),121        ChatMessage(role="assistant", content="PostgreSQL 접속을 위한 SQLAlchemy 설정을 준비합니다."),122        ChatMessage(role="tool", content="{" * 100 + "DB_CONNECTION_SUCCESS_FULL_LOG" + "}" * 100),123        ChatMessage(role="user", content="커넥션 풀 개수는 몇 개로 설정해야 해?"),124        ChatMessage(role="assistant", content="기본적으로 10개를 권장합니다."),125        ChatMessage(role="user", content="비동기 asyncpg 설정 예제도 보여줘."),126    ]127 128    compressed_context = await memory_mgr.prepare_context_window(sys_prompt, history)129 130    print("\n=== 최종 LLM에 전달되는 압축 컨텍스트 윈도우 ===")131    for idx, msg in enumerate(compressed_context):132        print(f"[{idx}] {msg.role.upper()}: {msg.content[:80]}...")133 134 135if __name__ == "__main__":136    asyncio.run(main())

---

5. 프로덕션 에이전트의 방어적 엔지니어링 체크리스트

실제 운영 환경에서 메모리 및 상태 관리 미흡으로 발생하는 사고를 방지하기 위해 다음 체크리스트와 위험 대응 체계를 준수해야 합니다.

프로덕션 에이전트에서는 상태 동시성(Concurrency), 개인정보(PII) 누출, 무한 루프 토큰 소비에 대비한 방어적 프로그래밍 (Defensive Programming) 조치가 필수적입니다.

방어적 엔지니어링 리스크 및 대응 매트릭스

1. 무한 루프 토큰 누수 (Infinite Loop Memory Leaks) 방지

• 최대 스텝 제한: max_steps를 명시하여 특정 노드에서 무한히 재시도하는 것을 차단합니다.
• 중복 메시지 감지: 동일한 도구(Tool) 호출 및 결과가 연속 3회 이상 반복될 경우, 실행을 중단하고 상위 예외 처리기로 격리합니다.

2. 메모리 오염(Memory Contamination) 및 환각 차단

• 사용자 추측과 Fact 분리: 사용자의 미검증 주장을 장기 에피소딕 메모리에 그대로 영구 저장하지 마십시오. 반드시 시스템 검증 절차를 거친 사실(Fact)만 이관해야 합니다.
• System Prompt Anchor 보존: 슬라이딩 윈도우 처리 시 system 역할의 핵심 페르소나 및 안전 지침(Safety Guardrails)이 지워지지 않도록 항상 배열의 0번 인덱스로 고정하십시오.

3. 동시성(Concurrency) 및 분산 상태 관리

• 상태 원자성(Atomicity): Multi-Agent 시스템이나 분산 환경에서는 Redis Lock 또는 데이터베이스 트랜잭션을 활용하여 동일한 session_id의 상태 덮어쓰기 충돌(Race Condition)을 방지하십시오.
• PII 마스킹: 장기 메모리(Vector DB)에 대화 이력을 저장하기 전, 주민등록번호, API Key, 개인 이메일 등의 개인정보를 정규식이나 PII Detector 서비스로 마스킹하십시오.

---

6. 마치며

상태 스키마는 노드 사이에 전달할 데이터를 정하고, 메모리 계층은 보존할 내용을 나눕니다. 컨텍스트 압축은 그중 현재 호출에 보낼 분량을 조절합니다.

핵심을 요약하자면:

1. 상태(State)는 Pydantic v2와 TypedDict 및 Reducer 패턴을 통해 불변성과 런타임 안정성을 확보해야 합니다.
2. 단기 메모리의 과도한 토큰 팽창은 Tool Result Pruning, Sliding Window, LLM Summarization 파이프라인으로 해결해야 합니다.
3. 장기 메모리는 단순 문맥 검색을 넘어, 에이전트의 과거 성공 트레이스를 회상하는 에피소딕 메모리로 확장되어야 합니다.

다음 03편에서는 다양한 내부/외부 도구를 안전하고 정확하게 호출하는 "Tool Calling & Safe Execution Environment (샌드박스 보안 및 실패 복구)"를 다루겠습니다.