Flows and orchestration (Flow and Graph)

Jangada provides two ways to chain calls, both aggregating usage/cost.

Flow — sequential

Flow chains Steps: the output of one becomes the input of the next.

from jangada_ai import LLM, Flow, Step

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

flow = Flow([
    Step("draft",  "Write a paragraph about {{topic}}."),
    Step("review", "Review and improve:\n{{draft}}"),
])

result = flow.run(llm, topic="jangadas of the Northeast")
print(result.output)     # output of the last step
print(result.cost)       # aggregated cost of the whole chain

Each Step references previous outputs by name via the {{ }} template.

Graph — conditional routing + parallel

Graph lets you branch (conditional routing) and run nodes in parallel (async core), joining the results.

from jangada_ai import Graph

# conditional routing: choose the next node based on the output
# parallel + join: fire several nodes and combine the responses
g = Graph()
# ... define nodes, conditional edges, and joins ...
res = g.run(...)        # GraphResult aggregates usage/cost

See the runnable examples in examples/graph_example.py.

Robust, observable fan-out (parallel)

parallel takes production-grade options:

g.parallel(
    "analyses", branches, join="synthesis",
    on_error="skip",        # a failing branch won't tear down the run (default: "raise")
    max_concurrency=4,      # cap on simultaneous branches (avoids rate limits)
    summarize=True,         # each branch summarizes its own output before the join
)

• on_error="skip": a failing branch is left out of the join (empty string in the context) and recorded in GraphResult.failures ({branch: error}). The fan-out returns the partial result instead of throwing away the already-paid work of the other branches. With "raise" (default) the first exception tears down the run.
• max_concurrency=N: semaphore — at most N branches run at once.
• summarize: True (default instruction) or a str instruction. Trims the join's input (it reads every branch and grows O(N)); the summarization call is counted in usage/cost.

GraphResult also exposes durations ({node/branch: seconds}) — so you can measure max(branches) vs t_join directly, the diagnostic that matters in a fan-out.

Aggregated cost

FlowResult and GraphResult sum usage and cost across all stages — useful for observability. Details in Cost and tokens and, to inspect step by step, Debug.

What changed in 1.9.0

• Flow.arun(**context): async version of Flow.
• Graph(max_steps=50): cap on nodes executed in a run; a cyclic router that never ends raises RuntimeError instead of spending without limit.
• A revisited node (cycle) or a step with a repeated name no longer overwrites the previous one: visits show up as name#2, name#3 in completions/durations, and usage/cost add them all up. parsed("name") returns the last visit.
• Parallel branches with on_error="raise": when a branch fails, the pending sibling branches are cancelled (they used to keep running and spending tokens).
• Reserved names. Nodes, steps, branches and context variables can't be called system, history, params, tools, tool_choice, files, images, schema, prompt, mcp_servers or self — they clashed with complete's arguments and caused odd errors. They now raise ValueError right away.
• then()/route() on a parallel block raise an error (they used to be silently ignored); a join pointing to a missing node has its own message.

Example

examples/graph_example.py — runnable script.