From Chaos to Clarity: How the User-Driven Workflow Planner Makes Agentic AI Design Effortless

The hardest problem in agentic AI isn't building the agent — it's helping users tell the agent what they actually want.

Users know their goals. They don't know how to express them as agent workflows. The result: either overly rigid systems that can't handle real-world variation, or overly open systems that frustrate users with unpredictable behavior.

The User-Driven Workflow Planner (UDWP) is a design pattern that bridges this gap.

The Core Problem #

Traditional workflow tools require users to think in terms of:

If-then rules
Decision trees
API calls and data schemas

Most users can't — and shouldn't need to — think this way. They just know what they want to achieve:

"When a high-priority support ticket comes in, automatically triage it, assign it to the right team, and send an acknowledgment to the customer."

That's a workflow. But expressing it as a flowchart, API calls, or code is a barrier that prevents most users from unlocking the full potential of agentic AI.

The User-Driven Workflow Planner Pattern #

UDWP uses an LLM to translate natural language workflow descriptions into structured, executable agent plans — with user review and approval at each step.

Implementation #

Step 1: Intent Parsing #

The first component takes free-form user input and extracts structured intent:

from pydantic import BaseModel
from typing import Optional
 
class WorkflowIntent(BaseModel):
    trigger: str                    # What starts this workflow
    goal: str                       # What the user wants to achieve
    constraints: list[str]          # Rules and guardrails
    required_tools: list[str]       # Capabilities needed
    success_criteria: str           # How to know it worked
    human_checkpoints: list[str]    # Where to pause for approval
 
async def parse_intent(user_description: str) -> WorkflowIntent:
    prompt = f"""Extract a structured workflow intent from this description.
 
User description: "{user_description}"
 
Identify:
- trigger: What event or condition starts this workflow
- goal: The ultimate outcome the user wants  
- constraints: Any rules, limitations, or guardrails mentioned
- required_tools: What capabilities are needed (email, database, search, etc.)
- success_criteria: How will we know the workflow succeeded
- human_checkpoints: Where should a human review before proceeding
 
Return as JSON matching the WorkflowIntent schema."""
 
    response = await llm.structured_complete(prompt, schema=WorkflowIntent)
    return response

Step 2: Workflow Draft Generation #

Convert intent into a reviewable, executable plan:

class WorkflowStep(BaseModel):
    step_id: str
    name: str
    description: str
    tool: str
    inputs: dict
    outputs: list[str]
    on_failure: str
    requires_approval: bool
 
class WorkflowPlan(BaseModel):
    name: str
    description: str
    trigger: str
    steps: list[WorkflowStep]
    estimated_duration: str
    estimated_cost: str
 
async def generate_workflow_plan(intent: WorkflowIntent) -> WorkflowPlan:
    available_tools = get_available_tools()
    
    prompt = f"""Design a step-by-step workflow plan to achieve this goal.
 
Intent:
{intent.model_dump_json(indent=2)}
 
Available tools: {available_tools}
 
Design a workflow that:
1. Achieves the stated goal
2. Respects all constraints  
3. Only uses available tools
4. Has clear success/failure conditions for each step
5. Includes approval gates where specified
 
Be conservative — fewer steps with clear checkpoints is better than many automated steps."""
 
    return await llm.structured_complete(prompt, schema=WorkflowPlan)

Step 3: User Review Interface #

Present the plan in plain English before executing:

def format_plan_for_review(plan: WorkflowPlan) -> str:
    """Format workflow plan as readable English for user review."""
    lines = [
        f"# {plan.name}",
        f"\n{plan.description}",
        f"\n**Trigger:** {plan.trigger}",
        f"**Estimated time:** {plan.estimated_duration}",
        f"**Estimated cost:** {plan.estimated_cost}",
        "\n## Steps\n"
    ]
    
    for i, step in enumerate(plan.steps, 1):
        approval_note = " *(requires your approval)*" if step.requires_approval else ""
        lines.append(f"{i}. **{step.name}**{approval_note}")
        lines.append(f"   {step.description}")
        if step.on_failure:
            lines.append(f"   *If this fails: {step.on_failure}*")
        lines.append("")
    
    lines.append("\n---\nDoes this plan look correct? Reply 'yes' to run it, or describe what to change.")
    return "\n".join(lines)

Step 4: Adaptive Execution with Checkpoints #

Execute with human-in-the-loop at defined points:

class WorkflowExecutor:
    async def execute(
        self, 
        plan: WorkflowPlan,
        user_callback,  # Function to request user input
    ) -> ExecutionResult:
        
        context = {}
        
        for step in plan.steps:
            # Request approval if needed
            if step.requires_approval:
                approval = await user_callback(
                    message=f"Ready to execute: **{step.name}**\n{step.description}\n\nProceed?",
                    options=["Yes, proceed", "Skip this step", "Stop workflow"]
                )
                if approval == "Stop workflow":
                    return ExecutionResult(status="stopped", context=context)
                if approval == "Skip this step":
                    continue
            
            # Execute the step
            result = await self.execute_step(step, context)
            
            if not result.success:
                recovery = await self.handle_failure(step, result, user_callback)
                if recovery.status == "abort":
                    return ExecutionResult(status="failed", context=context, error=result.error)
            
            # Update context with outputs
            context.update(result.outputs)
        
        return ExecutionResult(status="success", context=context)

Design Principles #

1. Progressive Disclosure #

Don't overwhelm users with technical details. Start simple:

"Create a workflow that emails me when my server CPU goes above 90%"

Then offer to add complexity as users get comfortable:

Add more conditions
Include automatic remediation
Set up escalation paths

2. Reversibility by Default #

Every workflow should have an undo path:

class WorkflowStep(BaseModel):
    ...
    rollback_action: Optional[str] = None  # How to undo this step
    is_reversible: bool = True

3. Explain Before Acting #

Always show the user what will happen before it happens. Never surprise them.

4. Learn from Corrections #

When users modify a workflow, extract the preference:

async def learn_from_correction(
    original_plan: WorkflowPlan,
    corrected_plan: WorkflowPlan,
    user_explanation: str
):
    """Extract preference from user correction and store for future use."""
    preference = await extract_preference(
        original=original_plan,
        corrected=corrected_plan,
        explanation=user_explanation
    )
    await preference_store.save(user_id=current_user.id, preference=preference)

Impact on Adoption #

The UDWP pattern significantly lowers the barrier to agentic AI adoption. Users don't need to understand:

How APIs work
How to define data schemas
How to write conditional logic

They just need to describe what they want in their own words. The system handles the translation.

This is the missing link between powerful agentic AI capabilities and the users who could benefit most from them — but currently can't access them because the interface is too technical.

The future of agentic AI isn't just more capable agents. It's agents that more people can actually use.

Niteen Badgujar