AgentFlow

Step-based workflow execution with routing and context

AgentFlow enables complex multi-step pipelines with routing, parallel execution, and context sharing between steps.

AgentFlow replaces Workflow and Pipeline as the recommended class name. The old names still work as silent aliases.

Quick Start

1. Install

   npm install praisonai
   
2. Create Flow

   import { Agent, AgentFlow } from 'praisonai';
   
   const researcher = new Agent({
     instructions: 'Research the topic'
   });
   
   const writer = new Agent({
     instructions: 'Write based on research'
   });
   
   const flow = new AgentFlow('content-pipeline')
     .agent(researcher, 'Research AI trends')
     .agent(writer, 'Write article based on research');
   
   const { output } = await flow.run('AI in 2025');
   

---

How It Works

Component	Description
Steps	Individual processing units (agents or functions)
Context	Shared state between steps
Output	Final result from the last step
Results	All intermediate step results

---

Configuration Options

Step Configuration

flow.addStep({
  name: 'process',
  execute: async (input, context) => {
    return processData(input);
  },
  condition: (context) => context.get('shouldRun'),
  onError: 'retry',
  maxRetries: 3
});

Option	Type	Default	Description
name	string	required	Step identifier
execute	function	required	Async function to process input
condition	function	undefined	Condition to run the step
onError	'fail' | 'skip' | 'retry'	'fail'	Error handling behavior
maxRetries	number	0	Maximum retry attempts

---

Common Patterns

Agent Steps

import { Agent, AgentFlow } from 'praisonai';

const researcher = new Agent({ instructions: 'Research topics' });
const writer = new Agent({ instructions: 'Write content' });

const flow = new AgentFlow('research-pipeline')
  .agent(researcher, 'Research the topic')
  .agent(writer, 'Write based on research');

const { output } = await flow.run('AI trends');

Custom Steps

const flow = new AgentFlow('data-pipeline')
  .step('fetch', async (input) => {
    return await fetchData(input);
  })
  .step('process', async (data) => {
    return processData(data);
  })
  .step('save', async (result) => {
    return await saveResult(result);
  });

const { output } = await flow.run(inputData);

With Context

const flow = new AgentFlow('analysis')
  .step('analyze', async (input, context) => {
    const result = await agent.chat(input);
    context.set('analysis', result);
    return result;
  })
  .step('report', async (input, context) => {
    const analysis = context.get('analysis');
    return generateReport(analysis, input);
  });

Conditional Steps

const flow = new AgentFlow('conditional')
  .addStep({
    name: 'primary',
    execute: async (input) => primaryAgent.chat(input),
    onError: 'skip'
  })
  .addStep({
    name: 'fallback',
    condition: (context) => context.get('primary') === undefined,
    execute: async (input) => fallbackAgent.chat(input)
  });

---

Workflow Helpers

Parallel Execution

import { parallel } from 'praisonai';

const [sentiment, summary, keywords] = await parallel([
  async () => sentimentAgent.chat(document),
  async () => summaryAgent.chat(document),
  async () => keywordAgent.chat(document)
]);

Conditional Routing

import { route } from 'praisonai';

const result = await route([
  {
    condition: () => query.includes('technical'),
    execute: async () => techAgent.chat(query)
  },
  {
    condition: () => query.includes('billing'),
    execute: async () => billingAgent.chat(query)
  }
], async () => generalAgent.chat(query));  // Default fallback

Loop Pattern

import { loop } from 'praisonai';

const result = await loop(
  async (context) => improverAgent.chat(context.get('draft')),
  (result) => result.includes('DONE'),  // Condition to stop
  { maxIterations: 5 }
);

---

Best Practices

Use context for data sharing

Store intermediate results in context rather than return values.

context.set('research', researchData);
// Later steps can access via context.get('research')

Handle errors gracefully

Use onError and maxRetries for resilient flows.

.addStep({
  name: 'api-call',
  execute: async (input) => callApi(input),
  onError: 'retry',
  maxRetries: 3
})

Use conditions for branching

Skip steps conditionally based on context state.

.addStep({
  name: 'optional',
  condition: (ctx) => ctx.get('needsProcessing'),
  execute: async (input) => process(input)
})

---

Backward Compatibility

All old names work as silent aliases with no deprecation warnings.

// All of these are equivalent
import { AgentFlow, Workflow, Pipeline } from 'praisonai';

const flow1 = new AgentFlow('my-flow');
const flow2 = new Workflow('my-flow');
const flow3 = new Pipeline('my-flow');

// They are the same class
console.log(AgentFlow === Workflow);   // true
console.log(AgentFlow === Pipeline);   // true

---

Related

AgentTeam

Multi-agent orchestration

AgentOS

Deploy as web service

Workflow Hooks

Lifecycle hooks for workflows