How to Build an Atomic Task Graph Agent: A Practical Tutorial

This tutorial shows the shape of an Atomic Task Graph (ATG) agent. The code is intentionally language-agnostic pseudo-Python — the point is the structure, which you can port to TypeScript, Go, or whatever your agent runtime is. The design follows the four-phase loop described in the ATG paper (arXiv:2607.01942). (ArcKit analysis)

The mental model

An ATG agent keeps three things, all as data, not as free text:

• Task nodes — each one small and atomic, with a status (pending, ready, running, done, failed).
• Edges — dependencies. A node is ready only when all its predecessors are done.
• A state store — the output of each completed node, keyed by node id, so later nodes read only the slice they need.

Step 1: Plan

Ask the model to produce a high-level plan: a list of sub-goals with the dependencies between them. Keep this coarse — you will recurse into details next.

plan = llm.plan(
    goal="Summarize Q3 churn drivers from CRM + billing",
    context=system_prompt
)
# plan.tasks = [{id, description, depends_on: [ids]}]
graph = build_graph(plan.tasks)

Step 2: Decompose to atomic

Walk the graph. For any node that is still composite (more than one action), ask the model to split it into children. Repeat until every leaf is atomic — one tool call, one check, one transformation.

def decompose(node):
    if is_atomic(node):
        return
    children = llm.decompose(node)
    for c in children:
        graph.add(c, depends_on=node.dependencies)
    graph.replace(node, children)  # node becomes a group
    for c in children:
        decompose(c)  # recurse

Step 3: Execute in dependency order

Run a scheduler that picks every ready node, executes it, stores the result, and marks dependents ready. Independent nodes run in parallel.

while graph.has_pending():
    ready = graph.ready_nodes()       # all deps done
    results = parallel_run(ready, tools)
    for node, res in zip(ready, results):
        if res.ok:
            state[node.id] = res.value
            graph.mark_done(node)
        else:
            graph.mark_failed(node)

Crucially, each executor passes a node only the state of its direct predecessors — not the whole transcript. That bounded context is what keeps a small model from getting lost.

Step 4: Repair the failed sub-graph

When a node fails, do not restart the whole task. Replan just that node (and its descendants), then resume the scheduler from there. This localized repair is ATG's main reliability lever — the paper reports hallucinated actions dropping from about 42.86% to 12.14% under graph-based control. (arXiv:2607.01942)

def on_failed(node):
    sub = subgraph_from(node)         # node + descendants
    new_plan = llm.repair(node, error=last_error, state=state)
    graph.replace(sub, new_plan)
    resume_scheduler()

Step 5: Guardrails

• Tool allowlist per node. Atomic nodes should declare which tool they may call — reject anything else to stop hallucinated actions before they run.
• Idempotency keys. Mark side-effecting nodes so re-runs during repair are safe.
• Max recursion depth. Cap decomposition so a stuck model cannot loop forever.
• Observability. Log node transitions; a graph is auditable by construction — lean into that.

What you get

Parallelism for free, recovery without starting over, and a plan you can inspect and debug. The benchmark evidence is still early (ALFWorld, July 2026, with limited independent replication), but the engineering pattern is sound: move the plan into a structure, keep context small per node, and repair locally. That is the whole idea behind an Atomic Task Graph agent.

Sources: Atomic Task Graph paper — arXiv:2607.01942. ArcKit analysis (July 22, 2026). The pseudo-code is a conceptual implementation pattern derived from the paper's described mechanism; it is not the authors' official code. Benchmark figures are as reported; independent replication is still limited. This content is educational and is not affiliated with the paper's authors.