Skip to content

05 · Research Squad

Two complementary multi-Agent patterns in one example. First, a coordinator opts in to the auto-injected run_subtasks tool (via tasks=TaskConfig()) to fan out three independent factual lookups concurrently in a single tool call. Then a second coordinator delegates arithmetic to a math_expert Agent exposed via Agent.as_tool(). Together they show "fan out then collect" alongside "named delegate".

What it covers#

  • Opting in to run_subtask / run_subtasks via tasks=TaskConfig(...) — disabled by default; subtasks themselves never get them, so recursion is impossible by construction.
  • Calling run_subtasks(parallel=True) to dispatch many sub-questions concurrently from one tool call.
  • Wrapping an Agent with Agent.as_tool() to expose it as a named delegate (task_math-expert) on a parent's tool list.
  • Subscribing to TaskStarted / TaskCompleted lifecycle events to observe the fan-out from outside the Agent.

Primitives covered#

  • Agent with tasks=TaskConfig(...) to opt in to sub-task tools
  • run_subtasks(parallel=True) for concurrent fan-out
  • Agent.as_tool(description=...) for sibling delegation
  • TaskStarted / TaskCompleted events on the stream
  • MemoryStream + stream.where(EventType).subscribe(...)

Source#

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
"""05 · Research squad — parallel subtasks and sibling delegation

Two patterns for multi-Agent orchestration:

1. **Opt-in subtask tools.** Pass ``tasks=TaskConfig(...)`` and the Agent
   gains ``run_subtask`` / ``run_subtasks``. The coordinator uses
   ``run_subtasks`` with ``parallel=True`` to fan out three short
   investigations concurrently. Spawned subtasks have **no** ``run_subtask``
   tools (they default to ``tasks=False``), so recursion is structurally
   impossible — no depth limiter needed.

2. **``Agent.as_tool()``.** A second Agent (``math_expert``) is exposed to
   the coordinator as a callable tool. The wrapped Agent has no
   ``run_subtask`` tools either (default), so recursion is bounded by the
   call structure.

Run::

    .venv-beta/bin/python 05_research_squad.py
"""

import asyncio
import time

from autogen.beta import Agent
from autogen.beta.agent import TaskConfig
from autogen.beta.config import GeminiConfig
from autogen.beta.events import TaskCompleted, TaskStarted
from autogen.beta.stream import MemoryStream

def section(title: str) -> None:
    print(f"\n── {title} ───")

async def main() -> None:
    config = GeminiConfig(model="gemini-3-flash-preview", temperature=0)

    section("Parallel subtasks — fan out three lookups in one tool call")

    coordinator = Agent(
        "coordinator",
        prompt=(
            "You answer multi-part questions by dispatching run_subtasks "
            "with parallel=True. Use one tool call with every sub-question "
            "packed into the 'tasks' list. Be concise."
        ),
        config=config,
        tasks=TaskConfig(),  # Opt in to run_subtask / run_subtasks.
    )

    # Collect subtask lifecycle events so we can show the fan-out to the user
    starts: list[TaskStarted] = []
    completions: list[TaskCompleted] = []
    stream = MemoryStream()
    stream.where(TaskStarted).subscribe(lambda e: starts.append(e))
    stream.where(TaskCompleted).subscribe(lambda e: completions.append(e))

    start = time.monotonic()
    reply = await coordinator.ask(
        "Use run_subtasks(parallel=True) to answer, in one tool call: "
        "(a) what is the tallest waterfall in the world, "
        "(b) what year was the Eiffel Tower completed, "
        "(c) what is the boiling point of nitrogen in Celsius. "
        "Then list all three answers.",
        stream=stream,
    )
    elapsed = time.monotonic() - start

    print(reply.body)
    print()
    print(f"Subtasks dispatched: {len(starts)}")
    print(f"Subtasks finished:   {len(completions)}")
    print(f"Wall time:           {elapsed:.2f}s (3 concurrent LLM calls)")

    section("Sibling delegation — math_expert is a tool on coordinator2")

    math_expert = Agent(
        "math-expert",
        prompt="You are an arithmetic specialist. Reply with only the number.",
        config=config,
    )

    coordinator2 = Agent(
        "coordinator2",
        prompt=(
            "When arithmetic comes up, delegate to the task_math-expert tool "
            "rather than computing yourself. Then present the answer in a "
            "complete sentence."
        ),
        config=config,
        tools=[
            math_expert.as_tool(
                description="Delegate arithmetic problems to the math expert.",
            )
        ],
    )

    reply2 = await coordinator2.ask("What is 237 times 19?")
    print(reply2.body)

if __name__ == "__main__":
    asyncio.run(main())