Mastering Multi-Agent Systems: Building Sub-Agent Pipelines

A sub-agent demonstration implementation with IBM Bob Introduction Large Language Model (LLM) agents excel at multi-step tasks when focused on tight scope and explicit goals. However, as tasks expand, single-agent systems suffer from context window saturation, tool confusion, and non-deterministic behavior. To solve this, modern LLM architectures employ Sub-Agents: isolated, specialized agents spawned by a parent orchestrator to solve individual sub-problems. In this post, we explore a lightweight, framework-agnostic Python implementation demonstrating how an orchestrator agent delegates complex codebase documentation generation to two distinct sub-agents. The demonstration implemented is meant to showcase the following points; - Understanding Sub-Agent Lifecycles: How isolation, strict parameter passing ( SubAgentTask ), and unified results (SubAgentResult ) prevent state corruption. - Context Partitioning ( fork_context ): Why passing forward selective context payloads improves accuracy and reduces token usage over shared context windows. - Pluggable LLM Abstractions: How to decouple agent logic from model backends (Ollama, llama.cpp, OpenAI) via factory callables. Implementation and Architecture Overview The system consists of three main roles across five core Python modules: subagents-demo/src/ ├── models.py # Shared DTOs (SubAgentTask, SubAgentResult) ├── utils.py # Abstracted LLM client factory & formatting ├── code_explorer_agent.py # Sub-Agent 1: "explore" type (read-only AST parsing) ├── documentation_writer_agent.py # Sub-Agent 2: "general" type (Markdown generation) └── orchestrator.py # Parent Agent managing lifecycle & aggregation High-Level System Flow - ## The Sub-Agent Lifecycle Every sub-agent in this pattern transitions through four strict phases, ensuring total state isolation and predictable execution: Core Implementation Details In this codebase a sub-agent is a self-contained Python object that: - receives a single, clearly bounded task from the orchestrator (parent agent), - executes that task entirely within its own execute() method, - returns its findings as a SubAgentResult dataclass, and - shares no mutable state with any other agent. This mirrors the Bob IDE sub-agent concept described at https://bob.ibm.com/docs/ide/features/subagents: “A new, independent agent is created with its own context window. Bob passes it a description of the focused task to perform. The subagent executes the task using its available tools. The subagent returns a summary of its findings back to Bob.” The demo implements two concrete sub-agents - one of the explore type (read-only codebase analysis) and one of the general type (documentation generation) - and one orchestrator that manages them sequentially. What sub-agents are NOT in this codebase They are not OS processes, threads, or coroutines. They are synchronous Python class instances called sequentially within a single process. They do not have independent network identities or message queues. They do not share Python object references - the only channel of communication is the SubAgentTask the orchestrator hands in and theSubAgentResult the sub-agent hands back. Data Transfer Objects (models.py ) Communication between agents is strictly controlled using explicit Data Transfer Objects (DTOs) rather than shared memory or global state: """ models.py - Shared data-transfer objects used by the orchestrator and both sub-agents. These lightweight dataclasses mirror the conceptual model described in the Bob sub-agents documentation: • SubAgentTask - what the orchestrator hands off to a sub-agent • SubAgentResult - what a sub-agent hands back to the orchestrator • OrchestratorReport - the unified output the orchestrator assembles """ from future import annotations from dataclasses import dataclass, field from datetime import datetime from typing import Any @dataclass class SubAgentTask: """ Encapsulates a focused, self-contained task that the orchestrator delegates to a sub-agent. Mirrors the 'description' parameter passed when Bob spawns a subagent (see: https://bob.ibm.com/docs/ide/features/subagents). Attributes ---------- agent_id : str Logical name of the sub-agent being targeted. task_type : str Short label for the category of work (e.g. "explore", "general"). description : str Full natural-language description of the work to perform. fork_context : bool When True the sub-agent receives the orchestrator's accumulated context (mirrors the fork_context flag in the Bob spawn_subagent call). payload : dict[str, Any] Arbitrary data the sub-agent may need (file paths, config, etc.). """ agent_id: str task_type: str # "explore" | "general" description: str fork_context: bool = False payload: dict[str, Any] = field(default_factory=dict) @dataclass class SubAgentResult: """ Encapsulates the summary a sub-agent returns to the orchestrator after it finishes its isolated work. Attributes ---------- agent_id : str Identifies which sub-agent produced this result. success : bool True when the sub-agent completed without error. summary : str Human-readable summary of what was accomplished (the 'results back to the main conversation' as described by the Bob documentation). data : dict[str, Any] Structured output produced by the sub-agent. error : str | None Error message when success is False. duration_seconds : float Wall-clock time the sub-agent spent on its task. tools_used : list[str] Names of tools the sub-agent invoked during execution. """ agent_id: str success: bool summary: str data: dict[str, Any] = field(default_factory=dict) error: str | None = None duration_seconds: float = 0.0 tools_used: list[str] = field(default_factory=list) @dataclass class OrchestratorReport: """ The unified output the orchestrator assembles by aggregating every sub-agent result. This is written to the output/ directory as a timestamped Markdown file. Attributes ---------- generated_at : datetime Timestamp of report generation. target_directory : str Codebase directory that was analysed. results : list[SubAgentResult] One entry per sub-agent that ran. unified_documentation : str Final Markdown documentation produced by aggregation. total_duration_seconds : float Sum of all sub-agent durations. """ generated_at: datetime target_directory: str results: list[SubAgentResult] = field(default_factory=list) unified_documentation: str = "" total_duration_seconds: float = 0.0 # ── convenience helpers ─────────────────────────────────────────────────── @property def all_succeeded(self) -> bool: return all(r.success for r in self.results) @property def failed_agents(self) -> list[str]: return [r.agent_id for r in self.results if not r.success] Read-Only Code Exploration Agent (code-explorer_agent.py ) The CodeExplorerAgent is a specialized, read-only agent (explore type). It scans directory structures, parses files using Python's native ast module, extracts structural metadata (classes, methods, imports), and requests a natural-language summary from the LLM without modifying any files: """ code_explorer_agent.py - Sub-Agent 1: "explore" type Responsibility -------------- This sub-agent is responsible for read-only codebase exploration. It mirrors the 'explore' sub-agent type described in the Bob IDE documentation: "Read-only codebase exploration, runs on a lighter model. Best for: Searching and summarising code, finding relevant files, understanding structure." (https://bob.ibm.com/docs/ide/features/subagents#subagent-types) Lifecycle (per Bob documentation) ---------------------------------- 1. REGISTRATION - The orchestrator instantiates the agent and sets its task. 2. TASK ASSIGNMENT - The orchestrator calls assign_task() with a SubAgentTask. 3. EXECUTION - execute() runs in an isolated context: it scans the target directory, collects Python file metadata, and uses the LLM to summarise the code structure. 4. RESULT RETURN - execute() returns a SubAgentResult summary back to the orchestrator, which uses it to continue the main flow. Isolation guarantee ------------------- This agent does NOT write any files and does NOT share state with the documentation-writer agent. It only returns a summary + structured data. """ def execute(self) -> SubAgentResult: if self._task is None: return SubAgentResult(agent_id=self.agent_id, success=False, summary="", error="No task assigned.") start_time = time.monotonic() tools_used = [] try: # Step a: Discover files tools_used.append("directory_scan") target_dir = Path(self._task.payload.get("target_directory", "input")) py_files = sorted(target_dir.rglob("*.py")) # Step b: AST parsing tools_used.append("ast_parser") modules = [self._parse_python_file(p) for p in py_files] # Step c: LLM summarization tools_used.append("llm_summarise") summary_text = self._llm_summarise(modules) return SubAgentResult( agent_id=self.agent_id, success=True, summary=summary_text, data={"files_found": len(py_files), "modules": modules}, duration_seconds=time.monotonic() - start_time, tools_used=tools_used, ) except Exception as exc: return SubAgentResult( agent_id=self.agent_id, success=False, summary="", error=f"{type(exc).name}: {exc}", duration_seconds=time.monotonic() - start_time, tools_used=tools_used, ) Payload-Driven Documentation Writer (documentation_writer_agent.py ) The DocumentationWriterAgent belongs to the general type. Rather than re-reading the filesystem, it processes the structured payload provided in SubAgentTask.payload (implementing fork_context=True ): """ documentation_writer_agent.py - Sub-Agent 2: "general" type Responsibility -------------- This sub-agent is responsible for generating structured Markdown documentation from the code-exploration findings produced by Sub-Agent 1. It mirrors the 'general' sub-agent type described in the Bob IDE documentation: "Full tool access, runs on the default model. Best for: Any self-contained task requiring reads, writes, or commands." (https://bob.ibm.com/docs/ide/features/subagents#subagent-