jangada-ai 0.15.0

Camada fina e adaptável sobre SDKs de LLM: texto e vision, async, structured output, fluxos e fallback por erro

jangada 🛶

Uma camada fina e adaptável sobre os SDKs de LLM. Uma jangada leve que te leva entre Anthropic, OpenAI, Groq e Gemini sem trocar o leme: você muda provider / model / api_key e o resto do código continua igual.

from jangada_ai import LLM

llm = LLM("anthropic", "claude-opus-4-8")
print(llm.complete("Explique {{tema}} em 2 frases.", tema="MCP").text)

Por que existe (as nuances que ela resolve)

Wrappers "genéricos" costumam quebrar em produção por detalhes que só aparecem quando você troca de modelo. A jangada resolve estes:

• Modelos do mesmo provider têm contratos diferentes. gpt-5 rejeita temperature (400) e exige max_completion_tokens; gemini-3.x descarta temperature/top_p/top_k e troca thinking_budget por thinking_level. A jangada normaliza o payload por modelo — você só troca o nome.
• Nomes de parâmetro divergem entre SDKs. stop vira stop_sequences na Anthropic/Gemini; max_tokens vira max_output_tokens no Gemini. Você usa sempre o nome canônico.
• Erros são heterogêneos. Cada SDK levanta exceções diferentes; aqui tudo vira um conjunto único, com status_code, que alimenta retry e fallback.
• Falhas transitórias. Rate limit e 5xx ganham retry com backoff no mesmo provider antes de cair pro fallback (outro modelo/provider).
• Custo é invisível por padrão. Cada resposta volta com usage e cost estimado, e Flow/Graph agregam o total.
• Structured output é diferente em cada um. OpenAI .parse, Groq json_schema, Gemini response_schema, Anthropic tool-forcing — uma só chamada parse() cuida disso.
• Detecção de objetos sem amarrar a um provider. detect_objects() devolve bounding boxes em pixels e funciona em qualquer modelo com visão (vision + structured output), não só no Gemini.
• Transcrição de áudio onde dá. transcribe() cobre OpenAI, Groq e Gemini (Anthropic não aceita áudio na API) — com a mesma interface e o mesmo fallback das outras chamadas.
• Documentos nem sempre precisam de vision. docx/pdf/csv/xlsx têm o texto embutido: a jangada extrai localmente (mais barato, roda em modelo sem visão) e só usa vision quando você pede ou quando o PDF é escaneado.

Imports são preguiçosos: import jangada_ai funciona sem nenhum SDK instalado.

Documentação

A documentação completa fica no repositório público jangada-docs:

• 📚 Guias por tema: https://github.com/nerigleston/jangada-docs/tree/main/docs
• 🤖 Para LLMs — índice: llms.txt
• 🤖 Para LLMs — completo: llms-full.txt
• 📦 Pacote no PyPI: https://pypi.org/project/jangada-ai/

A fonte dos docs está aqui em docs/; o repo público é espelhado por scripts/sync-docs.sh.

Instalação

pip install -e ".[anthropic]"        # só Claude
pip install -e ".[openai,groq]"      # OpenAI + Groq
pip install -e ".[all]"              # todos
pip install -e ".[files]"            # ler docx/pdf/csv/xlsx
pip install -e ".[dev]"              # ambiente de testes (pytest + libs)

Chaves de API e .env

Precedência: api_key= explícito > variável de ambiente > arquivo .env.

LLM("openai", "gpt-4o", api_key="sk-...")   # explícito
LLM("openai", "gpt-4o")                       # lê OPENAI_API_KEY (ambiente ou .env)

O .env é detectado na importação, subindo a partir do diretório atual, de forma não-destrutiva (não sobrescreve variáveis já definidas). Usa python-dotenv se instalado, ou um parser embutido. Desligue com JANGADA_NO_DOTENV=1, ou carregue manualmente: jangada.load_env("caminho/.env").

Variáveis lidas: ANTHROPIC_API_KEY, OPENAI_API_KEY, GROQ_API_KEY, GEMINI_API_KEY (ou GOOGLE_API_KEY).

Parâmetros de geração

Os params comuns são argumentos nomeados de primeira classe, traduzidos para cada SDK e descartados quando o modelo não os aceita:

llm = LLM("openai", "gpt-4o",
          temperature=0.2, max_tokens=800, top_p=0.9,
          top_k=40, stop=["\n\n"], seed=42)

canônico	OpenAI / Groq	Anthropic	Gemini
temperature	temperature	temperature	temperature
max_tokens	max_tokens¹	max_tokens	max_output_tokens
top_p	top_p	top_p	top_p
top_k	(descartado)	top_k	top_k
stop	stop	stop_sequences	stop_sequences
seed	seed	(descartado)	seed

¹ vira max_completion_tokens em modelos de raciocínio (gpt-5).

Params específicos vão em extra={...} (ex.: reasoning_effort, thinking_level, verbosity). Override por chamada: llm.complete("...", params={"temperature": 0.7}).

Diferenças entre modelos (perfis automáticos)

Você troca o nome do modelo e segue — a jangada ajusta o payload:

LLM("openai", "gpt-4o",  temperature=0.3, max_tokens=500)   # vai como está
LLM("openai", "gpt-5.2", temperature=0.3, max_tokens=500)   # temperature removido; max_tokens -> max_completion_tokens
LLM("gemini", "gemini-2.5-flash", temperature=0.5)          # vai como está
LLM("gemini", "gemini-3.5-flash", temperature=0.5)          # sampling descartado (Gemini 3.x usa defaults)

Registre regras para modelos novos:

from jangada_ai import Profile, register_profile
register_profile("openai", r"^modelo-novo", Profile(drop=("temperature",), note="..."))

Nuance de function calling no Gemini 3.x. A API valida thought signatures de forma estrita: reconstruir o histórico de tools (em vez de devolver o histórico completo e deixar o SDK cuidar) causa 400. Foi isso que quebrou conversores genéricos ao migrar gemini-2.5 → gemini-3.5. A jangada não reconstrói histórico de tools (structured output é single-turn), então não esbarra nisso — mas guarde a regra se for montar loops agênticos.

Structured output

A mesma parse() em qualquer provider (por baixo: OpenAI .parse, Groq json_schema, Gemini response_schema, Anthropic tool-forcing):

from pydantic import BaseModel

class Fatura(BaseModel):
    fornecedor: str
    total: float
    itens: list[str]

fatura = llm.parse("Extraia a fatura:\n{{t}}", Fatura, t=texto).parsed  # instância validada

Async: await llm.aparse(...). Em fluxos/grafos, um passo com schema= guarda o .parsed em result.parsed("passo").

Vision

from jangada_ai import Image

llm.complete("O que aparece aqui?", images=["foto.jpg"])

img = Image.from_bytes(upload_bytes, "image/png")   # ou from_base64
recibo = llm.parse("Extraia o total.", Recibo, images=[img]).parsed

Imagens são bytes (path/bytes/base64) e viram o formato nativo de cada SDK. Use sempre um modelo com visão.

Tools (function calling)

from jangada_ai import LLM, Message

def get_weather(city: str) -> str:
    """Retorna o clima atual de uma cidade."""
    return "25°C, ensolarado"

llm = LLM("openai", "gpt-4o-mini")
comp = llm.complete("Tempo em Recife?", tools=[get_weather])   # OpenAI/Groq/Anthropic/Gemini
for call in comp.tool_calls:                                    # você executa
    out = get_weather(**call.args)
    final = llm.complete("Tempo em Recife?",
        history=[comp.assistant_message(), Message.tool_results(call.result(out))],
        tools=[get_weather])

Baixo nível (você executa e reenvia). tools= aceita função/Pydantic/dict. Tools prontas em jangada_ai.prebuilt (ex.: tavily_search). Veja docs/tools.md.

MCP (Model Context Protocol)

from jangada_ai import LLM, MCPServer

# remoto por URL — o provider executa as tools (Anthropic/OpenAI/Groq)
llm = LLM("anthropic", "claude-opus-4-8")
llm.complete("Liste as issues.", mcp_servers=[
    MCPServer(url="https://mcp.exemplo.com/sse", name="github", authorization_token="TOKEN")])

# Gemini é client-side (sessão), só no async: await llm.acomplete(..., mcp_servers=[session])

Dois modelos conforme o SDK: remoto/URL (Anthropic/OpenAI/Groq, server-side; OpenAI/Groq via Responses API) e client-side por sessão (Gemini, async). Veja docs/mcp.md.

Detecção de objetos

from jangada_ai import LLM, detect_objects

llm = LLM("gemini", "gemini-2.5-flash")   # funciona em qualquer modelo com visão
for d in detect_objects(llm, "foto.png"):
    print(d.label, d.box)                 # box = [x1, y1, x2, y2] em pixels

# acrescente contexto sem perder o formato de saída:
detect_objects(llm, "foto.png", target="todos os carros",
               instructions="Ignore os desfocados; rotule em inglês.")

Bounding boxes em pixels absolutos. É vision + structured output, então roda em todos os providers (o Gemini é o mais preciso). Veja docs/detect.md.

RAG (embeddings + busca vetorial/híbrida)

from jangada_ai import LLM
from jangada_ai.rag import RAG, vector_store      # pip install "jangada-ai[rag]"

emb  = LLM("openai", "text-embedding-3-small")    # embed: OpenAI ou Gemini
store = vector_store("postgresql://...")          # ou "mongodb+srv://..." (pela conn string)
rag = RAG(emb, store, chat=LLM("openai", "gpt-4o-mini"))

rag.add_document("manual.pdf")
print(rag.ask("Como faço backup?", mode="hybrid").text)   # vetorial + texto (RRF)

embed() está em OpenAI/Gemini (Anthropic/Groq não têm). O vector store é escolhido pela string de conexão (pgvector ou Mongo) e cria tabelas/índices sozinho. Veja docs/rag.md.

Transcrição de áudio

from jangada_ai import LLM

LLM("openai", "gpt-4o-transcribe").transcribe("entrevista.mp3").text
LLM("groq", "whisper-large-v3-turbo").transcribe("entrevista.mp3", language="pt").text
LLM("gemini", "gemini-2.5-flash").transcribe("entrevista.mp3").text

Suportado em OpenAI, Groq e Gemini; o Anthropic não aceita áudio na API (levanta UnsupportedError). Honra retry e fallback como qualquer chamada. Veja docs/audio.md.

Documentos (docx, pdf, csv, xlsx)

from jangada_ai import Document

# Por padrão EXTRAI O TEXTO localmente (não usa vision): mais barato e
# funciona em qualquer modelo. Tabelas viram markdown.
llm.complete("Resuma:", files=["relatorio.pdf", "contrato.docx"])

# xlsx: todas as abas entram, cada uma rotulada (## Aba: ...).
# max_rows limita planilhas gigantes para economizar tokens.
llm.complete("Maior total?", files=[Document("vendas.xlsx", max_rows=200)])

# Bytes em memória (upload/fila) — informe o nome para detectar o tipo.
llm.parse("Há duplicadas?", Relatorio, files=[Document(blob, name="x.csv")])

# Forçar vision (PDF escaneado, sem camada de texto / quando o layout importa):
llm.complete("Transcreva:", files=[Document("scan.pdf", mode="vision")])

Regra do mode="auto" (padrão): csv/xlsx/docx e PDF com camada de texto são extraídos como texto; imagens vão para vision. Um PDF sem texto (escaneado) levanta DocumentError sugerindo mode="vision" — nunca devolve um bloco vazio em silêncio. files= existe em complete/parse/stream (sync e async) e convive com images=.

Requer o extra: pip install "jangada[files]" (pypdf, python-docx, openpyxl).

Streaming

for token in llm.stream("Conte sobre {{x}}", x="João Pessoa"):
    print(token, end="")

async for token in llm.astream("..."):   # FastAPI
    ...

Retry + fallback

Duas camadas, com semântica clara:

1. Retry no mesmo provider (com backoff exponencial + jitter) para erros transitórios: rate limit (429), timeout, conexão e 5xx.
2. Fallback para o próximo candidato quando o retry esgota — pode ser outro modelo ou outro provider.

llm = LLM(
    "groq", "llama-3.3-70b-versatile",
    max_retries=3, backoff_base=0.5, backoff_max=8.0, jitter=True,
).with_fallback(
    LLM("openai", "gpt-5.2"),
    LLM("anthropic", "claude-opus-4-8"),
)

resp = llm.complete("...")
print(resp.provider)   # quem de fato respondeu

O que não dispara retry nem fallback por padrão: AuthError (401/403) e BadRequestError (400/422) — trocar de provider ou repetir não resolveria. NotFoundError (modelo inexistente) não faz retry, mas faz fallback (ótimo para "modelo de reserva"). Tudo configurável via retry_on= (failover) e backoff_on= (quais erros repetem). Vale em sync, async e streaming (o failover de stream ocorre antes do 1º token).

Custo e tokens (na resposta)

Toda resposta volta com usage e cost (USD estimado):

r = llm.complete("...")
r.usage   # {'input_tokens': 120, 'output_tokens': 84}
r.cost    # 0.00042  (None se o modelo não estiver na tabela)

Flow e Graph agregam o total da execução:

res = flow.run(...)
res.usage   # soma de todos os passos
res.cost    # custo total estimado

Preços mudam direto. A tabela embutida é um retrato aproximado — verifique a página de preços e sobrescreva quando precisar de exatidão: jangada.register_price(r"gpt-5\.2", 1.75, 14.00) (USD por 1M tokens).

Fluxos (sequencial)

A saída de cada passo vira variável {{ }} dos próximos:

from jangada_ai import Flow

flow = (
    Flow(llm)
    .step("resumo",   "Resuma:\n{{texto}}")
    .step("traducao", "Traduza para {{idioma}}:\n{{resumo}}")
)
r = flow.run(texto="...", idioma="inglês")
r["traducao"]; r.cost; r.usage

Cada passo pode ter llm= próprio (misturar providers) e schema= (parse).

Orquestração (Graph: roteamento + paralelo)

Quando o Flow linear não basta — um agente decide o próximo, ou vários rodam em paralelo:

from jangada_ai import Graph

# roteamento condicional
g = Graph()
g.node("triagem", clf, "Responda 'tecnico' ou 'geral': {{pergunta}}")
g.node("tecnico", tec, "Responda técnico: {{pergunta}}")
g.node("geral",   ger, "Responda simples: {{pergunta}}")
g.route("triagem", lambda ctx: "tecnico" if "tecnico" in ctx["triagem"].lower() else "geral")
r = g.run("triagem", pergunta="...")
r.path   # ['triagem', 'tecnico']

# paralelo + junção
g = Graph()
g.parallel("pesquisas", {
    "mercado": (llm_a, "Mercado de: {{tema}}"),
    "tecnica": (llm_b, "Viabilidade de: {{tema}}"),
}, join="sintese")
g.node("sintese", llm_s, "Combine:\n{{mercado}}\n{{tecnica}}")
r = g.run("pesquisas", tema="...")

Compõe nos dois sentidos (rota → paralelo, ou paralelo → rota). O núcleo é async: em FastAPI use await g.arun(...); fora de loop, g.run(...).

Debug passo a passo (por agente)

debug=True narra a cadeia: prompt, params, resposta (tempo, tokens, custo), retries, erros e trocas de fallback. Cada LLM tem seu debug e name, então em fluxos/grafos o trace sai por agente.

┌─ [resumidor] groq/llama-3.3-70b-versatile
│  user: Resuma: ...
│  params: temperature=0.2, max_tokens=1024
│  ↻ tentativa 1 após 0.5s (RateLimitError)
│  ✗ RateLimitError (429): quota exceeded
│  ↪ fallback → anthropic/claude-opus-4-8
┌─ [resumidor] anthropic/claude-opus-4-8
│  ← 412ms · ↑12 ↓84 tok · $0.006330: Resumo do texto...
└─

Para mandar pro log: llm.debug.sink = logging.getLogger("jangada").info.

Boas práticas e nuances

• system templatizado é strict. Se você definir system="Aja como {{persona}}." no construtor, passe persona em toda chamada. Em fluxos, prefira system fixo no LLM e ponha a parte variável nos prompts dos passos.
• Modelos de raciocínio (gpt-5, gemini-3.x) ignoram temperature. Não adianta ajustar sampling; use extra={"reasoning_effort": "..."} / extra={"thinking_level": "..."}. A jangada já descarta o que não cabe.
• Cadeia de fallback barato → forte. Coloque o modelo rápido/barato como primário e os caros como reserva; o retry segura picos de 429 sem trocar.
• NotFoundError é seu amigo no fallback de modelo. Aponte um modelo novo como primário e um estável como fallback: se o nome ainda não existir na sua região, ele cai pro estável sem quebrar.
• Flow para pipeline fixo; Graph quando há decisão ou paralelismo. Não use Graph se a ordem é sempre a mesma — Flow é mais simples de ler.
• Em FastAPI use sempre as versões async (acomplete/aparse/astream e graph.arun) para não bloquear o event loop.
• Cheque o custo em produção. Logue resp.cost e result.cost; sobrescreva a tabela de preços com os valores do seu contrato.
• Endpoints compatíveis com OpenAI (Ollama, OpenRouter, vLLM) funcionam pelo provider openai passando base_url: LLM("openai", "llama3", base_url="http://localhost:11434/v1", api_key="ollama").
• Vision é só bytes. URLs remotas não são baixadas automaticamente; carregue com Image.from_path/bytes/base64 para uniformidade entre os 4 providers.

Erros

Em jangada.errors, todos com .provider, .status_code e .original: RateLimitError, APITimeoutError, APIConnectionError, ServerError, OverloadedError, AuthError, BadRequestError, NotFoundError, ProviderError. Conjuntos prontos: TRANSIENT (retry) e DEFAULT_FAILOVER.

Estendendo (novo provider)

Herde de jangada.Provider, implemente os 6 métodos (complete/acomplete/ parse/aparse/stream/astream) + os dois _build_*_client, e registre com jangada.register("nome", lambda: SuaClasse). Veja CLAUDE.md para as invariantes (imports preguiçosos, tradução de erro, ordem translate→profile).