beautifulsoups.dev
Published on

AutoGPT 메모리 폭주 해결 - 벡터DB+요약 TTL

Authors

AutoGPT류 에이전트를 운영하다 보면, 처음에는 잘 돌아가다가도 어느 순간부터 프롬프트가 비대해지고 응답이 느려지며 비용이 폭발합니다. 원인은 대개 단순합니다. 대화/작업 로그를 그대로 누적해 매 턴마다 LLM 컨텍스트에 밀어 넣기 때문입니다. 이 방식은 시간이 지날수록 토큰이 선형으로 증가하고, 검색도 없이 "전부 넣기"를 반복하므로 결국 메모리 폭주로 이어집니다.

이 글에서는 이를 구조적으로 해결하는 패턴인 벡터DB 기반 장기기억 + 요약 TTL(Time To Live) 조합을 다룹니다. 핵심은 다음 두 가지입니다.

  • 장기기억은 원문을 모두 넣지 말고 벡터DB에 저장하고 필요할 때만 검색해 주입한다.
  • 단기기억(최근 대화/상태)은 요약본으로 압축하되 TTL을 두고 주기적으로 재요약·만료시켜 컨텍스트를 일정하게 유지한다.

관련해 에이전트가 무한 루프나 토큰 폭탄으로 터지는 문제는 별도 설계 포인트가 많습니다. 함께 보면 좋은 글로 LangChain Agent 무한루프·토큰폭탄 차단 5팁도 추천합니다. 벡터DB만으로 비용을 줄이는 기본 방향은 AutoGPT 메모리 폭주? 벡터DB로 비용 절감에서 더 확장해 볼 수 있습니다.

왜 "메모리"가 폭주하는가: 3가지 병목

AutoGPT의 메모리 폭주는 보통 아래 3가지가 겹쳐서 발생합니다.

1) 컨텍스트 윈도우를 "로그 저장소"로 착각

LLM 컨텍스트는 DB가 아닙니다. 그런데 많은 구현이 다음처럼 동작합니다.

  • 매 턴마다 messages에 모든 히스토리를 append
  • 다음 호출에서 messages 전체를 그대로 전송

대화가 200턴만 넘어도 프롬프트 토큰이 수만 단위가 되고, 모델이 큰 컨텍스트를 지원하더라도 지연(latency) 과 비용(cost) 이 같이 증가합니다.

2) 검색 없는 장기기억은 결국 "전부 넣기"로 귀결

장기기억을 파일로 저장해도, 검색을 못 하면 결국 다시 전부 읽어서 넣게 됩니다. 이는 디스크가 아니라 컨텍스트를 폭주시킬 뿐입니다.

3) 요약을 하더라도 "요약의 요약"이 누적됨

요약은 좋은데, 요약을 매 턴 갱신하면서 이전 요약을 계속 덧붙이면 요약도 비대해집니다. 또한 오래된 요약이 계속 남아 있으면 현재 목표와 무관한 정보가 컨텍스트를 오염시킵니다.

따라서 검색 기반 장기기억(벡터DB) 과 만료 개념이 있는 단기요약(TTL) 을 같이 써야 합니다.

목표 아키텍처: 2-레벨 메모리 + TTL

권장하는 메모리 레이어는 다음처럼 나눕니다.

  • 단기기억(Working Memory)
    • 최근 N턴 원문(작게 유지)
    • 현재 목표/제약/체크리스트
    • 요약본(압축된 상태)
    • TTL 기반 재요약 및 만료
  • 장기기억(Long-term Memory)
    • 원문 이벤트(대화, 도구 결과, 파일 요약, 웹 리서치 등)
    • 임베딩 벡터 + 메타데이터
    • 필요 시 top-k 검색으로만 주입

즉, LLM에게 항상 주는 컨텍스트는 "최근 원문 + 최신 요약 + 검색 결과"의 합으로 고정되고, 전체 로그는 벡터DB에 안전하게 저장됩니다.

설계 1: 벡터DB 스키마(메타데이터가 절반)

벡터DB는 단순히 텍스트를 넣는 곳이 아니라, 검색 품질을 결정하는 메타데이터 저장소입니다. 최소한 아래 필드는 갖추는 것이 좋습니다.

  • id: UUID
  • project_id: 에이전트/작업 단위
  • type: chat, tool_result, web_snippet, file_note 등
  • text: 저장할 원문(또는 정제된 원문)
  • embedding: 임베딩 벡터
  • created_at: 생성 시각
  • ttl_at: 만료 시각(선택)
  • importance: 중요도(0~1)
  • tags: billing, auth, deadline 같은 키워드
  • source: URL, 파일 경로 등(부등호가 있으면 반드시 인라인 코드 처리)

메타데이터가 있어야 다음이 가능합니다.

  • 프로젝트별 필터링
  • 타입별 검색(예: 도구 결과만)
  • 시간 가중치(최신 우선)
  • 중요도 기반 리랭킹

설계 2: 요약 TTL의 핵심 규칙

요약 TTL을 도입할 때 중요한 규칙은 "요약은 영구 저장물이 아니라 캐시"라는 점입니다.

권장 규칙

  1. 요약은 일정 토큰 예산을 넘으면 재요약한다.
  • 예: 요약이 800토큰을 넘으면 400토큰 목표로 재요약
  1. 요약은 일정 시간이 지나면 만료된다.
  • 예: 30분 또는 2시간
  • 만료 시: 최신 원문 N턴 + 벡터DB 검색 결과로 다시 요약 생성
  1. 요약에는 "사실"과 "미결 과제"를 분리해서 담는다.
  • 사실: 변경 가능성 낮음(요구사항, 결정사항)
  • 미결: 다음 액션, 막힌 지점, 확인 필요
  1. 요약 생성 시 "금지" 목록을 둔다.
  • 개인식별정보, 토큰/키, 장황한 로그 등

이렇게 하면 요약이 오래된 상태를 고착시키지 않고, 컨텍스트도 일정하게 유지됩니다.

파이프라인: 저장 → 요약 → 검색 → 주입

아래는 한 턴이 끝날 때 수행할 파이프라인 예시입니다.

  1. 이벤트 수집
  • 사용자 입력, 모델 응답, 도구 호출 결과를 이벤트로 표준화
  1. 장기기억 저장(벡터DB upsert)
  • 원문 또는 정제 텍스트를 임베딩 후 저장
  1. 단기요약 갱신
  • 토큰 예산 초과 또는 TTL 만료 시 재요약
  1. 다음 턴 프롬프트 구성
  • 시스템/정책
  • 작업 목표
  • 최근 N턴 원문
  • 최신 요약(짧게)
  • 벡터DB 검색 top-k(짧게)

코드 예제: Python으로 벡터DB + 요약 TTL (미니멀)

아래 예시는 개념을 보여주는 목적의 간단한 구현입니다. 벡터DB는 예시로 SQLite 기반의 간단 저장소처럼 보이게 만들었고, 실제 운영에서는 Chroma, Qdrant, Weaviate, pgvector 등을 쓰면 됩니다.

from __future__ import annotations

import time
import uuid
from dataclasses import dataclass
from typing import List, Optional, Dict, Any


def now_ms() -> int:
    return int(time.time() * 1000)


@dataclass
class MemoryItem:
    id: str
    project_id: str
    type: str
    text: str
    embedding: List[float]
    created_at_ms: int
    ttl_at_ms: Optional[int] = None
    importance: float = 0.5
    tags: Optional[List[str]] = None


class VectorStore:
    """예시용 인터페이스. 실제로는 Qdrant/Chroma/pgvector 등으로 대체."""

    def __init__(self):
        self.items: List[MemoryItem] = []

    def upsert(self, item: MemoryItem) -> None:
        self.items.append(item)

    def search(self, project_id: str, query_emb: List[float], k: int = 5) -> List[MemoryItem]:
        # 예시: 코사인 유사도 대신 임의 점수. 실제 구현에서는 벡터 유사도 사용.
        def score(it: MemoryItem) -> float:
            return it.importance

        candidates = [it for it in self.items if it.project_id == project_id]
        candidates.sort(key=score, reverse=True)
        return candidates[:k]


class SummaryCache:
    def __init__(self, ttl_ms: int, max_tokens: int):
        self.ttl_ms = ttl_ms
        self.max_tokens = max_tokens
        self.summary: str = ""
        self.updated_at_ms: int = 0

    def is_expired(self) -> bool:
        return (now_ms() - self.updated_at_ms) > self.ttl_ms

    def token_estimate(self, text: str) -> int:
        # 매우 러프한 추정. 실제로는 tiktoken 같은 토크나이저 사용 권장.
        return max(1, len(text) // 4)

    def should_resummarize(self) -> bool:
        return self.token_estimate(self.summary) > self.max_tokens


# ---- 아래는 LLM/임베딩 호출을 대체하는 스텁 ----

def embed(text: str) -> List[float]:
    # 실제로는 text-embedding 모델 호출
    return [0.0, 1.0, 2.0]


def llm_summarize(prompt: str) -> str:
    # 실제로는 요약 프롬프트로 LLM 호출
    return "[요약] " + prompt[:200]


def build_summary_prompt(prev_summary: str, recent_events: List[Dict[str, Any]], retrieved: List[MemoryItem]) -> str:
    recent_text = "\n".join([f"- {e['type']}: {e['text']}" for e in recent_events])
    retrieved_text = "\n".join([f"- ({m.type}) {m.text}" for m in retrieved])

    return (
        "다음 정보를 바탕으로 작업 진행에 필요한 핵심만 요약하라. "
        "결정사항/사실과 미결 과제를 분리하라.\n\n"
        f"이전 요약:\n{prev_summary}\n\n"
        f"최근 이벤트:\n{recent_text}\n\n"
        f"참고(검색된 장기기억):\n{retrieved_text}\n"
    )


def on_turn_end(
    project_id: str,
    events: List[Dict[str, Any]],
    vector_store: VectorStore,
    summary_cache: SummaryCache,
) -> None:
    # 1) 벡터DB 저장
    for e in events:
        item = MemoryItem(
            id=str(uuid.uuid4()),
            project_id=project_id,
            type=e["type"],
            text=e["text"],
            embedding=embed(e["text"]),
            created_at_ms=now_ms(),
            ttl_at_ms=None,
            importance=e.get("importance", 0.5),
            tags=e.get("tags", []),
        )
        vector_store.upsert(item)

    # 2) 요약 TTL / 토큰 예산 체크
    if summary_cache.is_expired() or summary_cache.should_resummarize() or not summary_cache.summary:
        query = "현재 작업 목표와 관련된 핵심 이력"
        retrieved = vector_store.search(project_id=project_id, query_emb=embed(query), k=5)
        prompt = build_summary_prompt(summary_cache.summary, events[-10:], retrieved)
        summary_cache.summary = llm_summarize(prompt)
        summary_cache.updated_at_ms = now_ms()


def build_prompt(
    system_policy: str,
    goal: str,
    recent_messages: List[Dict[str, str]],
    summary_cache: SummaryCache,
    vector_store: VectorStore,
    project_id: str,
) -> str:
    query = goal
    retrieved = vector_store.search(project_id=project_id, query_emb=embed(query), k=5)

    retrieved_block = "\n".join([f"- ({m.type}) {m.text}" for m in retrieved])
    recent_block = "\n".join([f"{m['role']}: {m['content']}" for m in recent_messages])

    return (
        f"[SYSTEM]\n{system_policy}\n\n"
        f"[GOAL]\n{goal}\n\n"
        f"[SUMMARY]\n{summary_cache.summary}\n\n"
        f"[RECENT]\n{recent_block}\n\n"
        f"[RETRIEVED]\n{retrieved_block}\n"
    )

이 예제에서 중요한 지점은 다음입니다.

  • 원문 이벤트는 벡터DB에 계속 쌓이지만, LLM에 매번 전부 주입하지 않는다.
  • 요약은 ttl_ms 또는 max_tokens 기준으로만 갱신한다.
  • 다음 프롬프트는 SUMMARY 와 RETRIEVED 를 통해 필요한 정보만 가져온다.

검색 품질을 올리는 실전 팁: "무엇을" 저장하고 "어떻게" 자를 것인가

벡터 검색이 잘 되려면 저장 텍스트가 적절히 정제되어야 합니다.

1) Chunking(분할) 기준

  • 도구 결과: 표/로그 전체를 한 덩어리로 넣지 말고, 섹션별로 나누기
  • 웹 문서: 제목/소제목 단위로 나누기
  • 코드/스택트레이스: 원문을 그대로 넣되, 앞뒤 맥락을 2~3줄만 포함

너무 작은 청크는 문맥이 사라지고, 너무 큰 청크는 검색이 둔해집니다. 경험적으로는 200~600 토큰 정도가 무난합니다.

2) 메타데이터 필터를 먼저 걸고 top-k

예를 들어 "도구 실행 결과"만 필요하면 type=tool_result 로 필터한 뒤 top-k를 뽑는 편이 정확합니다.

3) 시간 가중치 + 중요도 가중치

검색 결과는 유사도만으로 정렬하면 오래된 정보가 계속 뜰 수 있습니다. 다음처럼 점수에 가중치를 주면 안정적입니다.

  • 최종점수 = 유사도 * 0.7 + 최신도 * 0.2 + 중요도 * 0.1

요약 TTL을 더 강하게 만드는 패턴: "상태머신" 분리

요약에 모든 것을 넣으려 하면 다시 비대해집니다. 요약을 다음 3파트로 분리하면 유지보수가 쉬워집니다.

  • facts: 변하지 않는 사실(요구사항, 결정)
  • open_tasks: 해야 할 일 목록(체크박스 형태 권장)
  • current_context: 최근 맥락(짧게)

그리고 open_tasks 는 TTL을 길게, current_context 는 TTL을 짧게 주는 식으로 차등 적용할 수 있습니다.

운영 관점 체크리스트: 폭주를 "탐지"해야 막는다

메모리 폭주는 사후 대응이 어렵습니다. 아래 지표를 로깅하면 빨리 감지할 수 있습니다.

  • 턴당 입력 토큰 수, 출력 토큰 수
  • 프롬프트 구성 요소별 토큰 비중
    • recent, summary, retrieved 각각
  • 벡터DB 검색 top-k의 평균 길이(문자수/토큰수)
  • 요약 갱신 빈도(분당/시간당)
  • 에이전트 루프 횟수(동일 목표에서 반복 호출)

특히 프롬프트 지연이 체감될 때는 스트리밍 환경에서 더 크게 느껴집니다. 로컬 LLM 스트리밍이 끊기거나 지연되는 현상과도 연결되므로, 필요하면 Transformers 로컬 LLM 스트리밍 끊김·지연 해결도 함께 점검해보는 것이 좋습니다.

흔한 실패 5가지와 처방

1) 요약이 계속 커진다

  • 처방: 요약 목표 토큰을 명시하고, 초과 시 재요약. 요약에 "근거 로그"를 넣지 않기.

2) 검색 결과가 엉뚱하다

  • 처방: 저장 시 타입/태그를 강화하고, 검색 시 메타데이터 필터를 먼저 적용. chunk 크기 재조정.

3) 중요한 결정이 TTL로 사라진다

  • 처방: 결정사항은 facts로 분리하고 TTL을 길게 또는 영구로. 또는 별도 "결정 레지스트리" 문서로 승격.

4) 비용은 줄었는데 성능이 나빠진다

  • 처방: top-k를 무작정 줄이지 말고, 검색 결과를 짧게 재요약해서 주입(검색 요약 레이어 추가).

5) 에이전트가 같은 질문을 반복한다

  • 처방: open_tasks에 "이미 시도한 것"과 "실패 원인"을 남기고, 반복 감지 시 강제 중단/휴리스틱 적용.

결론: "전부 넣기"를 버리고, 기억을 계층화하라

AutoGPT 메모리 폭주의 본질은 컨텍스트를 저장소로 쓰는 데서 시작합니다. 해결책은 단순한 최적화가 아니라 아키텍처 전환입니다.

  • 장기기억은 벡터DB로 보내고 검색으로만 가져오기
  • 단기기억은 요약하되 TTL과 토큰 예산으로 강제 수축
  • 프롬프트는 recent + summary + retrieved의 고정 예산으로 운영

이 조합을 적용하면 비용과 지연이 안정화되고, 에이전트가 장시간 실행되는 워크플로에서도 컨텍스트 품질을 유지할 수 있습니다.

Discuss on Twitter • View on GitHub