Flows Flows — Go examples backed by real provider calls. go examples examples/flows src/examples/go/flows example Flows
ax
Go github.com/ax-llm/ax/packages/go
TypeScript Python Java C++ Go Rust

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

Go Sequential Flow

Runs a two-step Ax flow against OpenAI.

Go 
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"time"

	ax "github.com/ax-llm/ax/packages/go"
)


func openAIClient() *ax.OpenAICompatibleClient {
	apiKey := os.Getenv("OPENAI_API_KEY")
	if apiKey == "" { apiKey = os.Getenv("OPENAI_APIKEY") }
	if apiKey == "" { panic("Set OPENAI_API_KEY or OPENAI_APIKEY to run this example.") }
	model := os.Getenv("AX_OPENAI_MODEL")
	if model == "" { model = "gpt-5.4-mini" }
	return ax.NewOpenAICompatibleClient(map[string]ax.Value{"api_key": apiKey, "model": model, "model_config": ax.Object("temperature", 0)})
}

func printJSON(value ax.Value) {
	data, err := json.MarshalIndent(value, "", "  ")
	if err != nil { panic(err) }
	fmt.Println(string(data))
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	client := openAIClient()
	step := ax.NewAx("documentText:string -> summaryText:string", nil)
	program := ax.NewFlow(map[string]ax.Value{"id": "examples.sequentialFlow"}).
		Execute("step", step, nil).
		Returns(map[string]ax.Value{"step": "step"})
	output, err := program.Forward(ctx, client, map[string]ax.Value{"documentText": "Ax gives developers signatures, provider clients, agents, flows, tracing, and optimization."}, nil)
	if err != nil { panic(err) }
	printJSON(output)
}

Go Branching Flow

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

Go 
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"time"

	ax "github.com/ax-llm/ax/packages/go"
)

func openAIClient() *ax.OpenAICompatibleClient {
	apiKey := os.Getenv("OPENAI_API_KEY")
	if apiKey == "" {
		apiKey = os.Getenv("OPENAI_APIKEY")
	}
	if apiKey == "" {
		panic("Set OPENAI_API_KEY or OPENAI_APIKEY to run this example.")
	}
	model := os.Getenv("AX_OPENAI_MODEL")
	if model == "" {
		model = "gpt-5.4-mini"
	}
	return ax.NewOpenAICompatibleClient(map[string]ax.Value{"api_key": apiKey, "model": model, "model_config": ax.Object("temperature", 0)})
}

func printJSON(value ax.Value) {
	data, err := json.MarshalIndent(value, "", "  ")
	if err != nil {
		panic(err)
	}
	fmt.Println(string(data))
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	client := openAIClient()
	classifier := ax.NewAx("request:string -> route:class \"support, sales, engineering\"", nil)
	responder := ax.NewAx("request:string, route:string -> response:string", nil)
	program := ax.NewFlow(map[string]ax.Value{"id": "examples.branchFlow"}).
		Execute("classifier", classifier, map[string]ax.Value{
			"reads": ax.Array("request"), "writes": ax.Array("classifierResult", "route"),
		}).
		Execute("responder", responder, map[string]ax.Value{
			"reads": ax.Array("request", "route"), "writes": ax.Array("responderResult", "response"),
		}).
		Returns(map[string]ax.Value{"route": "route", "response": "response"})
	output, err := program.Forward(ctx, client, map[string]ax.Value{"request": "A customer says checkout is down for their enterprise account."}, nil)
	if err != nil {
		panic(err)
	}
	printJSON(output)
}

Go Parallel Flow

Runs two independent OpenAI-backed steps in parallel before joining their results.

Go 
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"time"

	ax "github.com/ax-llm/ax/packages/go"
)

func openAIClient() *ax.OpenAICompatibleClient {
	apiKey := os.Getenv("OPENAI_API_KEY")
	if apiKey == "" {
		apiKey = os.Getenv("OPENAI_APIKEY")
	}
	if apiKey == "" {
		panic("Set OPENAI_API_KEY or OPENAI_APIKEY to run this example.")
	}
	model := os.Getenv("AX_OPENAI_MODEL")
	if model == "" {
		model = "gpt-5.4-mini"
	}
	return ax.NewOpenAICompatibleClient(map[string]ax.Value{
		"api_key":      apiKey,
		"model":        model,
		"model_config": ax.Object("temperature", 0),
	})
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	research := ax.NewAx("topicText:string -> factList:string[]", nil)
	audience := ax.NewAx("topicText:string -> audienceAngle:string", nil)
	join := ax.NewAx("factList:string[], audienceAngle:string -> briefText:string", nil)
	program := ax.NewFlow(map[string]ax.Value{"id": "examples.parallelFlow"}).
		Execute("research", research, map[string]ax.Value{
			"reads": ax.Array("topicText"), "writes": ax.Array("researchResult", "factList"),
		}).
		Execute("audience", audience, map[string]ax.Value{
			"reads": ax.Array("topicText"), "writes": ax.Array("audienceResult", "audienceAngle"),
		}).
		Execute("join", join, map[string]ax.Value{
			"reads": ax.Array("factList", "audienceAngle"), "writes": ax.Array("joinResult", "briefText"),
		}).
		Returns(map[string]ax.Value{"briefText": "briefText"})
	output, err := program.Forward(
		ctx,
		openAIClient(),
		map[string]ax.Value{"topicText": "Why typed contracts make multi-step LLM systems easier to maintain"},
		nil,
	)
	if err != nil {
		panic(err)
	}
	data, err := json.MarshalIndent(output, "", "  ")
	if err != nil {
		panic(err)
	}
	fmt.Println(string(data))
}

Go Controlled Background Flow

Uses ordinary generation with background tools, steering, and a reasoning update.

Go 
package main

import (
	"context"
	"fmt"
	ax "github.com/ax-llm/ax/packages/go"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

func main() {
	key := os.Getenv("OPENAI_API_KEY")
	if key == "" {
		key = os.Getenv("OPENAI_APIKEY")
	}
	if key == "" {
		panic("Set OPENAI_API_KEY or OPENAI_APIKEY.")
	}
	client := ax.NewAI("openai", ax.Object("api_key", key, "model", "gpt-6-astra", "model_config", ax.Object("thinkingTokenBudget", "low", "max_tokens", 4096)))
	pending := make(chan struct{})
	var finished, overlap atomic.Bool
	var queued sync.Once
	var applied atomic.Int32
	control := ax.RunControl()
	control.OnEvent(func(event map[string]ax.Value) {
		if event["type"] == "tool.started" {
			queued.Do(func() {
				if err := control.Steer("Include the word VERIFIED in the final answer."); err != nil {
					panic(err)
				}
				if err := control.SetThinkingTokenBudget("medium"); err != nil {
					panic(err)
				}
			})
		}
		if event["type"] == "applied" {
			applied.Add(1)
		}
	})
	var pendingOnce sync.Once
	slow := ax.Fn("slow_reference").Execution("background").WithContextHandler(func(ctx context.Context, _ map[string]ax.Value) (ax.Value, error) {
		pendingOnce.Do(func(){close(pending)})
		select {
		case <-time.After(6 * time.Second):
			finished.Store(true)
			return "REF-42", nil
		case <-ctx.Done():
			return nil, ctx.Err()
		}
	})
	label := ax.Fn("local_label").WithHandler(func(_ map[string]ax.Value) (ax.Value, error) {
		select {
		case <-pending:
			if !finished.Load() {
				overlap.Store(true)
			}
		case <-time.After(3 * time.Second):
		}
		return "LAUNCH", nil
	})
	program := ax.NewAx("question -> answer", nil)
	program.Functions = []ax.Tool{slow, label}
	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
	defer cancel()
	workflow:=ax.NewFlow(nil).Execute("lookup",program,ax.Object("writes",ax.Array("answer"))).Execute("verify",ax.NewAx(`answer -> report "Repeat the exact reference, label, and verification word from the answer."`,nil),ax.Object("reads",ax.Array("answer"))).Returns(ax.Object("answer","report"))
    result, err := workflow.Forward(ctx, client, ax.Object("question", "First call slow_reference. While it is pending, call local_label. Call each tool only once; do not call a tool again while its result is pending. If a required tool result is still pending, end this response with a brief progress message. The application will continue with the result when it arrives; do not spend reasoning tokens waiting for it. Once both results arrive, return them in one sentence."), ax.Object("control", control, "serviceTier", "standard", "maxSteps", 6))
	if err != nil {
		panic(err)
	}
	answer := fmt.Sprint(result)
	for _, word := range []string{"REF-42", "LAUNCH", "VERIFIED"} {
		if !strings.Contains(answer, word) {
			panic("Missing final result: " + answer)
		}
	}
	if !overlap.Load() {
		panic("No independent model work while background tool was pending")
	}
	if applied.Load() != 4 {
		panic("Control updates were not applied")
	}
	fmt.Println(answer)
	fmt.Println("Background overlap verified; steering and reasoning applied at the next response.")
}

Go Concurrent Astra Flow

Independent conversations overlap, retain their tool results, and receive scoped controls.

Go 
package main

import (
    "context"
    "fmt"
    "os"
    "strings"
    "sync/atomic"
    "time"
    ax "github.com/ax-llm/ax/packages/go"
)

func main() {
    key:=os.Getenv("OPENAI_API_KEY");if key==""{key=os.Getenv("OPENAI_APIKEY")};if key==""{panic("Set OPENAI_API_KEY or OPENAI_APIKEY.")}
    client:=ax.NewAI("openai",ax.Object("api_key",key,"model","gpt-6-astra","model_config",ax.Object("thinkingTokenBudget","low","max_tokens",4096)))
    control:=ax.RunControl();both,updatesReady:=make(chan struct{}),make(chan struct{});var calls atomic.Int32
    paths:=map[string]bool{};var applied []map[string]ax.Value
    control.OnEvent(func(event map[string]ax.Value){
        if event["type"]=="tool.started"{paths[fmt.Sprint(event["path"])]=true;if len(paths)==2{
            if err:=control.Steer("Include VERIFIED with the exact reference in your final answer.");err!=nil{panic(err)}
            if err:=control.SetThinkingTokenBudget("medium","root/left");err!=nil{panic(err)}
            close(updatesReady)
        }}
        if event["type"]=="applied"{applied=append(applied,event)}
    })
    lookup:=ax.Fn("lookup").Execution("background").WithContextHandler(func(ctx context.Context,_ map[string]ax.Value)(ax.Value,error){
        count:=calls.Add(1);if count>2{return nil,fmt.Errorf("lookup was called more than once per node")};if count==2{close(both)}
        select{case <-both:case <-ctx.Done():return nil,ctx.Err()};select{case <-updatesReady:case <-ctx.Done():return nil,ctx.Err()};return "REF-42",nil
    })
    program:=ax.NewAx("question -> answer",nil);program.Functions=[]ax.Tool{lookup}
    workflow:=ax.NewFlow(nil).Execute("left",program,nil).Execute("right",program,nil).Returns(ax.Object("left","leftResult","right","rightResult"))
    ctx,cancel:=context.WithTimeout(context.Background(),90*time.Second);defer cancel()
    result,err:=workflow.Forward(ctx,client,ax.Object("question","Call lookup exactly once. If its result is pending, return a brief progress message without calling it again. Return the exact reference when its result arrives."),ax.Object("control",control,"serviceTier","standard","maxSteps",6));if err!=nil{panic(err)}
    if !paths["root/left"]||!paths["root/right"]||len(applied)!=3{panic(fmt.Sprint("Missing scoped controls: ",paths,applied))}
    for _,node:=range []string{"left","right"}{answer:=fmt.Sprint(result.(map[string]ax.Value)[node]);if !strings.Contains(answer,"REF-42")||!strings.Contains(answer,"VERIFIED"){panic("Missing final result: "+answer)}}
    fmt.Println(result);fmt.Println("Parallel overlap verified; root steering and targeted reasoning applied.")
}

Go Composed Flow

Composes multiple typed programs into one OpenAI-backed flow.

Go 
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"time"

	ax "github.com/ax-llm/ax/packages/go"
)


func openAIClient() *ax.OpenAICompatibleClient {
	apiKey := os.Getenv("OPENAI_API_KEY")
	if apiKey == "" { apiKey = os.Getenv("OPENAI_APIKEY") }
	if apiKey == "" { panic("Set OPENAI_API_KEY or OPENAI_APIKEY to run this example.") }
	model := os.Getenv("AX_OPENAI_MODEL")
	if model == "" { model = "gpt-5.4-mini" }
	return ax.NewOpenAICompatibleClient(map[string]ax.Value{"api_key": apiKey, "model": model, "model_config": ax.Object("temperature", 0)})
}

func printJSON(value ax.Value) {
	data, err := json.MarshalIndent(value, "", "  ")
	if err != nil { panic(err) }
	fmt.Println(string(data))
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	client := openAIClient()
	step := ax.NewAx("topic:string -> outline:string[]", nil)
	program := ax.NewFlow(map[string]ax.Value{"id": "examples.composedFlow"}).
		Execute("step", step, nil).
		Returns(map[string]ax.Value{"step": "step"})
	output, err := program.Forward(ctx, client, map[string]ax.Value{"topic": "How Ax moves from typed generation to agents, flows, and optimization"}, nil)
	if err != nil { panic(err) }
	printJSON(output)
}

Go Refinement Flow

Drafts, critiques, and revises an answer through three OpenAI-backed steps.

Go 
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"time"

	ax "github.com/ax-llm/ax/packages/go"
)

func openAIClient() *ax.OpenAICompatibleClient {
	apiKey := os.Getenv("OPENAI_API_KEY")
	if apiKey == "" {
		apiKey = os.Getenv("OPENAI_APIKEY")
	}
	if apiKey == "" {
		panic("Set OPENAI_API_KEY or OPENAI_APIKEY to run this example.")
	}
	model := os.Getenv("AX_OPENAI_MODEL")
	if model == "" {
		model = "gpt-5.4-mini"
	}
	return ax.NewOpenAICompatibleClient(map[string]ax.Value{
		"api_key":      apiKey,
		"model":        model,
		"model_config": ax.Object("temperature", 0),
	})
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	draft := ax.NewAx("topicText:string -> draftText:string", nil)
	critique := ax.NewAx("draftText:string -> critiqueText:string", nil)
	revise := ax.NewAx("draftText:string, critiqueText:string -> revisedText:string", nil)
	program := ax.NewFlow(map[string]ax.Value{"id": "examples.refineFlow"}).
		Execute("draft", draft, map[string]ax.Value{
			"reads": ax.Array("topicText"), "writes": ax.Array("draftResult", "draftText"),
		}).
		Execute("critique", critique, map[string]ax.Value{
			"reads": ax.Array("draftText"), "writes": ax.Array("critiqueResult", "critiqueText"),
		}).
		Execute("revise", revise, map[string]ax.Value{
			"reads": ax.Array("draftText", "critiqueText"), "writes": ax.Array("reviseResult", "revisedText"),
		}).
		Returns(map[string]ax.Value{"revisedText": "revisedText"})
	output, err := program.Forward(
		ctx,
		openAIClient(),
		map[string]ax.Value{"topicText": "Explain automatic flow parallelism to a backend engineer."},
		nil,
	)
	if err != nil {
		panic(err)
	}
	data, err := json.MarshalIndent(output, "", "  ")
	if err != nil {
		panic(err)
	}
	fmt.Println(string(data))
}
Docs
Start
Quick Start How Ax Fits Together Advanced Start FAQ
Agents
Overview Micro Agents Standard Agents Long-Horizon Agents Performance Internals
Examples
Overview Generation Agents Long-Horizon Agents Flows Audio MCP Optimization
Concepts
DSPy Signatures Tools MCP MCP Subscriptions Event Runtime Optimization Playbook LLMs Telemetry
Subsystems
ai() LLM Models ax() Generation s() Signatures agent() Agents flow() Workflows optimize() GEPA
Skills
Overview Quick Reference AI Typesafe / Jev Audio Signatures Generation Flow Agent Agent RLM Memory & Skills Observability Agent Optimize Agent Context GEPA Refinement Playbook
API Docs
ai() API ax() API s() API agent() API flow() API optimize() API