Workflow Patterns

Choose the smallest workflow that makes your application's decisions explicit. A pipeline, router or bounded review loop often needs less autonomy than a general agent. Graph nodes can combine real model calls with deterministic validation and data access.

Complete Installation . The live examples below use OpenAI and facts supplied in the input; they do not invent a customer's order status.

Run this page’s examples in order in one IEx session; later blocks reuse the modules and values defined earlier.

Prompt chaining

Separate extraction from writing so application code can validate the intermediate result. This module performs two real calls:

defmodule MyApp.TicketWorkflow do
  use BeamWeaver.Graph
  alias BeamWeaver.Core.{ChatModel, Message}

  graph do
    node :summarize, &__MODULE__.summarize/1, timeout: 70_000
    node :draft, &__MODULE__.draft/1, timeout: 70_000
    edge start(), :summarize
    edge :summarize, :draft
    edge :draft, finish()
  end

  def summarize(%{ticket: ticket}) do
    with {:ok, reply} <-
           ChatModel.invoke(model(), "Summarize this request in one sentence, preserving the order ID: #{ticket}") do
      %{summary: Message.text(reply)}
    end
  end

  def draft(%{summary: summary, facts: facts}) do
    with {:ok, reply} <-
           ChatModel.invoke(model(), [
             Message.system(
               "Write a brief customer reply. Use only supplied facts. Include the order ID and delivery day."
             ),
             Message.user("Request: #{summary}\nVerified facts: #{facts}")
           ]) do
      %{reply: Message.text(reply)}
    end
  end

  defp model, do: BeamWeaver.Models.init_chat_model!("openai:gpt-5.4-mini", max_tokens: 256, timeout: 60_000)
end

{:ok, result} =
  MyApp.TicketWorkflow.invoke(%{
    ticket: "Where is my package A-1001?",
    facts: "Order A-1001 arrives Thursday."
  })

true = result.summary =~ "A-1001"
true = result.reply =~ "A-1001"
true = String.downcase(result.reply) =~ "thursday"

A node propagates provider errors rather than converting them into successful text. Add a validation node or guarded edge between stages when missing facts should stop the pipeline. For machine routing, prefer structured output to parsing arbitrary prose.

Routing

A classifier writes a constrained label; guarded edges select the handler. Choose an explicit fallback such as manual review for an invalid/unknown label. Do not silently route every unexpected model answer to a business category. Validate confidence and required facts where your use case needs them.

The Graph DSL demonstrates routing without a model. Substitute a structured classifier only when the input requires one. Decision models are another option; see Models .

Use a schema for the classifier and give unexpected/uncertain input an explicit manual-review label:

defmodule MyApp.TicketTeam do
  use BeamWeaver.Schema
  field :team, :string, enum: ["billing", "shipping", "manual"]
end

defmodule MyApp.TeamClassifier do
  use BeamWeaver.Agent
  model "openai:gpt-5.4-mini", max_tokens: 256, timeout: 60_000

  system_prompt "Classify the request. Payment/refund questions are billing, delivery questions are shipping. Use manual when neither fits."

  response_schema MyApp.TicketTeam
end

alias BeamWeaver.Graph
alias BeamWeaver.Graph.Compiled

routing =
  Graph.new()
  |> Graph.add_node(:classify, fn %{ticket: ticket} ->
    with {:ok, answer} <- MyApp.TeamClassifier.invoke(%{messages: [ticket]}),
         do: %{team: answer.structured_response["team"]}
  end)
  |> Graph.add_node(:billing, fn _ -> %{queue: "billing"} end)
  |> Graph.add_node(:shipping, fn _ -> %{queue: "shipping"} end)
  |> Graph.add_node(:manual, fn _ -> %{queue: "manual"} end)
  |> Graph.add_edge(Graph.start(), :classify)
  |> Graph.add_edge(:classify, :billing, when: %{team: "billing"})
  |> Graph.add_edge(:classify, :shipping, when: %{team: "shipping"})
  |> Graph.add_edge(:classify, :manual, default: true)
  |> Graph.add_edge(:billing, Graph.end_node())
  |> Graph.add_edge(:shipping, Graph.end_node())
  |> Graph.add_edge(:manual, Graph.end_node())
  |> Graph.compile!()

{:ok, %{team: "billing", queue: "billing"}} =
  Compiled.invoke(routing, %{ticket: "I was charged twice. Please review the duplicate payment."})

Parallelization

Run independent checks concurrently and join before deciding. Each branch should write a different key or contribute through a reducer. The join must wait for every required check; a successful tone check cannot stand in for a failed policy check.

Use a map keyed by check name to preserve provenance. Do not assume several model opinions constitute independent factual evidence. Supply source facts to each reviewer and define how disagreement affects the workflow. Graph State shows a complete parallel merge.

These independent checks are ordinary application rules. A reducer collects their named results and a join waits for both:

defmodule MyApp.ParallelReview do
  use BeamWeaver.Graph

  graph do
    state do
      channel :checks, merge: :map
    end

    node :order_id, fn %{draft: text} -> %{checks: %{order_id: text =~ "A-1001"}} end
    node :delivery, fn %{draft: text} -> %{checks: %{delivery: text =~ "Thursday"}} end
    node :decide, fn %{checks: checks} -> %{approved: Enum.all?(checks, fn {_, ok} -> ok end)} end
    edge start(), :order_id
    edge start(), :delivery
    join [:order_id, :delivery], :decide
    edge :decide, finish()
  end
end

{:ok, %{approved: false, checks: %{order_id: true, delivery: false}}} =
  MyApp.ParallelReview.invoke(%{draft: "A-1001 has shipped."})

{:ok, %{approved: true}} = MyApp.ParallelReview.invoke(%{draft: "A-1001 arrives Thursday."})

Orchestrator-workers

When task count is dynamic, dispatch with Send. This real-model example summarizes two supplied records and retains the source ID with each answer:

alias BeamWeaver.Graph
alias BeamWeaver.Graph.{Compiled, Send}
alias BeamWeaver.Core.{ChatModel, Message}
model = BeamWeaver.Models.init_chat_model!("openai:gpt-5.4-mini", max_tokens: 128, timeout: 60_000)

workers =
  Graph.new()
  |> Graph.add_reducer(:summaries, &Kernel.++/2)
  |> Graph.add_node(
    :dispatch,
    fn %{records: records} ->
      Enum.map(records, &%Send{node: :summarize, update: %{record: &1}})
    end,
    destinations: [:summarize]
  )
  |> Graph.add_node(
    :summarize,
    fn %{record: record} ->
      with {:ok, reply} <- ChatModel.invoke(model, "Summarize only this fact briefly: #{record.text}") do
        %{summaries: [%{id: record.id, text: Message.text(reply)}]}
      end
    end,
    timeout: 70_000
  )
  |> Graph.add_edge(Graph.start(), :dispatch)
  |> Graph.add_edge(:summarize, Graph.end_node())
  |> Graph.compile!()

{:ok, batch} =
  Compiled.invoke(workers, %{
    records: [%{id: "A-1001", text: "Delivery is Thursday."}, %{id: "A-1002", text: "Delivery is Friday."}],
    summaries: []
  })

["A-1001", "A-1002"] = batch.summaries |> Enum.map(& &1.id) |> Enum.sort()
true = Enum.all?(batch.summaries, &(byte_size(&1.text) > 0))

A worker can instead be an agent with tools. Bound task count and concurrency at the application boundary so a model-generated plan cannot create unbounded provider work. Subagents offers model-selected delegation; Send expresses application-selected work.

Evaluator-optimizer

A writer produces a candidate; an evaluator returns an explicit accepted or revision-required result. A guarded back-edge carries feedback to the writer. Bound revisions with max_runs: and the whole run with a timeout.

Require an explicit accepted verdict. Malformed output or an exhausted budget must not be treated as acceptance. Preserve the final verdict alongside the candidate so downstream code knows whether it met the criteria. Structured schemas help represent verdict, feedback and evidence consistently.

Begin with an incomplete draft, validate required facts in application code, and ask a real model to revise it. Acceptance remains an explicit result:

alias BeamWeaver.Graph
alias BeamWeaver.Graph.Compiled
alias BeamWeaver.Core.{ChatModel, Message}
model = BeamWeaver.Models.init_chat_model!("openai:gpt-5.4-mini", max_tokens: 256, timeout: 60_000)

improve =
  Graph.new()
  |> Graph.add_node(:evaluate, fn %{draft: draft} ->
    accepted = draft =~ "A-1001" and String.downcase(draft) =~ "thursday"
    %{verdict: if(accepted, do: :accept, else: :revise), feedback: "Include order A-1001 and delivery Thursday."}
  end)
  |> Graph.add_node(
    :write,
    fn state ->
      with {:ok, reply} <-
             ChatModel.invoke(
               model,
               "Revise this customer reply using only these facts: A-1001 arrives Thursday. #{state.feedback} Draft: #{state.draft}"
             ),
           do: %{draft: Message.text(reply)}
    end,
    timeout: 70_000
  )
  |> Graph.add_edge(Graph.start(), :evaluate)
  |> Graph.add_edge(:evaluate, :write, when: %{verdict: :revise}, max_runs: 3)
  |> Graph.add_edge(:evaluate, Graph.end_node(), default: true)
  |> Graph.add_edge(:write, :evaluate)
  |> Graph.compile!()

{:ok, result} = Compiled.invoke(improve, %{draft: "Your order has shipped."}, run_timeout: 240_000)
:accept = result.verdict
true = result.draft =~ "A-1001"
true = String.downcase(result.draft) =~ "thursday"

An agent is a graph

An agent coordinates model turns and tool execution with additional hooks, state management, structured output and review. Use use BeamWeaver.Agent for that standard loop. Building it by hand means owning tool correlation, error handling, persistence and termination yourself.

An agent module can be one node in a workflow. Subgraphs shows how to map the parent input and extract the agent's answer while keeping nested lifecycle behavior.

For advanced orchestration, the same model/tool cycle can be built explicitly. This example uses a real provider and a local lookup:

alias BeamWeaver.Agent.ToolNode
alias BeamWeaver.{Models, Graph}
alias BeamWeaver.Graph.Compiled
alias BeamWeaver.Core.{ChatModel, Message}

defmodule MyApp.Tools.LookupOrder do
  use BeamWeaver.Tool

  name "lookup_order"
  description "Look up the shipping status of an order by its id."

  schema do
    field :order_id, :string, description: "The order id"
  end

  @impl true
  def invoke(_tool, %{"order_id" => order_id}, _opts), do: {:ok, "Order #{order_id} shipped on Monday."}
end

model = Models.init_chat_model!("openai:gpt-5.4-mini", max_tokens: 256, timeout: 60_000)

call_model = fn %{messages: messages} ->
  {:ok, reply} = ChatModel.invoke(model, messages, tools: [MyApp.Tools.LookupOrder])
  %{messages: [reply]}
end

agent_loop =
  Graph.new(state_schema: BeamWeaver.Graph.Messages.state_schema())
  |> Graph.add_node(:model, call_model)
  |> Graph.add_node(:tools, ToolNode.new([MyApp.Tools.LookupOrder]))
  |> Graph.add_edge(Graph.start(), :model)
  |> Graph.add_edge(:model, :tools, when: fn %{messages: [reply]} -> reply.tool_calls != [] end)
  |> Graph.add_edge(:model, Graph.end_node(), default: true)
  |> Graph.add_edge(:tools, :model)
  |> Graph.compile!()

{:ok, state} =
  Compiled.invoke(agent_loop, %{messages: ["Use lookup_order to check A-1001, then report its shipping status."]})

true = Enum.any?(state.messages, &(&1.role == :tool and &1.name == "lookup_order"))
true = Message.text(List.last(state.messages)) =~ "Monday"

Choosing a pattern