If your users interact directly with a Workflow, it is often useful to make it a Conversational Workflow. Users will then be able to chat with the Workflow,
in the same way you’d interact with an Agent or a Team.This feature allows you to add a WorkflowAgent to your workflow that intelligently decides whether to:
Answer directly based on the current input and past workflow results
Run the workflow when the input cannot be answered based on past results
What is a WorkflowAgent?WorkflowAgent is a restricted version of the Agent class specifically designed for workflow orchestration.
This is how you can add a WorkflowAgent to your workflow:
Copy
Ask AI
from agno.workflow import WorkflowAgentfrom agno.workflow.workflow import Workflowfrom agno.models.openai import OpenAIChatworkflow_agent = WorkflowAgent( model=OpenAIChat(id="gpt-4o-mini"), # Set the model that should be used num_history_runs=4 # How many of the previous runs should it take into account)workflow = Workflow( name="Story Generation Workflow", description="A workflow that generates stories, formats them, and adds references", agent=workflow_agent,)
Similar to workflow history for steps, the WorkflowAgent has access to the full history of workflow runs for the current session.
This makes it possible to answer questions about previous results, compare outputs from multiple runs, and maintain conversation continuity.
How to control the number of previous runs the workflow agent can see?The num_history_runs parameter controls how many previous workflow runs the agent can see when making decisions. This is crucial for:
Context awareness: The agent needs to see past runs to answer follow-up questions
Memory limits: Including too many runs can exceed the model context window
Performance: Fewer runs mean faster processing and lower input tokens
You can provide custom instructions to the WorkflowAgent to control its behavior. Although default instructions are provided which instruct the agent to answer directly from history
or to run the workflow when new processing is needed, you can override them by providing your own instructions.
We recommend using the default instructions unless you have a specific use case where you want the agent to answer in a particular way or include some specific information in the response.
The default instructions should be sufficient for most use cases.
Copy
Ask AI
workflow_agent = WorkflowAgent( model=OpenAIChat(id="gpt-4o-mini"), num_history_runs=4, instructions="You are a helpful assistant that can answer questions and run workflows when new processing is needed.",)
from agno.agent import Agentfrom agno.db.postgres import PostgresDbfrom agno.models.openai import OpenAIChatfrom agno.workflow import WorkflowAgentfrom agno.workflow.types import StepInputfrom agno.workflow.workflow import Workflowdb_url = "postgresql+psycopg://ai:ai@localhost:5532/ai"story_writer = Agent( model=OpenAIChat(id="gpt-4o-mini"), instructions="You are tasked with writing a 100 word story based on a given topic",)story_formatter = Agent( model=OpenAIChat(id="gpt-4o-mini"), instructions="You are tasked with breaking down a short story in prelogues, body and epilogue",)def add_references(step_input: StepInput): """Add references to the story""" previous_output = step_input.previous_step_content if isinstance(previous_output, str): return previous_output + "\n\nReferences: https://www.agno.com"# Create a WorkflowAgent that will decide when to run the workflowworkflow_agent = WorkflowAgent(model=OpenAIChat(id="gpt-4o-mini"), num_history_runs=4)# Create workflow with the WorkflowAgentworkflow = Workflow( name="Story Generation Workflow", description="A workflow that generates stories, formats them, and adds references", agent=workflow_agent, steps=[story_writer, story_formatter, add_references], db=PostgresDb(db_url),)# First call - will run the workflow (new topic)workflow.print_response( "Tell me a story about a dog named Rocky", stream=True)# Second call - will answer directly from historyworkflow.print_response( "What was Rocky's personality?", stream=True)# Third call - will run the workflow (new topic)workflow.print_response( "Now tell me a story about a cat named Luna", stream=True)# Fourth call - will answer directly from historyworkflow.print_response( "Compare Rocky and Luna", stream=True)