Unlocking Agentic Workflows with Oban Pro

New workflow engines keep popping up everywhere lately (no, we're not linking you to them, keep your eyes up here). Some focus on chaining LLM calls, others on orchestration across APIs, and a few sprinkle in human-in-the-loop checkpoints. They're fun to explore, but most fall short on the fundamentals we think are table stakes for real workloads—self-hosting, transactional guarantees, reliability, ergonomics.

The good news? With the release of Oban v2.20 and Oban Pro v1.6, you get it all. Anything they can do, we can do better too. LLM call chaining? Dynamic expansion at runtime? Pausing for human feedback? Yes we can and did.

In fact, you can compose functional workflows, graft sub-workflows on the fly, and pause mid-flow to get a human involved—all without giving up durability and transparency from the comfort of your own app.

Cascading Workflows

Cascading workflows are undoubtedly the most useful tool in the agentic workflow toolbox. But we wrote an entire article touting the power of cascading workflows and why you should use them, so we won't bore you by repeating ourselves.

They're a dead simple way to link functions together with shared context, automatic retries, and distributed across nodes. They look like this:

def invoke(%{user_id: user_id, query: query} = source) do
  Workflow.new()
  |> Workflow.put_context(%{source: source, user_id: user_id, model: "gpt-4o-mini"})
  |> Workflow.add_cascade(:plan, &plan/1)
  |> Workflow.add_cascade(:draft, &draft/1, deps: :plan)
  |> Workflow.add_cascade(:revise, &revise/1, deps: :draft)
  |> Workflow.add_cascade(:persist, &persist/1, deps: :revise)
  |> Oban.insert_all()
end

Each of the steps is a plain elixir function that receives a cumulative context map. Bosh. That's it. On to the other new stuff!

Hold for Human

Not every workflow can, or should, be fully automated. Sometimes you need to pause to get a human in the loop. There are high-stakes, high-risk, or irreversible actions where mistakes become costly in money or trust. Or maybe you're one of those control freaks that like to verify citations before publishing a legal document.

Thanks to the recent addition of Oban.update_job/3, it's trivial to pause an in-progress workflow and await human intervention.

Imagine a marketing campaign pipeline that will draft an email, pass it off to a human for review, and deliver it upon approval. For the initial hold, we can make use of tags on the current job and start by notifying somebody that the job needs attention then enter a snooze loop to wait:

# This injects the `current_job/0` helper used to retrieve the job later
use Oban.Pro.Decorator

def hold_for_human(_context) do
  job = current_job()

  case job.tags do
    ["approved"] ->
      :ok

    ["denied"] ->
      {:cancel, "the human said so"}

    _ ->
      MyApp.Campaign.notify_a_human(job)

      {:snooze, {1, :hour}}
  end
end

Later, after the human has reviewed the details, they will approve or deny the rest of the workflow by tagging the job accordingly, then tell it to stop snoozing:

def resume_by_human(job_id, status) when status in ~w(approved denied) do
  with {:ok, job} <- Oban.update_job(job_id, &%{tags: [status | &1.tags]}) do
    Oban.retry_job(job)
  end
end

At that point, the job will complete and the rest of the workflow can continue on its merry way.

This is a powerful pattern that's useful for agent and non-agentic workflows alike.

Workflow Grafting

Grafting lets you define a placeholder in a workflow, then expand it into a sub-workflow after the workflow has already started. Those placeholders are called grafts, and the expansion is the grafting bit.

The beauty of grafting is that downstream jobs that depend on the graft will automatically depend on the expanded sub-workflow as well. That means downstream jobs won't run until both the graft and the grafted sub-workflow have completed.

Let's expand on the "human in the loop" example above to demonstrate. Imagine that we don't know the recipients of the marketing email until the workflow is already running:

def start(campaign_id) do
  Workflow.new()
  |> Workflow.put_context(%{campaign_id: campaign_id})
  |> Workflow.add_cascade(:draft, &draft_campaign/1)
  |> Workflow.add_graft(:load, &load_accounts/1, deps: :draft)
  |> Workflow.add_cascade(:finish, &finish_campaign/1, deps: [:draft, :load])
  |> Workflow.insert()
end

That builds out a cascade, and sticks a graft into the middle. When that runs, it calls load_accounts/1 to create a sub-workflow of all the notification jobs for each account dynamically:

def load_accounts(%{campaign_id: campaign_id, draft: draft}) do
  account_ids = MyApp.Campaign.recipients_for_campaign(campaign_id, draft)

  {account_ids, &notify_account/2}
  |> Workflow.apply_graft()
  |> Oban.insert_all()
end

The final job will still wait for both the load_accounts/1 job and the grafted notify_account/2 sub-workflow to complete.

So grafting allows dynamic expansion of workflows-you don't have to know up front which (or how many) jobs need to run, only that at some point during the workflow you'll figure it out.

For more involved pipelines, this type of dynamically constructed workflow is virtually impossible (or highly invasive) to manage without grafting!

The More You Know

With cascades, complex workflows become a reliable chain of functions. With human-in-the-loop holds, you can instill human expertise and comingle it with automation. And with grafting, a new type of dynamic workflow is possible sans duct-tape and black magic.

The patterns we've highlighted are necessary for agentic workflows, but they're equally powerful for regular, non-agentic workloads if "determinism" is your thang.

We're excited to see how you build your workflows. How have you worked humans into them? Has grafting changed what you build? Drop us a line and share—your patterns might inspire the next wave of features.

As usual, if you have any questions or comments, ask in the Elixir Forum. For future announcements and insight into what we're working on next, subscribe to our newsletter.