Skip to main content

​
Steps

Steps are the individual actions that make up agent workflows. Each step performs a specific operation and can depend on other steps.

​
What is a Step?

A step defines:
  • What action to perform (LLM call, code execution, memory operation, etc.)
  • When to execute (dependencies, conditions)
  • How to handle results (success, error, retry logic)
  • What data to use (inputs, parameters, previous step outputs)

​
Basic Steps

from erdo import Agent, state
from erdo.actions import llm

agent = Agent(name="data_analyzer")

# Create a step
analysis_step = agent.step(
    llm.message(
        model="claude-sonnet-4",
        query=f"Analyze this data: {state.input_data}"
    )
)

​
Step Dependencies

​
Sequential Execution

Use depends_on to control execution order: 
from erdo.actions import websearch, llm

# Step 1: Collect data
collect_step = agent.step(
    websearch.search(query=f"{state.topic} research")
)

# Step 2: Analyze (waits for collect_step)
analyze_step = agent.step(
    llm.message(
        model="claude-sonnet-4",
        query=f"Analyze this data: {collect_step.output.results}"
    ),
    depends_on=collect_step
)

​
Parallel Execution

Steps without dependencies run in parallel: 
# These run simultaneously
sales_step = agent.step(llm.message(query=f"Analyze sales: {state.sales_data}"))
marketing_step = agent.step(llm.message(query=f"Analyze marketing: {state.marketing_data}"))

# This waits for both to complete
summary_step = agent.step(
    llm.message(query="Summarize findings"),
    depends_on=[sales_step, marketing_step]
)

​
Multiple Dependencies

Steps can depend on multiple previous steps: 
validation_step = agent.step(
    llm.message(
        model="claude-sonnet-4",
        query=f"Validate results from: {data_step.output}, {api_step.output}"
    ),
    depends_on=[data_step, api_step, query_step]
)

​
Step Outputs

​
Accessing Results

Reference step outputs in subsequent steps: 
analysis_step = agent.step(
    llm.message(
        model="claude-sonnet-4",
        query=f"Analyze: {state.data}",
        response_format={
            "Type": "json_schema",
            "Schema": {
                "type": "object",
                "properties": {
                    "insights": {"type": "array"},
                    "confidence": {"type": "number"}
                }
            }
        }
    )
)

# Use outputs in next step
storage_step = agent.step(
    memory.store(memory={
        "content": analysis_step.output.insights,
        "confidence": analysis_step.output.confidence,
        "type": "analysis_result"
    }),
    depends_on=analysis_step
)

​
Output Structure

Step outputs follow a consistent structure: 
# For LLM steps
step.output.response      # The main response
step.output.model_used    # Which model was used
step.output.tokens_used   # Token consumption

# For code execution steps
step.output.result        # Execution result
step.output.stdout        # Console output
step.output.stderr        # Error output

# For memory steps
step.output.memories      # Retrieved memories
step.output.stored_id     # ID of stored memory

# For web search steps
step.output.results       # Search results
step.output.urls          # Found URLs

​
Step Configuration

​
Basic Configuration

step = agent.step(
    llm.message(model="claude-sonnet-4", query=state.query),
    depends_on=previous_step,
    running_status="Processing your request...",
    finished_status="Analysis complete!"
)

​
Visibility Settings

Control what users see: 
step = agent.step(
    llm.message(model="claude-sonnet-4", query=state.query),
    user_output_visibility="visible",    # visible, hidden
    bot_output_visibility="hidden"       # visible, hidden
)

​
Step Metadata in Result Handlers

To configure step attributes in result handlers, use StepMetadata: 
from erdo import StepMetadata

main_step = agent.step(
    llm.message(model="claude-sonnet-4", query="Analyze data")
)

# Result handler with step metadata
main_step.on(
    IsSuccess(),
    llm.message(
        model="claude-sonnet-4",
        query="Create detailed report",
        step_metadata=StepMetadata(
            key="detailed_report",
            depends_on=["main_step"],
            user_output_visibility="visible",
            running_status="Generating report..."
        )
    )
)

​
Common Patterns

​
Data Processing Pipeline

# 1. Load data
load_step = agent.step(
    codeexec.execute(code=f"load_data('{state.file_path}')")
)

# 2. Clean data
clean_step = agent.step(
    codeexec.execute(code=f"clean_data({load_step.output.result})"),
    depends_on=load_step
)

# 3. Analyze
analyze_step = agent.step(
    llm.message(
        model="claude-sonnet-4",
        query=f"Analyze this cleaned data: {clean_step.output.result}"
    ),
    depends_on=clean_step
)

​
Research and Analysis

# Research phase
search_step = agent.step(
    websearch.search(query=f"{state.topic} latest research")
)

# Analysis phase
analysis_step = agent.step(
    llm.message(
        model="claude-sonnet-4",
        query=f"Analyze research findings: {search_step.output.results}"
    ),
    depends_on=search_step
)

# Storage phase
storage_step = agent.step(
    memory.store(memory={
        "content": analysis_step.output.response,
        "type": "research_analysis",
        "tags": [state.topic, "research"]
    }),
    depends_on=analysis_step
)

​
Validation Workflow

# Process data
process_step = agent.step(
    llm.message(model="claude-sonnet-4", query=f"Process: {state.input}")
)

# Validate results
validate_step = agent.step(
    llm.message(
        model="claude-sonnet-4",
        query=f"Validate this processing result: {process_step.output.response}"
    ),
    depends_on=process_step
)

# Store if valid
validate_step.on(
    IsSuccess() & TextContains("valid", "true"),
    memory.store(memory={
        "content": process_step.output.response,
        "type": "validated_result"
    })
)

​
Code Execution with External Files

For steps that run Python code from external files, use the @agent.exec decorator: 
from erdo.types import PythonFile

@agent.exec(
    code_files=[PythonFile(filename="analysis_files/analyze.py")]
)
def analyze_data():
    """Execute analysis using external code files."""
    from analysis_files.analyze import analyze_data
    return analyze_data(context)

# Works with dependencies like regular steps
next_step = agent.step(
    llm.message(query=f"Summarize: {analyze_data.output.result}"),
    depends_on=analyze_data
)
See the Code Execution Guide for detailed examples and file organization patterns.

​
Step Lifecycle

  1. Created: Step is defined but not yet executed
  2. Waiting: Step is waiting for dependencies to complete
  3. Running: Step is currently executing
  4. Completed: Step finished successfully
  5. Failed: Step encountered an error

​
Best Practices

  1. Clear Dependencies: Use depends_on to control execution order
  2. Meaningful Names: Give steps descriptive names for debugging
  3. Handle Outputs: Always plan how step outputs will be used
  4. Error Handling: Use result handlers for error scenarios
  5. Resource Management: Consider timeout and resource limits for long-running steps

​
Step Types by Action

Action TypePurposeOutput
llm.message()AI text generationresponse, model_used
codeexec.execute()Code executionresult, stdout, stderr
memory.store()Store datastored_id, status
memory.search()Retrieve datamemories, count
websearch.search()Web searchresults, urls
bot.invoke()Call other agentsresult, status