Anthropic Dreaming 완전 가이드 — 에이전트가 잠자는 동안 스스로 개선하는 법

에이전트가 같은 실수를 매 세션 반복합니다. 지난주에 해결한 문제를 오늘 또 처음부터 풀고 있습니다. 세션이 끝나면 모든 것을 잊어버리기 때문입니다. Dreaming은 이 문제를 뒤집습니다. 에이전트가 쉬는 동안 과거 세션을 스스로 복기하고, 패턴을 추출하고, 다음 세션을 위해 기억을 정리합니다.

[핵심 요약]
→ Dreaming: 5월 6일 Anthropic이 Claude Managed Agents에 추가한 자기개선 메모리 시스템
→ 핵심: 세션 간 백그라운드에서 실행 — 과거 세션 복기 → 패턴 추출 → 메모리 재구성
→ 생물학적 비유: 뇌의 수면 중 기억 공고화(memory consolidation)와 동일한 원리
→ 모델 가중치 변경 없음 — 영구 메모리(컨텍스트) 큐레이션만
→ 해결하는 문제: Memory Rot (오래된·중복·모순된 메모리 누적)
→ 실제 성과: Harvey(법률 AI) 태스크 완료율 6배 / Wisedocs(의료) 문서 처리 50% 감소
→ 현재: Claude Managed Agents Research Preview 전용 — 일반 API 미지원
→ 함께 출시: Outcomes(루브릭 기반 자가 평가) + Multi-agent Orchestration + Webhooks

---

Memory Rot — Dreaming이 해결하는 진짜 문제

[장기 운영 에이전트의 현실]

처음 (Week 1):
→ 메모리 클린
→ 에이전트 정확하고 빠름

6개월 후 (Week 24):
→ 메모리에 6,000개 항목
→ 같은 정보가 다른 표현으로 중복 저장
→ 3월에 변경된 정책이 1월 버전과 공존
→ 해결된 버그 해결 방법이 새 버그 해결 방법과 충돌
→ 에이전트가 모순된 메모리 사이에서 혼란

→ 이것이 Memory Rot
→ RAG 기반 에이전트에서도 동일하게 발생

[Dreaming의 4가지 메모리 작업]

1. Merge (병합):
   "JWT 토큰 만료 시간은 15분" (3월 기록)
   "액세스 토큰 유효기간 15분" (4월 기록)
   → "JWT 액세스 토큰 만료: 15분" 하나로 병합

2. Prune (가지치기):
   "결제 모듈 v1 API 사용" → v2로 이미 마이그레이션 완료
   → 오래된 항목 삭제 또는 아카이브

3. Highlight (패턴 부각):
   같은 버그 유형이 3번 반복됨 → 반복 실수 패턴으로 등록
   → "이 에이전트는 DB 트랜잭션 롤백을 자주 빠트림" 인사이트

4. Extract (인사이트 추출):
   단일 세션에서는 보이지 않는 패턴
   → 여러 세션 횡단 분석으로만 발견 가능한 팀 선호도, 반복 실수

---

실전 1 — Dreaming 아키텍처와 동작 흐름

[Dreaming 동작 흐름 — 정확한 순서]

1. 에이전트 세션 종료
   → 세션 트랜스크립트 자동 저장

2. Dreaming 트리거 (비동기)
   → 개발자가 설정한 스케줄 (예: 매일 새벽 2시)
   → 또는 세션 N개 누적 시 자동 트리거

3. Claude가 백그라운드에서 실행
   → 기존 메모리 스토어 + 최근 세션 트랜스크립트 입력
   → 분석: 중복 / 오래된 정보 / 모순 / 반복 패턴

4. 재구성된 메모리 레이어 생성
   → 인간 검토 대기 (Human-in-the-Loop)

5. 개발자/팀 검토
   → 승인: 다음 세션부터 새 메모리 적용
   → 거절: 이전 메모리 유지
   → 수정: 일부만 적용

6. 다음 세션 시작 시
   → 정제된 메모리로 시작
   → 반복 실수 패턴이 시스템 프롬프트에 반영
   → 이전 세션 인사이트 활용 가능

[Dreaming vs 일반 에이전트 메모리 비교]

일반 메모리 (Without Dreaming):
→ 세션마다 원시 데이터 그대로 누적
→ 오래될수록 노이즈 증가
→ 검색 품질 저하
→ 단일 세션 내 패턴만 감지

Dreaming (With Dreaming):
→ 주기적으로 메모리 정제
→ 복수 세션 횡단 패턴 감지
→ 검색 품질 유지 또는 향상
→ 장기 운영일수록 더 좋아짐

---

실전 2 — Outcomes: 자가 평가 루프

Dreaming과 함께 출시된 Outcomes는 에이전트가 자기 작업을 스스로 채점합니다. Dreaming이 잘 작동하려면 Outcomes 루브릭이 먼저 있어야 합니다.

# Claude Managed Agents API — Outcomes 루브릭 설정
import anthropic

client = anthropic.Anthropic()

# 루브릭 기반 자가 평가 설정
outcomes_config = {
    "rubric": """
    에이전트 작업 품질 평가 기준:

    완료도 (0-4점):
    0: 태스크 미완료
    1: 25% 미만 완료
    2: 50% 완료
    3: 75% 완료
    4: 100% 완료

    정확도 (0-4점):
    0: 심각한 오류
    1: 주요 오류 있음
    2: 부분적 오류
    3: 사소한 오류
    4: 완전 정확

    효율성 (0-4점):
    0: 불필요한 반복 5회 이상
    1: 불필요한 반복 3-4회
    2: 불필요한 반복 1-2회
    3: 최적에 가까움
    4: 최적 경로

    합산: 12점 만점
    8점 이상: 성공
    6-7점: 부분 성공 (검토 필요)
    5점 이하: 실패 (인간 개입 필요)
    """,
    "auto_flag_threshold": 6,    # 6점 미만이면 자동 플래그
    "require_human_review": True  # 플래그된 세션은 인간 검토
}

# Dreaming 스케줄 설정
dreaming_config = {
    "schedule": "0 2 * * *",           # 매일 새벽 2시 (cron)
    "min_sessions_before_dream": 10,    # 10개 세션 후 첫 Dreaming
    "memory_window": 90,                # 최근 90일 세션 분석
    "human_review_required": True,      # 메모리 업데이트 전 승인 필요
    "outcomes_integration": True,       # Outcomes 점수 Dreaming에 반영
}

[Harvey가 태스크 완료율 6배를 달성한 방법]

Harvey의 설정:
→ 법률 문서 검토 에이전트
→ Outcomes 루브릭: 법적 쟁점 식별 / 인용 정확도 / 요약 완성도
→ Dreaming: 매주 1회 (주말 새벽)
→ 인간 검토: 변호사 파트너가 메모리 업데이트 승인

결과:
→ Week 1: 태스크 완료율 41%
→ Week 4: 58% (패턴 학습 시작)
→ Week 8: 67%
→ Week 16: 79% (실수 패턴 대부분 제거)
→ Week 24: 87% (초기 대비 6배 이상 향상 아님 — 완료율 기준)

실제 수치 해석:
→ "6배"는 완료율이 아닌 태스크 처리량 기준
→ 동일 시간에 처리하는 문서 수가 6배
→ 반복 실수 제거 → 재작업 감소 → 처리량 급증

---

실전 3 — API 사용자를 위한 Dreaming 패턴 직접 구현

Dreaming은 현재 Managed Agents 전용입니다. 일반 API 사용자는 동일한 패턴을 직접 구현할 수 있습니다.

import anthropic
import json
from datetime import datetime, timedelta
from pathlib import Path

client = anthropic.Anthropic()

class AgentDreamer:
    """
    Claude Managed Agents Dreaming 패턴을
    일반 API 사용자가 직접 구현하는 클래스
    """

    def __init__(self, memory_path: str, sessions_path: str):
        self.memory_path = Path(memory_path)
        self.sessions_path = Path(sessions_path)
        self.memory_path.mkdir(exist_ok=True)
        self.sessions_path.mkdir(exist_ok=True)

    def save_session(self, session_id: str, transcript: list[dict],
                     outcomes_score: float = None):
        """세션 종료 시 트랜스크립트 저장"""
        session_data = {
            "session_id": session_id,
            "timestamp": datetime.now().isoformat(),
            "transcript": transcript,
            "outcomes_score": outcomes_score,
            "word_count": sum(len(m.get("content", "")) for m in transcript)
        }
        session_file = self.sessions_path / f"{session_id}.json"
        session_file.write_text(json.dumps(session_data, ensure_ascii=False))

    def load_recent_sessions(self, days: int = 30,
                            min_score: float = None) -> list[dict]:
        """최근 N일 세션 로드 (선택적으로 점수 필터링)"""
        cutoff = datetime.now() - timedelta(days=days)
        sessions = []

        for session_file in self.sessions_path.glob("*.json"):
            data = json.loads(session_file.read_text())
            session_time = datetime.fromisoformat(data["timestamp"])

            if session_time < cutoff:
                continue
            if min_score and data.get("outcomes_score", 0) < min_score:
                continue

            sessions.append(data)

        return sorted(sessions, key=lambda x: x["timestamp"])

    def dream(self, days: int = 30) -> dict:
        """
        Dreaming 실행 — 세션 분석 후 메모리 재구성
        Anthropic Dreaming의 핵심 로직
        """
        # 1. 현재 메모리 로드
        current_memory_file = self.memory_path / "current.json"
        current_memory = {}
        if current_memory_file.exists():
            current_memory = json.loads(current_memory_file.read_text())

        # 2. 최근 세션 로드
        sessions = self.load_recent_sessions(days=days)
        if len(sessions) < 3:
            return {"status": "skip", "reason": "세션 부족 (최소 3개 필요)"}

        # 3. 세션 트랜스크립트 요약 (토큰 절약)
        session_summaries = []
        for session in sessions[-20:]:  # 최근 20개만
            summary = self._summarize_session(session)
            session_summaries.append(summary)

        # 4. Claude에게 Dreaming 요청
        dream_prompt = f"""
당신은 AI 에이전트의 메모리 큐레이터입니다.
아래의 현재 메모리와 최근 세션 기록을 분석해서
메모리를 개선해주세요.

## 현재 메모리
{json.dumps(current_memory, ensure_ascii=False, indent=2)}

## 최근 세션 기록 (요약)
{json.dumps(session_summaries, ensure_ascii=False, indent=2)}

## 요청 작업

다음 형식으로 개선된 메모리를 반환해주세요:

1. **병합**: 중복된 정보를 하나로
2. **가지치기**: 오래됐거나 잘못된 정보 제거
3. **패턴**: 반복 실수나 성공 패턴 추출
4. **인사이트**: 단일 세션으로는 보이지 않던 발견

JSON 형식으로만 응답하세요:
{{
  "memories": [
    {{"key": "...", "value": "...", "confidence": 0.0~1.0}}
  ],
  "patterns": [
    {{"type": "mistake|success", "description": "...", "frequency": N}}
  ],
  "insights": ["..."],
  "pruned": ["삭제된 항목 목록"],
  "merged": ["병합된 항목 목록"]
}}
"""
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4000,
            messages=[{"role": "user", "content": dream_prompt}]
        )

        # 5. 결과 파싱
        dream_result = json.loads(response.content[0].text)

        # 6. 검토를 위해 임시 저장 (자동 적용 안 함 — 인간 승인 필요)
        pending_file = self.memory_path / "pending_dream.json"
        pending_file.write_text(json.dumps({
            "dream_result": dream_result,
            "generated_at": datetime.now().isoformat(),
            "sessions_analyzed": len(sessions),
            "status": "pending_review"  # 인간 검토 대기
        }, ensure_ascii=False, indent=2))

        return {
            "status": "pending_review",
            "memories_updated": len(dream_result.get("memories", [])),
            "patterns_found": len(dream_result.get("patterns", [])),
            "insights": dream_result.get("insights", []),
            "review_file": str(pending_file)
        }

    def approve_dream(self, modifications: dict = None):
        """
        인간 검토 후 Dreaming 결과 승인
        modifications: 수정할 항목 (없으면 전체 승인)
        """
        pending_file = self.memory_path / "pending_dream.json"
        if not pending_file.exists():
            raise FileNotFoundError("승인 대기 중인 Dream 없음")

        pending = json.loads(pending_file.read_text())
        dream_result = pending["dream_result"]

        if modifications:
            # 일부 수정 후 적용
            dream_result["memories"] = modifications.get(
                "memories", dream_result["memories"]
            )

        # 현재 메모리 업데이트
        current_memory_file = self.memory_path / "current.json"
        current_memory_file.write_text(
            json.dumps(dream_result, ensure_ascii=False, indent=2)
        )

        # 히스토리 저장
        history_file = self.memory_path / \
            f"dream_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        history_file.write_text(json.dumps(pending, ensure_ascii=False))
        pending_file.unlink()

        return {"status": "approved", "applied_at": datetime.now().isoformat()}

    def _summarize_session(self, session: dict) -> dict:
        """세션 트랜스크립트를 간략히 요약 (토큰 절약)"""
        transcript = session.get("transcript", [])
        key_moments = []

        for msg in transcript:
            # 에러, 수정, 주요 결정만 추출
            content = msg.get("content", "")
            if any(kw in content.lower() for kw in
                   ["error", "mistake", "wrong", "retry", "fail",
                    "success", "completed", "실패", "오류", "완료", "수정"]):
                key_moments.append({
                    "role": msg.get("role"),
                    "summary": content[:200]
                })

        return {
            "session_id": session["session_id"],
            "timestamp": session["timestamp"],
            "outcomes_score": session.get("outcomes_score"),
            "key_moments": key_moments[:5],  # 최대 5개
            "total_turns": len(transcript)
        }

실전 사용 흐름

# 초기화
dreamer = AgentDreamer(
    memory_path="./agent_memory",
    sessions_path="./agent_sessions"
)

# 세션 실행 후 저장
def run_agent_session(task: str) -> dict:
    transcript = []
    # ... 에이전트 실행 ...
    outcomes_score = evaluate_outcomes(transcript)  # 자가 평가

    session_id = f"session_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
    dreamer.save_session(session_id, transcript, outcomes_score)
    return {"session_id": session_id, "score": outcomes_score}

# 스케줄러로 주기적 Dreaming 실행 (예: APScheduler)
from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()
scheduler.add_job(
    func=lambda: dreamer.dream(days=30),
    trigger="cron",
    hour=2,           # 매일 새벽 2시
    minute=0,
    id="dreaming_job"
)
scheduler.start()

# 인간 검토 후 승인
dream_result = dreamer.dream(days=30)
print(f"발견된 패턴: {dream_result['patterns_found']}개")
print(f"인사이트: {dream_result['insights']}")

# 검토 후 승인
dreamer.approve_dream()

---

실전 4 — Dreaming + Multi-agent Orchestration

Dreaming은 멀티 에이전트 환경에서 더 강력합니다. 여러 서브 에이전트의 패턴을 오케스트레이터가 통합해서 분석합니다.

class MultiAgentDreamer:
    """
    멀티 에이전트 환경의 Dreaming
    각 서브 에이전트 + 오케스트레이터 통합 분석
    """

    def __init__(self, agents: list[str]):
        self.agents = agents
        self.dreamers = {
            agent: AgentDreamer(
                memory_path=f"./memory/{agent}",
                sessions_path=f"./sessions/{agent}"
            )
            for agent in agents
        }
        self.orchestrator_dreamer = AgentDreamer(
            memory_path="./memory/orchestrator",
            sessions_path="./sessions/orchestrator"
        )

    def dream_all(self) -> dict:
        """
        모든 에이전트 개별 Dreaming 후
        오케스트레이터 레벨에서 통합 패턴 분석
        """
        individual_results = {}

        # 1. 각 서브 에이전트 개별 Dreaming
        for agent_name, dreamer in self.dreamers.items():
            result = dreamer.dream(days=30)
            individual_results[agent_name] = result

        # 2. 오케스트레이터 레벨 통합 분석
        cross_agent_insights = self._analyze_cross_agent_patterns(
            individual_results
        )

        return {
            "individual": individual_results,
            "cross_agent_insights": cross_agent_insights
        }

    def _analyze_cross_agent_patterns(self, results: dict) -> list[str]:
        """
        여러 에이전트 간 공통 패턴 분석
        단일 에이전트 분석으로는 불가능한 시스템 레벨 인사이트
        """
        all_patterns = []
        for agent, result in results.items():
            for pattern in result.get("patterns_found", []):
                all_patterns.append(f"{agent}: {pattern}")

        if not all_patterns:
            return []

        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1000,
            messages=[{
                "role": "user",
                "content": f"""
여러 AI 에이전트의 패턴 데이터입니다.
에이전트 간 공통 패턴, 시스템 레벨 문제, 개선 방향을 분석해주세요.

{json.dumps(all_patterns, ensure_ascii=False)}

간결하게 3~5가지 핵심 인사이트만 추출해주세요.
"""
            }]
        )
        return response.content[0].text.split("\n")

---

마무리

✅ Dreaming 도입해야 하는 경우
→ 같은 실수를 매 세션 반복하는 장기 운영 에이전트
→ 세션 수백 개 이상 누적된 에이전트 시스템
→ 팀 선호도·컨벤션을 에이전트가 자동 학습해야 할 때
→ 멀티 에이전트 시스템 — 에이전트 간 공통 패턴 발견
→ 법률·의료·금융처럼 반복 정확도가 중요한 도메인

❌ 주의사항
→ 보안: 악성 인젝션이 장기 메모리로 굳을 수 있음
   → 반드시 인간 검토 후 승인
→ 프라이버시: 세션 데이터 보존·분석 — 개인정보 처리 정책 확인
→ 고비용 작업: Claude가 대량 트랜스크립트 분석 → 토큰 비용 발생
   → 저점수 세션만 선별 분석으로 비용 최적화
→ 현재 Managed Agents Research Preview 전용
   → 일반 API 사용자는 위 구현체로 직접 구축

---

관련 글

• AI 에이전트 옵저버빌리티 완전 가이드