LLM APIs em Produção: streaming, structured output, batch e cache

# Streaming com Anthropic SDK
from anthropic import Anthropic
client = Anthropic()

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Explique quantum tunneling."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final = stream.get_final_message()

# Passando adiante em FastAPI/Hono como SSE para o browser
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.post("/chat")
async def chat(body: dict):
    async def gen():
        async with client.messages.stream(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=body["messages"],
        ) as stream:
            async for text in stream.text_stream:
                yield f"data: {json.dumps({'delta': text})}\n\n"
            yield "data: [DONE]\n\n"
    return StreamingResponse(gen(), media_type="text/event-stream")

# OpenAI Structured Outputs com Pydantic
from openai import OpenAI
from pydantic import BaseModel
from typing import Literal

client = OpenAI()

class Extract(BaseModel):
    customer_name: str
    sentiment: Literal["positive", "neutral", "negative"]
    urgency: int   # 1-5
    categories: list[str]

resp = client.responses.parse(
    model="gpt-5",
    input=[
        {"role": "system", "content": "Extraia dados estruturados do ticket."},
        {"role": "user",   "content": ticket_text},
    ],
    text_format=Extract,
)
extracted: Extract = resp.output_parsed     # tipado, válido

# Anthropic: usar tool_use como mecanismo (pattern comum)
schema = {
    "name": "extract_ticket",
    "description": "Extrai dados estruturados do ticket.",
    "input_schema": Extract.model_json_schema(),
}
r = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=500,
    tools=[schema],
    tool_choice={"type": "tool", "name": "extract_ticket"},   # força uso
    messages=[{"role": "user", "content": ticket_text}],
)
parsed = next(b for b in r.content if b.type == "tool_use").input

# Anthropic Batch API (Message Batches)
from anthropic import Anthropic
client = Anthropic()

batch = client.messages.batches.create(
    requests=[
        {
            "custom_id": f"item-{i}",
            "params": {
                "model": "claude-sonnet-4-6",
                "max_tokens": 512,
                "messages": [{"role": "user", "content": prompt}],
            },
        }
        for i, prompt in enumerate(prompts)      # pode ter 10k+ itens
    ],
)

# Poll até terminar
import time
while True:
    status = client.messages.batches.retrieve(batch.id)
    if status.processing_status in ("ended", "failed", "canceled"):
        break
    time.sleep(30)

# Ler resultados (JSONL streamed)
for line in client.messages.batches.results(batch.id):
    if line.result.type == "succeeded":
        save_result(line.custom_id, line.result.message)

# Retry robusto com tenacity
from tenacity import retry, stop_after_attempt, wait_exponential_jitter, retry_if_exception_type
from anthropic import APIStatusError, APITimeoutError, RateLimitError

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential_jitter(initial=1, max=30, jitter=2),
    retry=retry_if_exception_type((APITimeoutError, RateLimitError, APIStatusError)),
    reraise=True,
)
def llm_call(messages, **kwargs):
    try:
        return client.messages.create(messages=messages, **kwargs)
    except APIStatusError as e:
        # Retry só em 5xx e 429; 4xx persistente não resolve
        if e.status_code < 500 and e.status_code != 429:
            raise StopRetry() from e
        raise

// Token bucket simples (RPM=60) para não bater 429 no provider
class TokenBucket {
  private tokens: number;
  private last: number;
  constructor(private rate: number, private capacity: number) {
    this.tokens = capacity;
    this.last = Date.now();
  }
  async acquire(n = 1): Promise<void> {
    while (true) {
      this.refill();
      if (this.tokens >= n) { this.tokens -= n; return; }
      const need = (n - this.tokens) / this.rate;
      await new Promise(r => setTimeout(r, need * 1000));
    }
  }
  private refill() {
    const now = Date.now();
    const add = ((now - this.last) / 1000) * this.rate;
    this.tokens = Math.min(this.capacity, this.tokens + add);
    this.last = now;
  }
}

const bucket = new TokenBucket(60 / 60, 60);  // 60 RPM
await bucket.acquire();
const r = await client.messages.create(...);

import uuid
from anthropic import Anthropic
client = Anthropic()

def generate_for_request(request_id: str, prompt: str):
    # idempotency_key estável → provider dedupe no seu lado
    r = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
        extra_headers={"anthropic-idempotency-key": request_id},
    )
    return r

# Em jobs async: persista (request_id, response) no seu lado.
# Antes de chamar LLM, consulte cache; só chama se não tiver.