Flows Flows — C++ examples backed by real provider calls. cpp examples examples/flows src/examples/cpp/flows example Flows
ax
C++ axllm::axllm
TypeScript Python Java C++ Go Rust

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

C++ Sequential Flow

Runs a two-step Ax flow against OpenAI.

C++ 
#include "axllm/axllm.hpp"
#include <cstdlib>
#include <fstream>
#include <iostream>
#include <sstream>


int main() {
  const char* key = std::getenv("OPENAI_API_KEY");
  if (key == nullptr || std::string(key).empty()) key = std::getenv("OPENAI_APIKEY");
  if (key == nullptr || std::string(key).empty()) {
    std::cerr << "Set OPENAI_API_KEY or OPENAI_APIKEY to run this example.\n";
    return 2;
  }
  const char* model = std::getenv("AX_OPENAI_MODEL");
  axllm::OpenAICompatibleClient client(axllm::object({
      {"api_key", key},
      {"model", model == nullptr || std::string(model).empty() ? "gpt-5.4-mini" : model},
      {"model_config", axllm::object({{"temperature", 0}})},
  }));
  axllm::AxGen step = axllm::ax("documentText:string -> summaryText:string");
  axllm::AxFlow program = axllm::flow(axllm::object({{"id", "examples.sequentialFlow"}}))
      .execute("step", step)
      .map("note", [](axllm::Value) { return axllm::object({{"note", "Mapped flow state after the provider-backed step."}}); })
      .returns(axllm::object({{"step", "step"}, {"note", "note"}}));
  axllm::Value output = program.forward(client, axllm::object({{"documentText", "Ax gives developers signatures, provider clients, agents, flows, tracing, and optimization."}}));
  std::cout << axllm::stringify(output) << "\n";
}

C++ Branching Flow

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

C++ 
#include "axllm/axllm.hpp"
#include <cstdlib>
#include <fstream>
#include <iostream>
#include <sstream>


int main() {
  const char* key = std::getenv("OPENAI_API_KEY");
  if (key == nullptr || std::string(key).empty()) key = std::getenv("OPENAI_APIKEY");
  if (key == nullptr || std::string(key).empty()) {
    std::cerr << "Set OPENAI_API_KEY or OPENAI_APIKEY to run this example.\n";
    return 2;
  }
  const char* model = std::getenv("AX_OPENAI_MODEL");
  axllm::OpenAICompatibleClient client(axllm::object({
      {"api_key", key},
      {"model", model == nullptr || std::string(model).empty() ? "gpt-5.4-mini" : model},
      {"model_config", axllm::object({{"temperature", 0}})},
  }));
  axllm::AxGen classifier =
      axllm::ax("request:string -> route:class \"support, sales, engineering\"");
  axllm::AxGen responder = axllm::ax("request:string, route:string -> response:string");
  axllm::AxFlow program = axllm::flow(axllm::object({{"id", "examples.branchFlow"}}))
      .execute("classifier", classifier,
               axllm::object({{"reads", axllm::array({"request"})},
                              {"writes", axllm::array({"classifierResult", "route"})}}))
      .execute("responder", responder,
               axllm::object({{"reads", axllm::array({"request", "route"})},
                              {"writes", axllm::array({"responderResult", "response"})}}))
      .returns(axllm::object({{"route", "route"}, {"response", "response"}}));
  axllm::Value output = program.forward(client, axllm::object({{"request", "A customer says checkout is down for their enterprise account."}}));
  std::cout << axllm::stringify(output) << "\n";
}

C++ Parallel Flow

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

C++ 
#include "axllm/axllm.hpp"
#include <cstdlib>
#include <iostream>

int main() {
  const char* key = std::getenv("OPENAI_API_KEY");
  if (key == nullptr || std::string(key).empty()) key = std::getenv("OPENAI_APIKEY");
  if (key == nullptr || std::string(key).empty()) {
    std::cerr << "Set OPENAI_API_KEY or OPENAI_APIKEY to run this example.\n";
    return 2;
  }
  const char* model = std::getenv("AX_OPENAI_MODEL");
  axllm::OpenAICompatibleClient client(axllm::object({
      {"api_key", key},
      {"model", model == nullptr || std::string(model).empty() ? "gpt-5.4-mini" : model},
      {"model_config", axllm::object({{"temperature", 0}})},
  }));
  axllm::AxGen research = axllm::ax("topicText:string -> factList:string[]");
  axllm::AxGen audience = axllm::ax("topicText:string -> audienceAngle:string");
  axllm::AxGen join = axllm::ax("factList:string[], audienceAngle:string -> briefText:string");
  axllm::AxFlow program = axllm::flow(axllm::object({{"id", "examples.parallelFlow"}}))
      .execute("research", research,
               axllm::object({{"reads", axllm::array({"topicText"})},
                              {"writes", axllm::array({"researchResult", "factList"})}}))
      .execute("audience", audience,
               axllm::object({{"reads", axllm::array({"topicText"})},
                              {"writes", axllm::array({"audienceResult", "audienceAngle"})}}))
      .execute("join", join,
               axllm::object({{"reads", axllm::array({"factList", "audienceAngle"})},
                              {"writes", axllm::array({"joinResult", "briefText"})}}))
      .returns(axllm::object({{"briefText", "briefText"}}));
  axllm::Value output = program.forward(
      client,
      axllm::object({{"topicText", "Why typed contracts make multi-step LLM systems easier to maintain"}}));
  std::cout << axllm::stringify(output) << "\n";
}

C++ Controlled Background Flow

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

C++ 
#include "axllm/axllm.hpp"
#include <iostream>
using namespace axllm;

struct RunState { std::atomic<bool> pending{false},finished{false},overlap{false},steered{false};std::atomic<int> applied{0};AxRunControl control; };

int main(){
  const char* key=std::getenv("OPENAI_API_KEY");if(!key||!*key)key=std::getenv("OPENAI_APIKEY");if(!key||!*key)throw std::runtime_error("Set OPENAI_API_KEY or OPENAI_APIKEY.");
  auto client=ai("openai",object({{"api_key",key},{"model","gpt-6-astra"},{"model_config",object({{"thinkingTokenBudget","low"},{"max_tokens",4096}})}}));
  auto state=std::make_shared<RunState>();auto control=state->control;std::weak_ptr<RunState> weak=state;
  control.on_event([weak](Value event){if(auto state=weak.lock();state&&stringify(Core::get(event,"type"))=="\"applied\"")++state->applied;});
  Value schema=object({{"type","object"},{"properties",Value::object()},{"additionalProperties",false}});
  Tool slow("slow_reference","Look up a reference; takes a few seconds.",schema,[state](Value){state->pending.store(true);if(!state->steered.exchange(true)){state->control.steer("Include the word VERIFIED in the final answer.");state->control.set_thinking_token_budget("medium");}std::this_thread::sleep_for(std::chrono::seconds(6));state->finished.store(true);return Value("REF-42");});slow.execution("background");
  Tool label("local_label","Read an independent local label immediately.",schema,[state](Value){for(int i=0;i<300&&!state->pending.load();++i)std::this_thread::sleep_for(std::chrono::milliseconds(10));if(state->pending.load()&&!state->finished.load())state->overlap.store(true);return Value("LAUNCH");});
  auto program=ax("question -> answer");program.add_tool(slow).add_tool(label);
  auto verifier=ax("answer -> report \"Repeat the exact reference, label, and verification word from the answer.\"");
  auto workflow=flow().execute("lookup",program,object({{"writes",Value(Array{"answer"})}})).execute("verify",verifier,object({{"reads",Value(Array{"answer"})}})).returns(object({{"answer","report"}}));
  Value result=workflow.forward(*client,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."}}),object({{"control",control.value()},{"serviceTier","standard"},{"maxSteps",6}}));
  std::string answer=stringify(result);for(const std::string& word:{"REF-42","LAUNCH","VERIFIED"})if(answer.find(word)==std::string::npos)throw std::runtime_error("Missing final result: "+answer);
  if(!state->overlap.load())throw std::runtime_error("No independent work while background tool was pending");if(state->applied.load()!=4)throw std::runtime_error("Control updates were not applied");
  std::cout<<answer<<"\nBackground overlap verified; steering and reasoning applied at the next response.\n";
}

C++ Concurrent Astra Flow

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

C++ 
#include "axllm/axllm.hpp"
#include <cstdlib>
#include <iostream>
#include <set>
using namespace axllm;

int main(){
  const char* key=std::getenv("OPENAI_API_KEY");if(!key||!*key)key=std::getenv("OPENAI_APIKEY");if(!key||!*key)throw std::runtime_error("Set OPENAI_API_KEY or OPENAI_APIKEY.");
  auto client=ai("openai",object({{"api_key",key},{"model","gpt-6-astra"},{"model_config",object({{"thinkingTokenBudget","low"},{"max_tokens",4096}})}}));
  struct Gate{std::mutex mutex;std::condition_variable ready;int calls=0;bool updates=false;};auto gate=std::make_shared<Gate>();
  auto control=run_control();std::set<std::string> paths;std::vector<Value> applied;
  control.on_event([&control,&paths,&applied,gate](Value event){
    if(display(Core::get(event,"type"))=="tool.started"){
      paths.insert(display(Core::get(event,"path")));if(paths.size()==2){control.steer("Include VERIFIED with the exact reference in your final answer.");control.set_thinking_token_budget("medium","root/left");std::lock_guard<std::mutex> lock(gate->mutex);gate->updates=true;gate->ready.notify_all();}
    }
    if(display(Core::get(event,"type"))=="applied")applied.push_back(event);
  });
  Tool lookup("lookup","Look up the exact reference once.",Value::object(),[gate](Value){
    std::unique_lock<std::mutex> lock(gate->mutex);if(++gate->calls>2)throw std::runtime_error("Lookup was called more than once per node");gate->ready.notify_all();
    if(!gate->ready.wait_for(lock,std::chrono::seconds(45),[&]{return gate->calls==2&&gate->updates;}))throw std::runtime_error("Both controlled nodes did not overlap");return Value("REF-42");
  });lookup.execution("background");
  auto program=ax("question -> answer");program.add_tool(lookup);
  auto workflow=flow().execute("left",program).execute("right",program).returns(object({{"left","leftResult"},{"right","rightResult"}}));
  auto result=workflow.forward(*client,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."}}),object({{"control",control.value()},{"serviceTier","standard"},{"maxSteps",6}}));
  if(paths!=std::set<std::string>{"root/left","root/right"}||applied.size()!=3)throw std::runtime_error("Scoped controls did not apply");
  for(const auto* node:{"left","right"}){auto answer=stringify(Core::get(result,node));if(answer.find("REF-42")==std::string::npos||answer.find("VERIFIED")==std::string::npos)throw std::runtime_error("Missing final result: "+answer);}
  std::cout<<stringify(result)<<"\nParallel overlap verified; root steering and targeted reasoning applied.\n";
}

C++ Composed Flow

Composes multiple typed programs into one OpenAI-backed flow.

C++ 
#include "axllm/axllm.hpp"
#include <cstdlib>
#include <fstream>
#include <iostream>
#include <sstream>


int main() {
  const char* key = std::getenv("OPENAI_API_KEY");
  if (key == nullptr || std::string(key).empty()) key = std::getenv("OPENAI_APIKEY");
  if (key == nullptr || std::string(key).empty()) {
    std::cerr << "Set OPENAI_API_KEY or OPENAI_APIKEY to run this example.\n";
    return 2;
  }
  const char* model = std::getenv("AX_OPENAI_MODEL");
  axllm::OpenAICompatibleClient client(axllm::object({
      {"api_key", key},
      {"model", model == nullptr || std::string(model).empty() ? "gpt-5.4-mini" : model},
      {"model_config", axllm::object({{"temperature", 0}})},
  }));
  axllm::AxGen step = axllm::ax("topic:string -> outline:string[]");
  axllm::AxFlow program = axllm::flow(axllm::object({{"id", "examples.composedFlow"}}))
      .execute("step", step)
      .map("note", [](axllm::Value) { return axllm::object({{"note", "Mapped flow state after the provider-backed step."}}); })
      .returns(axllm::object({{"step", "step"}, {"note", "note"}}));
  axllm::Value output = program.forward(client, axllm::object({{"topic", "How Ax moves from typed generation to agents, flows, and optimization"}}));
  std::cout << axllm::stringify(output) << "\n";
}

C++ Refinement Flow

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

C++ 
#include "axllm/axllm.hpp"
#include <cstdlib>
#include <iostream>

int main() {
  const char* key = std::getenv("OPENAI_API_KEY");
  if (key == nullptr || std::string(key).empty()) key = std::getenv("OPENAI_APIKEY");
  if (key == nullptr || std::string(key).empty()) {
    std::cerr << "Set OPENAI_API_KEY or OPENAI_APIKEY to run this example.\n";
    return 2;
  }
  const char* model = std::getenv("AX_OPENAI_MODEL");
  axllm::OpenAICompatibleClient client(axllm::object({
      {"api_key", key},
      {"model", model == nullptr || std::string(model).empty() ? "gpt-5.4-mini" : model},
      {"model_config", axllm::object({{"temperature", 0}})},
  }));
  axllm::AxGen draft = axllm::ax("topicText:string -> draftText:string");
  axllm::AxGen critique = axllm::ax("draftText:string -> critiqueText:string");
  axllm::AxGen revise = axllm::ax("draftText:string, critiqueText:string -> revisedText:string");
  axllm::AxFlow program = axllm::flow(axllm::object({{"id", "examples.refineFlow"}}))
      .execute("draft", draft,
               axllm::object({{"reads", axllm::array({"topicText"})},
                              {"writes", axllm::array({"draftResult", "draftText"})}}))
      .execute("critique", critique,
               axllm::object({{"reads", axllm::array({"draftText"})},
                              {"writes", axllm::array({"critiqueResult", "critiqueText"})}}))
      .execute("revise", revise,
               axllm::object({{"reads", axllm::array({"draftText", "critiqueText"})},
                              {"writes", axllm::array({"reviseResult", "revisedText"})}}))
      .returns(axllm::object({{"revisedText", "revisedText"}}));
  axllm::Value output = program.forward(
      client,
      axllm::object({{"topicText", "Explain automatic flow parallelism to a backend engineer."}}));
  std::cout << axllm::stringify(output) << "\n";
}
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