메뉴
Shows how an orchestrator agent spawns multiple sub-agents and aggregates their results. Use when building hierarchical multi-agent systems where a coordinator delegates specialised tasks to sub-agents, either sequentially or in parallel via asyncio.gather.
Codex 또는 Claude로 설치 이 Prompt를 복사해 Codex, Claude 또는 다른 어시스턴트에 붙여 넣으면 Skill 페이지를 검토하고 설치를 진행할 수 있습니다.
SOC 직업 분류 기준
| name | sub-agent-spawning |
| description | Shows how an orchestrator agent spawns multiple sub-agents and aggregates their results. Use when building hierarchical multi-agent systems where a coordinator delegates specialised tasks to sub-agents, either sequentially or in parallel via asyncio.gather. |
Use
codemap find "SymbolName"to locate any symbol before reading — it gives exact file + line range and is faster than grep across the whole repo.
AgentRunnerBase.run() can be called multiple times with different @agent
classes. An orchestrator function calls runner.run(sub_agent, prompt) for
each delegated task and collects AgentResponse.content strings.
Sequential and parallel patterns are both straightforward:
# orchestrator.py — from __future__ import annotations is safe here
from __future__ import annotations
import asyncio
from lauren_ai import agent
from lauren_ai._agents._runner import AgentRunnerBase as AgentRunner
@agent(model=None, system="You are a research specialist.")
class ResearchSubAgent: ...
@agent(model=None, system="You are a writing specialist.")
class WritingSubAgent: ...
async def spawn_sequential(runner: AgentRunner, prompts: list[tuple]) -> list[str]:
"""Run sub-agents one at a time and collect results."""
results = []
for sub_agent, prompt in prompts:
response = await runner.run(sub_agent, prompt)
results.append(response.content)
return results
async def spawn_parallel(runner: AgentRunner, prompts: list[tuple]) -> list[str]:
"""Run sub-agents concurrently and collect all results."""
tasks = [runner.run(sub_agent, prompt) for sub_agent, prompt in prompts]
responses = await asyncio.gather(*tasks)
return [r.content for r in responses]
asyncio.gatherruns all coroutines on the same event loop. BecauseMockTransportuses a shared dequeue, parallel tests must queue enough responses for every concurrent call.
from lauren_ai._agents._runner import AgentRunnerBase as AgentRunner
from lauren_ai._config import LLMConfig
from lauren_ai._transport import Completion, TokenUsage
from lauren_ai._transport._mock import MockTransport
def _make_runner(mock=None):
if mock is None:
mock = MockTransport()
cfg = LLMConfig(provider="anthropic", model="mock-model", api_key="mock")
runner = AgentRunner(transport=mock)
return runner, mock
def _completion(content="OK", *, n=1):
return Completion(
id=f"c{n}", model="mock-model", content=content,
tool_calls=[], stop_reason="end_turn",
usage=TokenUsage(input_tokens=10, output_tokens=5),
)
When sub-agents live in different AgentModules, inject via
AgentRunner[SubAgentClass]:
class OrchestratorService:
def __init__(self, research: AgentRunner[ResearchSubAgent]) -> None:
self._research = research
async def run_task(self, topic: str) -> str:
response = await self._research.run(ResearchSubAgent(), topic)
return response.content
| File | Contents |
|---|---|
src/lauren_ai/_agents/_runner.py | AgentRunnerBase.run() |
src/lauren_ai/_agents/__init__.py | @agent(), AgentResponse |
src/lauren_ai/_transport/_mock.py | MockTransport |
Adds content safety filters to lauren-ai agents using @use_guardrails for input and output checks. Use when an agent needs prompt injection protection, PII redaction, topic filtering, length limiting, or LLM-powered content evaluation. Also covers creating custom guardrail classes injectable via @guardrail.
Aggregates per-model telemetry from SignalBus events using CostTracker and a custom PerformanceMetrics collector. Use when tracking total LLM calls, token consumption, average latency, error rate, and cost per model across agent runs.
Submit agent tasks to an asyncio queue and process them with a background worker, returning results by task ID. Use when you need non-blocking agent execution in web handlers, batch processing pipelines, or any situation where callers should not wait for agent completion.
Builds AI agents with the lauren-ai framework using @agent, @use_tools, and AgentRunner. Use when defining an agent class, setting up lifecycle hooks, streaming responses, implementing agent delegation, or configuring the agentic loop with tools, guardrails, and memory.
Creates tools for lauren-ai agents using @tool() on async functions or classes. Use when implementing a @tool() function or class, working with ToolContext for DI injection, adding caching, HITL approval, or using built-in skills like WebSearchTool, HttpFetchTool, or CodeExecutionTool.
Implements an in-memory calendar management tool for agents using @tool() class-form. Use when building agents that create, query, or cancel calendar events, with ISO 8601 time handling and date-based filtering. Supports @set_metadata for tool-type annotation.