AgentTeam Batch Runs

Run the same AgentTeam once per input with start_for_each

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

1. 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"])
   
2. 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)

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"])

fail_fast

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")

---

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.

---

Related

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.