Flows | Ax

These Python examples are real runnable files. Edit the source file first; this page is rebuilt from the checked-in example and its metadata header.

Python Sequential Flow

Runs a two-step Ax flow against OpenAI.

Provider: openai
Env: OPENAI_API_KEY, OPENAI_APIKEY
Level: beginner
Run: npm run example -- python src/examples/python/flows/flow-openai.py
Source: src/examples/python/flows/flow-openai.py

Python

import json
import os

from axllm import OpenAICompatibleClient, ax, flow


api_key = os.getenv("OPENAI_API_KEY") or os.getenv("OPENAI_APIKEY")
if not api_key:
    raise SystemExit("Set OPENAI_API_KEY or OPENAI_APIKEY to run this example.")

client = OpenAICompatibleClient(
    api_key=api_key,
    model=os.getenv("AX_OPENAI_MODEL", "gpt-5.4-mini"),
    model_config={"temperature": 0},
)
step = ax('documentText:string -> summaryText:string')
program = (
    flow({"id": "examples.sequentialFlow"})
    .execute("step", step)
    .map("note", lambda state: {"note": "Mapped flow state after the provider-backed step."})
    .returns({"summary": "step", "note": "note"})
)
output = program.forward(client, {"documentText": "Ax gives developers signatures, provider clients, agents, flows, tracing, and optimization."})
print(json.dumps(output, indent=2, sort_keys=True))

Python Branching Flow

Routes a classification through follow-up flow logic backed by OpenAI.

Provider: openai
Env: OPENAI_API_KEY, OPENAI_APIKEY
Level: intermediate
Run: npm run example -- python src/examples/python/flows/branch-flow.py
Source: src/examples/python/flows/branch-flow.py

Python

import json
import os

from axllm import OpenAICompatibleClient, ax, flow


api_key = os.getenv("OPENAI_API_KEY") or os.getenv("OPENAI_APIKEY")
if not api_key:
    raise SystemExit("Set OPENAI_API_KEY or OPENAI_APIKEY to run this example.")

client = OpenAICompatibleClient(
    api_key=api_key,
    model=os.getenv("AX_OPENAI_MODEL", "gpt-5.4-mini"),
    model_config={"temperature": 0},
)
step = ax('request:string -> route:class "support, sales, engineering"')
program = (
    flow({"id": "examples.branchFlow"})
    .execute("step", step)
    .map("note", lambda state: {"note": "Mapped flow state after the provider-backed step."})
    .returns({"route": "step", "response": "note"})
)
output = program.forward(client, {"request": "A customer says checkout is down for their enterprise account."})
print(json.dumps(output, indent=2, sort_keys=True))

Python Composed Flow

Composes multiple typed programs into one OpenAI-backed flow.

Provider: openai
Env: OPENAI_API_KEY, OPENAI_APIKEY
Level: advanced
Run: npm run example -- python src/examples/python/flows/composed-flow.py
Source: src/examples/python/flows/composed-flow.py

Python

import json
import os

from axllm import OpenAICompatibleClient, ax, flow


api_key = os.getenv("OPENAI_API_KEY") or os.getenv("OPENAI_APIKEY")
if not api_key:
    raise SystemExit("Set OPENAI_API_KEY or OPENAI_APIKEY to run this example.")

client = OpenAICompatibleClient(
    api_key=api_key,
    model=os.getenv("AX_OPENAI_MODEL", "gpt-5.4-mini"),
    model_config={"temperature": 0},
)
step = ax('topic:string -> outline:string[]')
program = (
    flow({"id": "examples.composedFlow"})
    .execute("step", step)
    .map("note", lambda state: {"note": "Mapped flow state after the provider-backed step."})
    .returns({"outline": "step", "brief": "note"})
)
output = program.forward(client, {"topic": "How Ax moves from typed generation to agents, flows, and optimization"})
print(json.dumps(output, indent=2, sort_keys=True))