How AI agents plan multi-step tasks

AI agent multi-step task planning is where most agent builds either become reliable systems or fall apart under real inputs. Treating planning as a single prompt that emits a to-do list ignores dependency tracking, partial failure, and the need to replan when the world changes. This guide lays out an ordered path for engineering that process with explicit state and separation of concerns.

1. Define the task graph, not just a prompt

The first mistake is asking the model to “make a plan” and then parsing prose. You want a structured directed acyclic graph (DAG) where each node is a unit of work with explicit dependencies. This makes execution mechanical and lets you detect deadlocks, parallelize independent steps, and resume after crashes.

A planner should return something like:

{
  "nodes": [
    {"id": "fetch", "task": "Fetch order 12345 from API", "depends_on": []},
    {"id": "validate", "task": "Validate order items against inventory", "depends_on": ["fetch"]},
    {"id": "refund", "task": "Issue refund if validation fails", "depends_on": ["validate"]}
  ]
}

If you let the LLM emit natural language, you will spend the next week writing fragile regex to extract step 3 from “Next, we should probably check the inventory maybe”. For AI agent multi-step task planning, the contract between planner and executor must be typed.

Pitfall: allowing cycles. The planner will occasionally emit A depends on B and B depends on A. Validate the graph before execution.

2. Separate the planner from the executor

These are different jobs. The planner needs strong reasoning, low temperature, and often a larger context window. The executor needs fast tool-calling, cheaper inference, and higher availability. Coupling them to one model forces bad tradeoffs.

In code, treat them as separate calls:

from openai import OpenAI

client = OpenAI(base_url="https://api.n4n.ai/v1", api_key="KEY")

def plan(goal: str) -> dict:
    resp = client.chat.completions.create(
        model="anthropic/claude-3.5-sonnet",
        temperature=0.2,
        messages=[
            {"role": "system", "content": "Output a JSON task DAG. No prose."},
            {"role": "user", "content": goal}
        ]
    )
    return json.loads(resp.choices[0].message.content)

def act(node: dict, state: dict) -> dict:
    resp = client.chat.completions.create(
        model="mistralai/mixtral-8x7b-instruct",
        temperature=0,
        messages=[{"role": "system", "content": "Execute step using tools."},
                  {"role": "user", "content": f"{node['task']} | state: {state}"}]
    )
    return resp.choices[0].message.content

A gateway like n4n.ai exposes one OpenAI-compatible endpoint for 240+ models and honors routing directives, so you can send planner traffic to a reasoning model and executor traffic to a cheaper one without restructuring your client code. Automatic fallback also keeps the executor running when a provider is rate-limited.

3. Execute with a state machine, not recursion

Recursion hides control flow and blows the stack on long plans. Use an explicit loop that tracks pending nodes and only runs those whose dependencies are satisfied.

def execute(graph: dict, state: dict) -> dict:
    pending = {n["id"] for n in graph["nodes"]}
    while pending:
        ready = [n for n in graph["nodes"]
                 if n["id"] in pending
                 and all(d not in pending for d in n["depends_on"])]
        if not ready:
            raise ValueError("Deadlock: cyclic or missing dependencies")
        for node in ready:
            state[node["id"]] = act(node, state)
            pending.remove(node["id"])
    return state

This loop is trivially parallelizable: ready is your parallel batch. The tradeoff is that a plain loop lacks the observability of a workflow engine like Temporal. For most teams shipping their first agent, the loop is enough; adopt a engine later if audit trails become mandatory.

4. Handle failure with retries and compensation

Steps fail. The network blips, a tool returns 500, the schema drifts. Wrap each act call with a retry policy, but only if the operation is idempotent.

import time

def act_with_retry(node, state, max_attempts=3):
    for attempt in range(max_attempts):
        try:
            return act(node, state)
        except TransientError as e:
            if attempt == max_attempts - 1:
                raise
            time.sleep(2 ** attempt)

Non-idempotent actions—sending an email, charging a card—must use an idempotency key stored in state. Otherwise a retry duplicates side effects. For AI agent multi-step task planning, you should design compensation steps: if refund succeeds but notify fails, you may need a reverse_refund path. That is a graph edge, not an afterthought.

5. Use reflection to repair the plan

When a node fails irrecoverably, do not crash the whole agent. Capture the error and the current state, then call the planner again with a “repair” directive.

def replan_on_failure(original_goal, state, failed_node, error):
    prompt = f"Goal: {original_goal}\nState: {state}\nFailed: {failed_node}\nError: {error}\nOutput revised DAG."
    return plan(prompt)

This closes the loop on AI agent multi-step task planning: the system adapts instead of halting. The tradeoff is latency and cost—a replan is a full LLM call. Cap repairs at two attempts before escalating to a human.

6. Persist state and make it observable

In-memory state dies with the process. Write each state transition to a durable store:

def execute_persisted(graph, run_id, store):
    state = store.get(run_id, {})
    for node_id, result in run_loop(graph, state):
        store.set(f"{run_id}:{node_id}", result)
        log.info("completed", node=node_id, run=run_id)

Use Redis for speed or Postgres for relational audit. Either way, you need to answer “what did the agent do at 2am?” without scraping logs. Emit one event per node completion with the input state hash and output hash.

7. Common pitfalls and tradeoffs

• Over-decomposition. Splitting “send report” into 12 nodes creates orchestration overhead and more failure points. Keep a node at the granularity of one tool call or one decision.
• Under-specifying interfaces. If the planner emits "task": "handle customer", the executor guesses. Nodes must reference concrete tools and parameters.
• Ignoring token limits. A 50-step DAG serialized into the planner context on every repair will overflow. Summarize state before replanning.
• Trusting planner output. Always validate the JSON schema and check for cycles. A single malformed node blocks the loop.
• No fallback model. If your executor model is down and you have no secondary, the agent stalls. Use a gateway with automatic provider fallback.

AI agent multi-step task planning is not a prompting trick; it is a distributed systems problem wearing an LLM costume. Build the graph, separate concerns, persist everything, and design for the step that will inevitably fail.