AutoGPT 메모리 폭주 막기 - SQLite RAG 압축

-- SQLite
PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;
PRAGMA temp_store = MEMORY;

CREATE TABLE IF NOT EXISTS documents (
  id INTEGER PRIMARY KEY,
  source TEXT NOT NULL,          -- 예: tool:web, tool:db, agent:obs
  source_id TEXT,                -- URL, 파일명, 요청 ID 등
  sha256 TEXT NOT NULL UNIQUE,   -- 중복 제거 키
  created_at INTEGER NOT NULL
);

CREATE TABLE IF NOT EXISTS chunks (
  id INTEGER PRIMARY KEY,
  document_id INTEGER NOT NULL,
  chunk_index INTEGER NOT NULL,
  content_zstd BLOB NOT NULL,    -- 압축된 텍스트
  content_len INTEGER NOT NULL,  -- 압축 전 문자 수
  token_est INTEGER NOT NULL,    -- 대략 토큰 수
  importance REAL NOT NULL DEFAULT 0.5,
  created_at INTEGER NOT NULL,
  FOREIGN KEY(document_id) REFERENCES documents(id)
);

CREATE TABLE IF NOT EXISTS embeddings (
  chunk_id INTEGER PRIMARY KEY,
  dim INTEGER NOT NULL,
  vector_f32 BLOB NOT NULL,      -- float32 배열
  FOREIGN KEY(chunk_id) REFERENCES chunks(id)
);

CREATE TABLE IF NOT EXISTS summaries (
  id INTEGER PRIMARY KEY,
  scope TEXT NOT NULL,           -- session, document, topic
  scope_id TEXT NOT NULL,
  summary_zstd BLOB NOT NULL,
  token_est INTEGER NOT NULL,
  created_at INTEGER NOT NULL,
  UNIQUE(scope, scope_id)
);

CREATE INDEX IF NOT EXISTS idx_chunks_doc_time ON chunks(document_id, created_at);
CREATE INDEX IF NOT EXISTS idx_docs_time ON documents(created_at);

import re
import hashlib

def normalize_text(text: str) -> str:
    text = text.strip()
    text = re.sub(r"\b\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z\b", "<ts>", text)
    text = re.sub(r"\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\b", "<uuid>", text, flags=re.I)
    text = re.sub(r"\s+", " ", text)
    return text

def sha256_hex(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()

import time
import zstandard as zstd

_cctx = zstd.ZstdCompressor(level=6)
_dctx = zstd.ZstdDecompressor()

def compress_text(text: str) -> bytes:
    return _cctx.compress(text.encode("utf-8"))

def decompress_text(blob: bytes) -> str:
    return _dctx.decompress(blob).decode("utf-8")

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

def estimate_tokens_rough(text: str) -> int:
    # 매우 러프한 휴리스틱: 한글/영문 혼합 환경에서 보수적으로 잡기
    # 실제 운영에서는 모델별 토크나이저 사용 권장
    return max(1, len(text) // 3)

def chunk_text(text: str, max_tokens: int = 400) -> list[str]:
    words = text.split(" ")
    chunks, buf = [], []
    buf_tokens = 0

    for w in words:
        t = estimate_tokens_rough(w + " ")
        if buf and buf_tokens + t > max_tokens:
            chunks.append(" ".join(buf).strip())
            buf, buf_tokens = [], 0
        buf.append(w)
        buf_tokens += t

    if buf:
        chunks.append(" ".join(buf).strip())
    return chunks

import sqlite3

def upsert_session_summary(conn: sqlite3.Connection, session_id: str, summary_text: str):
    blob = compress_text(summary_text)
    token_est = estimate_tokens_rough(summary_text)
    conn.execute(
        """
        INSERT INTO summaries(scope, scope_id, summary_zstd, token_est, created_at)
        VALUES(?, ?, ?, ?, ?)
        ON CONFLICT(scope, scope_id)
        DO UPDATE SET summary_zstd=excluded.summary_zstd,
                      token_est=excluded.token_est,
                      created_at=excluded.created_at
        """,
        ("session", session_id, blob, token_est, now_ts())
    )
    conn.commit()

def insert_document_if_new(conn, source: str, source_id: str | None, text: str) -> int | None:
    norm = normalize_text(text)
    h = sha256_hex(norm)

    try:
        cur = conn.execute(
            "INSERT INTO documents(source, source_id, sha256, created_at) VALUES(?, ?, ?, ?)",
            (source, source_id, h, now_ts())
        )
        conn.commit()
        return int(cur.lastrowid)
    except sqlite3.IntegrityError:
        return None

import numpy as np

def cosine(a: np.ndarray, b: np.ndarray) -> float:
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-9))

def mmr_select(query_vec, cand_vecs, cand_ids, k=4, lam=0.7):
    selected = []
    selected_vecs = []

    scores = [cosine(query_vec, v) for v in cand_vecs]
    remaining = list(range(len(cand_ids)))

    for _ in range(min(k, len(remaining))):
        best_i, best_val = None, -1e9
        for idx in remaining:
            relevance = scores[idx]
            diversity = 0.0
            if selected_vecs:
                diversity = max(cosine(cand_vecs[idx], sv) for sv in selected_vecs)
            val = lam * relevance - (1 - lam) * diversity
            if val > best_val:
                best_val, best_i = val, idx
        selected.append(cand_ids[best_i])
        selected_vecs.append(cand_vecs[best_i])
        remaining.remove(best_i)

    return selected

-- 30일 이전의 로그성 문서 삭제 (예시)
DELETE FROM documents
WHERE source IN ('agent:obs', 'tool:web', 'tool:api')
  AND created_at < strftime('%s','now') - 30*24*60*60;

-- 고아 chunk/embedding 정리
DELETE FROM chunks WHERE document_id NOT IN (SELECT id FROM documents);
DELETE FROM embeddings WHERE chunk_id NOT IN (SELECT id FROM chunks);