Research Squad - AG2

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/bin/python 05_research_squad.py """ import asyncio import time from ag2 import Agent from ag2.agent import TaskConfig from ag2.config import GeminiConfig from ag2.events import TaskCompleted, TaskStarted from ag2.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())


<br> 1<br> 2<br> 3<br> 4<br> 5<br> 6<br> 7<br> 8<br> 9<br> 10<br> 11<br> 12<br> 13<br> 14<br> 15<br> 16<br> 17<br> 18<br> 19<br> 20<br> 21<br> 22<br> 23<br> 24<br> 25<br> 26<br> 27<br> 28<br> 29<br> 30<br> 31<br> 32<br> 33<br> 34<br> 35<br> 36<br> 37<br> 38<br> 39<br> 40<br> 41<br> 42<br> 43<br> 44<br> 45<br> 46<br> 47<br> 48<br> 49<br> 50<br> 51<br> 52<br> 53<br> 54<br> 55<br> 56<br> 57<br> 58<br> 59<br> 60<br> 61<br> 62<br> 63<br> 64<br> 65<br> 66<br> 67<br> 68<br> 69<br> 70<br> 71<br> 72<br> 73<br> 74<br> 75<br> 76<br> 77<br> 78<br> 79<br> 80<br> 81<br> 82<br> 83<br> 84<br> 85<br> 86<br> 87<br> 88<br> 89<br> 90<br> 91<br> 92<br> 93<br> 94<br> 95<br> 96<br> 97<br> 98<br> 99<br>100<br>101<br>	<br>"""05 · Research squad — parallel subtasks and sibling delegation<br>Two patterns for multi-Agent orchestration:<br>1. Opt-in subtask tools. Pass ``tasks=TaskConfig(...)`` and the Agent<br> gains ``run_subtask`` / ``run_subtasks``. The coordinator uses<br> ``run_subtasks`` with ``parallel=True`` to fan out three short<br> investigations concurrently. Spawned subtasks have no ``run_subtask``<br> tools (they default to ``tasks=False``), so recursion is structurally<br> impossible — no depth limiter needed.<br>2. ``Agent.as_tool()``. A second Agent (``math_expert``) is exposed to<br> the coordinator as a callable tool. The wrapped Agent has no<br> ``run_subtask`` tools either (default), so recursion is bounded by the<br> call structure.<br>Run::<br> .venv/bin/python 05_research_squad.py<br>"""<br>import asyncio<br>import time<br>from ag2 import Agent<br>from ag2.agent import TaskConfig<br>from ag2.config import GeminiConfig<br>from ag2.events import TaskCompleted, TaskStarted<br>from ag2.stream import MemoryStream<br>def section(title: str) -> None:<br> print(f"\n── {title} ───")<br>async def main() -> None:<br> config = GeminiConfig(model="gemini-3-flash-preview", temperature=0)<br> section("Parallel subtasks — fan out three lookups in one tool call")<br> coordinator = Agent(<br> "coordinator",<br> prompt=(<br> "You answer multi-part questions by dispatching run_subtasks "<br> "with parallel=True. Use one tool call with every sub-question "<br> "packed into the 'tasks' list. Be concise."<br> ),<br> config=config,<br> tasks=TaskConfig(), # Opt in to run_subtask / run_subtasks.<br> )<br> # Collect subtask lifecycle events so we can show the fan-out to the user<br> starts: list[TaskStarted] = []<br> completions: list[TaskCompleted] = []<br> stream = MemoryStream()<br> stream.where(TaskStarted).subscribe(lambda e: starts.append(e))<br> stream.where(TaskCompleted).subscribe(lambda e: completions.append(e))<br> start = time.monotonic()<br> reply = await coordinator.ask(<br> "Use run_subtasks(parallel=True) to answer, in one tool call: "<br> "(a) what is the tallest waterfall in the world, "<br> "(b) what year was the Eiffel Tower completed, "<br> "(c) what is the boiling point of nitrogen in Celsius. "<br> "Then list all three answers.",<br> stream=stream,<br> )<br> elapsed = time.monotonic() - start<br> print(reply.body)<br> print()<br> print(f"Subtasks dispatched: {len(starts)}")<br> print(f"Subtasks finished: {len(completions)}")<br> print(f"Wall time: {elapsed:.2f}s (3 concurrent LLM calls)")<br> section("Sibling delegation — math_expert is a tool on coordinator2")<br> math_expert = Agent(<br> "math-expert",<br> prompt="You are an arithmetic specialist. Reply with only the number.",<br> config=config,<br> )<br> coordinator2 = Agent(<br> "coordinator2",<br> prompt=(<br> "When arithmetic comes up, delegate to the task_math-expert tool "<br> "rather than computing yourself. Then present the answer in a "<br> "complete sentence."<br> ),<br> config=config,<br> tools=[<br> math_expert.as_tool(<br> description="Delegate arithmetic problems to the math expert.",<br> )<br> ],<br> )<br> reply2 = await coordinator2.ask("What is 237 times 19?")<br> print(reply2.body)<br>if __name__ == "__main__":<br> asyncio.run(main())<br>