BabyAGI 메모리 폭증 해결 - 벡터DB TTL·요약체인

BabyAGI류 에이전트를 오래 돌리다 보면 어느 순간부터 속도가 급격히 느려지고, 프로세스 RSS 메모리와 벡터DB 디스크 사용량이 계속 증가합니다. 원인은 단순합니다. 작업(Task)과 관찰(Observation), 도구 호출 결과, 중간 추론 산출물이 계속 “기억”으로 축적되는데, 대부분은 다시는 쓰이지 않거나, 쓰이더라도 원문 전체가 아니라 핵심만 필요하기 때문입니다.

이 글에서는 BabyAGI 메모리 폭증을 벡터DB TTL(만료 정책) 과 요약 체인(summarization chain) 으로 제어하는 방법을 정리합니다. 목표는 다음 3가지입니다.

장시간 실행해도 메모리/디스크가 선형 증가하지 않게 만들기
검색 품질을 크게 떨어뜨리지 않으면서 기억을 압축하기
운영 관점에서 “왜 삭제/요약됐는지” 추적 가능하게 설계하기

관련해서 에이전트 무한 루프나 툴 호출 난사까지 겹치면 비용과 상태가 함께 폭주합니다. 그 부분은 LangChain 에이전트 무한루프·툴난사 차단법, 비용 상한 설계는 AutoGPT 비용 폭주 막기 - 토큰·툴 호출 한도 설계도 함께 참고하면 좋습니다.

1) BabyAGI 메모리 폭증의 구조적 원인

BabyAGI 패턴은 보통 아래 흐름을 반복합니다.

목표를 기반으로 작업을 생성
작업을 실행하면서 도구를 호출
결과를 메모리(벡터DB 포함)에 저장
저장된 메모리를 검색해 다음 작업에 반영

문제는 3번이 “append-only”로 구현되는 경우가 많다는 점입니다.

벡터DB에 문서가 계속 쌓임(중복, 저품질 로그, 임시 결과 포함)
메타데이터 없이 저장해 삭제/정리 불가
매 반복마다 검색 대상이 커져서 쿼리 지연 증가
상위 K 검색을 하더라도 인덱스 스캔/리랭크 비용 증가

즉, 저장 정책이 없어서 생기는 운영 문제입니다. 해결은 “저장하지 말자”가 아니라 저장 수명과 압축 정책을 명시하는 것입니다.

2) 해결 전략 개요: TTL + 요약 체인 + 계층 메모리

권장하는 설계는 3계층입니다.

핫 메모리(Hot): 최근 N분~N시간의 원문. 빠른 회상용.
웜 메모리(Warm): 요약된 기억. 장기 문맥 유지용.
콜드 메모리(Cold): 원문 아카이브(옵션). 감사/재현 목적.

핵심은 다음 두 가지 제어 장치입니다.

TTL(만료): 핫 메모리는 자동 삭제되게 해서 무한 성장을 끊습니다.
요약 체인: TTL로 지우기 전에 “필요한 정보만” 웜 메모리로 압축해 남깁니다.

이때 요약은 단순 요약이 아니라, 검색 가능한 형태로 만들어야 합니다.

사실/결정/수치/링크/파일명 등 키 포인트를 구조화
태스크와의 연결(어떤 목표/작업에서 나온 기억인지)을 메타데이터로 보존

3) 벡터DB TTL 설계: 무엇을 언제 지울 것인가

3.1 TTL이 필요한 데이터 유형

다음 데이터는 TTL 대상입니다.

도구 호출의 원시 출력(특히 대용량 JSON, HTML, 로그)
실패한 시도/재시도 로그
중간 생각/스텝별 메모(최종 결론이 이미 요약에 반영된 경우)
동일 태스크에서 중복 저장되는 관찰

반대로 TTL을 길게 두거나 콜드로 보내야 하는 데이터는 다음입니다.

사용자 요구사항의 핵심(변하지 않는 제약)
최종 결정과 근거(나중에 재현 필요)
시스템 설정/정책(툴 사용 규칙 등)

3.2 TTL 정책 예시

핫 메모리 TTL: 6h 또는 24h
실패/노이즈 문서 TTL: 30m~2h
웜 메모리 TTL: 길게(예: 30d) 또는 무기한(단, 주기적 재요약)

운영에서는 “문서 타입별 TTL”이 중요합니다. 문서마다 doc_type을 메타데이터로 넣고, 타입별로 다른 TTL을 적용하세요.

3.3 TTL 구현 포인트

벡터DB마다 TTL 지원 방식이 다릅니다.

네이티브 TTL 지원: 컬렉션/레코드에 만료 필드를 두고 자동 삭제
미지원: 애플리케이션 레벨에서 expires_at 메타데이터를 두고 주기적으로 purge

여기서는 “미지원”을 가정한 범용 패턴을 제시합니다.

예시: 문서 저장 스키마(메타데이터)

memory_id: UUID
run_id: 실행 세션
task_id: 작업 식별자
doc_type: tool_output, observation, decision, summary
importance: 0~1
created_at: epoch seconds
expires_at: epoch seconds
source: browser, filesystem, sql, user

이 메타데이터가 있어야 “언제 무엇을 지울지”를 자동화할 수 있습니다.

4) 요약 체인 설계: 지우기 전에 압축해 남기기

TTL만 적용하면 중요한 문맥까지 사라져 다음 루프에서 품질이 급락할 수 있습니다. 그래서 TTL로 삭제하기 전에 “핵심만 남기는” 요약 체인이 필요합니다.

4.1 요약의 목표는 압축이 아니라 재검색 가능성

좋은 요약은 다음을 만족합니다.

검색 힌트가 풍부: 키워드, 엔티티, 수치, 파일명, API명
행동 가능: 다음 태스크로 이어질 수 있는 TODO/결정사항 포함
근거 추적: 원문 메모리 ID 목록을 남겨 감사 가능

따라서 출력 포맷을 구조화하는 것이 중요합니다. 예를 들어 JSON 스키마로 요약을 만들고, 그 요약 텍스트를 다시 임베딩해 벡터DB에 넣습니다.

4.2 요약 트리거(언제 요약할까)

5) 구현 예제: TTL purge + 요약 체인 파이프라인

아래 코드는 개념을 전달하기 위한 Python 예시입니다. 실제 벡터DB SDK(Chroma, Pinecone, Weaviate, Qdrant 등)에 맞게 VectorStore 부분만 바꾸면 됩니다.

5.1 데이터 모델과 저장 함수

from __future__ import annotations

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


@dataclass
class MemoryDoc:
    memory_id: str
    run_id: str
    task_id: str
    doc_type: str  # tool_output | observation | decision | summary
    text: str
    importance: float
    created_at: int
    expires_at: int
    meta: Dict[str, Any]


def now_ts() -> int:
    return int(time.time())


def new_memory(
    run_id: str,
    task_id: str,
    doc_type: str,
    text: str,
    ttl_sec: int,
    importance: float = 0.3,
    meta: Optional[Dict[str, Any]] = None,
) -> MemoryDoc:
    ts = now_ts()
    return MemoryDoc(
        memory_id=str(uuid.uuid4()),
        run_id=run_id,
        task_id=task_id,
        doc_type=doc_type,
        text=text,
        importance=importance,
        created_at=ts,
        expires_at=ts + ttl_sec,
        meta=meta or {},
    )


class VectorStore:
    """벡터DB 어댑터 인터페이스(예시)."""

    def upsert(self, doc: MemoryDoc) -> None:
        raise NotImplementedError

    def query(self, run_id: str, query_text: str, k: int) -> List[MemoryDoc]:
        raise NotImplementedError

    def list_expired(self, run_id: str, before_ts: int, limit: int = 500) -> List[MemoryDoc]:
        raise NotImplementedError

    def delete_ids(self, ids: List[str]) -> None:
        raise NotImplementedError

5.2 TTL purge 작업(주기 실행)

def purge_expired_memories(store: VectorStore, run_id: str, batch: int = 500) -> int:
    expired = store.list_expired(run_id=run_id, before_ts=now_ts(), limit=batch)
    if not expired:
        return 0

    ids = [d.memory_id for d in expired]
    store.delete_ids(ids)
    return len(ids)

운영 팁:

purge는 크론 또는 워커로 분리하세요.
한 번에 너무 많이 지우면 벡터DB compaction 비용이 튈 수 있으니 배치로 나누세요.

5.3 요약 체인: “핫 메모리 묶음”을 “웜 요약”으로 변환

아래는 LLM 호출부를 추상화한 요약 체인입니다. 출력은 구조화된 텍스트(또는 JSON 문자열)로 만들고, 원문 ID 목록을 메타데이터로 남깁니다.

from typing import Callable


def build_summarization_prompt(docs: List[MemoryDoc]) -> str:
    joined = "\n\n".join(
        f"[doc_id={d.memory_id} type={d.doc_type}]\n{d.text}" for d in docs
    )

    # MDX 렌더링 안전을 위해 본문에 부등호를 직접 쓰지 않고,
    # 코드 블록 내부에서만 사용합니다.

    return (
        "너는 에이전트 메모리 압축기다. 아래 문서 묶음을 장기기억으로 압축하라.\n"
        "요구사항:\n"
        "1) 사실/결정/수치/파일명/API명/에러코드 등 검색 힌트를 최대한 보존\n"
        "2) 다음 행동(TODO)와 미해결 이슈를 분리\n"
        "3) 중복과 노이즈 제거\n"
        "4) 마지막에 source_doc_ids 배열을 포함\n\n"
        "출력 포맷(JSON 문자열):\n"
        "{\n"
        "  \"facts\": [ ... ],\n"
        "  \"decisions\": [ ... ],\n"
        "  \"todos\": [ ... ],\n"
        "  \"open_questions\": [ ... ],\n"
        "  \"keywords\": [ ... ],\n"
        "  \"source_doc_ids\": [ ... ]\n"
        "}\n\n"
        "입력 문서:\n" + joined
    )


def summarize_hot_memories(
    llm_call: Callable[[str], str],
    store: VectorStore,
    run_id: str,
    task_id: str,
    hot_docs: List[MemoryDoc],
    warm_ttl_sec: int = 60 * 60 * 24 * 30,
) -> MemoryDoc:
    prompt = build_summarization_prompt(hot_docs)
    summary_json_text = llm_call(prompt)

    src_ids = [d.memory_id for d in hot_docs]

    warm_doc = new_memory(
        run_id=run_id,
        task_id=task_id,
        doc_type="summary",
        text=summary_json_text,
        ttl_sec=warm_ttl_sec,
        importance=0.8,
        meta={
            "source_doc_ids": src_ids,
            "summary_level": 1,
            "summarized_at": now_ts(),
        },
    )

    store.upsert(warm_doc)
    return warm_doc

5.4 “요약 후 삭제” 안전장치

요약을 만들었으면 원문을 지워도 되지만, 바로 삭제하면 요약 품질이 낮을 때 복구가 어렵습니다. 그래서 다음 전략을 권합니다.

요약 생성 후 원문 TTL을 더 짧게 재설정(예: 10m)하고, 그 사이에 품질 문제가 없으면 purge
또는 콜드 스토리지(S3, 파일)로 원문을 아카이브하고 벡터DB에서는 제거

아카이브를 한다면 벡터DB에는 “아카이브 포인터”만 남기는 방식이 좋습니다.

6) 검색 품질을 지키는 요약/TTL 튜닝 체크리스트

6.1 중요도(importance) 기반 보존

모든 문서를 동일 TTL로 지우지 마세요. 예를 들어:

decision 타입은 TTL을 길게
tool_output은 짧게
observation은 importance가 높을 때만 길게

importance는 휴리스틱으로도 충분히 쓸 수 있습니다.

에러/예외/실패 원인 포함 시 가중치 상승
숫자/버전/API명/파일명 포함 시 가중치 상승
“최종 결론”, “결정”, “합의” 같은 키워드 포함 시 상승

6.2 요약의 과압축 방지: 계층 요약

한 번에 너무 많이 요약하면 중요한 디테일이 날아갑니다. 다음처럼 “요약의 요약”을 단계적으로 하세요.

레벨1: 최근 50개 문서를 요약
레벨2: 레벨1 요약 20개를 다시 요약

메타데이터에 summary_level을 넣어 계층을 추적하면, 검색 시 레벨별로 가중치를 다르게 줄 수 있습니다.

6.3 검색 시점 필터링

벡터 검색은 유사도만으로는 노이즈가 섞이기 쉽습니다. 메타데이터 필터를 적극 활용하세요.

같은 run_id만 검색
필요 시 같은 task_id 우선
doc_type in (summary, decision) 우선
created_at 최신 가중

이렇게 하면 TTL로 핫 메모리가 줄어도 웜 요약이 안정적으로 히트합니다.

7) 운영 관점: 관측(Observability) 없이는 다시 폭증한다

TTL과 요약을 넣어도, 운영 지표가 없으면 어느 순간 다시 커집니다. 아래 지표를 최소로 잡으세요.

mem_docs_total (run_id별 문서 수)
mem_bytes_total (텍스트 총량 추정)
purge_deleted_count (주기당 삭제 수)
summary_created_count (주기당 요약 수)
query_latency_ms (검색 지연)
hit_rate_by_type (검색 결과의 doc_type 분포)

또한 레이트리밋/재시도 폭주로 로그가 급증하면 메모리도 함께 오염됩니다. API 호출이 많은 구조라면 OpenAI Responses API 429 레이트리밋 토큰버킷으로 끝내기 같은 방식으로 호출량을 안정화하는 것이 간접적으로 메모리 폭증도 줄입니다.

8) 실전 적용 순서(가장 안전한 롤아웃)

메타데이터 스키마 추가: doc_type, created_at, expires_at부터
purge 워커 도입: 삭제는 배치로, 로그 남기기
요약 체인 도입: 먼저 “요약만 만들고 삭제는 안 함”으로 검증
요약 품질 확인 후: “요약 생성된 원문만 TTL 단축” 적용
계층 요약: 장기 실행에서 웜 메모리도 커지면 레벨2 요약 추가

이 순서로 가면, 한 번에 시스템을 바꾸지 않고도 안정적으로 메모리 성장 곡선을 꺾을 수 있습니다.

9) 마무리: 핵심은 ‘기억의 수명’과 ‘기억의 형태’

BabyAGI 메모리 폭증은 벡터DB나 임베딩 모델 문제가 아니라, 대부분 정책 부재에서 시작합니다. 해결의 요점은 두 가지입니다.

TTL로 “기억의 수명”을 정의해 무한 성장을 차단
요약 체인으로 “기억의 형태”를 장기 검색에 맞게 재구성

이 조합을 적용하면 장시간 실행에서도 성능이 안정되고, 비용과 디스크 사용량을 예측 가능하게 만들 수 있습니다.