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 is a Python package that combines these ideas into a general-purpose framework for building deep agents. It is heavily inspired by Claude Code.
DeepAgents provides a framework for building AI agents that can plan, delegate to sub-agents, and maintain context across complex tasks. One enhancement that can significantly improve the user experience is real-time streaming: seeing responses arrive token by token rather than waiting for complete outputs.
This post demonstrates how to implement streaming capabilities on top of DeepAgents' package with multi-agent setup, with practical code examples and architectural patterns you can apply to your own projects.
Deep Agents move beyond simple LLM wrappers to create sophisticated multi-agent systems. The architecture enables:
This implementation guide focuses on adding real-time streaming to this architecture.
The key difference between traditional and streaming agents lies in how responses are handled:
python# 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.
By design, our implementation combines agent specialization, task delegation, and real-time streaming. When processing a complex task, the flow looks like this:
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:
pythonclass 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
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:
This isn't just about dividing work or optimizing throughput — it's about matching the structure of the agent system to the structure of the 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.
At the lowest level, we intercept tokens as they're generated by the language model:
pythonasync 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.
Streaming callback uses an async queue to decouple token generation from consumption:
pythonclass 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.
The orchestrator's role is crucial. When a complex task arrives, it doesn't split it randomly — it reads the semantic structure:
pythonasync 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.
Each agent has a carefully crafted system prompt that shapes its behavior:
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.
Deepagents has several non-obvious benefits:
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.
When tasks are independent, they execute simultaneously. This isn't just about speed — it's about utilizing computational resources efficiently. While one agent researches best practices, another can already start drafting implementation templates.
The reviewer agent provides a natural feedback mechanism. Its analysis can trigger refinements in other agents' outputs, creating an iterative improvement cycle.
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.
Building a system with deepagents requires careful attention to several aspects:
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.
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.
bash# 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"
When you give the CLI a complex task, you can watch the entire orchestration process:
bash$ 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.
The CLI also allows you to create agents with specific parameters:
python# 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.
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 can create systems that are not just more capable, but more understandable and controllable.
Key takeaways from this implementation:
The shift from monolithic to orchestrated agent systems parallels the evolution we've seen in software architecture. Just as microservices revolutionized how we build scalable applications, agent orchestration is reshaping how we build AI systems.
Want to experiment with streaming deep agents? Check out our open-source implementation: github.com/dtunai/streaming-deepagents
The repository includes a complete implementation with LangChain integration, real-time streaming, parallel task execution and a CLI for interactive experimentation. The architecture described in this post is fully implemented and ready to extend for your own use cases.