Getting previous step content in Workflow Agno v2

[aleonsan]

Hello all,

I've been using Workflows v2 since Agno v1 released it.
I had the following running workflow example with Agno v1 & workflows v2:

from typing import Optional
from agno.workflow.v2 import Parallel, Workflow, StepInput, StepOutput, Steps, Step
from pydantic import BaseModel

from wealth_ai.agents.investments.steps import investments, investments_review


## Base model for step results
class StepResult(BaseModel):
    one: str
    two: Optional[dict] = {}
    three: Optional[list] = []

## Model to combine results from parallel steps
class CombinedResult(BaseModel):
    parallel_1: StepResult
    parallel_2: StepResult


## Steps executors
async def parallel_step_1_1(step_input: StepInput) -> StepOutput:
    result = StepResult(
        one="result from step 1.1",
        two={"key": "value"})
    return StepOutput(content=result)

async def parallel_step_1_2(step_input: StepInput) -> StepOutput:
    result = step_input.previous_step_content
    result.one += "and step 1.2"
    result.three = ["item1", "item2"]
    return StepOutput(content=result)

async def parallel_step_2_1(step_input: StepInput) -> StepOutput:
    result = StepResult(
        one="result from step 2.1",
        two={"key": "value"})
    return StepOutput(content=result)

async def parallel_step_2_2(step_input: StepInput) -> StepOutput:
    result = step_input.previous_step_content
    result.one += "and step 2.2"
    result.three = ["item1", "item2"]
    return StepOutput(content=result)

async def gather_step_results(step_input: StepInput) -> StepOutput:
    import ipdb; ipdb.set_trace()
    previous_step_content = step_input.get_step_content("parallel_steps")
    combined_result = {
        "parallel_1": previous_step_content.get("parallel_step_1_2"),
        "parallel_2": previous_step_content.get("parallel_step_2_2")
    }
    return StepOutput(content=combined_result)


## Define parallel steps and main workflow
parallel_steps_1 = Steps(
    name="parallel_steps_1",
    description="First set of parallel steps",
    steps=[
        Step(name="parallel_step_1_1", executor=parallel_step_1_1),
        Step(name="parallel_step_1_2", executor=parallel_step_1_2)
    ]
)

parallel_steps_2 = Steps(
    name="parallel_steps_2",
    description="Second set of parallel steps",
    steps=[
        Step(name="parallel_step_2_1", executor=parallel_step_2_1),
        Step(name="parallel_step_2_2", executor=parallel_step_2_2)
    ]
)

workflow = Workflow(
    name="example_workflow",
    description="Example workflow",
    steps=[
        Parallel(
            parallel_steps_1,
            parallel_steps_2,
            name="parallel_steps"
        ),
        Step(name="gather_step_results", executor=gather_step_results)
    ]
)

There, in the last step of the workflow, gather_step_results, I could access directly to the structured output (StepResult model), with no transformation, using StepInput.get_step_content, which code in Agno v1 was this.

Now in Agno v2, this implementation has changed the output, and now it is not possible to access to the original structured output but a stringified version of it, see StepInput.get_step_content code here.

This change, obviously, has broken the workflow.

So,

• Althought it is not a life or death issue, I have not found any mention in the v2 migration guide or v2 release Changelog. Is there any comment on this?
• I know I can solve the problem by providing a "stringified" version of the structured output using .model_dump() in each of the last parallel steps, but this also produces an unnecesary overhead dumping/parsing. Is there any reason for this change in StepInput.get_step_content?

Thanks for such amazing framework and for your support in advance.

[yuvarajsl]

Hi @aleonsan, thanks for reaching out and supporting Agno!
I’ve shared this with the team, we’re working through all requests one by one and will get back to you soon.
If it’s urgent, please let us know. We appreciate your patience!

[ysolanky]

Hi @aleonsan ! Thank you for the kind words.

Sorry, but I think I am missing your concern here. The return type of get_step_content fn is still Optional[Union[str, Dict[str, str]]].

That said, I do see the gap that Basemodel should also be allowed as a return type of get_step_content. But please do share more about your concern

[aleonsan]

Hello @ysolanky, thanks for your reply.

This is get_step_content current implementation, code:

As commented in my first message, my use case is to retrieve different step outputs from a Parallel step, so I'm going to focus on that:

    def get_step_content(self, step_name: str) -> Optional[Union[str, Dict[str, str]]]:

        step_output = self.get_step_output(step_name)
        if not step_output:
            return None

        # <<<< THIS IS MY CASE >>>>>> Check if this is a parallel step with nested steps
        if step_output.step_type == "Parallel" and step_output.steps:
            # Return dict with {step_name: content} for each sub-step
            parallel_content = {}
            for sub_step in step_output.steps:
                if sub_step.step_name and sub_step.content:
                    # Check if this sub-step has its own nested steps (like Condition -> Research Step)
                    if sub_step.steps and len(sub_step.steps) > 0:
                        # This is a composite step (like Condition) - get content from its nested steps
                        for nested_step in sub_step.steps:
                            if nested_step.step_name and nested_step.content:
                                # <<<< BREAKPOINT-1 This returns STRINGIFIED repr of the substep CONTENT
                                parallel_content[nested_step.step_name] = str(nested_step.content)
                    else:
                        # This is a direct step - use its content
                        # <<<< BREAKPOINT-2 This returns STRINGIFIED repr of the substep CONTENT
                        # in my example this breakpoint won't be reached
                        parallel_content[sub_step.step_name] = str(sub_step.content)
                        
            # <<<< BREAKPOINT-3 This ALSO returns STRINGIFIED repr of the substep CONTENT            
            return parallel_content if parallel_content else str(step_output.content)

        # <<<< NOT MY CASE >>>> For other nested step types (Condition, Router, Loop, Steps), get the deepest content
        elif step_output.steps and len(step_output.steps) > 0:
            # ...

        # THIS RETURNS the output of a Parallel step, nesting all subcontents but their STRINGIFIED representation
        return step_output.content  # type: ignore[return-value]

Debugging with Agno v2 from my example, I set breakpoints into get_step_content function, and after the call of the function implemented in my example:

async def gather_step_results(step_input: StepInput) -> StepOutput:
    import ipdb; ipdb.set_trace()
    previous_step_content = step_input.get_step_content("parallel_steps")
    combined_result = {
        "parallel_1": previous_step_content.get("parallel_step_1_2"),
        "parallel_2": previous_step_content.get("parallel_step_2_2")
    }
    # <<<< BREAKPOINT-4 to see results
    return StepOutput(content=combined_result)

to easily see the step-by-step execution and clarify the problem.

<<<< BREAKPOINT-1
1st iteration (first substep from parallel)

2nd iteration (second substep from parallel)

3rd iteration and 4th iteration, has the same result, summarised here for sake of simplicity.

<<<< BREAKPOINT-2
was not reached as per my comment in the code above.

<<<< BREAKPOINT-3
with all substeps executed the STRINGIFIED substep contents are returned by get_step_content into parallel_content

<<<< BREAKPOINT-4
Back from get_step_content to my code this is the output:

As you commented, there is a gap between Agno v1 and Agno v2 get_step_content implementation, as in v2 I can only get a stringified version (and no the original BaseModel type) of the substeps output content.

So, finally, summarizing

My concern here is that this was a breaking change that AFAIK is not documented in neither in the migration guide of the changelog.

Possible solution ?

For compatibility with v1, maybe it would be nice to add a flag temporarily to avoid the string casting. To be fair, It would not be an ellegant solution but it would leave the detail of how the step's outputs are represented to another "layer" which IMHO is not get_step_output responsability.

    from typing import Callable
    def get_step_content(self, step_name: str, output_format_callable: Callable[[Any], str] = str) -> Optional[Union[str, Dict[str, str]]]:
        # ... 
        # Check if this is a parallel step with nested steps
        if step_output.step_type == "Parallel" and step_output.steps:
            # Return dict with {step_name: content} for each sub-step
            parallel_content = {}
            for sub_step in step_output.steps:
                if sub_step.step_name and sub_step.content:
                    # Check if this sub-step has its own nested steps (like Condition -> Research Step)
                    if sub_step.steps and len(sub_step.steps) > 0:
                        # This is a composite step (like Condition) - get content from its nested steps
                        for nested_step in sub_step.steps:
                            if nested_step.step_name and nested_step.content:
                                ## <<<< quick proposal                            
                                nested_step_content = nested_step.content
                                if output_format_callable is not None:
                                    nested_step_content = output_format_callable(nested_step.content)
                                parallel_content[nested_step.step_name] = nested_step_content

Hope this has clarified the concern.
BR