用Claude API构建简单AI应用 零基础上手案例

一、先量化：你的响应时间正常吗？

在说”Claude 慢”之前，先看看参考基准。Claude API 在正常状态下的响应时间大概是这样的：

TTFT = Time To First Token，即开始生成第一个字的等待时间。以上为正常网络、非高峰期的参考值，实际会有波动。

如果你的响应时间显著超过上表，说明确实有问题需要排查。如果在范围内，感觉慢可能是”体验问题”而不是”性能问题”——这两种有不同的解法。

延迟测试代码

import anthropic
import time

client = anthropic.Anthropic()

def measure_latency(
    model:   str,
    prompt:  str = "用三句话介绍人工智能",
    runs:    int = 3,
) -> dict:
    """测量指定模型的响应延迟"""
    ttfts      = []
    totals     = []
    token_rates= []

    for i in range(runs):
        start     = time.time()
        first_tok = None

        with client.messages.stream(
            model      = model,
            max_tokens = 256,
            messages   = [{"role": "user", "content": prompt}],
        ) as stream:
            for text in stream.text_stream:
                if first_tok is None:
                    first_tok = time.time()
                    ttfts.append(first_tok - start)

            final = stream.get_final_message()
            total = time.time() - start
            totals.append(total)

            out_tok = final.usage.output_tokens
            if total > 0:
                token_rates.append(out_tok / total)

        time.sleep(1)   # 避免触发速率限制

    return {
        "model":          model,
        "avg_ttft_s":     round(sum(ttfts) / len(ttfts), 2),
        "avg_total_s":    round(sum(totals) / len(totals), 2),
        "avg_tokens_per_s": round(sum(token_rates) / len(token_rates), 1),
    }


# 对比三个模型的速度
for model in ["claude-haiku-4-5-20251001", "claude-sonnet-4-6"]:
    result = measure_latency(model)
    print(f"{result['model']}")
    print(f"  首字延迟：{result['avg_ttft_s']}s")
    print(f"  完整响应：{result['avg_total_s']}s")
    print(f"  生成速度：{result['avg_tokens_per_s']} tokens/s\n")

二、原因一：用了比任务需要更重的模型

这是最常见的”慢”——用 Opus 或 Sonnet 做完全可以用 Haiku 完成的任务。Opus 的响应速度大约是 Haiku 的 3-5 倍慢，但对于分类、提取、简单问答这类任务，Haiku 的准确率已经足够。

判断方法：把你的任务用 Haiku 跑一遍，看结果质量是否可接受。

"""
速度 vs 质量参考：

Haiku（最快）适合的任务：
  ✅ 文本分类、情感分析
  ✅ 关键词提取
  ✅ 简单问答（答案在提供的文本里）
  ✅ 格式转换（JSON/表格）
  ✅ 翻译（通用文本）

Sonnet（平衡）适合的任务：
  ✅ 代码生成和调试
  ✅ 内容创作（文章、邮件）
  ✅ 复杂的多轮对话
  ✅ 需要推理的分析

Opus（最慢）适合的任务：
  ✅ 架构设计和深度分析
  ✅ 需要长链条推理的任务
  ✅ 对质量要求极高且低频的场景
"""

# 速度优化：智能模型路由
def route_by_complexity(prompt: str) -> str:
    """根据任务复杂度选择模型，平衡速度和质量"""
    prompt_len = len(prompt)

    # 短提示 + 常见简单任务关键词 → Haiku
    simple_keywords = [
        "分类", "判断", "提取", "翻译", "是否",
        "classify", "extract", "translate", "yes or no"
    ]
    if prompt_len < 200 and any(kw in prompt.lower() for kw in simple_keywords):
        return "claude-haiku-4-5-20251001"

    # 默认用 Sonnet
    return "claude-sonnet-4-6"

三、原因二：输入内容太长

输入 token 数量对首字延迟影响很大——Claude 需要先"读完"整个输入，才能开始生成输出。输入越长，开始生成前的等待越长。

判断方法：用 count_tokens 查一下你的实际输入大小。

def diagnose_input_size(messages: list[dict], system: str = "") -> dict:
    """诊断输入大小对速度的影响"""
    count = client.messages.count_tokens(
        model    = "claude-sonnet-4-6",
        system   = system,
        messages = messages,
    )
    tok = count.input_tokens

    if tok < 1000:
        assessment = "✅ 输入较短，不是慢的原因"
    elif tok < 10000:
        assessment = "⚠️ 输入中等，首字延迟约 1-3s"
    elif tok < 50000:
        assessment = "⚠️ 输入较长，首字延迟约 3-10s"
    else:
        assessment = "❌ 输入很长（10万+ tokens），首字延迟可能超过 30s"

    return {"input_tokens": tok, "assessment": assessment}


# 三个减少输入长度的技巧：

# 技巧 A：多轮对话用滑动窗口，不保留所有历史
def trim_history(history: list[dict], max_turns: int = 8) -> list[dict]:
    return history[-(max_turns * 2):]    # 每轮 2 条消息

# 技巧 B：System Prompt 能短就短
VERBOSE = "你是一个非常专业的、经验丰富的、友善的助手，请帮助用户解决各种问题……"  # 约80 tokens
CONCISE = "你是专业助手。简洁、准确。"  # 约10 tokens

# 技巧 C：长文档任务用 RAG，不要整份文档塞进去
# 只提取和问题最相关的段落（通常200-500 tokens），不是整份文档（可能几万 tokens）

四、原因三：服务端负载（非你能控制，但能绕开）

Anthropic 服务器在高峰期（通常是北京时间晚上9点-凌晨1点，对应美国工作时间）响应会变慢。这不是你的代码问题，但有几个应对方法：

"""
服务端负载高的信号：
- 响应时间比平时长 2 倍以上
- 偶尔出现 529 Overloaded 错误
- status.anthropic.com 有黄色/红色状态

应对策略：
"""

import asyncio

# 策略 A：高峰期自动降级到更快的模型
async def adaptive_model_request(
    messages:       list[dict],
    preferred_model:str   = "claude-sonnet-4-6",
    fallback_model: str   = "claude-haiku-4-5-20251001",
    ttft_threshold: float = 3.0,    # 首字超过 3s 就认为服务慢
) -> str:
    """
    首先尝试首选模型，如果首字延迟过长，
    取消并改用更快的模型
    """
    async_client = anthropic.AsyncAnthropic()
    start        = asyncio.get_event_loop().time()

    try:
        # 设置超时：首字必须在 ttft_threshold 秒内到达
        async with async_client.messages.stream(
            model      = preferred_model,
            max_tokens = 1024,
            messages   = messages,
        ) as stream:
            full_text = []
            async for text in stream.text_stream:
                if not full_text:    # 第一个 token
                    ttft = asyncio.get_event_loop().time() - start
                    if ttft > ttft_threshold:
                        print(f"首字延迟 {ttft:.1f}s，超过阈值，但已开始流式，继续当前请求")
                full_text.append(text)
            return "".join(full_text)

    except Exception as e:
        # 如果出现 529 或连接超时，用 Haiku 重试
        print(f"首选模型失败（{e}），降级到 {fallback_model}")
        resp = await async_client.messages.create(
            model      = fallback_model,
            max_tokens = 1024,
            messages   = messages,
        )
        return resp.content[0].text


# 策略 B：离线任务避开高峰期
# 把批量处理任务安排在低峰期（凌晨或早上）自动执行
import schedule

def run_batch_job():
    print("执行批量任务...")
    # 你的批量处理逻辑

# 每天凌晨 3 点执行（美国西海岸下午，负载低）
schedule.every().day.at("03:00").do(run_batch_job)

五、原因四：代码和架构问题

有时候慢不是 Claude 的问题，而是你的代码串行做了本可以并行的事：

import asyncio
import anthropic

async_client = anthropic.AsyncAnthropic()

# ❌ 慢：串行处理多个独立请求
async def slow_multi_request(prompts: list[str]) -> list[str]:
    results = []
    for prompt in prompts:          # 一个一个等，总时间 = N × 单次时间
        resp = await async_client.messages.create(
            model      = "claude-haiku-4-5-20251001",
            max_tokens = 256,
            messages   = [{"role": "user", "content": prompt}],
        )
        results.append(resp.content[0].text)
    return results


# ✅ 快：并发处理多个独立请求（总时间 ≈ 最慢单次的时间）
async def fast_multi_request(
    prompts:     list[str],
    concurrency: int = 5,    # 控制并发数，避免触发速率限制
) -> list[str]:

    semaphore = asyncio.Semaphore(concurrency)

    async def one_request(prompt: str) -> str:
        async with semaphore:
            resp = await async_client.messages.create(
                model      = "claude-haiku-4-5-20251001",
                max_tokens = 256,
                messages   = [{"role": "user", "content": prompt}],
            )
            return resp.content[0].text

    return await asyncio.gather(*[one_request(p) for p in prompts])


# 速度对比演示
async def benchmark():
    prompts = [f"用一句话描述{topic}" for topic in
               ["Python", "机器学习", "云计算", "数据库", "API"]]

    start = asyncio.get_event_loop().time()
    await slow_multi_request(prompts)
    serial_time = asyncio.get_event_loop().time() - start

    start = asyncio.get_event_loop().time()
    await fast_multi_request(prompts, concurrency=5)
    parallel_time = asyncio.get_event_loop().time() - start

    print(f"串行：{serial_time:.1f}s")
    print(f"并行：{parallel_time:.1f}s")
    print(f"提速：{serial_time / parallel_time:.1f}x")

asyncio.run(benchmark())

六、体验提速：流式输出让用户感觉快很多

这是最容易被忽视的优化——总响应时间没变，但用户感受到的"等待时间"大幅缩短。原因很简单：用户看到第一个字出现，就不觉得在"等待"了。

"""
不开流式 vs 开流式的用户体验对比：

不开流式（等待完整响应）：
  0s -------- 等待 -------- 8s [一次性显示全部内容]
  用户感受：等了8秒

开流式（逐字显示）：
  0s - 0.8s [显示第一个字] - 8s [全部完成]
  用户感受：0.8秒就有反应了，然后一直在流

实际测试：相同的请求，开流式后用户满意度评分通常高 20-30%
"""

# 在 FastAPI 中开启流式输出（后端）
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json

app = FastAPI()

@app.post("/chat")
async def chat_stream(data: dict):
    async def generate():
        async with async_client.messages.stream(
            model      = "claude-sonnet-4-6",
            max_tokens = 1024,
            messages   = [{"role": "user", "content": data["message"]}],
        ) as stream:
            async for text in stream.text_stream:
                # SSE 格式
                yield f"data: {json.dumps({'text': text}, ensure_ascii=False)}\n\n"
        yield "data: {\"done\": true}\n\n"

    return StreamingResponse(
        generate(),
        media_type = "text/event-stream",
        headers    = {
            "Cache-Control":     "no-cache",
            "X-Accel-Buffering": "no",    # 关键：防止 Nginx 缓冲
        }
    )

七、网页端用户的提速方法

如果你是在 claude.ai 使用，不是调 API，响应慢的原因和解决方法稍有不同：

• 选对模型：在 claude.ai 里，Haiku 和 Sonnet 的速度差距很明显。如果当前任务不复杂，主动切换到 Sonnet（比 Opus 快得多）
• 避开高峰期：国内用户体感最慢的时间是晚上9点到11点（对应美国工作时间），早上和下午往往更快
• 减少单次输入长度：粘贴一份很长的文档再提问，首字延迟会很长。优先只粘贴相关段落，不是整份文档
• 浏览器问题：有时候是浏览器扩展（尤其是翻译、广告拦截）影响了 SSE 流的展示。无痕模式试一下，如果变快了说明是扩展干扰
• 网络质量：claude.ai 的服务器在美国，网络稳定性直接影响体验。节点不稳定时切换一下节点

八、一张提速决策表

常见问题

Q：用了流式输出，但前端显示还是有延迟，怎么回事？
最常见原因是 Nginx 缓冲——Nginx 默认会把上游的响应缓存后再发给客户端，导致流式效果消失。在 Nginx 配置里加 proxy_buffering off 和响应头 X-Accel-Buffering: no 可以解决。另一个常见原因是前端用了 fetch 但没有正确处理 ReadableStream，导致等到全部内容才渲染。

Q：Haiku 速度是快，但准确率差了很多，能不能有一个中间方案？
可以用"两阶段策略"：先用 Haiku 快速生成，再用 Sonnet 审查和补充。对于很多任务（比如摘要、分类），Haiku 的结果 80% 情况下已经够好，用 Sonnet 审查只针对 Haiku 标记了不确定的部分。整体比全用 Sonnet 快，比全用 Haiku 准。

Q：我的应用需要同时服务很多用户，并发高了会触发速率限制，怎么平衡？
三个方向：用 Semaphore 控制并发上限（本文代码示例）；对非实时请求转入队列平滑流量；联系 Anthropic 申请更高的速率限制配额（企业用量大时通常可以申请）。如果是批量任务，Batch API 没有 RPM 限制，是更好的选择。

总结

Claude 响应慢的四类原因优先级排序：模型选重了（最容易改，换 Haiku 立刻见效）、输入太长（裁剪历史和 System Prompt）、服务端负载（错峰或降级）、代码串行（改并发）。"感觉慢"和"实际慢"是两件事——开流式输出能在不改变任何底层逻辑的情况下，让用户感受到的响应时间从"等待完整内容"变成"0.8秒就开始出字"，是性价比最高的体验优化。