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

from dataclasses import dataclass, asdict
import json

@dataclass
class WorkingSummary:
    goal: str = ""
    plan: list[str] = None
    done: list[str] = None
    open_questions: list[str] = None
    constraints: list[str] = None
    artifacts: list[str] = None

    def __post_init__(self):
        self.plan = self.plan or []
        self.done = self.done or []
        self.open_questions = self.open_questions or []
        self.constraints = self.constraints or []
        self.artifacts = self.artifacts or []


def update_summary(llm, prev: WorkingSummary, new_events: str) -> WorkingSummary:
    prompt = f"""
너는 에이전트의 작업 요약을 갱신하는 모듈이다.

[이전 요약 JSON]
{json.dumps(asdict(prev), ensure_ascii=False)}

[이번 턴 이벤트]
{new_events}

규칙:
- 목표/계획/완료/미해결/제약/산출물을 최신 상태로 갱신
- 중복 항목은 제거
- 불확실하면 open_questions에 남긴다
- JSON만 출력
"""
    out = llm(prompt)
    data = json.loads(out)
    return WorkingSummary(**data)

import time
from qdrant_client import QdrantClient
from qdrant_client.http.models import Filter, FieldCondition, Range

client = QdrantClient(url="http://localhost:6333")
COLLECTION = "autogpt_memory"


def purge_expired(ttl_seconds: int):
    now = int(time.time())
    cutoff = now - ttl_seconds

    flt = Filter(
        must=[
            FieldCondition(
                key="created_at",
                range=Range(lt=cutoff)
            )
        ]
    )

    client.delete(
        collection_name=COLLECTION,
        points_selector=flt
    )

def should_retrieve(user_text: str, summary: WorkingSummary) -> bool:
    keywords = ["전에", "기억", "지난", "어제", "이전", "로그", "참조"]
    if any(k in user_text for k in keywords):
        return True
    # 계획 단계가 바뀌는 경우(예: plan에 새로운 단계 추가)도 트리거
    if "다음 단계" in user_text or "이제" in user_text:
        return True
    return False


def build_context(user_text, summary, recent_messages, retrieved_docs):
    # retrieved_docs는 top-k + score threshold를 통과한 것만
    parts = []
    parts.append("[WORKING_SUMMARY]\n" + json.dumps(asdict(summary), ensure_ascii=False))
    parts.append("[RECENT]\n" + "\n".join(recent_messages[-6:]))
    if retrieved_docs:
        parts.append("[RETRIEVED]\n" + "\n\n".join(d["text"] for d in retrieved_docs))
    parts.append("[USER]\n" + user_text)
    return "\n\n".join(parts)

def should_promote_to_long_term(event: dict) -> bool:
    t = event.get("type")
    if t in ["final_answer", "decision", "config", "error_analysis", "api_schema"]:
        return True

    text = event.get("text", "")
    # 너무 짧으면 정보 가치가 낮다고 가정
    if len(text) < 200:
        return False

    # 로그/진행 메시지 제외
    noisy_markers = ["thinking", "progress", "step", "retry"]
    if any(m in text.lower() for m in noisy_markers):
        return False

    return True

class AgentMemory:
    def __init__(self, llm, vector_store, summary_store):
        self.llm = llm
        self.vector_store = vector_store
        self.summary_store = summary_store  # Redis 같은 KV를 가정

    def step(self, session_id: str, user_text: str, recent_messages: list[str]):
        summary = self.summary_store.get(session_id) or WorkingSummary(goal="")

        retrieved = []
        if should_retrieve(user_text, summary):
            retrieved = self.vector_store.search(
                query=user_text,
                top_k=5,
                score_threshold=0.25,
                filters={"session_id": session_id}
            )

        prompt = build_context(user_text, summary, recent_messages, retrieved)
        answer, events = self.llm.run(prompt)  # events: tool 결과/결정 등 구조화 로그

        # 작업 기억 갱신(증분)
        new_events_text = "\n".join(e.get("text", "") for e in events)
        summary = update_summary(self.llm, summary, new_events_text)
        self.summary_store.set(session_id, summary)

        # 장기 기억 승격
        for e in events:
            if should_promote_to_long_term(e):
                self.vector_store.upsert(
                    text=e["text"],
                    metadata={
                        "session_id": session_id,
                        "type": e.get("type", "unknown"),
                        "created_at": int(time.time()),
                    }
                )

        return answer