Accessing Multiple Previous Steps

1def create_comprehensive_report(step_input: StepInput) -> StepOutput:
2    """
3    Custom function that creates a report using data from multiple previous steps.
4    This function has access to ALL previous step outputs and the original workflow message.
5    """
6
7    # Access original workflow input
8    original_topic = step_input.workflow_message or ""
9
10    # Access specific step outputs by name
11    hackernews_data = step_input.get_step_content("research_hackernews") or ""
12    web_data = step_input.get_step_content("research_web") or ""
13
14    # Or access ALL previous content
15    all_research = step_input.get_all_previous_content()
16
17    # Create a comprehensive report combining all sources
18    report = f"""
19        # Comprehensive Research Report: {original_topic}
20
21        ## Executive Summary
22        Based on research from HackerNews and web sources, here's a comprehensive analysis of {original_topic}.
23
24        ## HackerNews Insights
25        {hackernews_data[:500]}...
26
27        ## Web Research Findings  
28        {web_data[:500]}...
29    """
30
31    return StepOutput(
32        step_name="comprehensive_report", 
33        content=report.strip(), 
34        success=True
35    )
36
37# Use in workflow
38workflow = Workflow(
39    name="Enhanced Research Workflow",
40    steps=[
41        Step(name="research_hackernews", agent=hackernews_agent),
42        Step(name="research_web", agent=web_agent),
43        Step(name="comprehensive_report", executor=create_comprehensive_report),  # Accesses both previous steps
44        Step(name="final_reasoning", agent=reasoning_agent),
45    ],
46)

1def aggregator(step_input: StepInput) -> StepOutput:
2    # Access the entire Parallel group output (returns a dict)
3    parallel_data = step_input.get_step_content("parallel_processing")
4    
5    # Direct access to individual nested steps (recursive search)
6    step_a_output = step_input.get_step_output("step_a")  # Returns StepOutput object
7    step_a_content = step_input.get_step_content("step_a")  # Returns content string
8    
9    return StepOutput(
10        step_name="aggregator",
11        content=f"Step A: {step_a_content}",
12        success=True
13    )
14
15workflow = Workflow(
16    name="Parallel Access Example",
17    steps=[
18        Parallel(
19            step_a,  # Can be accessed directly by name
20            step_b,
21            step_c,
22            name="parallel_processing",
23        ),
24        Step(name="aggregator", executor=aggregator),
25    ],
26)

1def deep_nested_step(step_input: StepInput) -> StepOutput:
2    """Step nested deep inside Parallel -> Condition -> Steps."""
3    return StepOutput(step_name="deep_nested_step", content="Deep nested content", success=True)
4
5def verify_nested_access(step_input: StepInput) -> StepOutput:
6    """Verify we can access deeply nested step."""
7    nested_output = step_input.get_step_output("deep_nested_step")
8    nested_content = step_input.get_step_content("deep_nested_step")
9    
10    return StepOutput(
11        step_name="verifier",
12        content=f"Nested output found: {nested_output is not None}, Content: {nested_content}",
13        success=True
14    )
15
16def condition_evaluator(step_input: StepInput) -> bool:
17    """Evaluator for the condition."""
18    return True
19
20workflow = Workflow(
21    name="Multiple Depth Access",
22    steps=[
23        Parallel(
24            Condition(
25                name="condition_layer",
26                evaluator=condition_evaluator,
27                steps=[
28                    Steps(
29                        name="steps_layer",
30                        steps=[deep_nested_step],
31                    )
32                ],
33            ),
34            name="Parallel Processing",
35        ),
36        Step(name="verifier", executor=verify_nested_access),
37    ],
38)

1parallel_output = step_input.get_step_content("parallel_processing")
2# Returns:
3# {
4#     "step_a": "output from step_a",
5#     "step_b": "output from step_b",
6#     "step_c": "output from step_c",
7# }