Streaming Deep Agents and Task Delegation with Real-Time Output

Using an LLM to call tools in a loop is the simplest form of an agent. This architecture, however, can yield agents that are "shallow" and fail to plan and act over longer, more complex tasks.

Applications like "Deep Research", "Manus" and "Claude Code" have gotten around this limitation by implementing a combination of four things: a planning tool, sub agents, access to a file system, and a detailed prompt.

DeepAgents packages that combination (a planning tool, sub-agents, a file system, a detailed prompt) into a general-purpose framework, heavily inspired by Claude Code.

What none of that solves is what it feels like to watch one run. A deep agent can spend thirty seconds planning, delegating and drafting before it says anything back, and a blank terminal for thirty seconds looks the same whether the agent is working or stuck. This post covers a streaming layer built on top of DeepAgents: token-level output, per-agent attribution, and task delegation you can watch happen instead of wait for.

What Are Deep Agents?

Deep Agents move beyond simple LLM wrappers to create sophisticated multi-agent systems. The architecture enables:

• task planning: breaking down complex requests into manageable subtasks
• hierarchical delegation: assigning work to specialized sub-agents
• context management: maintaining state across operations via middleware
• parallel execution: running multiple agents simultaneously

The rest of this post adds real-time streaming on top of that architecture.

Implementing the Streaming Layer

The key difference between traditional and streaming agents lies in how responses are handled:

pythonCopy
# Traditional approach - blocking
agent = create_deep_agent(...)
result = agent.run(task)  # Wait for complete response
print(result)

# Streaming approach - async generator
agent = StreamingDeepAgent(role=AgentRole.COORDINATOR)
async for token in agent.stream_response(task):
    print(token, end="", flush=True)  # Display each token immediately

This pattern provides immediate feedback and allows users to see the agent's reasoning unfold in real-time.

How a Task Moves Through the System

The implementation combines agent specialization, task delegation, and real-time streaming. When a complex task arrives, here is the shape it takes:

User: "Build a REST API with authentication and rate limiting"
         ↓
[Coordinator Agent] → Breaks down into subtasks
         ↓
    ┌────┴────┬──────────┬──────────┐
    ↓         ↓          ↓          ↓
Researcher  Coder   Documenter  Reviewer
(Best       (FastAPI  (API       (Security
practices)   impl.)    docs)      check)

Each agent has a specific role and optimized parameters:

pythonCopy
class StreamingDeepAgent:
    def __init__(self, role: AgentRole, temperature: float):
        # Coder uses low temperature (0.3) for consistency
        # Researcher uses higher (0.7) for exploration
        # Each agent optimized for its specific task

A Real Example: Building a REST API

The core insight is that complex tasks naturally decompose into specialized subtasks. Instead of forcing a single model to handle everything, we can create a system where:

• A Coordinator agent breaks down complex requests into subtasks
• Specialized agents (Researcher, Coder, Reviewer, Documenter) handle their own domains
• Responses stream in real-time as tokens are generated
• Tasks execute in parallel when dependencies allow

It's less about dividing work or optimizing throughput than about matching the structure of the agent system to the structure of the problem.

The Spinner Problem

Real-time streaming substantially changes how it feels to interact with agent systems and LLMs.

Instead of staring at a loading spinner for 30 seconds, users can see attributed output arrive as agents draft code, report progress, and produce results.

Token-Level Streaming

At the lowest level, we intercept tokens as they're generated by the language model:

pythonCopy
async def stream_response(self, prompt: str) -> AsyncIterator[str]:
    # Create callback to capture tokens
    self.stream_callback = StreamingCallback()
    self.llm.callbacks = [self.stream_callback]

    # Start generation asynchronously
    generation_task = asyncio.create_task(
        self.llm.ainvoke([HumanMessage(content=prompt)])
    )

    # Stream tokens as they arrive
    async for token in self.stream_callback.get_stream():
        yield token

    await generation_task

The key here is the async generator pattern: we yield tokens immediately as they become available rather than waiting for completion.

Async Queue for Decoupling Token Generation

Streaming callback uses an async queue to decouple token generation from consumption:

pythonCopy
class StreamingCallback(AsyncCallbackHandler):
    def __init__(self):
        self.streaming_queue = asyncio.Queue()
        self.error = None

    async def on_llm_new_token(self, token: str, **kwargs):
        await self.streaming_queue.put(token)

    async def on_llm_end(self, response, **kwargs):
        await self.streaming_queue.put(None)

    async def on_llm_error(self, error: BaseException, **kwargs):
        self.error = error
        await self.streaming_queue.put(None)

    async def get_stream(self) -> AsyncIterator[str]:
        while True:
            token = await self.streaming_queue.get()
            if token is None:  # End signal
                break
            yield token

        if self.error is not None:
            raise self.error

The completion and error hooks matter: without a sentinel, the consumer can wait forever after generation ends. The queue provides decoupling and basic backpressure, but each stream should have one consumer unless you add an explicit broadcast layer.

Task Delegation and Parallel Execution

The orchestrator's role matters here: when a complex task arrives, it doesn't split it randomly. It reads the semantic structure:

pythonCopy
async def collect_stream(stream: AsyncIterator[str]) -> str:
    chunks = []
    async for token in stream:
        chunks.append(token)
    return "".join(chunks)

async def process_complex_task(self, main_task: str):
    # Coordinator analyzes and breaks down the task
    breakdown = await collect_stream(
        coordinator.stream_response(
            f"Break down this task into subtasks: {main_task}"
        )
    )

    # Parse the coordinator's actual output into executable subtasks
    subtasks = self._create_subtasks(breakdown)

    # Execute in parallel where possible
    results = await asyncio.gather(
        self.delegate_task(subtasks[0], "Researcher"),
        self.delegate_task(subtasks[1], "Coder"),
        self.delegate_task(subtasks[2], "Documenter")
    )

When instructed, Deep Agents reason about dependencies: research might need to finish before problem-solving begins, but documentation can start in parallel with implementation.

Agent Specialization Through Prompting

Each agent has a carefully crafted system prompt that shapes its behavior:

• Coordinator: focuses on decomposition and delegation
• Researcher: prioritizes accuracy and comprehensiveness
• Coder: emphasizes clean, production-ready implementations
• Reviewer: looks for bugs, security issues and optimizations
• Documenter: creates clear, user-friendly documentation

Specialization can also extend to model parameters and context engineering. On the parameter side, code generation uses a lower temperature for consistency, while research uses a higher temperature for creative exploration.

What Delegation Actually Buys You

DeepAgents has several non-obvious benefits:

1. Cognitive Load Distribution

Just as humans work better in specialized teams, AI agents perform better when focused on specific domains. A coding agent doesn't need to worry about documentation style and a reviewer doesn't need to generate implementations.

2. Parallel Processing

When tasks are independent, they execute simultaneously, which uses computational resources more efficiently on top of the speed gain. While one agent researches best practices, another can already start drafting implementation templates.

3. Feedback Loops

The reviewer agent provides a natural feedback mechanism. Its analysis can trigger refinements in other agents' outputs, creating an iterative improvement cycle.

4. Transparency

Streaming responses with clear agent attribution make the system's execution trace easier to inspect. Users can see which agent is contributing what without treating generated output as a faithful window into private model reasoning.

What Building This Actually Requires

Four aspects of a system like this needed careful attention:

state management: tasks need persistent state to track progress, dependencies, and results. We use a simple but effective Task dataclass with status tracking and timestamps.

error handling: distributed systems fail in distributed ways. Each agent needs explicit failure handling. If the Reviewer fails, the system should surface that missing check while preserving any usable implementation output.

rate limiting: with multiple agents potentially making parallel API calls, rate limiting becomes critical. The system implements configurable parallel execution limits.

token economics: multiple agents mean multiple API calls. The system needs to balance thoroughness with token consumption, using techniques like response summarization and selective delegation.

Interactive CLI Implementation

To make these concepts accessible, I've built a command-line interface that demonstrates streaming deep agents in action. The CLI provides both interactive and direct command modes for experimenting with the architecture.

Basic Usage

bashCopy
# Start interactive mode
python cli.py

# Chat directly with a specific agent
python cli.py chat -a "Coder" -p "Write a binary search function"

# Process a complex task with automatic delegation
python cli.py task -p "Create a REST API with authentication"

Task Delegation in Action

When you give the CLI a complex task, you can watch the entire orchestration process:

bashCopy
$ python cli.py task -p "Build a user authentication system"

✓ Added agent: Coordinator (coordinator)
✓ Added agent: Researcher (researcher)
✓ Added agent: Coder (coder)
✓ Added agent: Reviewer (reviewer)
✓ Added agent: Documenter (documenter)

Processing Complex Task: Build a user authentication system

Coordinator analyzing task...
[Streams task breakdown in real-time]

→ Delegating to Researcher: Research best practices...
→ Delegating to Coder: Implement authentication logic...
→ Delegating to Documenter: Create API documentation...
→ Delegating to Reviewer: Review security implementation...

The key difference from traditional CLIs is that you see everything happening in real time: the Coordinator's analysis streams token by token, then multiple agents work in parallel, each streaming their outputs as they generate them.

Creating Custom Agents

The CLI also allows you to create agents with specific parameters:

pythonCopy
# In the CLI's interactive mode
Command: add
Agent name: DataAnalyst
Agent role: researcher
Model name: gpt-4
Temperature (0.0-1.0): 0.5
Enable streaming? [y/n]: y

This flexibility lets you experiment with different agent configurations and see how temperature, role, and model selection affect the streaming outputs.

Where To Take This Next

Building streaming deep agent systems is about recognizing that complex problems have inherent structure. By matching our computational architecture to this structure, through specialization, parallelization, and real-time feedback, we get systems that are more capable, and also easier to understand and control.

Key takeaways from this implementation:

1. Async generators enable token-by-token streaming without blocking
2. Queue-based architecture decouples generation from consumption
3. Role specialization improves output quality by focusing agents on specific domains
4. Parallel execution reduces overall response time for complex tasks

None of this is finished, and here's where it stops. The queue-based pattern works for one producer and one consumer; multiple listeners on the same agent stream, a dashboard next to a CLI, say, would need an explicit broadcast layer this implementation doesn't have. And the rate limiting is a configured concurrency cap, not a shared budget the coordinator reasons about. That's the harder version of the problem, and it's still open.