Anthropic เปิดตัว Claude Haiku 5.5 โมเดล AI จิ๋วแต่แจ๋ว

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

# haiku_tasks.py
"""Fast, cheap Claude Haiku 5.5 calls: classification and extraction with JSON outputs."""
import json
 
import anthropic
 
MODEL = "claude-haiku-5-5"
client = anthropic.Anthropic()    # reads ANTHROPIC_API_KEY from the environment
 
# USD per million tokens (input, output) for prompts up to 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 date, 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"):
    """One request whose reply must be JSON matching `schema`. Returns (data, usage)."""
    reply = client.messages.create(
        model=model,
        max_tokens=1000,    # thinking tokens count toward this limit, so leave headroom
        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"No usable answer, stop reason: {reply.stop_reason}")
    # A reply can start with a thinking block, so pick the text block by type.
    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, "Classify the customer message.", text, model)
    return data["category"], usage
 
def extract_invoice(text, model=MODEL):
    data, usage = ask_json(INVOICE_SCHEMA, "Extract the invoice fields from the text.", 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})")

# subagents.py
"""A lead model delegates narrow jobs to parallel Haiku 5.5 workers, then writes the answer."""
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:    # cap how many workers run at once
        reply = await client.messages.create(
            model=WORKER,
            max_tokens=1500,
            system="Summarize the document in two sentences. Keep names, dates and amounts exact.",
            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="You are the lead analyst. Answer using only the worker notes.",
        messages=[{"role": "user", "content": f"Question: {question}\n\nWorker notes:\n{brief}"}],
        output_config={"effort": "medium"},
    )
    return text_of(reply), worker_cost, cost_usd(LEAD, reply.usage)
 
if __name__ == "__main__":
    documents = {
        "contract-a": "Northwind renewed its support contract on 3 March 2026 for 48,000 USD per year.",
        "contract-b": "Contoso signed a 12-month license on 9 May 2026. The fee is 15,500 USD, billed quarterly.",
        "contract-c": "Fabrikam's agreement ends on 30 June 2026 and renews only if either party gives notice.",
    }
    text, worker_cost, lead_cost = asyncio.run(answer(documents, "Which agreements need action before July 2026?"))
    print(text)
    print(f"workers: ${worker_cost:.6f}  lead: ${lead_cost:.6f}")