Claude Haiku 5.5: What’s New and How to Use It for Fast, Cheap AI Workflows

pip install -U anthropic
export ANTHROPIC_API_KEY="your-key-here"

# haiku_tasks.py
"""快速、低成本的 Claude Haiku 5.5 调用：使用 JSON 输出进行分类和提取。"""
import json

import anthropic

MODEL = "claude-haiku-5-5"
client = anthropic.Anthropic()    # 从环境变量中读取 ANTHROPIC_API_KEY

# 每百万 tokens 的价格（输入、输出），适用于最多 100K tokens 的提示
PRICES = {
    "claude-haiku-5-5": (0.10, 0.50),
    "claude-sonnet-5-5": (2.00, 10.00),
    "claude-opus-5-5": (4.00, 20.00),
}

CATEGORY_SCHEMA = {
    "type": "object",
    "properties": {"category": {"type": "string", "enum": ["billing", "shipping", "technical", "other"]}},
    "required": ["category"],
    "additionalProperties": False,
}

INVOICE_SCHEMA = {
    "type": "object",
    "properties": {
        "vendor": {"type": "string"},
        "invoice_number": {"type": "string"},
        "total": {"type": "number"},
        "currency": {"type": "string"},
        "due_date": {"type": "string", "description": "ISO 日期格式，YYYY-MM-DD"},
    },
    "required": ["vendor", "invoice_number", "total", "currency", "due_date"],
    "additionalProperties": False,
}

def cost_usd(model, usage):
    price_in, price_out = PRICES[model]
    return (usage.input_tokens * price_in + usage.output_tokens * price_out) / 1_000_000

def ask_json(schema, instruction, text, model=MODEL, effort="low"):
    """一个要求回复必须符合 `schema` 的 JSON 格式的请求。返回 (数据, 用法)。"""
    reply = client.messages.create(
        model=model,
        max_tokens=1000,    # 思考 tokens 会计入此限制，因此预留空间
        system=instruction,
        messages=[{"role": "user", "content": text}],
        output_config={"effort": effort, "format": {"type": "json_schema", "schema": schema}},
    )
    if reply.stop_reason in ("refusal", "max_tokens"):
        raise RuntimeError(f"无可用答案，停止原因: {reply.stop_reason}")
    # 回复可能以思考块开头，因此按类型选择文本块
    answer = "".join(block.text for block in reply.content if block.type == "text")
    return json.loads(answer), reply.usage

def classify(text, model=MODEL):
    data, usage = ask_json(CATEGORY_SCHEMA, "对客户消息进行分类。", text, model)
    return data["category"], usage

def extract_invoice(text, model=MODEL):
    data, usage = ask_json(INVOICE_SCHEMA, "从文本中提取发票字段。", text, model)
    return data, usage

if __name__ == "__main__":
    for message in ["I was charged twice this month.", "My parcel never arrived.", "The app crashes on login."]:
        category, usage = classify(message)
        print(f"{category:<10} {usage.input_tokens} in / {usage.output_tokens} out  <- {message}")

    invoice = "Acme Cloud Ltd, invoice AC-2291. Total due: 1,240.50 USD by 15 November 2026."
    fields, usage = extract_invoice(invoice)
    print(fields, f"(${cost_usd(MODEL, usage):.6f})")

python haiku_tasks.py

# subagents.py
"""主模型将具体任务委托给并行的Haiku 5.5工作者，然后生成最终答案。"""
import asyncio

import anthropic

from haiku_tasks import MODEL as WORKER, cost_usd

LEAD = "claude-sonnet-5-5"
client = anthropic.AsyncAnthropic()

def text_of(reply):
    return "".join(block.text for block in reply.content if block.type == "text")

async def summarize(name, document, limit):
    async with limit:    # 限制同时运行的工作者数量
        reply = await client.messages.create(
            model=WORKER,
            max_tokens=1500,
            system="用两句话总结文档。保留人名、日期和金额的准确性。",
            messages=[{"role": "user", "content": document}],
            output_config={"effort": "low"},
        )
    return name, text_of(reply), cost_usd(WORKER, reply.usage)

async def answer(documents, question):
    limit = asyncio.Semaphore(8)
    notes = await asyncio.gather(*(summarize(n, d, limit) for n, d in documents.items()))
    worker_cost = sum(cost for _, _, cost in notes)

    brief = "\n".join(f"[{name}] {summary}" for name, summary, _ in notes)
    reply = await client.messages.create(
        model=LEAD,
        max_tokens=2000,
        system="你是主分析员。仅使用工作者笔记来回答问题。",
        messages=[{"role": "user", "content": f"问题: {question}\n\n工作者笔记:\n{brief}"}],
        output_config={"effort": "medium"},
    )
    return text_of(reply), worker_cost, cost_usd(LEAD, reply.usage)

if __name__ == "__main__":
    documents = {
        "contract-a": "Northwind于2026年3月3日以每年48,000美元的价格续签了支持合同。",
        "contract-b": "Contoso于2026年5月9日签署了一项为期12个月的许可协议。费用为15,500美元，按季度计费。",
        "contract-c": "Fabrikam的协议将于2026年6月30日到期，只有在任一方发出通知的情况下才会续签。",
    }
    text, worker_cost, lead_cost = asyncio.run(answer(documents, "哪些协议需要在2026年7月前采取行动？"))
    print(text)
    print(f"工作者: ${worker_cost:.6f}  主模型: ${lead_cost:.6f}")

# compare_models.py
"""使用三个模型执行相同的分类任务，比较准确率、延迟和成本。"""
import statistics
import time

from haiku_tasks import classify, cost_usd

MODELS = ["claude-haiku-5-5", "claude-sonnet-5-5", "claude-opus-5-5"]

TICKETS = [
    ("I was charged twice for my last order.", "billing"),
    ("Please send me a copy of my invoice.", "billing"),
    ("Where is my package? It was due on Monday.", "shipping"),
    ("My order was delivered to the wrong building.", "shipping"),
    ("The app crashes every time I open settings.", "technical"),
    ("I cannot log in even after resetting my password.", "technical"),
    ("Can you tell me about your company history?", "other"),
    ("My card was declined but the money left my account.", "billing"),
]

def run(model):
    seconds, cost, hits, output_tokens = [], 0.0, 0, 0
    for text, label in TICKETS:
        start = time.perf_counter()
        category, usage = classify(text, model)
        seconds.append(time.perf_counter() - start)
        hits += category == label
        cost += cost_usd(model, usage)
        output_tokens += usage.output_tokens
    n = len(TICKETS)
    return hits / n, statistics.median(seconds), cost / n * 1000, output_tokens / n

if __name__ == "__main__":
    print(f"{'model':<20}{'accuracy':>9}{'median s':>10}{'$/1k calls':>12}{'out tok':>9}")
    for model in MODELS:
        try:
            accuracy, median_s, per_thousand, out_tokens = run(model)
        except Exception as error:    # 例如模型或平台拒绝请求的情况
            print(f"{model:<20} failed: {error}")
            continue
        print(f"{model:<20}{accuracy:>9.0%}{median_s:>10.2f}{per_thousand:>12.3f}{out_tokens:>9.0f}")

python compare_models.py