AgentTeam Batch Runs

Run the same team many times over a list of inputs with a single call.

AgentTeam / PraisonAIAgents accepts guardrails, web, reflection, caching, learn, and knowledge kwargs for API symmetry with Agent, but does not yet apply them at the team level. Since PraisonAI PR #4004, passing a truthy value for any of the six now logs a warning telling you to set them on each individual Agent(...) instance instead. autonomy= on AgentTeam remains fully wired — this caveat does not apply to it.

Quick Start

Run the team over a list of inputs

Each input dict fills the {{placeholder}} slots in your task templates.

from praisonaiagents import Agent, AgentTeam, Task

writer = Agent(
    name="Writer",
    instructions="Write a short, factual bio"
)

bio_task = Task(
    description="Write a short bio for {{name}}",
    expected_output="A 2-3 sentence bio for {{name}}",
    agent=writer,
)

team = AgentTeam(agents=[writer], tasks=[bio_task], process="sequential")

batch = team.start_for_each(inputs=[
    {"name": "Ada Lovelace"},
    {"name": "Alan Turing"},
])

print(batch["succeeded"], "/", batch["total"])
for item in batch["items"]:
    print(item["index"], item["input"]["name"], "→", item["success"])

Run it asynchronously

astart_for_each is the async twin — same inputs, same result shape.

import asyncio
from praisonaiagents import Agent, AgentTeam, Task

writer = Agent(name="Writer", instructions="Write a short, factual bio")
task = Task(
    description="Write a short bio for {{name}}",
    expected_output="A 2-3 sentence bio for {{name}}",
    agent=writer,
)
team = AgentTeam(agents=[writer], tasks=[task])

async def main():
    batch = await team.astart_for_each(
        inputs=[{"name": "Ada"}, {"name": "Bob"}],
        on_error="continue",
    )
    for item in batch["items"]:
        print(item["index"], "ok" if item["success"] else item["error"])

asyncio.run(main())

How It Works

The team runs once per input dict, restoring its original templates afterwards so it stays reusable.

Step	What happens
Snapshot	Original `self.variables` and per-task state are saved.
Per item	The item dict merges over saved variables; task state resets so each item runs fresh.
Interpolate	`{{key}}` placeholders in `Task.description` / `Task.expected_output` are filled.
Aggregate	Per-item results collect into a single batch dict.
Restore	Variables and task templates are restored — the team is safe to reuse.

Result Shape

start_for_each returns a plain dict summarising the whole batch.

Key	Type	Description
`batch_id`	`str`	`"batch_"` + 12 hex chars from `uuid4`.
`items`	`list[dict]`	Per-input results in input order. Each has `index`, `input`, `success`, `output`, `error`.
`outputs`	`list`	Every item’s `output`, aligned with `items` (`None` for failures).
`succeeded`	`int`	Count of items with `success=True`.
`failed`	`int`	Count of items with `success=False`.
`total`	`int`	`len(items)`.
`token_usage_total`	`dict`	This team’s cumulative token usage from `PraisonAIAgents.get_token_usage_summary()` across the whole run — not batch-item-scoped. Since PraisonAI PR #4462, finalized in #4470 (fixes #4446), it is scoped to this instance’s own named agents, so concurrent teams do not leak into each other.

Each entry in items has this shape:

Key	Type	Description
`index`	`int`	0-based position in `inputs`.
`input`	`dict` \| `None`	The original input dict passed for this item.
`success`	`bool`	Whether the run completed without raising.
`output`	`Any`	Return value of `start()` / `astart()`; `None` on failure.
`error`	`str` \| `None`	`str(exception)` on failure; `None` on success.

Parameters

Parameter	Type	Default	Description
`inputs`	`list[dict]`	Required	One dict per run. Keys fill `{{placeholder}}` slots in `Task.description` / `Task.expected_output`. `None` items are treated as `{}`. Non-dict items raise `ValueError`.
`on_error`	`"continue"` \| `"fail_fast"`	`"continue"`	`continue` captures per-item errors and keeps going. `fail_fast` re-raises the first exception. Any other value raises `ValueError`.
`output`	`str`	`"silent"`	Forwarded to `start()`. Available on `start_for_each` only.
`**kwargs`	—	—	Forwarded verbatim to `start()` / `astart()`.

Error Handling

Choose whether a failing item stops the batch or is recorded and skipped.

continue (default)
fail_fast

Errors are captured per item; the batch runs to completion.

batch = team.start_for_each(
    inputs=[{"name": "Ada"}, {"name": "Bob"}],
    on_error="continue",
)

for item in batch["items"]:
    if not item["success"]:
        print("failed:", item["index"], item["error"])

The first failing item re-raises immediately.

batch = team.start_for_each(
    inputs=[{"name": "Ada"}, {"name": "Bob"}],
    on_error="fail_fast",   # re-raise on first failing item
)

Common Patterns

Generate bios for many names in one call.

from praisonaiagents import Agent, AgentTeam, Task

writer = Agent(name="Writer", instructions="Write a short, factual bio")
task = Task(
    description="Write a short bio for {{name}}",
    expected_output="A 2-3 sentence bio for {{name}}",
    agent=writer,
)
team = AgentTeam(agents=[writer], tasks=[task])

batch = team.start_for_each(inputs=[
    {"name": "Ada Lovelace"},
    {"name": "Grace Hopper"},
    {"name": "Katherine Johnson"},
])

for output in batch["outputs"]:
    print(output)

Run the same batch asynchronously.

import asyncio
from praisonaiagents import Agent, AgentTeam, Task

writer = Agent(name="Writer", instructions="Write a short, factual bio")
task = Task(
    description="Write a short bio for {{name}}",
    expected_output="A 2-3 sentence bio for {{name}}",
    agent=writer,
)
team = AgentTeam(agents=[writer], tasks=[task])

batch = asyncio.run(team.astart_for_each(
    inputs=[{"name": "Ada"}, {"name": "Grace"}],
))
print(batch["succeeded"], "/", batch["total"])

Map rows from a CSV into inputs.

import csv
from praisonaiagents import Agent, AgentTeam, Task

writer = Agent(name="Writer", instructions="Write a short, factual bio")
task = Task(
    description="Write a short bio for {{name}}",
    expected_output="A 2-3 sentence bio for {{name}}",
    agent=writer,
)
team = AgentTeam(agents=[writer], tasks=[task])

with open("people.csv", newline="") as f:
    inputs = [{"name": row["name"]} for row in csv.DictReader(f)]

batch = team.start_for_each(inputs=inputs)
print(batch["succeeded"], "of", batch["total"], "succeeded")

Concurrent-run safety

A single AgentTeam / PraisonAIAgents instance is single-run: it shares its Task objects and instance state across runs, so two runs on the same object at once would corrupt each other. Running the same instance concurrently — from two threads, or two branches of asyncio.gather — now raises RuntimeError instead of silently duplicating tasks or swapping one caller’s result for another’s.

import asyncio
from praisonaiagents import Agent, AgentTeam, Task

writer = Agent(name="Writer", instructions="Write a short, factual bio")
task = Task(description="Write a short bio for {{name}}",
            expected_output="A 2-3 sentence bio", agent=writer)
team = AgentTeam(agents=[writer], tasks=[task])

async def main():
    # ❌ Same instance twice at once → RuntimeError
    await asyncio.gather(team.astart(), team.astart())

asyncio.run(main())

Construct one instance per concurrent run instead — they are cheap to build:

import asyncio
from praisonaiagents import Agent, AgentTeam, Task

def make_team():
    writer = Agent(name="Writer", instructions="Write a short, factual bio")
    task = Task(description="Write a short bio for {{name}}",
                expected_output="A 2-3 sentence bio", agent=writer)
    return AgentTeam(agents=[writer], tasks=[task])

async def main():
    # ✅ One instance per concurrent run
    await asyncio.gather(make_team().astart(), make_team().astart())

asyncio.run(main())

start_for_each and astart_for_each are unaffected: they hold the guard for the whole batch and re-enter it per item on the same thread, so batched runs keep working exactly as before.

The exact error, so you can grep for it:

This AgentTeam instance is already running; an AgentTeam is not safe to run concurrently on the same object. Create a separate AgentTeam/PraisonAIAgents instance per concurrent run.

See Concurrency for the full list of concurrency guarantees.

Best Practices

Use double braces in templates

Placeholders use {{name}} (double braces). This runs through the team’s built-in variables interpolator, not Python’s str.format. Single braces are left untouched.

Pick on_error by job type

Use on_error="continue" for bulk jobs where one bad row shouldn’t stop the rest. Use on_error="fail_fast" for evaluation harnesses that must halt on the first failure.

token_usage_total is team-cumulative, not batch-item-scoped

token_usage_total comes from PraisonAIAgents.get_token_usage_summary() and reflects this team’s cumulative usage across the whole run (not just the batch item). Since PraisonAI PR #4462, finalized in #4470 (fixes #4446), it is scoped to this instance’s own named agents, so concurrent teams do not leak into each other. For per-item accounting, still reset the team between batches or inspect per-item outputs.

The team stays reusable

self.variables and the task templates are restored in a finally block after every batch, so you can call start_for_each again — or start() — on the same team without side effects.

AgentTeam

The team class behind batch runs.

Multi-Agent Execution

How teams execute tasks.

Dynamic Variables

How {{placeholder}} interpolation works.

Async Crew Kickoff

Running teams asynchronously.

Getting Started

Install

Core Concepts

Guides

Features

Models

Databases

Observability

Memory

Telemetry & Analytics

Knowledge

RAG

Persistence

Tools

Other Features

Developers

Configuration

Best Practices

Getting Started (No Code)

AgentTeam Batch Runs

Quick Start

How It Works

Result Shape

Parameters

Error Handling

Common Patterns

Concurrent-run safety

Best Practices

AgentTeam

Multi-Agent Execution

Dynamic Variables

Async Crew Kickoff

​Quick Start

​How It Works

​Result Shape

​Parameters

​Error Handling

​Common Patterns

​Concurrent-run safety

​Best Practices

​Related

AgentTeam

Multi-Agent Execution

Dynamic Variables

Async Crew Kickoff

Quick Start

How It Works

Result Shape

Parameters

Error Handling

Common Patterns

Concurrent-run safety

Best Practices

Related