Multi-Step Reasoning and Agentic Workflows: Building AI That Plans and Executes

Source: Dev.to

Quick Reference: Terms You’ll Encounter

Technical Acronyms

• DAG: Directed Acyclic Graph—workflow structure with no circular dependencies
• FSM: Finite State Machine—system with defined states and transitions
• CoT: Chain of Thought—prompting technique for step‑by‑step reasoning
• ReAct: Reasoning + Acting—pattern combining thinking with tool use
• LLM: Large Language Model—transformer‑based text generation system

Statistical & Mathematical Terms

• State: Current snapshot of all variables in a workflow
• Transition: Movement from one state to another based on conditions
• Topological Sort: Ordering DAG nodes so dependencies come first
• Idempotent: Operation that produces the same result if executed multiple times

Introduction: From Single Prompts to Orchestrated Workflows

Imagine you’re planning a cross‑country road trip. You wouldn’t just say “drive to California” and start moving. You’d:

• Decompose: Break it into legs (Chicago → Denver → Las Vegas → LA)
• Plan: Identify gas stops, hotels, attractions
• Execute: Drive each segment, adjusting for traffic and weather
• Track State: Know where you are, how much gas you have, what’s completed

Single LLM calls are like asking “how do I get to California?” You get directions, but no execution. Agentic workflows actually make the trip—handling detours, flat tires, and closed roads along the way.

Analogy 1: Single prompts are functions; agentic workflows are programs.
A function computes one thing. A program orchestrates many functions, manages state, handles errors, and produces complex outcomes.

Analogy 2: Traditional LLM use is a calculator; agentic AI is a spreadsheet.
The calculator answers one question. The spreadsheet maintains state, has interdependent cells, and updates automatically when inputs change.

Task Decomposition: Breaking Problems Apart

from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
import json

class TaskStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    BLOCKED = "blocked"

@dataclass
class Task:
    """Represents a single task in a workflow."""
    id: str
    description: str
    dependencies: List[str] = field(default_factory=list)
    status: TaskStatus = TaskStatus.PENDING
    result: Optional[Any] = None
    error: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)

@dataclass
class TaskPlan:
    """A decomposed plan with multiple tasks."""
    goal: str
    tasks: List[Task]
    created_at: str = ""

    def get_ready_tasks(self) -> List[Task]:
        """Get tasks whose dependencies are all completed."""
        completed_ids = {t.id for t in self.tasks if t.status == TaskStatus.COMPLETED}
        ready = []
        for task in self.tasks:
            if task.status == TaskStatus.PENDING:
                if all(dep in completed_ids for dep in task.dependencies):
                    ready.append(task)
        return ready

    def is_complete(self) -> bool:
        """Check if all tasks are completed."""
        return all(t.status == TaskStatus.COMPLETED for t in self.tasks)

    def has_failed(self) -> bool:
        """Check if any task has failed."""
        return any(t.status == TaskStatus.FAILED for t in self.tasks)

class TaskDecomposer:
    """
    Decompose complex goals into executable task plans.

    Two approaches:
    1. LLM‑based: Let the model break down the task
    2. Template‑based: Use predefined patterns for known task types
    """

    DECOMPOSITION_PROMPT = """Break down this goal into specific, executable tasks.

Goal: {goal}

Context: {context}

Rules:
1. Each task should be atomic (one clear action)
2. Identify dependencies between tasks
3. Tasks should be ordered logically
4. Include validation/verification tasks where appropriate

Output JSON format:
{
    "tasks": [
        {"id": "task_1", "description": "...", "dependencies": []},
        {"id": "task_2", "description": "...", "dependencies": ["task_1"]}
    ]
}

Output only valid JSON."""

    def __init__(self, llm_client):
        self.llm = llm_client

    def decompose(self, goal: str, context: str = "") -> TaskPlan:
        """Decompose a goal into tasks using LLM."""
        prompt = self.DECOMPOSITION_PROMPT.format(goal=goal, context=context)

        response = self.llm.generate(prompt)

        # Parse response
        try:
            # Handle markdown code blocks
            if "```" in response:
                response = response.split("```")[1]
                if response.startswith("json"):
                    response = response[4:]

            data = json.loads(response.strip())

            tasks = [
                Task(
                    id=t["id"],
                    description=t["description"],
                    dependencies=t.get("dependencies", [])
                )
                for t in data["tasks"]
            ]

            return TaskPlan(goal=goal, tasks=tasks)

        except (json.JSONDecodeError, KeyError) as e:
            # Fallback: single task
            return TaskPlan(
                goal=goal,
                tasks=[Task(id="task_1", description=goal)]
            )

    def decompose_with_template(
        self,
        goal: str,
        template: str
    ) -> TaskPlan:
        """Use predefined templates for common task patterns."""

        templates = {
            "research": [
                Task(id="search", description="Search for relevant sources", dependencies=[]),
                Task(id="extract", description="Extract key information", dependencies=["search"]),
                Task(id="synthesize", description="Synthesize findings", dependencies=["extract"]),
                Task(id="validate", description="Validate accuracy", dependencies=["synthesize"])
            ],
            "data_pipeline": [
                Task(id="extract", description="Extract data from source", dependencies=[]),
                Task(id="validate_input", description="Validate input data", dependencies=["extract"]),
                Task(id="transform", description="Transform data", dependencies=["validate_input"]),
                Task(id="validate_output", description="Validate output data", dependencies=["transform"]),
                Task(id="load", description="Load to destination", dependencies=["validate_output"])
            ],
            "analysis": [
                Task(id="gather", description="Gather relevant data", dependencies=[]),
                Task(id="clean", description="Clean and prepare data", dependencies=["gather"]),
                Task(id="analyze", description="Perform analysis", dependencies=["clean"]),
                Task(id="interpret", description="Interpret results", dependencies=["analyze"]),
                Task(id="report", description="Generate report", dependencies=["interpret"])
            ]
        }

        if template not in templates:
            raise ValueError(f"Unknown template: {template}")

        # Customize task descriptions with goal
        tasks = []
        for t in templates[template]:
            tasks.append(Task(
                id=t.id,
                description=f"{t.description} for: {goal}",
                dependencies=t.dependencies.copy()
            ))

        return TaskPlan(goal=goal, tasks=tasks)

# Simple LLM client interface
class LLMClient:
    """Provider‑agnostic LLM client."""

    def __init__(self, provider: str = "openai", model: str = None):
        self.provider = provider
        self.model = model or self._default_model()

    def _default_model(self) -> str:
        return {
            "openai": "gpt-4o-mini",
            "anthropic": "claude-3-haiku-20240307"
        }.get(self.provider, "gpt-4o-mini")

    def generate(self, prompt: str, temperature: float = 0) -> str:
        if self.provider == "openai":
            # Implementation would call OpenAI API
            pass
        elif self.provider == "anthropic":
            # Implementation would call Anthropic API
            pass
        else:
            raise NotImplementedError(f"Provider {self.provider} not supported")