1M 上下文窗口时代的 LLM 应用策略: 大规模上下文处理实战指南

# 换算 1M token 实际相当于多少数据量
# 基于 Anthropic 的分词器(claude-tokenizer)

TOKEN_ESTIMATES = {
    "英文单词数": "约 750,000 个单词",
    "韩文字符数": "约 500,000 个字符(token 效率低于英文)",
    "英文小说": "约 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 = 4096
    results["Sliding Window (w=4096)"] = (
        seq_length * window_size * d_model
    )

    # 3. Ring Attention(分布式)
    # 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 风格)
    # 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
    # token 估算:英文约 4 字符/token，代码约 3 字符/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 的日志，也能在 token 上限内处理
with open("/var/log/app/production.log", "r") as f:
    # 最近约 100 万行大致相当于 1M token
    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}。请提问。",
        })
        # token 估算(粗略)
        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 token 刷新逻辑中是否可能出现竞态条件(race condition)?"))
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 阶段:只加载所有文件的摘要。
        既能把握整体结构，又能节省 token。
        """
        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("=== 各场景的 token 使用量 ===")
for s in scenarios:
    print(
        f"\n{s['name']}:"
        f"\n  输入: {s['input_tokens']:,} tokens"
        f"\n  输出: {s['output_tokens']:,} tokens"
        f"\n  总 token: {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 = [
    "请说明这个项目的认证系统架构。",
    "数据库 schema 存在哪些潜在问题?",
    "请列出 API 端点清单及各自的作用。",
    "哪些模块的测试覆盖率不足?",
]

# 第一次查询:创建缓存(正常成本)
# 第二次及以后:命中缓存(输入成本节省 90%)

# 反例:无差别地把整个项目倾倒进去
# bad_context = dump_entire_project("/path/to/monorepo")
# 结果:浪费 token + 无关信息成为噪声

# 正例:只挑选与问题相关的文件
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 上下文并不合适
# 利用流式传输(streaming)来降低体感延迟是必要的

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)

# token 使用量监控与预算管理
class TokenBudgetManager:
    """管理日/月 token 预算。"""

    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 -> 执行
                (工具调用)                (无限制)