Claude API高级开发技巧完全教程

本教程面向中高级开发者，系统讲解Claude API的核心架构、高级功能与企业级集成实践。阅读后你将掌握Extended Thinking、200K上下文优化、Tool Use进阶、多模态理解等关键能力，并能构建生产级AI应用。

目录

1. Claude API架构与模型选择
2. Extended Thinking深度思考
3. 200K上下文窗口优化
4. Tool Use工具调用进阶
5. 多模态视觉理解
6. System Prompt设计
7. 流式输出实现
8. 批处理API
9. 与OpenAI API对比
10. 企业级集成最佳实践

1. Claude API架构与模型选择

1.1 API调用基础架构

Claude API采用RESTful架构，基于HTTPS进行通信。所有请求均需携带API Key进行认证，响应格式为JSON。

┌──────────────┐     HTTPS/JSON      ┌──────────────────┐
│  你的应用服务  │ ──────────────────▶ │  Claude API 网关  │
│              │ ◀────────────────── │                  │
└──────────────┘     流式/非流式       └──────────────────┘
                                           │
                                           ▼
                                    ┌──────────────────┐
                                    │   模型推理集群    │
                                    │  (Haiku/Sonnet/  │
                                    │   Opus)          │
                                    └──────────────────┘

1.2 模型家族对比

1.3 模型选择决策框架

def select_model(task_type: str, complexity: str, budget: str) -> str:
    """
    根据任务特征选择最优模型
    
    Args:
        task_type: 任务类型 (code/research/chat/vision/batch)
        complexity: 复杂度 (low/medium/high)
        budget: 预算约束 (tight/moderate/flexible)
    """
    # 高复杂度任务：优先 Opus
    if complexity == "high" and budget != "tight":
        return "claude-opus-4-20250514"
    
    # 批量简单任务：Haiku 性价比最高
    if task_type == "batch" and complexity == "low":
        return "claude-3-5-haiku-20241022"
    
    # 通用场景：Sonnet 是最佳平衡点
    return "claude-sonnet-4-20250514"


# 实际调用示例
import anthropic

client = anthropic.Anthropic()

def call_claude(prompt: str, model: str = "claude-sonnet-4-20250514",
                max_tokens: int = 4096) -> str:
    """标准化的Claude API调用封装"""
    message = client.messages.create(
        model=model,
        max_tokens=max_tokens,
        messages=[{"role": "user", "content": prompt}]
    )
    return message.content[0].text

1.4 API版本管理

Claude API通过日期版本号进行管理。建议在请求头中显式指定版本：

# 推荐：使用环境变量管理版本
import os

client = anthropic.Anthropic(
    api_key=os.environ["ANTHROPIC_API_KEY"],
    default_headers={
        "anthropic-version": "2023-06-01"  # 使用稳定版本
    }
)

2. Extended Thinking深度思考

2.1 什么是Extended Thinking

Extended Thinking是Claude的"深度思考"能力。开启后，模型会在生成最终回答前进行内部推理链（chain-of-thought），显著提升数学、编程、逻辑推理等复杂任务的表现。

2.2 基本用法

import anthropic

client = anthropic.Anthropic()

def deep_think(prompt: str, thinking_budget: int = 10000) -> dict:
    """
    使用Extended Thinking处理复杂推理任务
    
    Args:
        prompt: 用户问题
        thinking_budget: 思考token预算 (1024-128000)
    
    Returns:
        包含thinking过程和最终答案的字典
    """
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=16000,
        thinking={
            "type": "enabled",
            "budget_tokens": thinking_budget
        },
        messages=[{
            "role": "user",
            "content": prompt
        }]
    )
    
    result = {"thinking": "", "answer": ""}
    for block in response.content:
        if block.type == "thinking":
            result["thinking"] = block.thinking
        elif block.type == "text":
            result["answer"] = block.text
    
    return result


# 示例：复杂数学推理
result = deep_think(
    "一个水池有A、B两个进水管和C排水管。A单独注满需要6小时，"
    "B单独注满需要8小时，C单独排空需要12小时。如果三管同时打开，"
    "需要多少小时注满水池？请详细推导。",
    thinking_budget=5000
)

print("思考过程：", result["thinking"][:200], "...")
print("最终答案：", result["answer"])

2.3 流式Extended Thinking

对于需要实时展示思考过程的场景，使用流式模式：

def stream_deep_think(prompt: str, thinking_budget: int = 10000):
    """流式输出思考过程和最终答案"""
    thinking_text = ""
    answer_text = ""
    
    with client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=16000,
        thinking={
            "type": "enabled",
            "budget_tokens": thinking_budget
        },
        messages=[{"role": "user", "content": prompt}]
    ) as stream:
        for event in stream:
            if event.type == "content_block_start":
                if event.content_block.type == "thinking":
                    print("\n🧠 [思考中...]\n")
            elif event.type == "content_block_delta":
                if event.delta.type == "thinking_delta":
                    thinking_text += event.delta.thinking
                    print(event.delta.thinking, end="", flush=True)
                elif event.delta.type == "text_delta":
                    if not answer_text:
                        print("\n\n💬 [回答]\n")
                    answer_text += event.delta.text
                    print(event.delta.text, end="", flush=True)
    
    return {"thinking": thinking_text, "answer": answer_text}

2.4 Extended Thinking最佳实践

注意：Extended Thinking的思考过程不计入对话历史中的上下文，但会消耗API调用的token额度。budget_tokens是上限而非固定消耗。

3. 200K上下文窗口优化

3.1 上下文窗口的价值

Claude支持200K tokens的上下文窗口（约15万字中文），可以一次性处理整本书籍、大型代码库或长对话历史。但"能塞进去"不等于"应该塞进去"。

3.2 上下文成本计算

def estimate_cost(input_tokens: int, output_tokens: int, 
                  model: str = "claude-sonnet-4-20250514") -> float:
    """估算API调用成本（美元）"""
    pricing = {
        "claude-sonnet-4-20250514": {"input": 3.0, "output": 15.0},
        "claude-opus-4-20250514": {"input": 15.0, "output": 75.0},
        "claude-3-5-haiku-20241022": {"input": 0.80, "output": 4.0},
    }
    p = pricing[model]
    return (input_tokens * p["input"] + output_tokens * p["output"]) / 1_000_000


# 200K上下文的单次调用成本（Sonnet）
cost = estimate_cost(200_000, 4096, "claude-sonnet-4-20250514")
print(f"单次调用成本: ${cost:.2f}")  # ~$0.66

3.3 上下文优化策略

策略一：分层摘要法

def hierarchical_summarize(documents: list[str], 
                           chunk_size: int = 50000) -> str:
    """
    分层摘要：先对每个文档块摘要，再对摘要进行综合
    适用于超长文档集的处理
    """
    summaries = []
    
    # 第一层：对每个chunk生成摘要
    for i, doc in enumerate(documents):
        chunk = doc[:chunk_size]
        summary = call_claude(
            f"请用500字概括以下文档的核心内容，"
            f"保留关键数据、结论和决策要点：\n\n{chunk}"
        )
        summaries.append(f"## 文档块 {i+1} 摘要\n{summary}")
    
    # 第二层：综合所有摘要
    combined = "\n\n".join(summaries)
    final = call_claude(
        f"基于以下多个文档块的摘要，请生成一份综合分析报告，"
        f"识别共同主题、矛盾点和关键洞察：\n\n{combined}",
        max_tokens=8192
    )
    return final

策略二：智能上下文裁剪

import tiktoken

class ContextManager:
    """智能上下文管理器"""
    
    def __init__(self, max_context_tokens: int = 180000,
                 reserve_output: int = 4096):
        self.max_input_tokens = max_context_tokens - reserve_output
        self.message_history: list[dict] = []
    
    def add_message(self, role: str, content: str):
        self.message_history.append({"role": role, "content": content})
        self._trim_if_needed()
    
    def _trim_if_needed(self):
        """保留system和最近对话，裁剪中间历史"""
        total = sum(self._count_tokens(m["content"]) 
                   for m in self.message_history)
        
        while total > self.max_input_tokens and len(self.message_history) > 3:
            # 保留第一条（通常是system）和最近两条
            removed = self.message_history.pop(1)
            total -= self._count_tokens(removed["content"])
    
    def _count_tokens(self, text: str) -> int:
        """近似token计数（中文约1.5字/token）"""
        return len(text) // 2  # 粗略估计
    
    def get_messages(self) -> list[dict]:
        return self.message_history.copy()

策略三：Prompt Caching（提示缓存）

def cached_long_context_call(document: str, question: str) -> str:
    """
    利用Prompt Caching减少重复长文档的处理成本
    缓存的前缀在后续请求中直接复用，节省约90%输入token费用
    """
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        system=[
            {
                "type": "text",
                "text": f"以下是一份需要分析的文档：\n\n{document}",
                "cache_control": {"type": "ephemeral"}  # 标记为可缓存
            }
        ],
        messages=[{
            "role": "user",
            "content": question
        }]
    )
    return response.content[0].text


# 首次调用建立缓存（费用较高）
answer1 = cached_long_context_call(long_doc, "文档的主旨是什么？")

# 后续调用复用缓存（费用大幅降低）
answer2 = cached_long_context_call(long_doc, "有哪些关键数据？")
answer3 = cached_long_context_call(long_doc, "结论是什么？")

3.4 上下文使用效率对照表

4. Tool Use工具调用进阶

4.1 基础Tool Use

Tool Use允许Claude调用外部工具来获取信息或执行操作。定义工具schema后，Claude会自主决定何时调用、传递什么参数。

import anthropic
import json

client = anthropic.Anthropic()

# 定义工具
tools = [
    {
        "name": "get_weather",
        "description": "获取指定城市的当前天气信息",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "城市名称，如'北京'、'上海'"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "温度单位"
                }
            },
            "required": ["city"]
        }
    },
    {
        "name": "search_products",
        "description": "搜索商品数据库",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "搜索关键词"},
                "category": {"type": "string", "description": "商品类别"},
                "max_price": {"type": "number", "description": "最高价格"},
                "limit": {"type": "integer", "description": "返回数量", "default": 5}
            },
            "required": ["query"]
        }
    }
]

# 工具执行器
def execute_tool(name: str, params: dict) -> str:
    """根据工具名执行对应逻辑"""
    if name == "get_weather":
        # 实际项目中调用天气API
        return json.dumps({
            "city": params["city"],
            "temperature": 22,
            "condition": "晴",
            "humidity": 45
        }, ensure_ascii=False)
    elif name == "search_products":
        # 实际项目中查询数据库
        return json.dumps({
            "results": [
                {"name": "商品A", "price": 99.9},
                {"name": "商品B", "price": 149.0}
            ]
        }, ensure_ascii=False)
    return json.dumps({"error": f"未知工具: {name}"})

4.2 多轮工具调用循环

def chat_with_tools(user_message: str, max_rounds: int = 5) -> str:
    """
    处理可能涉及多轮工具调用的对话
    Claude可能在一次回复中请求多个工具，或链式调用多个工具
    """
    messages = [{"role": "user", "content": user_message}]
    
    for round_num in range(max_rounds):
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            tools=tools,
            messages=messages
        )
        
        # 检查是否需要调用工具
        if response.stop_reason == "tool_use":
            # 将assistant的回复（包含tool_use块）加入消息历史
            messages.append({"role": "assistant", "content": response.content})
            
            # 执行所有请求的工具调用
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    print(f"  🔧 调用工具: {block.name}({block.input})")
                    result = execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result
                    })
            
            # 将工具结果返回给Claude
            messages.append({"role": "user", "content": tool_results})
        
        elif response.stop_reason == "end_turn":
            # Claude给出了最终回答
            return next(
                (b.text for b in response.content if b.type == "text"), ""
            )
    
    return "达到最大工具调用轮数"

4.3 并行工具调用

Claude可以在一次响应中请求多个独立的工具调用，实现并行执行：

import concurrent.futures

def parallel_chat_with_tools(user_message: str) -> str:
    """支持并行工具调用的对话处理"""
    messages = [{"role": "user", "content": user_message}]
    
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        tools=tools,
        messages=messages
    )
    
    if response.stop_reason == "tool_use":
        messages.append({"role": "assistant", "content": response.content})
        
        # 收集所有工具调用
        tool_calls = [
            (block.id, block.name, block.input)
            for block in response.content
            if block.type == "tool_use"
        ]
        
        # 并行执行工具调用
        tool_results = []
        with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
            futures = {
                executor.submit(execute_tool, name, params): (tool_id, name)
                for tool_id, name, params in tool_calls
            }
            for future in concurrent.futures.as_completed(futures):
                tool_id, name = futures[future]
                result = future.result()
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": tool_id,
                    "content": result
                })
        
        messages.append({"role": "user", "content": tool_results})
        
        # 获取最终回答
        final = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            tools=tools,
            messages=messages
        )
        return next(
            (b.text for b in final.content if b.type == "text"), ""
        )
    
    return next(
        (b.text for b in response.content if b.type == "text"), ""
    )

4.4 强制工具调用

通过 tool_choice 参数控制Claude的工具调用行为：

# 强制调用特定工具（无论是否有必要）
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=4096,
    tools=tools,
    tool_choice={"type": "tool", "name": "get_weather"},  # 强制调用
    messages=[{"role": "user", "content": "你好"}]
)

# 禁止工具调用
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=4096,
    tools=tools,
    tool_choice={"type": "none"},  # 禁止调用工具
    messages=[{"role": "user", "content": "直接告诉我天气怎么样"}]
)

5. 多模态视觉理解

5.1 图片分析基础

Claude支持直接理解图片内容，包括照片、截图、图表、文档扫描件等。

import base64
from pathlib import Path

def analyze_image(image_path: str, question: str) -> str:
    """分析本地图片"""
    # 读取并编码图片
    image_data = Path(image_path).read_bytes()
    base64_image = base64.b64encode(image_data).decode("utf-8")
    
    # 判断MIME类型
    suffix = Path(image_path).suffix.lower()
    mime_map = {".jpg": "image/jpeg", ".jpeg": "image/jpeg", 
                ".png": "image/png", ".gif": "image/gif", 
                ".webp": "image/webp"}
    media_type = mime_map.get(suffix, "image/png")
    
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": media_type,
                        "data": base64_image
                    }
                },
                {
                    "type": "text",
                    "text": question
                }
            ]
        }]
    )
    return response.content[0].text


# 分析URL图片
def analyze_url_image(image_url: str, question: str) -> str:
    """分析网络图片"""
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "url",
                        "url": image_url
                    }
                },
                {"type": "text", "text": question}
            ]
        }]
    )
    return response.content[0].text

5.2 多图对比分析

def compare_images(image_paths: list[str], question: str) -> str:
    """对比分析多张图片"""
    content = []
    
    for i, path in enumerate(image_paths):
        image_data = Path(path).read_bytes()
        base64_image = base64.b64encode(image_data).decode("utf-8")
        
        content.append({
            "type": "image",
            "source": {
                "type": "base64",
                "media_type": "image/png",
                "data": base64_image
            }
        })
        content.append({
            "type": "text",
            "text": f"图片{i+1}："
        })
    
    content.append({"type": "text", "text": question})
    
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        messages=[{"role": "user", "content": content}]
    )
    return response.content[0].text

5.3 文档OCR与结构化提取

def extract_document_data(image_path: str) -> dict:
    """
    从文档图片中提取结构化数据
    适用于发票、合同、表格等场景
    """
    image_data = Path(image_path).read_bytes()
    base64_image = base64.b64encode(image_data).decode("utf-8")
    
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": "image/png",
                        "data": base64_image
                    }
                },
                {
                    "type": "text",
                    "text": (
                        "请从这份文档中提取所有信息，以JSON格式返回。"
                        "包含以下字段：\n"
                        "- document_type: 文档类型\n"
                        "- date: 日期\n"
                        "- parties: 相关方列表\n"
                        "- amounts: 金额列表\n"
                        "- key_terms: 关键条款摘要\n"
                        "- raw_text: 原始文本内容\n\n"
                        "仅返回JSON，不要其他说明文字。"
                    )
                }
            ]
        }]
    )
    
    import json
    # 提取JSON（可能被markdown代码块包裹）
    text = response.content[0].text
    if "```json" in text:
        text = text.split("```json")[1].split("```")[0]
    elif "```" in text:
        text = text.split("```")[1].split("```")[0]
    
    return json.loads(text.strip())

5.4 多模态最佳实践

• 图片分辨率：建议不超过1568×1568像素，过大的图片会自动缩放
• 图片数量：单次请求支持多张图片，但总token消耗会增加
• 格式选择：优先使用WebP（体积小），其次PNG/JPEG
• 成本控制：图片token按像素面积计算，适当裁剪无关区域

6. System Prompt设计

6.1 System Prompt的作用

System Prompt是定义Claude行为边界和风格的核心机制。好的System Prompt能显著提升输出质量和一致性。

6.2 分层设计模式

def create_system_prompt(
    role: str,
    context: str,
    constraints: list[str],
    output_format: str,
    examples: list[dict] = None
) -> str:
    """
    构建结构化的System Prompt
    采用分层设计：角色 → 上下文 → 约束 → 格式 → 示例
    """
    prompt_parts = []
    
    # 第一层：角色定义
    prompt_parts.append(f"# 角色\n你是{role}。")
    
    # 第二层：业务上下文
    if context:
        prompt_parts.append(f"# 背景信息\n{context}")
    
    # 第三层：行为约束
    if constraints:
        constraints_text = "\n".join(f"- {c}" for c in constraints)
        prompt_parts.append(f"# 行为约束\n{constraints_text}")
    
    # 第四层：输出格式
    if output_format:
        prompt_parts.append(f"# 输出格式\n{output_format}")
    
    # 第五层：示例（Few-shot）
    if examples:
        examples_text = ""
        for i, ex in enumerate(examples, 1):
            examples_text += f"\n## 示例{i}\n"
            examples_text += f"输入：{ex['input']}\n"
            examples_text += f"输出：{ex['output']}\n"
        prompt_parts.append(f"# 参考示例{examples_text}")
    
    return "\n\n".join(prompt_parts)


# 实际使用
system_prompt = create_system_prompt(
    role="一位资深的Python代码审查专家，拥有10年大型项目经验",
    context="你正在为一个金融科技公司的核心交易系统进行代码审查。"
            "该系统每天处理数百万笔交易，对准确性和性能要求极高。",
    constraints=[
        "仅关注代码质量、安全性和性能问题，不评论代码风格偏好",
        "对每个问题给出严重等级：Critical / Warning / Suggestion",
        "提供具体的修复代码，而非泛泛建议",
        "如果代码没有明显问题，直接说明'代码质量良好，无需修改'",
        "使用中文回复，代码注释使用英文"
    ],
    output_format=(
        "## 审查报告\n"
        "### 概要\n"
        "（一句话总结代码质量）\n\n"
        "### 发现的问题\n"
        "每个问题按以下格式：\n"
        "**[等级] 问题标题**\n"
        "- 位置：文件名:行号\n"
        "- 问题描述\n"
        "- 修复建议（含代码）\n\n"
        "### 总结\n"
        "（整体评价和优先修复建议）"
    ),
    examples=[
        {
            "input": "def get_user(id): return db.query(f'SELECT * FROM users WHERE id={id}')",
            "output": "## 审查报告\n### 概要\n发现1个Critical级别的SQL注入漏洞。\n\n"
                      "### 发现的问题\n**[Critical] SQL注入漏洞**\n"
                      "- 位置：user.py:1\n- 使用字符串拼接构建SQL语句..."
        }
    ]
)

6.3 动态System Prompt

def get_dynamic_system(user_role: str, conversation_stage: str) -> str:
    """根据用户角色和对话阶段动态调整System Prompt"""
    
    base = "你是智能客服助手，负责解答用户问题。"
    
    role_configs = {
        "vip": "当前用户是VIP客户，优先处理其需求，可以提供专属优惠。",
        "new": "当前用户是新用户，需要更详细的引导和耐心的解释。",
        "enterprise": "当前用户是企业客户，提供专业的技术方案和商务报价。"
    }
    
    stage_configs = {
        "greeting": "当前处于问候阶段，简短友好地自我介绍。",
        "inquiry": "当前处于需求了解阶段，多提问以明确用户需求。",
        "solution": "当前处于方案推荐阶段，提供具体的产品或解决方案。",
        "closing": "当前处于收尾阶段，确认用户满意度，提供后续支持信息。"
    }
    
    parts = [base]
    if user_role in role_configs:
        parts.append(role_configs[user_role])
    if conversation_stage in stage_configs:
        parts.append(stage_configs[conversation_stage])
    
    return "\n".join(parts)

6.4 System Prompt设计原则

1. 明确优先：告诉Claude"做什么"比"不做什么"更有效
2. 约束具体：用可量化的标准代替模糊描述（"回答不超过200字"优于"简短回答"）
3. 示例驱动：复杂的输出格式用Few-shot示例比纯文字描述更可靠
4. 分层隔离：角色、上下文、约束、格式分开管理，便于维护
5. 版本控制：System Prompt应该像代码一样版本管理，每次修改记录变更原因

7. 流式输出实现

7.1 基础流式输出

流式输出可以显著改善用户体验，减少首次响应的等待时间。

import anthropic

client = anthropic.Anthropic()

def stream_chat(prompt: str, system: str = None):
    """基础流式输出"""
    kwargs = {
        "model": "claude-sonnet-4-20250514",
        "max_tokens": 4096,
        "messages": [{"role": "user", "content": prompt}]
    }
    if system:
        kwargs["system"] = system
    
    with client.messages.stream(**kwargs) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
    print()  # 换行


# 使用事件流进行更精细的控制
def stream_with_events(prompt: str):
    """使用事件流获取更详细的状态信息"""
    with client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        messages=[{"role": "user", "content": prompt}]
    ) as stream:
        for event in stream:
            if event.type == "message_start":
                print(f"[开始] 模型: {event.message.model}")
            elif event.type == "content_block_start":
                print(f"[内容块] 类型: {event.content_block.type}")
            elif event.type == "content_block_delta":
                if event.delta.type == "text_delta":
                    print(event.delta.text, end="", flush=True)
            elif event.type == "message_delta":
                print(f"\n[结束] 停止原因: {event.delta.stop_reason}")
                usage = event.usage
                print(f"[用量] 输出tokens: {usage.output_tokens}")

7.2 WebSocket服务端流式转发

将Claude的流式输出转发给前端WebSocket客户端：

import asyncio
import websockets
import json
import anthropic

client = anthropic.Anthropic()

async def handle_websocket(websocket):
    """处理WebSocket连接，转发Claude流式输出"""
    async for raw_message in websocket:
        data = json.loads(raw_message)
        user_message = data.get("message", "")
        history = data.get("history", [])
        
        messages = history + [{"role": "user", "content": user_message}]
        
        # 在线程池中运行同步的Anthropic SDK
        loop = asyncio.get_event_loop()
        
        def generate_stream():
            chunks = []
            with client.messages.stream(
                model="claude-sonnet-4-20250514",
                max_tokens=4096,
                messages=messages
            ) as stream:
                for text in stream.text_stream:
                    chunks.append(text)
            return chunks
        
        # 流式发送给客户端
        try:
            with client.messages.stream(
                model="claude-sonnet-4-20250514",
                max_tokens=4096,
                messages=messages
            ) as stream:
                for text in stream.text_stream:
                    await websocket.send(json.dumps({
                        "type": "chunk",
                        "content": text
                    }))
                
                await websocket.send(json.dumps({
                    "type": "done"
                }))
        except Exception as e:
            await websocket.send(json.dumps({
                "type": "error",
                "message": str(e)
            }))


async def main():
    async with websockets.serve(handle_websocket, "localhost", 8765):
        await asyncio.Future()  # 永远运行

# asyncio.run(main())

7.3 Server-Sent Events (SSE) 实现

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import anthropic
import json

app = FastAPI()
client = anthropic.Anthropic()

@app.post("/chat/stream")
async def chat_stream(request: dict):
    """SSE流式聊天接口"""
    
    async def event_generator():
        with client.messages.stream(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            messages=[{"role": "user", "content": request["message"]}]
        ) as stream:
            for text in stream.text_stream:
                yield f"data: {json.dumps({'text': text})}\n\n"
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        }
    )

8. 批处理API

8.1 Message Batches API

对于不需要实时响应的大规模任务，批处理API可以将成本降低50%。

import anthropic
import json
import time

client = anthropic.Anthropic()

def create_batch_job(requests: list[dict]) -> str:
    """
    创建批处理任务
    
    Args:
        requests: 请求列表，每个元素包含 custom_id 和 params
    
    Returns:
        batch_id
    """
    batch_requests = []
    for req in requests:
        batch_requests.append({
            "custom_id": req["id"],
            "params": {
                "model": "claude-sonnet-4-20250514",
                "max_tokens": 2048,
                "messages": [
                    {"role": "user", "content": req["prompt"]}
                ]
            }
        })
    
    batch = client.messages.batches.create(requests=batch_requests)
    print(f"批处理任务已创建: {batch.id}")
    print(f"请求总数: {len(batch_requests)}")
    return batch.id


def wait_for_batch(batch_id: str, poll_interval: int = 10) -> dict:
    """等待批处理完成并返回结果"""
    while True:
        batch = client.messages.batches.retrieve(batch_id)
        print(f"状态: {batch.processing_status} | "
              f"已完成: {batch.request_counts.succeeded}/{batch.request_counts.processing + batch.request_counts.succeeded}")
        
        if batch.processing_status == "ended":
            results = {}
            for result in client.messages.batches.results(batch_id):
                if result.result.type == "succeeded":
                    results[result.custom_id] = \
                        result.result.message.content[0].text
                else:
                    results[result.custom_id] = f"错误: {result.result.type}"
            return results
        
        time.sleep(poll_interval)


# 使用示例
requests = [
    {"id": f"task_{i}", "prompt": f"用一句话解释什么是{topic}"}
    for i, topic in enumerate(["量子计算", "区块链", "机器学习", "容器化", "微服务"])
]

batch_id = create_batch_job(requests)
results = wait_for_batch(batch_id)

for task_id, answer in results.items():
    print(f"{task_id}: {answer}")

8.2 批处理最佳实践

9. 与OpenAI API对比

9.1 API设计差异

9.2 迁移指南：从OpenAI到Claude

# ===== OpenAI 风格 =====
from openai import OpenAI
client_openai = OpenAI()

response = client_openai.chat.completions.create(
    model="gpt-4o",
    max_tokens=4096,
    temperature=0.7,
    messages=[
        {"role": "system", "content": "你是一个助手"},
        {"role": "user", "content": "你好"}
    ]
)
answer = response.choices[0].message.content


# ===== Claude 等价写法 =====
from anthropic import Anthropic
client_claude = Anthropic()

response = client_claude.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=4096,
    temperature=0.7,
    system="你是一个助手",       # system是独立参数
    messages=[
        {"role": "user", "content": "你好"}
    ]
)
answer = response.content[0].text   # content是block数组

9.3 统一抽象层

from typing import Literal
import anthropic
from openai import OpenAI

class UnifiedLLM:
    """统一的LLM调用抽象层，支持Claude和OpenAI"""
    
    def __init__(self, provider: Literal["claude", "openai"] = "claude"):
        self.provider = provider
        if provider == "claude":
            self.client = anthropic.Anthropic()
        else:
            self.client = OpenAI()
    
    def chat(self, prompt: str, system: str = None, 
             model: str = None, max_tokens: int = 4096,
             temperature: float = 0.7) -> str:
        """统一的聊天接口"""
        
        if self.provider == "claude":
            model = model or "claude-sonnet-4-20250514"
            kwargs = {
                "model": model,
                "max_tokens": max_tokens,
                "temperature": temperature,
                "messages": [{"role": "user", "content": prompt}]
            }
            if system:
                kwargs["system"] = system
            response = self.client.messages.create(**kwargs)
            return response.content[0].text
        
        else:
            model = model or "gpt-4o"
            messages = []
            if system:
                messages.append({"role": "system", "content": system})
            messages.append({"role": "user", "content": prompt})
            
            response = self.client.chat.completions.create(
                model=model,
                max_tokens=max_tokens,
                temperature=temperature,
                messages=messages
            )
            return response.choices[0].message.content


# 使用
llm = UnifiedLLM("claude")
answer = llm.chat("解释递归", system="用简单的语言回答")

10. 企业级集成最佳实践

10.1 错误处理与重试

import anthropic
import time
import logging
from functools import wraps

logger = logging.getLogger(__name__)

def with_retry(max_retries: int = 3, base_delay: float = 1.0):
    """带指数退避的重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except anthropic.RateLimitError as e:
                    wait_time = base_delay * (2 ** attempt)
                    logger.warning(f"速率限制，等待{wait_time}秒后重试 "
                                  f"(第{attempt+1}次)")
                    time.sleep(wait_time)
                    last_exception = e
                except anthropic.APIError as e:
                    if e.status_code >= 500:  # 服务端错误可重试
                        wait_time = base_delay * (2 ** attempt)
                        logger.warning(f"服务端错误 {e.status_code}，"
                                      f"等待{wait_time}秒后重试")
                        time.sleep(wait_time)
                        last_exception = e
                    else:
                        raise  # 客户端错误不重试
            raise last_exception
        return wrapper
    return decorator


@with_retry(max_retries=3)
def reliable_call(prompt: str) -> str:
    """带重试的可靠API调用"""
    client = anthropic.Anthropic()
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text

10.2 请求限流与队列

import asyncio
from collections import deque
from dataclasses import dataclass, field
from datetime import datetime, timedelta

@dataclass
class RateLimiter:
    """令牌桶限流器"""
    max_requests_per_minute: int = 60
    max_tokens_per_minute: int = 100_000
    _request_times: deque = field(default_factory=deque)
    _token_usage: deque = field(default_factory=deque)
    
    async def acquire(self, estimated_tokens: int = 1000):
        """等待直到可以发送请求"""
        now = datetime.now()
        cutoff = now - timedelta(minutes=1)
        
        # 清理过期记录
        while self._request_times and self._request_times[0] < cutoff:
            self._request_times.popleft()
        while self._token_usage and self._token_usage[0][0] < cutoff:
            self._token_usage.popleft()
        
        # 检查请求数限制
        if len(self._request_times) >= self.max_requests_per_minute:
            wait_until = self._request_times[0] + timedelta(minutes=1)
            wait_seconds = (wait_until - now).total_seconds()
            if wait_seconds > 0:
                await asyncio.sleep(wait_seconds)
        
        # 检查token限制
        current_tokens = sum(t for _, t in self._token_usage)
        if current_tokens + estimated_tokens > self.max_tokens_per_minute:
            wait_until = self._token_usage[0][0] + timedelta(minutes=1)
            wait_seconds = (wait_until - now).total_seconds()
            if wait_seconds > 0:
                await asyncio.sleep(wait_seconds)
        
        self._request_times.append(datetime.now())
        self._token_usage.append((datetime.now(), estimated_tokens))


# 使用示例
limiter = RateLimiter(max_requests_per_minute=50)

async def rate_limited_call(prompt: str) -> str:
    await limiter.acquire(estimated_tokens=len(prompt) * 2)
    return reliable_call(prompt)

10.3 成本监控与预算控制

from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass 
class CostTracker:
    """API成本追踪器"""
    daily_budget: float = 100.0  # 每日预算（美元）
    _daily_cost: float = 0.0
    _daily_date: str = ""
    _log_file: str = "api_costs.jsonl"
    
    def record_usage(self, input_tokens: int, output_tokens: int,
                     model: str):
        """记录一次API调用的用量和成本"""
        today = datetime.now().strftime("%Y-%m-%d")
        
        # 重置每日计数
        if today != self._daily_date:
            self._daily_cost = 0.0
            self._daily_date = today
        
        # 计算成本
        cost = estimate_cost(input_tokens, output_tokens, model)
        self._daily_cost += cost
        
        # 写入日志
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cost_usd": round(cost, 6),
            "daily_total": round(self._daily_cost, 4)
        }
        with open(self._log_file, "a") as f:
            f.write(json.dumps(log_entry) + "\n")
        
        # 预算检查
        if self._daily_cost > self.daily_budget:
            raise BudgetExceededError(
                f"日预算已超限: ${self._daily_cost:.2f} > ${self.daily_budget:.2f}"
            )
        
        return cost
    
    def get_daily_summary(self) -> dict:
        """获取当日成本汇总"""
        return {
            "date": self._daily_date,
            "total_cost": round(self._daily_cost, 4),
            "budget": self.daily_budget,
            "remaining": round(self.daily_budget - self._daily_cost, 4),
            "usage_pct": round(self._daily_cost / self.daily_budget * 100, 1)
        }


class BudgetExceededError(Exception):
    pass

10.4 安全最佳实践

import os
import hashlib
import re

class SecurityMiddleware:
    """API安全中间件"""
    
    @staticmethod
    def sanitize_input(user_input: str, max_length: int = 50000) -> str:
        """清理用户输入"""
        # 长度限制
        if len(user_input) > max_length:
            raise ValueError(f"输入超过最大长度限制 ({max_length}字符)")
        
        # 移除潜在的prompt injection标记
        dangerous_patterns = [
            r"ignore\s+(all\s+)?previous\s+instructions",
            r"system:\s*you\s+are",
            r"<\|im_start\|>system",
        ]
        for pattern in dangerous_patterns:
            if re.search(pattern, user_input, re.IGNORECASE):
                raise ValueError("检测到潜在的提示注入攻击")
        
        return user_input
    
    @staticmethod
    def mask_sensitive_data(text: str) -> str:
        """脱敏处理：遮盖敏感信息"""
        # 手机号
        text = re.sub(r'1[3-9]\d{9}', lambda m: m.group()[:3] + '****' + m.group()[-4:], text)
        # 身份证号
        text = re.sub(r'\d{17}[\dXx]', lambda m: m.group()[:6] + '********' + m.group()[-4:], text)
        # 邮箱
        text = re.sub(r'[\w.]+@[\w.]+\.\w+', 
                      lambda m: m.group().split('@')[0][:2] + '***@' + m.group().split('@')[1], text)
        return text
    
    @staticmethod
    def get_api_key() -> str:
        """安全获取API Key"""
        key = os.environ.get("ANTHROPIC_API_KEY")
        if not key:
            raise EnvironmentError("未设置 ANTHROPIC_API_KEY 环境变量")
        if not key.startswith("sk-ant-"):
            raise ValueError("API Key格式不正确")
        return key

10.5 完整的企业级封装

import anthropic
import logging
from contextlib import contextmanager

logger = logging.getLogger(__name__)

class ClaudeService:
    """企业级Claude API服务封装"""
    
    def __init__(self, 
                 daily_budget: float = 100.0,
                 max_retries: int = 3):
        self.client = anthropic.Anthropic()
        self.rate_limiter = RateLimiter()
        self.cost_tracker = CostTracker(daily_budget=daily_budget)
        self.security = SecurityMiddleware()
        self.max_retries = max_retries
    
    def chat(self, 
             prompt: str,
             system: str = None,
             model: str = "claude-sonnet-4-20250514",
             max_tokens: int = 4096,
             temperature: float = 0.7,
             tools: list = None,
             enable_thinking: bool = False,
             thinking_budget: int = 10000) -> dict:
        """
        统一的聊天接口，集成所有企业级功能
        
        Returns:
            {
                "content": str,       # 回答内容
                "thinking": str,      # 思考过程（如果开启）
                "usage": dict,        # token用量
                "cost": float,        # 本次成本
                "model": str,         # 实际使用的模型
            }
        """
        # 1. 安全检查
        prompt = self.security.sanitize_input(prompt)
        
        # 2. 构建请求参数
        kwargs = {
            "model": model,
            "max_tokens": max_tokens,
            "temperature": temperature,
            "messages": [{"role": "user", "content": prompt}]
        }
        if system:
            kwargs["system"] = self.security.sanitize_input(system)
        if tools:
            kwargs["tools"] = tools
        if enable_thinking:
            kwargs["thinking"] = {
                "type": "enabled",
                "budget_tokens": thinking_budget
            }
            kwargs["temperature"] = 1.0  # Extended Thinking要求temperature=1
        
        # 3. 限流等待
        # asyncio.get_event_loop().run_until_complete(
        #     self.rate_limiter.acquire(len(prompt))
        # )
        
        # 4. 带重试的API调用
        for attempt in range(self.max_retries):
            try:
                response = self.client.messages.create(**kwargs)
                break
            except anthropic.RateLimitError:
                import time
                wait = 2 ** attempt
                logger.warning(f"限流，等待{wait}秒")
                time.sleep(wait)
            except anthropic.APIError as e:
                if e.status_code >= 500 and attempt < self.max_retries - 1:
                    import time
                    time.sleep(2 ** attempt)
                    continue
                raise
        
        # 5. 解析响应
        result = {
            "content": "",
            "thinking": "",
            "usage": {
                "input_tokens": response.usage.input_tokens,
                "output_tokens": response.usage.output_tokens
            },
            "model": response.model,
            "stop_reason": response.stop_reason
        }
        
        for block in response.content:
            if block.type == "text":
                result["content"] = block.text
            elif block.type == "thinking":
                result["thinking"] = block.thinking
        
        # 6. 记录成本
        cost = self.cost_tracker.record_usage(
            response.usage.input_tokens,
            response.usage.output_tokens,
            model
        )
        result["cost"] = round(cost, 6)
        
        logger.info(f"API调用完成 | 模型: {model} | "
                    f"输入: {response.usage.input_tokens} | "
                    f"输出: {response.usage.output_tokens} | "
                    f"成本: ${cost:.4f}")
        
        return result


# 使用示例
service = ClaudeService(daily_budget=50.0)

result = service.chat(
    prompt="请分析这段代码的潜在问题",
    system="你是代码审查专家",
    enable_thinking=True,
    thinking_budget=8000
)

print(f"回答: {result['content'][:200]}...")
print(f"成本: ${result['cost']}")

总结

本教程覆盖了Claude API从基础到企业级的完整开发栈。关键要点：

1. 模型选择：Sonnet是通用首选，Opus处理复杂推理，Haiku负责高吞吐场景
2. Extended Thinking：在数学、编程、逻辑推理场景中显著提升质量
3. 上下文优化：Prompt Caching可节省80-95%的重复文档处理成本
4. Tool Use：支持并行调用和多轮循环，是构建Agent的基础
5. 多模态：直接分析图片、文档，支持OCR和结构化提取
6. 流式输出：SSE和WebSocket两种模式，覆盖Web和实时场景
7. 批处理：非实时任务可节省50%成本
8. 企业级：重试、限流、成本监控、安全防护缺一不可

掌握这些技巧，你就能构建出高质量、高可靠性的Claude API应用。