Fluxos e orquestração (Flow e Graph)

A jangada traz duas formas de encadear chamadas, ambas agregando usage/cost.

Flow — sequencial

Flow encadeia Steps: a saída de um vira entrada do próximo.

from jangada_ai import LLM, Flow, Step

llm = LLM("openai", "gpt-4o-mini")

flow = Flow([
    Step("rascunho", "Escreva um parágrafo sobre {{tema}}."),
    Step("revisao",  "Revise e melhore:\n{{rascunho}}"),
])

resultado = flow.run(llm, tema="jangadas do Nordeste")
print(resultado.output)     # saída do último step
print(resultado.cost)       # custo agregado de toda a cadeia

Cada Step referencia as saídas anteriores pelo nome via template {{ }}.

Graph — roteamento condicional + paralelo

Graph permite ramificar (roteamento condicional) e executar nós em paralelo (core async), juntando os resultados.

from jangada_ai import Graph

# roteamento condicional: escolhe o próximo nó conforme a saída
# paralelo + junção: dispara vários nós e combina as respostas
g = Graph()
# ... defina nós, arestas condicionais e junções ...
res = g.run(...)        # GraphResult agrega usage/cost

Veja os exemplos executáveis em examples/graph_example.py.

Fan-out robusto e observável (`parallel`)

parallel aceita opções para produção:

g.parallel(
    "analises", branches, join="sintese",
    on_error="skip",        # um ramo que falha não derruba o run (default: "raise")
    max_concurrency=4,      # teto de ramos simultâneos (evita rate limit)
    summarize=True,         # cada ramo resume a própria saída antes do join
)

on_error="skip": o ramo que falhar fica ausente do join (string vazia no contexto) e vai para GraphResult.failures ({ramo: erro}). O fan-out entrega o parcial em vez de perder o trabalho já pago dos outros ramos. Com "raise" (padrão) a primeira exceção derruba o run.
max_concurrency=N: semáforo — no máximo N ramos rodam ao mesmo tempo.
summarize: True (instrução padrão) ou uma instrução str. Corta o input do join (que lê todos os ramos e cresce O(N)); a chamada de resumo entra no usage/cost.

GraphResult também expõe durations ({nó/ramo: segundos}) — dá para medir max(ramos) vs t_join diretamente, o diagnóstico que importa num fan-out.

Custo agregado

FlowResult e GraphResult somam usage e cost de todas as etapas — útil para observabilidade. Detalhes em Custo e tokens e, para inspecionar passo a passo, Debug.

O que mudou na 1.9.0

Flow.arun(**contexto): versão assíncrona do Flow.
Graph(max_steps=50): teto de nós executados num run; um router cíclico que nunca termina levanta RuntimeError em vez de gastar sem limite.
Nó revisitado (ciclo) ou step com nome repetido não sobrescreve o anterior: as visitas aparecem como nome#2, nome#3 em completions/durations, e usage/cost somam todas. parsed("nome") devolve a última visita.
Ramos paralelos com on_error="raise": quando um ramo falha, os irmãos ainda pendentes são cancelados (antes continuavam rodando e gastando tokens).
Nomes reservados. Nós, steps, ramos e variáveis de contexto não podem se chamar system, history, params, tools, tool_choice, files, images, schema, prompt, mcp_servers ou self — eles colidiam com os argumentos de complete e causavam erros estranhos. Agora levantam ValueError na hora.
then()/route() num bloco paralelo levantam erro (antes eram ignorados em silêncio); join apontando para nó inexistente tem mensagem própria.

Exemplo

examples/graph_example.py — script executável.