Skip to main content
BaseWorkflow is the abstract foundation for all workflow types in the Agentic Patterns library. It manages a named registry of agents and an optional list of tools, and enforces a common run interface that subclasses must implement. Known subclasses: SequentialWorkflow, ParallelWorkflow

​
Import

from workflows.base_workflow import BaseWorkflow

​
Constructor

​
agents
Dict[str, BaseAgent]
required
A dictionary mapping agent names to agent instances. Used to look up agents by name at runtime.
agents = {
    "planner": PlanningAgent(...),
    "executor": ExecutionAgent(...)
}
​
tools
Optional[List[Any]]
default:"None"
A list of tools or skills available to workflow components. Stored as self.tools; empty list if not provided.
​
workflow_name
str
default:"BaseWorkflow"
A label used in log output to identify this workflow instance.

​
Methods

​
get_agent

Retrieves a registered agent by name. 
def get_agent(self, name: str) -> BaseAgent
​
name
str
required
The key used when the agent was registered in the agents dictionary.
​
returns
BaseAgent
The agent instance associated with the given name.
Raises ValueError if the requested name is not present in self.agents.

​
run (abstract)

The main entry point for executing the workflow. Must be implemented by every subclass. 
@abstractmethod
def run(self, task: str, **kwargs: Any) -> Any
​
task
str
required
The overarching task or objective to be executed by the workflow.
​
returns
Any
The final output of the workflow. Shape depends on the concrete subclass implementation.

​
Usage Example

from abc import abstractmethod
from typing import Any, Dict, List
from workflows.base_workflow import BaseWorkflow
from agents.base_agent import BaseAgent

class MyWorkflow(BaseWorkflow):
    def __init__(self, agents: Dict[str, BaseAgent]):
        super().__init__(
            agents=agents,
            workflow_name="MyWorkflow"
        )

    def run(self, task: str, **kwargs: Any) -> Any:
        executor = self.get_agent("executor")
        return executor.execute_step(task)

workflow = MyWorkflow(agents={"executor": ExecutionAgent(...)})
result = workflow.run("Summarize the quarterly report")

Build docs developers (and LLMs) love

Get started for freeTalk to us