agents.patterns.sequential_workflow_agent¶

Sequential Workflow Agent - Using MultiAgent with SimpleAgentV3 patterns.

This module demonstrates creating sequential multi-agent workflows using the MultiAgent as a base, with SimpleAgentV3 agents as components.

Shows various sequential patterns: 1. Simple linear workflows 2. Conditional branching workflows 3. Iterative refinement workflows 4. Pipeline-style processing

Classes¶

`ConditionalWorkflowAgent`	Conditional workflow with branching logic.
`FinalReport`	Final formatted report.
`IterativeRefinementAgent`	Iterative refinement workflow with feedback loops.
`PipelineAgent`	Pipeline-style agent for data transformation workflows.
`QualityAssessment`	Quality assessment for iterative refinement.
`ResearchBrief`	Research brief from analyzer.
`ResearchFindings`	Detailed research findings.
`SequentialWorkflowAgent`	Sequential workflow agent for multi-stage processing.

Functions¶

`create_conditional_workflow`([name, ...])	Create a conditional workflow with branching.
`create_iterative_workflow`([name, max_iterations, ...])	Create an iterative refinement workflow.
`create_pipeline`([name, debug])	Create a data processing pipeline.
`create_research_workflow`([name, stages, debug])	Create a research workflow with default stages.
`example_conditional_workflow`()	Example of conditional workflow with branching.
`example_iterative_refinement`()	Example of iterative refinement workflow.
`example_pipeline`()	Example of pipeline processing.
`example_research_workflow`()	Example of research workflow execution.

Module Contents¶

class agents.patterns.sequential_workflow_agent.ConditionalWorkflowAgent¶

Bases: SequentialWorkflowAgent

Conditional workflow with branching logic.

This variant adds conditional routing between stages based on intermediate results.

setup_agent()¶

Setup conditional workflow with routing.

Return type:: None

class agents.patterns.sequential_workflow_agent.FinalReport(/, **data)¶

Bases: pydantic.BaseModel

Final formatted report.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:: data (Any)

class agents.patterns.sequential_workflow_agent.IterativeRefinementAgent(**kwargs)¶

Bases: haive.agents.multi.agent.MultiAgent

Iterative refinement workflow with feedback loops.

This pattern implements iterative improvement through multiple passes.

Init .

class agents.patterns.sequential_workflow_agent.PipelineAgent(**kwargs)¶

Bases: haive.agents.multi.agent.MultiAgent

Pipeline-style agent for data transformation workflows.

Each stage transforms data for the next stage in a pipeline pattern.

Init .

class agents.patterns.sequential_workflow_agent.QualityAssessment(/, **data)¶

Bases: pydantic.BaseModel

Quality assessment for iterative refinement.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:: data (Any)

class agents.patterns.sequential_workflow_agent.ResearchBrief(/, **data)¶

Bases: pydantic.BaseModel

Research brief from analyzer.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:: data (Any)

class agents.patterns.sequential_workflow_agent.ResearchFindings(/, **data)¶

Bases: pydantic.BaseModel

Detailed research findings.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:: data (Any)

class agents.patterns.sequential_workflow_agent.SequentialWorkflowAgent¶

Bases: haive.agents.multi.agent.MultiAgent

Sequential workflow agent for multi-stage processing.

This agent orchestrates a sequence of SimpleAgentV3 agents to accomplish complex tasks through staged processing.

Examples

>>> workflow = SequentialWorkflowAgent(
...     name="research_pipeline",
...     stages=["analyze", "research", "synthesize", "format"],
...     debug=True
... )
>>> report = await workflow.arun("Research AI ethics implications")

setup_agent()¶

Setup workflow stages as SimpleAgentV3 instances.

Return type:: None

agents.patterns.sequential_workflow_agent.create_conditional_workflow(name='conditional_workflow', routing_conditions=None, debug=True)¶

Create a conditional workflow with branching.

Parameters:

name (str)
routing_conditions (dict[str, callable] | None)
debug (bool)

Return type:

ConditionalWorkflowAgent

agents.patterns.sequential_workflow_agent.create_iterative_workflow(name='iterative_workflow', max_iterations=3, quality_threshold=0.85, debug=True)¶

Create an iterative refinement workflow.

Parameters:

name (str)
max_iterations (int)
quality_threshold (float)
debug (bool)

Return type:

IterativeRefinementAgent

agents.patterns.sequential_workflow_agent.create_pipeline(name='data_pipeline', debug=True)¶

Create a data processing pipeline.

Parameters:

name (str)
debug (bool)

Return type:

PipelineAgent

agents.patterns.sequential_workflow_agent.create_research_workflow(name='research_workflow', stages=None, debug=True)¶

Create a research workflow with default stages.

Parameters:

name (str)
stages (list[str] | None)
debug (bool)

Return type:

SequentialWorkflowAgent

async agents.patterns.sequential_workflow_agent.example_conditional_workflow()¶: Example of conditional workflow with branching.

async agents.patterns.sequential_workflow_agent.example_iterative_refinement()¶: Example of iterative refinement workflow.

async agents.patterns.sequential_workflow_agent.example_pipeline()¶: Example of pipeline processing.

async agents.patterns.sequential_workflow_agent.example_research_workflow()¶: Example of research workflow execution.