07b - 百万 Token 上下文实战

决策树：长上下文任务模型选择
├── 需要最高推理质量 + 长上下文？ → Gemini 3.1 Pro（Preview）
├── 需要稳定 GA + 长上下文？ → Gemini 3 Pro
├── 需要速度 + 成本平衡？ → Gemini 3 Flash（1M 上下文，$0.50/1M 输入）
├── 预算极度敏感？ → Gemini 2.5 Flash（1M 上下文，$0.30/1M 输入）
└── 上下文 < 200K + 极低成本？ → Gemini 2.5 Flash-Lite（$0.10/1M 输入）

import os
from google import genai
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
def collect_project_files(project_dir, extensions=(".ts", ".tsx", ".py")):
    """收集项目文件并计算 token 数"""
    files = []
    for root, dirs, filenames in os.walk(project_dir):
        # 跳过不必要的目录
        dirs[:] = [d for d in dirs if d not in {
            "node_modules", ".git", "dist", "build",
            "__pycache__", ".next", "coverage"
        }]
        for f in filenames:
            if f.endswith(extensions):
                path = os.path.join(root, f)
                with open(path) as fh:
                    files.append((path, fh.read()))
    return files
 
def estimate_tokens(text):
    """粗略估算 token 数（中文约 1.5 token/字，英文约 1.3 token/词）"""
    return len(text) // 3  # 粗略估算
 
def build_context_under_budget(files, budget=190_000):
    """构建不超过 token 预算的上下文"""
    context_parts = []
    total_tokens = 0
    
    # 按文件重要性排序（入口文件优先）
    priority_patterns = ["index.", "main.", "app.", "config.", "types."]
    files.sort(key=lambda f: (
        not any(p in f[0].lower() for p in priority_patterns),
        len(f[1])  # 小文件优先
    ))
    
    for path, content in files:
        file_tokens = estimate_tokens(content)
        if total_tokens + file_tokens > budget:
            # 超预算，添加文件摘要而非全文
            context_parts.append(f"// File: {path} (摘要: {len(content)} 字符, 已省略)")
            continue
        context_parts.append(f"// File: {path}\n{content}")
        total_tokens += file_tokens
    
    return "\n\n".join(context_parts), total_tokens
 
# 使用示例
files = collect_project_files("./src")
context, token_count = build_context_under_budget(files, budget=190_000)
print(f"上下文大小: ~{token_count:,} tokens（在 200K 阶梯内）")

from google import genai
from google.genai import types
 
client = genai.Client(api_key="your-api-key")
 
# ========== 显式缓存：适合反复查询同一代码库 ==========
 
# 1. 上传大文件（代码库打包或长文档）
uploaded_file = client.files.upload(file="./project_codebase.txt")
 
# 2. 创建缓存（TTL 设为 1 小时，可按需调整）
cache = client.caches.create(
    model="gemini-3.1-pro-preview",
    config=types.CreateCachedContentConfig(
        display_name="my-project-codebase",
        system_instruction=(
            "你是一位资深软件架构师。以下是一个完整的项目代码库，"
            "请基于这些代码回答用户的问题。"
        ),
        contents=[uploaded_file],
        ttl="3600s",  # 1 小时
    )
)
print(f"缓存已创建: {cache.name}")
 
# 3. 使用缓存进行多次查询（每次查询只需支付缓存价格）
questions = [
    "这个项目的整体架构是什么？画出组件关系图。",
    "找出所有潜在的安全漏洞。",
    "哪些模块的测试覆盖率可能不足？",
    "给出重构建议，按优先级排序。",
]
 
for q in questions:
    response = client.models.generate_content(
        model="gemini-3.1-pro-preview",
        contents=q,
        config=types.GenerateContentConfig(cached_content=cache.name)
    )
    print(f"\n问题: {q}")
    print(f"回答: {response.text[:200]}...")
    # 查看缓存命中情况
    print(f"缓存 token: {response.usage_metadata.cached_content_token_count}")
    print(f"总输入 token: {response.usage_metadata.prompt_token_count}")

# 批处理 API 适合非实时的批量分析任务
# 所有模型的批处理价格为标准价格的 50%
 
# Gemini 3.1 Pro 批处理价格：
# 输入 ≤200K: $1.00/1M（标准 $2.00 的 50%）
# 输入 >200K: $2.00/1M（标准 $4.00 的 50%）
# 输出: $6.00/1M（标准 $12.00 的 50%）
 
# 批处理 + 缓存可以叠加使用，实现最大成本优化

import os
from pathlib import Path
from google import genai
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
def collect_codebase(project_dir, max_tokens=900_000):
    """智能收集代码库，按重要性排序"""
    files = []
    
    # 定义文件优先级
    high_priority = {
        "package.json", "tsconfig.json", "Cargo.toml",
        "pyproject.toml", "README.md", "ARCHITECTURE.md"
    }
    
    skip_dirs = {
        "node_modules", ".git", "dist", "build", "target",
        "__pycache__", ".next", "coverage", ".turbo", "vendor"
    }
    
    code_extensions = {
        ".ts", ".tsx", ".js", ".jsx", ".py", ".rs",
        ".go", ".java", ".kt", ".swift", ".vue", ".svelte"
    }
    
    for root, dirs, filenames in os.walk(project_dir):
        dirs[:] = [d for d in dirs if d not in skip_dirs]
        for fname in filenames:
            path = os.path.join(root, fname)
            ext = Path(fname).suffix
            if fname in high_priority or ext in code_extensions:
                try:
                    with open(path, encoding="utf-8") as f:
                        content = f.read()
                    files.append({
                        "path": os.path.relpath(path, project_dir),
                        "content": content,
                        "priority": 0 if fname in high_priority else 1,
                        "size": len(content)
                    })
                except (UnicodeDecodeError, PermissionError):
                    continue
    
    # 按优先级排序，然后按大小排序
    files.sort(key=lambda f: (f["priority"], f["size"]))
    
    # 组装上下文，控制总量
    context_parts = []
    total_chars = 0
    char_budget = max_tokens * 3  # 粗略：3 字符 ≈ 1 token
    
    for f in files:
        if total_chars + f["size"] > char_budget:
            context_parts.append(
                f"// File: {f['path']} [已省略, {f['size']} 字符]"
            )
            continue
        context_parts.append(f"// File: {f['path']}\n{f['content']}")
        total_chars += f["size"]
    
    return "\n\n".join(context_parts)
 
# 收集代码库
codebase = collect_codebase("./my-project")

# 精确计算 token 数
token_response = client.models.count_tokens(
    model="gemini-3.1-pro-preview",
    contents=codebase
)
token_count = token_response.total_tokens
print(f"代码库 token 数: {token_count:,}")
 
# 根据 token 数选择策略
if token_count <= 200_000:
    print(f"✅ 在 200K 阶梯内，输入价格: ${token_count / 1_000_000 * 2:.2f}")
    strategy = "direct"
elif token_count <= 1_000_000:
    print(f"⚠️ 超过 200K 阶梯，输入价格: ${token_count / 1_000_000 * 4:.2f}")
    print("建议：使用上下文缓存降低重复查询成本")
    strategy = "cache"
else:
    print(f"❌ 超过 1M 上下文窗口，需要分块处理")
    strategy = "chunk"

# 流式响应：边生成边输出，适合长上下文的交互式分析
response_stream = client.models.generate_content_stream(
    model="gemini-3.1-pro-preview",
    contents=f"""作为资深架构师，请对以下代码库进行全面分析：
 
{codebase}
 
请提供：
1. 架构概览（用 Mermaid 格式画组件关系图）
2. 模块间依赖关系分析
3. 代码质量评估（命名规范、结构清晰度、复杂度）
4. 潜在问题和技术债务（按严重程度排序）
5. 重构建议（按优先级和投入产出比排序）
""",
    config={"thinking_config": {"thinking_level": "HIGH"}}
)
 
# 逐块输出结果
full_response = ""
for chunk in response_stream:
    if chunk.text:
        print(chunk.text, end="", flush=True)
        full_response += chunk.text
 
print(f"\n\n--- 完成 ---")
print(f"总输出长度: {len(full_response)} 字符")

import time
from google import genai
from google.genai import types
 
client = genai.Client(api_key="your-api-key")
 
# 1. 上传 PDF 文件（支持最大 100MB）
pdf_file = client.files.upload(file="./contract_bundle.pdf")
 
# 等待文件处理完成
while pdf_file.state.name == "PROCESSING":
    time.sleep(2)
    pdf_file = client.files.get(name=pdf_file.name)
print(f"文件处理完成: {pdf_file.uri}")
 
# 2. 创建缓存（法律合同通常需要多轮审查）
cache = client.caches.create(
    model="gemini-3.1-pro-preview",
    config=types.CreateCachedContentConfig(
        display_name="contract-review",
        system_instruction=(
            "你是一位资深法律顾问，精通中国合同法和商业法。"
            "请基于上传的合同文件回答问题，引用具体条款编号。"
            "如果合同中存在风险条款，请明确标注风险等级（高/中/低）。"
        ),
        contents=[pdf_file],
        ttl="7200s",  # 2 小时，足够完成审查
    )
)
 
# 3. 多轮审查查询
review_queries = [
    "列出所有关键条款的摘要，包括：合同期限、付款条件、违约责任、知识产权归属、保密义务。",
    "找出所有可能对我方不利的条款，按风险等级排序，并给出修改建议。",
    "对比行业标准，这份合同的付款条件和违约金比例是否合理？",
    "生成一份合同审查报告，包含风险评估和修改建议清单。",
]
 
for i, query in enumerate(review_queries, 1):
    print(f"\n{'='*60}")
    print(f"审查轮次 {i}: {query[:50]}...")
    print(f"{'='*60}")
    
    response = client.models.generate_content(
        model="gemini-3.1-pro-preview",
        contents=query,
        config=types.GenerateContentConfig(
            cached_content=cache.name,
            thinking_config=types.ThinkingConfig(thinking_level="HIGH")
        )
    )
    print(response.text)

from google import genai
 
client = genai.Client(api_key="your-api-key")
 
# 多文件推理：同时分析需求文档、设计文档和代码实现
# 利用 Gemini 的多点注意力能力进行交叉验证
 
requirements = open("./docs/requirements.md").read()
design_doc = open("./docs/design.md").read()
api_code = open("./src/api/routes.ts").read()
db_schema = open("./prisma/schema.prisma").read()
test_files = open("./tests/api.test.ts").read()
 
cross_reference_prompt = f"""请对以下项目文档和代码进行交叉引用分析：
 
## 需求文档
{requirements}
 
## 设计文档
{design_doc}
 
## API 实现代码
{api_code}
 
## 数据库 Schema
{db_schema}
 
## 测试文件
{test_files}
 
请完成以下分析：
1. **需求覆盖度**：哪些需求已在代码中实现？哪些缺失？
2. **设计一致性**：代码实现是否与设计文档一致？有哪些偏差？
3. **Schema 匹配**：API 代码中的数据操作是否与 DB Schema 匹配？
4. **测试覆盖**：测试是否覆盖了所有 API 端点和关键业务逻辑？
5. **交叉引用问题**：文档间是否存在矛盾或不一致？
 
请用表格形式呈现需求追踪矩阵。
"""
 
response = client.models.generate_content(
    model="gemini-3.1-pro-preview",
    contents=cross_reference_prompt,
    config={"thinking_config": {"thinking_level": "HIGH"}}
)
print(response.text)

import time
from google import genai
from google.genai import types
 
client = genai.Client(api_key="your-api-key")
 
# ========== 视频分析（最长 ~3 小时） ==========
 
# 1. 上传视频文件
video_file = client.files.upload(file="./meeting_recording.mp4")
while video_file.state.name == "PROCESSING":
    time.sleep(5)
    video_file = client.files.get(name=video_file.name)
 
# 2. 分析完整会议录像
response = client.models.generate_content(
    model="gemini-3.1-pro-preview",
    contents=[
        video_file,
        "请分析这段会议录像，提供：\n"
        "1. 会议摘要（关键决策和行动项）\n"
        "2. 每位参与者的发言要点\n"
        "3. 讨论中的分歧点和最终共识\n"
        "4. 待跟进事项清单（含负责人和截止日期）\n"
        "5. 按时间线标注关键讨论节点"
    ]
)
print(response.text)
 
# ========== 音频分析（最长 ~11 小时） ==========
 
# 上传音频文件
audio_file = client.files.upload(file="./podcast_series.mp3")
while audio_file.state.name == "PROCESSING":
    time.sleep(5)
    audio_file = client.files.get(name=audio_file.name)
 
# 分析长音频
response = client.models.generate_content(
    model="gemini-3.1-pro-preview",
    contents=[
        audio_file,
        "请对这段播客内容进行深度分析：\n"
        "1. 主题大纲和关键论点\n"
        "2. 嘉宾观点对比\n"
        "3. 提到的工具、书籍、资源列表\n"
        "4. 可引用的精彩语录（标注时间戳）"
    ]
)
print(response.text)

你是一位拥有 15 年经验的 [语言/框架] 架构师。

以下是一个完整的项目代码库（[文件数] 个文件，约 [token 数] tokens）：

[完整代码库内容]

请完成以下分析任务：

1. **架构概览**：用 Mermaid 格式画出模块依赖关系图
2. **分层分析**：识别表示层、业务逻辑层、数据访问层的边界是否清晰
3. **代码质量**：
   - 命名规范一致性（1-10 分）
   - 模块内聚度（1-10 分）
   - 耦合度评估（1-10 分）
4. **技术债务**：列出 Top 5 技术债务，每项包含：
   - 问题描述
   - 影响范围
   - 修复难度（高/中/低）
   - 建议优先级
5. **重构路线图**：给出分阶段重构计划

请用中文回答，使用 Markdown 格式。

你是一位 [领域] 专家。以下是 [文档数量] 份相关文档：

## 文档 1：[文档名称]
[文档内容]

## 文档 2：[文档名称]
[文档内容]

...

请完成交叉审查：
1. **一致性检查**：文档间是否存在矛盾或不一致的描述？
2. **完整性检查**：是否有文档 A 提到但文档 B 缺失的内容？
3. **依赖分析**：文档间的引用关系是否正确？
4. **风险识别**：基于所有文档的综合分析，识别 [项目/合同/方案] 的主要风险
5. **改进建议**：给出具体的修改建议，引用具体文档和段落

输出格式：使用表格汇总发现，按严重程度排序。

请对上传的 [视频/音频] 文件进行深度分析：

分析维度：
1. **内容摘要**：按时间线列出关键内容节点
2. **主题提取**：识别 [数量] 个核心主题
3. **[特定需求]**：[如：识别所有提到的技术工具和框架]
4. **情感分析**：[如：评估演讲者的语气变化和关键情绪转折点]
5. **行动项提取**：[如：列出所有提到的待办事项和决策]

输出要求：
- 使用中文
- 关键引用标注时间戳（格式：[HH:MM:SS]）
- 使用 Markdown 表格汇总结构化信息

import os
from google import genai
from google.genai import types
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
# 1. 收集代码库
files = collect_codebase("./ecommerce-platform")  # 使用前面定义的函数
 
# 2. 精确计算 token
token_info = client.models.count_tokens(
    model="gemini-3.1-pro-preview",
    contents=files
)
print(f"总 token: {token_info.total_tokens:,}")  # 输出: 总 token: 612,345
 
# 3. 超过 200K，创建缓存以支持多轮审查
cache = client.caches.create(
    model="gemini-3.1-pro-preview",
    config=types.CreateCachedContentConfig(
        display_name="ecommerce-audit",
        system_instruction="你是一位资深全栈架构师，精通 React 和 Node.js。",
        contents=[files],
        ttl="3600s",
    )
)
 
# 4. 第一轮：架构审查
arch_review = client.models.generate_content(
    model="gemini-3.1-pro-preview",
    contents="请分析整体架构，画出模块依赖图，识别循环依赖和过度耦合。",
    config=types.GenerateContentConfig(
        cached_content=cache.name,
        thinking_config=types.ThinkingConfig(thinking_level="HIGH")
    )
)
 
# 5. 第二轮：安全审查
security_review = client.models.generate_content(
    model="gemini-3.1-pro-preview",
    contents="请进行安全审查：SQL 注入、XSS、CSRF、认证绕过、敏感数据泄露。",
    config=types.GenerateContentConfig(cached_content=cache.name)
)
 
# 6. 第三轮：性能审查
perf_review = client.models.generate_content(
    model="gemini-3.1-pro-preview",
    contents="请分析性能瓶颈：N+1 查询、不必要的重渲染、缺失的索引、内存泄漏风险。",
    config=types.GenerateContentConfig(cached_content=cache.name)
)

import os
import time
from google import genai
from google.genai import types
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
# 1. 上传所有 PDF 文件
pdf_dir = "./papers/"
uploaded_files = []
for fname in sorted(os.listdir(pdf_dir)):
    if fname.endswith(".pdf"):
        f = client.files.upload(file=os.path.join(pdf_dir, fname))
        uploaded_files.append(f)
        print(f"已上传: {fname}")
 
# 等待所有文件处理完成
for f in uploaded_files:
    while f.state.name == "PROCESSING":
        time.sleep(2)
        f = client.files.get(name=f.name)
 
# 2. 创建缓存
cache = client.caches.create(
    model="gemini-3.1-pro-preview",
    config=types.CreateCachedContentConfig(
        display_name="llm-security-papers",
        system_instruction=(
            "你是一位 AI 安全领域的研究员。"
            "以下是 15 篇关于大语言模型安全性的研究论文。"
            "请基于这些论文回答问题，引用具体论文标题和页码。"
        ),
        contents=uploaded_files,
        ttl="7200s",
    )
)
 
# 3. 综合分析
response = client.models.generate_content(
    model="gemini-3.1-pro-preview",
    contents="""请对这 15 篇论文进行系统性综述：
 
1. **研究主题分类**：将论文按研究方向分组（如：对抗攻击、防御机制、评估框架等）
2. **方法论对比**：用表格对比各论文的研究方法、数据集、评估指标
3. **关键发现汇总**：每篇论文的核心贡献（1-2 句话）
4. **共识与分歧**：研究者们在哪些问题上达成共识？哪些问题存在分歧？
5. **研究空白**：基于现有研究，识别尚未充分探索的方向
6. **未来方向建议**：给出 3-5 个值得深入研究的方向
 
请用学术论文的综述风格撰写，引用格式为 [论文标题, 年份]。
""",
    config=types.GenerateContentConfig(
        cached_content=cache.name,
        thinking_config=types.ThinkingConfig(thinking_level="HIGH")
    )
)
print(response.text)