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.

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.

Result Shape

start_for_each returns a plain dict summarising the whole batch.

Each entry in items has this shape:

Parameters

Error Handling

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

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

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 and task, so batched runs keep working exactly as before.

Best Practices

Related