第十章：生产级 Harness 构建¶

class PrefixCacheAwarePrompt:
    """缓存感知的提示词组装"""

    def __init__(self, static_modules: list, cache_boundary_tokens: int = 47000):
        self.static_prefix = "\n\n".join(static_modules)
        self.cache_boundary = cache_boundary_tokens

    def assemble(self, dynamic_context: str) -> dict:
        """组装提示词,标记缓存边界"""
        return {
            "system": [
                {
                    "type": "text",
                    "text": self.static_prefix,
                    "cache_control": {"type": "ephemeral"}  # 标记可缓存
                },
                {
                    "type": "text",
                    "text": dynamic_context  # 动态部分不缓存
                }
            ]
        }

from typing import Dict, Any, List
from dataclasses import dataclass
from enum import Enum
import hashlib

class PromptModule(Enum):
    """提示词模块类型"""
    CORE_IDENTITY = "core_identity"
    CAPABILITIES = "capabilities"
    DOMAIN_KNOWLEDGE = "domain_knowledge"
    CONTEXT_SPECIFIC = "context_specific"

@dataclass
class PromptSection:
    """提示词段落"""
    module: PromptModule
    content: str
    cacheable: bool
    priority: int

class ModularSystemPrompt:
    """模块化系统提示词管理器"""

    MAX_TOKENS = 54000
    CACHE_BOUNDARY = 47000

    def __init__(self):
        self.sections: Dict[PromptModule, List[PromptSection]] = {
            module: [] for module in PromptModule
        }
        self.cache_metadata: Dict[str, str] = {}

    def add_section(
        self,
        module: PromptModule,
        content: str,
        cacheable: bool = True,
        priority: int = 0,
    ) -> None:
        """添加提示词段落"""
        section = PromptSection(
            module=module,
            content=content,
            cacheable=cacheable,
            priority=priority,
        )
        self.sections[module].append(section)

        if cacheable:
            content_hash = hashlib.sha256(content.encode()).hexdigest()
            self.cache_metadata[content_hash] = content

    def build_prompt(
        self,
        context_variables: Dict[str, str] = None,
        max_tokens: int = MAX_TOKENS,
    ) -> tuple[str, Dict[str, Any]]:
        """构建最终系统提示词"""
        context_variables = context_variables or {}

        ordered_modules = [
            PromptModule.CORE_IDENTITY,
            PromptModule.CAPABILITIES,
            PromptModule.DOMAIN_KNOWLEDGE,
            PromptModule.CONTEXT_SPECIFIC,
        ]

        cacheable_content = []
        non_cacheable_content = []
        metadata = {
            "static_tokens": 0,
            "dynamic_tokens": 0,
            "caching_enabled": False,
            "module_breakdown": {},
        }

        for module in ordered_modules:
            sections = self.sections.get(module, [])
            sections.sort(key=lambda s: s.priority, reverse=True)

            for section in sections:
                token_estimate = len(section.content) * 0.4

                content = section.content
                if module == PromptModule.CONTEXT_SPECIFIC:
                    for var_name, var_value in context_variables.items():
                        content = content.replace(f"{{{{{var_name}}}}}", var_value)

                if section.cacheable:
                    cacheable_content.append(content)
                    metadata["static_tokens"] += int(token_estimate)
                else:
                    non_cacheable_content.append(content)
                    metadata["dynamic_tokens"] += int(token_estimate)

                metadata["module_breakdown"][module.value] = {
                    "sections": len(sections),
                    "tokens": int(token_estimate),
                }

        static_part = "\n\n".join(cacheable_content)
        dynamic_part = "\n\n".join(non_cacheable_content)
        total_tokens = metadata["static_tokens"] + metadata["dynamic_tokens"]

        if total_tokens > max_tokens:
            print(f"[Warning] Prompt exceeds {max_tokens} tokens")

        metadata["caching_enabled"] = len(static_part) > 1024

        if metadata["caching_enabled"]:
            static_hash = hashlib.sha256(static_part.encode()).hexdigest()
            metadata["cache_key"] = static_hash
            metadata["cache_boundary"] = len(static_part)

        final_prompt = static_part + "\n\n" + dynamic_part

        return final_prompt, metadata

    def optimize_for_token_budget(
        self,
        token_budget: int,
        context_variables: Dict[str, str] = None,
    ) -> str:
        """在Token预算限制下优化提示词"""
        sections_by_priority = []

        for module in PromptModule:
            for section in self.sections.get(module, []):
                token_estimate = len(section.content) * 0.4
                sections_by_priority.append((section, token_estimate, module))

        sections_by_priority.sort(
            key=lambda x: (x[2].value, x[0].priority),
            reverse=True
        )

        selected_sections = []
        total_tokens = 0

        for section, tokens, module in sections_by_priority:
            if total_tokens + tokens <= token_budget:
                selected_sections.append(section)
                total_tokens += tokens
            elif section.priority >= 10:
                selected_sections.append(section)
                total_tokens += tokens

        content = "\n\n".join(s.content for s in selected_sections)
        return content

    def get_cache_stats(self) -> Dict[str, Any]:
        """获取缓存统计信息"""
        total_cacheable = sum(
            len(s.content)
            for module_sections in self.sections.values()
            for s in module_sections
            if s.cacheable
        )

        total_non_cacheable = sum(
            len(s.content)
            for module_sections in self.sections.values()
            for s in module_sections
            if not s.cacheable
        )

        return {
            "total_modules": len(self.sections),
            "total_sections": sum(len(sections) for sections in self.sections.values()),
            "cacheable_tokens": int(total_cacheable * 0.4),
            "non_cacheable_tokens": int(total_non_cacheable * 0.4),
            "cache_potential_savings": int(total_cacheable * 0.4 * 0.35),
        }

def create_default_system_prompt() -> ModularSystemPrompt:
    """创建默认的系统提示词"""
    prompt = ModularSystemPrompt()

    # Core Identity
    prompt.add_section(
        PromptModule.CORE_IDENTITY,
        """你是Claude,一个由Anthropic创建的AI助手。

你的核心职责是:
1. 准确、有帮助地回答用户问题
2. 承认你的限制和不确定性
3. 拒绝可能有害的请求
4. 遵循所有安全指南""",
        cacheable=False,
        priority=100,
    )

    # Capabilities
    prompt.add_section(
        PromptModule.CAPABILITIES,
        """## 工具使用指南

你可以使用以下工具:
1. web_search - 搜索网络信息
2. code_execution - 执行代码
3. file_operations - 文件操作

### 思维过程
遵循以下思维框架:
1. 分解问题
2. 制定策略
3. 执行步骤
4. 验证结果""",
        cacheable=True,
        priority=50,
    )

    # Domain Knowledge
    prompt.add_section(
        PromptModule.DOMAIN_KNOWLEDGE,
        """## 编程最佳实践

1. 代码质量
   - 遵循编码标准
   - 包含错误处理
   - 添加文档

2. 性能
   - 考虑复杂度
   - 缓存计算结果

### 常见算法
...""",
        cacheable=True,
        priority=40,
    )

    # Context-Specific
    prompt.add_section(
        PromptModule.CONTEXT_SPECIFIC,
        """## 当前任务信息

用户任务:{{task}}

背景信息:
{{context}}

特定约束:
{{constraints}}""",
        cacheable=False,
        priority=30,
    )

    return prompt

if __name__ == "__main__":
    # 创建系统提示词
    system_prompt = create_default_system_prompt()

    # 构建提示词
    context = {
        "task": "写一个Python排序算法",
        "context": "学生练习项目",
        "constraints": "不使用内置排序函数",
    }

    final_prompt, metadata = system_prompt.build_prompt(context)

    print("=== 最终提示词 ===")
    print(final_prompt[:500] + "...\n")

    print("=== 元数据 ===")
    print(f"静态Token: {metadata['static_tokens']}")
    print(f"动态Token: {metadata['dynamic_tokens']}")
    print(f"缓存启用: {metadata['caching_enabled']}")

    print("\n=== 缓存统计 ===")
    stats = system_prompt.get_cache_stats()
    print(f"可缓存Token: {stats['cacheable_tokens']}")
    print(f"潜在节省: {stats['cache_potential_savings']}Token")