出海团队大模型应用与 API 架构百科：ChatGPT、Claude、DeepSeek 成本优化、上下文工程与工业级落地

import os
import time
import anthropic
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

client = anthropic.Anthropic(
    api_key=os.environ.get("ANTHROPIC_API_KEY"),
    # 工业实践：生产环境可配置企业网关反代，避免本地直接连接断流
    base_url=os.environ.get("ANTHROPIC_BASE_URL", "https://api.anthropic.com")
)

# 巨量长篇系统准则 (包含品牌人设、100条FAQ与拒付防御红线)
STATIC_BRAND_KNOWLEDGE = """
[SYSTEM ENTERPRISE PROMPT: ~5,000 TOKENS]
You are the Chief Brand Voice Officer for a high-end DTC apparel brand.
Rules: Never use corporate jargon (delve, robust, testament, multifaceted).
Always reply in natural native tone, empathetic and human-written.
[Full FAQ database and refund matrix...]
"""

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type((anthropic.RateLimitError, anthropic.APIConnectionError))
)
def generate_brand_content(user_query: str) -> str:
    start_time = time.time()
    
    # 核心：利用 cache_control: {"type": "ephemeral"} 声明显式缓存
    response = client.beta.prompt_caching.messages.create(
        model="claude-3-7-sonnet-20250219",
        max_tokens=2048,
        system=[
            {
                "type": "text",
                "text": STATIC_BRAND_KNOWLEDGE,
                "cache_control": {"type": "ephemeral"}
            }
        ],
        messages=[
            {"role": "user", "content": user_query}
        ]
    )
    
    # 性能与成本审计追踪
    usage = response.usage
    cache_read = getattr(usage, 'cache_read_input_tokens', 0)
    cache_creation = getattr(usage, 'cache_creation_input_tokens', 0)
    
    print(f"[LLM Audit] 耗时: {time.time() - start_time:.2f}s | "
          f"缓存命中: {cache_read} tokens (省90%) | 新增缓存: {cache_creation} tokens")
    
    return response.content[0].text

# 生产测试调用
if __name__ == "__main__":
    result = generate_brand_content("Write a personalized apology email for delayed custom boots order #8921.")
    print("\nResponse:\n", result)