Accessing Multiple Previous Steps

How to access multiple previous steps

Advanced workflows often require data from multiple previous steps beyond just the immediate predecessor. The StepInput object provides powerful methods to access any previous step’s output by name or retrieve all accumulated content.

Example

def create_comprehensive_report(step_input: StepInput) -> StepOutput:
    """
    Custom function that creates a report using data from multiple previous steps.
    This function has access to ALL previous step outputs and the original workflow message.
    """

    # Access original workflow input
    original_topic = step_input.workflow_message or ""

    # Access specific step outputs by name
    hackernews_data = step_input.get_step_content("research_hackernews") or ""
    web_data = step_input.get_step_content("research_web") or ""

    # Or access ALL previous content
    all_research = step_input.get_all_previous_content()

    # Create a comprehensive report combining all sources
    report = f"""
        # Comprehensive Research Report: {original_topic}

        ## Executive Summary
        Based on research from HackerNews and web sources, here's a comprehensive analysis of {original_topic}.

        ## HackerNews Insights
        {hackernews_data[:500]}...

        ## Web Research Findings  
        {web_data[:500]}...
    """

    return StepOutput(
        step_name="comprehensive_report", 
        content=report.strip(), 
        success=True
    )

# Use in workflow
workflow = Workflow(
    name="Enhanced Research Workflow",
    steps=[
        Step(name="research_hackernews", agent=hackernews_agent),
        Step(name="research_web", agent=web_agent),
        Step(name="comprehensive_report", executor=create_comprehensive_report),  # Accesses both previous steps
        Step(name="final_reasoning", agent=reasoning_agent),
    ],
)

Available Methods

• step_input.get_step_content("step_name") - Get content from specific step by name
• step_input.get_step_output("step_name") - Get the full StepOutput object from a specific step
• step_input.get_all_previous_content() - Get all previous step content combined
• step_input.workflow_message - Access the original workflow input message
• step_input.previous_step_content - Get content from immediate previous step

Accessing Nested Steps

You can directly access steps nested inside Parallel, Condition, Router, Loop, and Steps groups using their step name. The lookup performs a recursive search to find nested steps at any depth.

Direct Access to Steps in Parallel Groups

def aggregator(step_input: StepInput) -> StepOutput:
    # Access the entire Parallel group output (returns a dict)
    parallel_data = step_input.get_step_content("parallel_processing")
    
    # Direct access to individual nested steps (recursive search)
    step_a_output = step_input.get_step_output("step_a")  # Returns StepOutput object
    step_a_content = step_input.get_step_content("step_a")  # Returns content string
    
    return StepOutput(
        step_name="aggregator",
        content=f"Step A: {step_a_content}",
        success=True
    )

workflow = Workflow(
    name="Parallel Access Example",
    steps=[
        Parallel(
            step_a,  # Can be accessed directly by name
            step_b,
            step_c,
            name="parallel_processing",
        ),
        Step(name="aggregator", executor=aggregator),
    ],
)

Deeply Nested Steps

The recursive search works at any depth level:

def deep_nested_step(step_input: StepInput) -> StepOutput:
    """Step nested deep inside Parallel -> Condition -> Steps."""
    return StepOutput(step_name="deep_nested_step", content="Deep nested content", success=True)

def verify_nested_access(step_input: StepInput) -> StepOutput:
    """Verify we can access deeply nested step."""
    nested_output = step_input.get_step_output("deep_nested_step")
    nested_content = step_input.get_step_content("deep_nested_step")
    
    return StepOutput(
        step_name="verifier",
        content=f"Nested output found: {nested_output is not None}, Content: {nested_content}",
        success=True
    )

def condition_evaluator(step_input: StepInput) -> bool:
    """Evaluator for the condition."""
    return True

workflow = Workflow(
    name="Multiple Depth Access",
    steps=[
        Parallel(
            Condition(
                name="condition_layer",
                evaluator=condition_evaluator,
                steps=[
                    Steps(
                        name="steps_layer",
                        steps=[deep_nested_step],
                    )
                ],
            ),
            name="Parallel Processing",
        ),
        Step(name="verifier", executor=verify_nested_access),
    ],
)

Lookup Priority: When a step name exists at both top-level and nested within a group, the direct (top-level) lookup takes priority over the nested search.

Parallel Group Output Format

When accessing a Parallel group by its name, the output is a dictionary with each nested step’s name as the key:

parallel_output = step_input.get_step_content("parallel_processing")
# Returns:
# {
#     "step_a": "output from step_a",
#     "step_b": "output from step_b",
#     "step_c": "output from step_c",
# }

Developer Resources

• Access Multiple Previous Steps Output