1M 컨텍스트 윈도우 시대의 LLM 활용 전략: 대규모 문맥 처리의 실전 가이드

# 1M 토큰이 실제로 어느 정도의 데이터인지 환산
# Anthropic의 토크나이저 기준 (claude-tokenizer)

TOKEN_ESTIMATES = {
    "영문 단어 수": "약 750,000 단어",
    "한국어 글자 수": "약 500,000 글자 (토큰 효율이 영어 대비 낮음)",
    "영문 소설": "약 7~8권 (평균 10만 단어/권)",
    "소스 코드 (Python)": "약 50,000~80,000 줄",
    "소스 코드 (Java)": "약 40,000~60,000 줄",
    "A4 문서": "약 1,500~2,000 페이지",
    "GitHub 저장소": "중소규모 프로젝트 전체 코드베이스",
    "Kubernetes 매니페스트": "약 3,000~5,000개 리소스 정의",
    "로그 파일": "약 500MB~1GB 텍스트 로그",
}

for item, estimate in TOKEN_ESTIMATES.items():
    print(f"  {item}: {estimate}")

# Lost in the Middle 현상 시각화 (개념적 코드)
import numpy as np

def simulate_retrieval_accuracy(context_length, num_positions=20):
    """
    컨텍스트 내 정보 위치에 따른 검색 정확도 시뮬레이션
    실제 논문의 실험 결과를 기반으로 한 근사 모델
    """
    positions = np.linspace(0, 1, num_positions)
    # U자형 곡선: 시작과 끝이 높고, 중간이 낮음
    # 2026년 모델들은 이 곡선이 상당히 평탄해짐
    accuracy_old = 0.5 + 0.4 * np.abs(2 * positions - 1) ** 1.5
    accuracy_new = 0.85 + 0.12 * np.abs(2 * positions - 1) ** 2.0

    return {
        "positions": positions,
        "accuracy_2023": accuracy_old,  # GPT-4 128K 수준
        "accuracy_2026": accuracy_new,  # Claude Opus 4.6 1M 수준
    }

result = simulate_retrieval_accuracy(1_000_000)
for i in range(0, 20, 5):
    pos = result["positions"][i]
    old = result["accuracy_2023"][i]
    new = result["accuracy_2026"][i]
    print(
        f"Position {pos:.0%}: "
        f"2023 model={old:.1%}, "
        f"2026 model={new:.1%}"
    )

# 어텐션 복잡도 비교
def attention_complexity(seq_length, d_model=4096):
    """다양한 어텐션 메커니즘의 복잡도를 비교한다."""
    results = {}

    # 1. 풀 어텐션 (Vanilla Transformer)
    # O(n^2 * d)
    results["Full Attention"] = seq_length ** 2 * d_model

    # 2. Sliding Window Attention
    # O(n * w * d), w = window size
    window_size = 4096
    results["Sliding Window (w=4096)"] = (
        seq_length * window_size * d_model
    )

    # 3. Ring Attention (distributed)
    # O(n^2 * d / num_devices)
    num_devices = 8
    results["Ring Attention (8 GPUs)"] = (
        seq_length ** 2 * d_model / num_devices
    )

    # 4. Sparse Attention (BigBird style)
    # O(n * sqrt(n) * d)
    results["Sparse Attention"] = (
        seq_length * int(seq_length ** 0.5) * d_model
    )

    return results

for length_name, length in [
    ("128K", 128_000),
    ("200K", 200_000),
    ("1M", 1_000_000),
]:
    print(f"\n=== Sequence Length: {length_name} ===")
    complexities = attention_complexity(length)
    base = complexities["Full Attention"]
    for method, value in complexities.items():
        ratio = value / base * 100
        print(f"  {method}: {value:.2e} ({ratio:.1f}% of full)")

# KV 캐시 메모리 계산
def kv_cache_memory(
    seq_length,
    num_layers=80,
    num_heads=64,
    head_dim=128,
    dtype_bytes=2,  # FP16
):
    """KV 캐시에 필요한 메모리를 계산한다."""
    # K와 V 각각 저장
    # 메모리 = 2 * layers * heads * head_dim * seq_len * dtype
    memory_bytes = (
        2 * num_layers * num_heads * head_dim
        * seq_length * dtype_bytes
    )
    memory_gb = memory_bytes / (1024 ** 3)
    return memory_gb

for name, length in [
    ("8K", 8_000),
    ("128K", 128_000),
    ("200K", 200_000),
    ("1M", 1_000_000),
]:
    mem = kv_cache_memory(length)
    print(f"  {name} tokens: {mem:.1f} GB KV cache")

# 출력 예시:
#   8K tokens: 2.4 GB KV cache
#   128K tokens: 38.1 GB KV cache
#   200K tokens: 59.6 GB KV cache
#   1M tokens: 298.0 GB KV cache

import os
import anthropic

def collect_codebase(root_dir, extensions=None, max_tokens=900_000):
    """코드베이스의 모든 파일을 수집하여 단일 문자열로 반환한다."""
    if extensions is None:
        extensions = {'.py', '.ts', '.js', '.yaml', '.yml', '.json', '.md'}

    files_content = []
    total_chars = 0
    # 토큰 추정: 영문 기준 약 4 chars/token, 코드는 약 3 chars/token
    char_limit = max_tokens * 3

    for dirpath, dirnames, filenames in os.walk(root_dir):
        # node_modules, .git 등 제외
        dirnames[:] = [
            d for d in dirnames
            if d not in {'node_modules', '.git', '__pycache__', 'dist', 'build'}
        ]
        for filename in sorted(filenames):
            if any(filename.endswith(ext) for ext in extensions):
                filepath = os.path.join(dirpath, filename)
                relative_path = os.path.relpath(filepath, root_dir)
                try:
                    with open(filepath, 'r', encoding='utf-8') as f:
                        content = f.read()
                    entry = f"\n### FILE: {relative_path}\n```\n{content}\n```\n"
                    if total_chars + len(entry) > char_limit:
                        break
                    files_content.append(entry)
                    total_chars += len(entry)
                except (UnicodeDecodeError, PermissionError):
                    continue

    return '\n'.join(files_content), len(files_content)

def review_codebase(root_dir):
    """전체 코드베이스를 LLM에 전달하여 종합 리뷰를 수행한다."""
    codebase, file_count = collect_codebase(root_dir)

    client = anthropic.Anthropic()
    response = client.messages.create(
        model="claude-opus-4-6-20260314",
        max_tokens=8192,
        messages=[
            {
                "role": "user",
                "content": f"""다음은 프로젝트의 전체 코드베이스입니다 ({file_count}개 파일).
종합적인 코드 리뷰를 수행해 주세요.

분석 항목:
1. 아키텍처 패턴 및 설계 원칙 준수 여부
2. 보안 취약점 (인증, 인가, 입력 검증, 비밀 관리)
3. 성능 병목 지점
4. 테스트 커버리지 및 품질
5. 의존성 관리 및 버전 호환성
6. 코드 중복 및 리팩토링 기회
7. 에러 처리 일관성
8. API 설계 및 인터페이스 일관성

{codebase}""",
            }
        ],
    )
    return response.content[0].text

import anthropic

def analyze_logs(log_content, query):
    """대규모 로그 파일을 직접 분석한다."""
    client = anthropic.Anthropic()

    response = client.messages.create(
        model="claude-sonnet-4-6-20260314",
        max_tokens=4096,
        messages=[
            {
                "role": "user",
                "content": f"""다음은 프로덕션 서버의 로그 파일입니다.

분석 요청: {query}

주의사항:
- 타임스탬프 기반 시간 순서를 정확히 파악하세요
- 에러 패턴의 빈도와 상관관계를 분석하세요
- 근본 원인(Root Cause)을 추론하세요
- 구체적인 로그 라인을 인용하여 근거를 제시하세요

로그:
{log_content}""",
            }
        ],
    )
    return response.content[0].text

# 사용 예시
# 500MB 로그도 토큰 한도 내에서 처리 가능
with open("/var/log/app/production.log", "r") as f:
    # 최근 100만 줄 정도가 1M 토큰에 해당
    logs = f.read()

result = analyze_logs(
    logs,
    "지난 24시간 동안 발생한 5xx 에러의 근본 원인을 분석하고, "
    "에러 발생 패턴과 선행 이벤트의 상관관계를 파악해 주세요."
)

import anthropic

def compare_documents(documents, comparison_criteria):
    """여러 문서를 동시에 비교 분석한다."""
    client = anthropic.Anthropic()

    docs_text = ""
    for i, doc in enumerate(documents, 1):
        docs_text += f"\n\n=== 문서 {i}: {doc['title']} ===\n{doc['content']}\n"

    response = client.messages.create(
        model="claude-opus-4-6-20260314",
        max_tokens=8192,
        messages=[
            {
                "role": "user",
                "content": f"""다음 {len(documents)}개의 문서를 비교 분석해 주세요.

비교 기준:
{comparison_criteria}

각 문서의 핵심 주장, 공통점, 차이점, 모순점을 구조화하여 정리하세요.
인용 시 문서 번호와 해당 부분을 명시하세요.

{docs_text}""",
            }
        ],
    )
    return response.content[0].text

# 사용 예시: 경쟁사 기술 문서 비교
documents = [
    {"title": "AWS Well-Architected Framework", "content": aws_doc},
    {"title": "GCP Architecture Framework", "content": gcp_doc},
    {"title": "Azure Well-Architected Framework", "content": azure_doc},
]

result = compare_documents(
    documents,
    "보안 원칙, 비용 최적화, 운영 우수성 관점에서 세 클라우드의 "
    "아키텍처 프레임워크를 비교하세요."
)

import anthropic

class LongConversationManager:
    """1M 컨텍스트를 활용한 장기 대화 관리자."""

    def __init__(self, model="claude-sonnet-4-6-20260314"):
        self.client = anthropic.Anthropic()
        self.model = model
        self.messages = []
        self.system_prompt = ""
        self.estimated_tokens = 0

    def set_system(self, prompt):
        self.system_prompt = prompt

    def add_context(self, context, label="배경 정보"):
        """대화에 대규모 컨텍스트를 추가한다."""
        self.messages.append({
            "role": "user",
            "content": f"[{label}]\n{context}",
        })
        self.messages.append({
            "role": "assistant",
            "content": f"{label}를 확인했습니다. 질문해 주세요.",
        })
        # 토큰 추정 (대략적)
        self.estimated_tokens += len(context) // 3

    def chat(self, user_message):
        """대화를 이어간다."""
        self.messages.append({
            "role": "user",
            "content": user_message,
        })

        response = self.client.messages.create(
            model=self.model,
            max_tokens=4096,
            system=self.system_prompt,
            messages=self.messages,
        )

        assistant_message = response.content[0].text
        self.messages.append({
            "role": "assistant",
            "content": assistant_message,
        })

        self.estimated_tokens += len(user_message) // 3
        self.estimated_tokens += len(assistant_message) // 3

        return assistant_message

    def get_usage_info(self):
        return {
            "message_count": len(self.messages),
            "estimated_tokens": self.estimated_tokens,
            "remaining_capacity": f"{(1_000_000 - self.estimated_tokens) / 1_000_000:.1%}",
        }

# 사용 예시
manager = LongConversationManager()
manager.set_system("당신은 시니어 아키텍트입니다. 코드베이스를 기반으로 질문에 답하세요.")

# 전체 코드베이스를 컨텍스트로 추가
with open("codebase_dump.txt", "r") as f:
    manager.add_context(f.read(), "프로젝트 코드베이스")

# 이후 여러 차례 질문 가능
print(manager.chat("이 프로젝트의 인증 시스템은 어떻게 구현되어 있나요?"))
print(manager.chat("인증 시스템의 보안 취약점은 없나요?"))
print(manager.chat("JWT 토큰 갱신 로직에서 레이스 컨디션이 발생할 수 있나요?"))
print(manager.get_usage_info())

import anthropic

def one_shot_prototype(spec_document, tech_stack, output_format="complete"):
    """
    기획서 하나로 전체 프로토타입 코드를 생성한다.
    1M 컨텍스트 덕분에 기획서 + 참조 코드 + 기술 문서를
    모두 한 번에 전달할 수 있다.
    """
    client = anthropic.Anthropic()

    prompt = f"""다음 기획서를 기반으로 완전한 프로토타입을 생성해 주세요.

기술 스택: {tech_stack}

요구사항:
1. 프로덕션 수준의 에러 처리
2. 타입 안전성 (TypeScript strict mode)
3. 테스트 코드 포함
4. API 문서 (OpenAPI spec)
5. Docker 및 docker-compose 설정
6. README.md

기획서:
{spec_document}

모든 파일을 파일 경로와 함께 완전한 코드로 출력하세요.
"""

    response = client.messages.create(
        model="claude-opus-4-6-20260314",
        max_tokens=16384,
        messages=[{"role": "user", "content": prompt}],
    )
    return response.content[0].text

import time
import anthropic

def benchmark_context_utilization(
    model, context_sizes, needle, haystack_generator
):
    """
    다양한 컨텍스트 크기에서 모델의 정보 검색 정확도를
    측정하는 벤치마크 프레임워크.
    """
    client = anthropic.Anthropic()
    results = []

    for size in context_sizes:
        # 건초더미 생성
        haystack = haystack_generator(size)

        # 바늘을 다양한 위치에 삽입하여 테스트
        positions = [0.0, 0.25, 0.5, 0.75, 1.0]
        position_results = []

        for pos in positions:
            insert_idx = int(len(haystack) * pos)
            test_text = (
                haystack[:insert_idx]
                + f"\n{needle}\n"
                + haystack[insert_idx:]
            )

            start_time = time.time()
            response = client.messages.create(
                model=model,
                max_tokens=256,
                messages=[
                    {
                        "role": "user",
                        "content": (
                            f"다음 텍스트에서 특별한 정보를 찾아 "
                            f"정확히 인용하세요:\n\n{test_text}"
                        ),
                    }
                ],
            )
            elapsed = time.time() - start_time

            found = needle.lower() in response.content[0].text.lower()
            position_results.append({
                "position": pos,
                "found": found,
                "latency_s": elapsed,
            })

        results.append({
            "context_size": size,
            "accuracy": sum(
                r["found"] for r in position_results
            ) / len(position_results),
            "avg_latency": sum(
                r["latency_s"] for r in position_results
            ) / len(position_results),
            "details": position_results,
        })

    return results

def structured_context_packing(files, metadata=None):
    """
    파일들을 구조화된 형태로 컨텍스트에 패킹한다.
    LLM이 정보를 효과적으로 탐색할 수 있도록 목차를 포함한다.
    """
    # 1. 목차(Table of Contents) 생성
    toc = "## TABLE OF CONTENTS\n\n"
    for i, file in enumerate(files, 1):
        toc += f"{i}. [{file['path']}] - {file.get('description', 'N/A')}\n"
    toc += "\n---\n"

    # 2. 메타데이터 섹션
    meta = ""
    if metadata:
        meta = "## PROJECT METADATA\n\n"
        for key, value in metadata.items():
            meta += f"- {key}: {value}\n"
        meta += "\n---\n"

    # 3. 파일 내용 (경계 구분자 포함)
    content = ""
    for i, file in enumerate(files, 1):
        content += f"\n{'='*60}\n"
        content += f"## FILE {i}: {file['path']}\n"
        content += f"Language: {file.get('language', 'unknown')}\n"
        content += f"Lines: {file['content'].count(chr(10)) + 1}\n"
        content += f"{'='*60}\n\n"
        content += file['content']
        content += "\n"

    return toc + meta + content

class ProgressiveContextLoader:
    """
    필요한 만큼만 컨텍스트를 점진적으로 로딩하는 전략.
    초기에는 요약/목차만 제공하고, 필요 시 상세 내용을 추가한다.
    """

    def __init__(self, client, model, max_context=900_000):
        self.client = client
        self.model = model
        self.max_context = max_context
        self.loaded_files = {}
        self.summaries = {}

    def create_summary(self, file_path, content):
        """파일의 요약본을 생성한다."""
        response = self.client.messages.create(
            model=self.model,
            max_tokens=512,
            messages=[
                {
                    "role": "user",
                    "content": (
                        f"다음 파일을 3~5줄로 요약하세요. "
                        f"주요 함수/클래스 이름을 포함하세요.\n\n"
                        f"파일: {file_path}\n{content}"
                    ),
                }
            ],
        )
        return response.content[0].text

    def load_initial(self, files):
        """
        1단계: 모든 파일의 요약만 로딩한다.
        전체 구조를 파악하되 토큰을 절약한다.
        """
        context = "## PROJECT OVERVIEW (Summaries)\n\n"
        for file_path, content in files.items():
            summary = self.create_summary(file_path, content)
            self.summaries[file_path] = summary
            self.loaded_files[file_path] = content
            context += f"### {file_path}\n{summary}\n\n"
        return context

    def expand_file(self, current_context, file_path):
        """
        2단계: 특정 파일의 전체 내용을 컨텍스트에 추가한다.
        """
        if file_path in self.loaded_files:
            full_content = self.loaded_files[file_path]
            return (
                current_context
                + f"\n\n## FULL CONTENT: {file_path}\n"
                + f"```\n{full_content}\n```\n"
            )
        return current_context

def compress_context(content, target_ratio=0.5):
    """
    컨텍스트를 압축하여 더 많은 정보를 담을 수 있게 한다.
    코드의 경우 주석/공백 제거, 문서의 경우 요약을 활용한다.
    """
    import re

    compressed = content

    # 1. 연속 공백 줄 제거
    compressed = re.sub(r'\n\s*\n\s*\n', '\n\n', compressed)

    # 2. 단일행 주석 제거 (코드 파일)
    # 주의: docstring이나 중요 주석은 보존 필요
    compressed = re.sub(
        r'^\s*//\s*(?!TODO|FIXME|HACK|NOTE|WARNING).*$',
        '',
        compressed,
        flags=re.MULTILINE,
    )

    # 3. 블록 주석 제거
    compressed = re.sub(
        r'/\*(?!\*)\s*(?!TODO|FIXME)[\s\S]*?\*/',
        '',
        compressed,
    )

    # 4. import 문 그룹화
    lines = compressed.split('\n')
    imports = [l for l in lines if l.strip().startswith(('import ', 'from '))]
    non_imports = [
        l for l in lines
        if not l.strip().startswith(('import ', 'from '))
    ]

    if imports:
        compressed = (
            "// Imports: "
            + ", ".join(
                i.strip().split()[-1] for i in imports[:20]
            )
            + "\n"
            + "\n".join(non_imports)
        )

    return compressed

def estimate_cost(
    input_tokens,
    output_tokens,
    input_price_per_1m,
    output_price_per_1m,
):
    """API 호출 비용을 추정한다."""
    input_cost = (input_tokens / 1_000_000) * input_price_per_1m
    output_cost = (output_tokens / 1_000_000) * output_price_per_1m
    total = input_cost + output_cost
    return {
        "input_cost": round(input_cost, 4),
        "output_cost": round(output_cost, 4),
        "total_cost": round(total, 4),
    }

# 시나리오별 비용 비교 (가상의 가격 기준)
scenarios = [
    {
        "name": "코드 리뷰 (전체 코드베이스)",
        "input_tokens": 800_000,
        "output_tokens": 4_000,
    },
    {
        "name": "로그 분석",
        "input_tokens": 500_000,
        "output_tokens": 2_000,
    },
    {
        "name": "문서 비교 (3개 문서)",
        "input_tokens": 300_000,
        "output_tokens": 8_000,
    },
    {
        "name": "대화형 코딩 세션 (2시간)",
        "input_tokens": 200_000,
        "output_tokens": 50_000,
    },
]

print("=== 시나리오별 토큰 사용량 ===")
for s in scenarios:
    print(
        f"\n{s['name']}:"
        f"\n  입력: {s['input_tokens']:,} tokens"
        f"\n  출력: {s['output_tokens']:,} tokens"
        f"\n  총 토큰: {s['input_tokens'] + s['output_tokens']:,} tokens"
    )

import anthropic

def use_prompt_caching(large_context, queries):
    """
    Anthropic의 프롬프트 캐싱을 활용하여 동일 컨텍스트에
    대한 반복 질의 비용을 크게 절감한다.
    """
    client = anthropic.Anthropic()

    results = []
    for query in queries:
        response = client.messages.create(
            model="claude-sonnet-4-6-20260314",
            max_tokens=4096,
            messages=[
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "text",
                            "text": large_context,
                            "cache_control": {"type": "ephemeral"},
                        },
                        {
                            "type": "text",
                            "text": query,
                        },
                    ],
                }
            ],
        )
        results.append({
            "query": query,
            "response": response.content[0].text,
            "input_tokens": response.usage.input_tokens,
            "cache_read_tokens": getattr(
                response.usage, 'cache_read_input_tokens', 0
            ),
            "cache_creation_tokens": getattr(
                response.usage, 'cache_creation_input_tokens', 0
            ),
        })

    return results

# 사용 예시: 동일 코드베이스에 대해 여러 질문
queries = [
    "이 프로젝트의 인증 시스템 아키텍처를 설명해 주세요.",
    "데이터베이스 스키마의 잠재적 문제점은 무엇인가요?",
    "API 엔드포인트 목록과 각각의 역할을 정리해 주세요.",
    "테스트 커버리지가 부족한 모듈은 어디인가요?",
]

# 첫 번째 질의: 캐시 생성 (정상 비용)
# 두 번째 이후: 캐시 히트 (입력 비용 90% 절감)

# 나쁜 예: 프로젝트 전체를 무차별적으로 덤프
# bad_context = dump_entire_project("/path/to/monorepo")
# 결과: 토큰 낭비 + 관련 없는 정보가 노이즈로 작용

# 좋은 예: 질문과 관련된 파일만 선별
def smart_file_selection(query, file_index):
    """
    질문의 의도에 맞는 파일만 선별하여 컨텍스트를 구성한다.
    """
    # 1단계: 파일 인덱스에서 관련 파일 식별
    relevant_files = []
    keywords = extract_keywords(query)

    for file_path, file_meta in file_index.items():
        relevance_score = calculate_relevance(
            keywords, file_meta
        )
        if relevance_score > 0.3:  # 임계값
            relevant_files.append(
                (file_path, relevance_score)
            )

    # 2단계: 관련성 순으로 정렬하여 상위 N개만 선택
    relevant_files.sort(key=lambda x: x[1], reverse=True)
    return [f[0] for f in relevant_files[:50]]

def extract_keywords(query):
    """쿼리에서 키워드를 추출한다."""
    # 간단한 키워드 추출 (실제로는 더 정교한 방법 사용)
    stopwords = {"the", "a", "is", "in", "to", "of", "and"}
    words = query.lower().split()
    return [w for w in words if w not in stopwords]

def calculate_relevance(keywords, file_meta):
    """파일의 메타데이터와 키워드의 관련성을 계산한다."""
    score = 0
    file_text = (
        file_meta.get("path", "")
        + " "
        + file_meta.get("description", "")
    ).lower()
    for kw in keywords:
        if kw in file_text:
            score += 1
    return score / max(len(keywords), 1)

# 나쁜 예: 중요한 지시사항을 컨텍스트 중간에 배치
messages = [
    {
        "role": "user",
        "content": (
            huge_context_part_1        # 400K tokens
            + "\n\nIMPORTANT: "
            + critical_instruction     # 여기가 중간에 묻힘
            + "\n\n"
            + huge_context_part_2      # 400K tokens
        ),
    }
]

# 좋은 예: 중요한 지시사항은 시작 또는 끝에 배치
messages = [
    {
        "role": "user",
        "content": (
            "## INSTRUCTIONS (READ FIRST)\n"
            + critical_instruction     # 시작 부분에 배치
            + "\n\n## CONTEXT\n"
            + huge_context             # 대량 데이터
            + "\n\n## REMINDER\n"
            + critical_instruction     # 끝에 다시 한번 상기
        ),
    }
]

# 컨텍스트 크기별 예상 지연시간 (서빙 환경에 따라 다름)
LATENCY_ESTIMATES = {
    "8K tokens": "1~3초",
    "32K tokens": "3~8초",
    "128K tokens": "10~30초",
    "500K tokens": "30~90초",
    "1M tokens": "60~180초",
}

# 실시간 응답이 필요한 서비스에서 1M 컨텍스트는 부적합
# 스트리밍을 활용하여 체감 지연을 줄이는 것이 필수

def stream_long_context_response(client, messages, model):
    """스트리밍으로 긴 컨텍스트의 체감 지연시간을 줄인다."""
    with client.messages.stream(
        model=model,
        max_tokens=8192,
        messages=messages,
    ) as stream:
        first_token_time = None
        for text in stream.text_stream:
            if first_token_time is None:
                first_token_time = time.time()
                print(f"Time to first token: {first_token_time:.1f}s")
            print(text, end="", flush=True)

# 토큰 사용량 모니터링 및 예산 관리
class TokenBudgetManager:
    """일일/월간 토큰 예산을 관리한다."""

    def __init__(self, daily_budget_tokens=10_000_000):
        self.daily_budget = daily_budget_tokens
        self.used_today = 0
        self.history = []

    def check_budget(self, estimated_tokens):
        """요청 전 예산 확인."""
        if self.used_today + estimated_tokens > self.daily_budget:
            remaining = self.daily_budget - self.used_today
            raise BudgetExceededError(
                f"Daily budget exceeded. "
                f"Remaining: {remaining:,} tokens, "
                f"Requested: {estimated_tokens:,} tokens"
            )
        return True

    def record_usage(self, input_tokens, output_tokens):
        """사용량 기록."""
        total = input_tokens + output_tokens
        self.used_today += total
        self.history.append({
            "timestamp": time.time(),
            "input": input_tokens,
            "output": output_tokens,
        })

    def get_report(self):
        """사용량 리포트 생성."""
        return {
            "used_today": f"{self.used_today:,}",
            "budget_remaining": f"{self.daily_budget - self.used_today:,}",
            "utilization": f"{self.used_today / self.daily_budget:.1%}",
            "request_count": len(self.history),
        }

class BudgetExceededError(Exception):
    pass

def choose_approach(requirements):
    """
    요구사항에 따라 RAG와 대규모 컨텍스트 중
    적합한 접근법을 추천한다.
    """
    score_rag = 0
    score_long_context = 0

    # 1. 데이터 규모
    if requirements.get("data_size_gb", 0) > 1:
        score_rag += 3
    elif requirements.get("data_size_gb", 0) > 0.1:
        score_rag += 1
    else:
        score_long_context += 2

    # 2. 업데이트 빈도
    if requirements.get("update_frequency") == "real-time":
        score_long_context += 2
    elif requirements.get("update_frequency") == "daily":
        score_rag += 1

    # 3. 멀티홉 추론 필요
    if requirements.get("multi_hop_reasoning"):
        score_long_context += 3

    # 4. 지연시간 요구사항
    if requirements.get("max_latency_sec", 999) < 5:
        score_rag += 2
    elif requirements.get("max_latency_sec", 999) < 30:
        score_rag += 1

    # 5. 구축 시간 제약
    if requirements.get("setup_time_days", 999) < 1:
        score_long_context += 3

    # 6. 비용 민감도
    if requirements.get("cost_sensitive"):
        score_rag += 2  # 반복 질의 시 RAG가 저렴

    # 7. 정확도 요구수준
    if requirements.get("accuracy_critical"):
        score_long_context += 2  # 전체 문맥 참조가 유리

    if score_long_context > score_rag:
        recommendation = "LONG_CONTEXT"
    elif score_rag > score_long_context:
        recommendation = "RAG"
    else:
        recommendation = "HYBRID"

    return {
        "recommendation": recommendation,
        "rag_score": score_rag,
        "long_context_score": score_long_context,
        "reasoning": f"RAG={score_rag}, LongCtx={score_long_context}",
    }

# 사용 예시
result = choose_approach({
    "data_size_gb": 0.05,
    "update_frequency": "real-time",
    "multi_hop_reasoning": True,
    "max_latency_sec": 60,
    "setup_time_days": 0,
    "cost_sensitive": False,
    "accuracy_critical": True,
})
print(f"추천: {result['recommendation']}")
print(f"근거: {result['reasoning']}")

class HybridRetriever:
    """RAG + 대규모 컨텍스트 하이브리드 검색 시스템."""

    def __init__(self, vector_store, llm_client, model):
        self.vector_store = vector_store
        self.client = llm_client
        self.model = model

    def retrieve_and_augment(self, query, max_context_tokens=500_000):
        """
        1단계: RAG로 관련 문서를 검색한다.
        2단계: 검색된 문서의 전체 원문을 대규모 컨텍스트로 전달한다.

        -> RAG의 검색 정확도 + 대규모 컨텍스트의 전체 문맥 이해를 결합
        """
        # RAG 1단계: 관련 문서 청크 검색
        chunks = self.vector_store.similarity_search(
            query, k=20
        )

        # 검색된 청크의 원본 문서 전체를 수집
        source_docs = set()
        for chunk in chunks:
            source_docs.add(chunk.metadata["source_document"])

        # 원본 문서 전체를 컨텍스트에 포함
        full_context = ""
        token_count = 0
        for doc_id in source_docs:
            doc = self.vector_store.get_full_document(doc_id)
            estimated_tokens = len(doc.content) // 3
            if token_count + estimated_tokens > max_context_tokens:
                break
            full_context += (
                f"\n\n=== {doc.title} ===\n{doc.content}"
            )
            token_count += estimated_tokens

        # 2단계: 전체 문서와 함께 LLM 호출
        response = self.client.messages.create(
            model=self.model,
            max_tokens=4096,
            messages=[
                {
                    "role": "user",
                    "content": (
                        f"다음은 질문과 관련된 문서들의 전문입니다.\n"
                        f"질문: {query}\n\n"
                        f"문서:\n{full_context}"
                    ),
                }
            ],
        )
        return response.content[0].text

import anthropic
import os

class CodeMigrationAssistant:
    """1M 컨텍스트를 활용한 코드 마이그레이션 어시스턴트."""

    def __init__(self):
        self.client = anthropic.Anthropic()
        self.model = "claude-opus-4-6-20260314"

    def analyze_migration_scope(self, source_dir, migration_spec):
        """
        1단계: 전체 코드베이스를 분석하여 마이그레이션 범위를 파악한다.
        """
        codebase = self._collect_files(source_dir)

        response = self.client.messages.create(
            model=self.model,
            max_tokens=8192,
            messages=[
                {
                    "role": "user",
                    "content": f"""전체 코드베이스를 분석하여 마이그레이션 계획을 수립해 주세요.

마이그레이션 사양:
{migration_spec}

분석 항목:
1. 영향받는 파일 목록과 변경 범위
2. 의존성 그래프에서의 변경 순서
3. 호환성이 깨지는 지점 (Breaking Changes)
4. 단계별 마이그레이션 계획
5. 예상 위험 요소

코드베이스:
{codebase}""",
                }
            ],
        )
        return response.content[0].text

    def generate_migration_code(self, source_dir, plan, file_batch):
        """
        2단계: 마이그레이션 계획에 따라 실제 변환 코드를 생성한다.
        """
        files_content = ""
        for filepath in file_batch:
            full_path = os.path.join(source_dir, filepath)
            if os.path.exists(full_path):
                with open(full_path, 'r') as f:
                    content = f.read()
                files_content += (
                    f"\n### {filepath}\n```\n{content}\n```\n"
                )

        response = self.client.messages.create(
            model=self.model,
            max_tokens=16384,
            messages=[
                {
                    "role": "user",
                    "content": f"""마이그레이션 계획에 따라 다음 파일들을 변환해 주세요.

마이그레이션 계획:
{plan}

변환 대상 파일:
{files_content}

각 파일에 대해:
1. 변환된 전체 코드
2. 변경 사항 요약
3. 수동 검토가 필요한 부분 표시
를 제공해 주세요.""",
                }
            ],
        )
        return response.content[0].text

    def _collect_files(self, directory, max_chars=2_700_000):
        """디렉토리의 파일을 수집한다."""
        extensions = {'.py', '.ts', '.js', '.java', '.go', '.rs'}
        result = []
        total_chars = 0

        for root, dirs, files in os.walk(directory):
            dirs[:] = [
                d for d in dirs
                if d not in {
                    'node_modules', '.git', '__pycache__',
                    'dist', 'build', 'vendor',
                }
            ]
            for fname in sorted(files):
                if any(fname.endswith(ext) for ext in extensions):
                    fpath = os.path.join(root, fname)
                    rel_path = os.path.relpath(fpath, directory)
                    try:
                        with open(fpath, 'r', encoding='utf-8') as f:
                            content = f.read()
                        entry = f"\n### {rel_path}\n```\n{content}\n```\n"
                        if total_chars + len(entry) > max_chars:
                            return '\n'.join(result)
                        result.append(entry)
                        total_chars += len(entry)
                    except (UnicodeDecodeError, PermissionError):
                        continue

        return '\n'.join(result)

# NIAH 벤치마크의 올바른 해석 방법

niah_results = {
    "claude_opus_4_6": {
        "8K": 99.8,
        "32K": 99.7,
        "128K": 99.5,
        "500K": 99.3,
        "1M": 99.2,
    },
    "gpt_5": {
        "8K": 99.5,
        "32K": 99.2,
        "128K": 98.0,
        "256K": 97.5,
    },
    "gemini_2_pro": {
        "8K": 99.3,
        "32K": 99.0,
        "128K": 98.8,
        "500K": 98.6,
        "1M": 98.5,
        "2M": 97.1,
    },
}

# NIAH 점수 해석 시 주의사항:
# 1. NIAH는 "특정 사실 하나를 찾는" 단순 태스크
# 2. 실제 업무는 "여러 정보를 종합하는" 복잡한 태스크
# 3. NIAH 99%라도 실제 업무 정확도는 다를 수 있음
# 4. 컨텍스트 유형(코드/문서/로그)에 따라 성능이 다름

for model, scores in niah_results.items():
    print(f"\n{model}:")
    for size, score in scores.items():
        bar = '#' * int(score / 2)
        print(f"  {size:>6}: {score}% {bar}")

[현재] 질문 -> 관련 파일 검색 -> 부분 컨텍스트 -> LLM -> 답변
              (수동/RAG)         (4K~128K)

[1M 시대] 질문 -> 전체 코드베이스 로딩 -> LLM -> 답변
                 (자동)               (1M)

[미래]   질문 -> 에이전트가 자율적으로 필요한 정보 수집 -> LLM -> 실행
                (도구 사용)                           (무제한)