Streaming Responses in Production with AutoGen: Complete Guide
Building conversational AI applications with Microsoft's AutoGen framework is powerful, but when you move from prototyping to production, one feature becomes non-negotiable: streaming responses. Users expect to see tokens appear in real time, not stare at a loading spinner for thirty seconds while a multi-agent system deliberates. This guide walks through everything you need to know to implement, optimize, and deploy streaming responses with AutoGen in a production environment.
What Is Response Streaming in AutoGen?
Streaming, in the context of LLM-based applications, means delivering the model's output incrementally — token by token or chunk by chunk — as it is generated, rather than waiting for the entire response to complete before sending it to the client. AutoGen, Microsoft's multi-agent conversation framework, supports streaming through its underlying integration with OpenAI-compatible APIs and other model providers.
In a multi-agent setup, streaming becomes more nuanced. You may have agents talking to each other, and you need to decide which streams surface to the user, how to label them, and how to handle intermediate reasoning versus final answers. AutoGen's streaming support lets you tap into these conversations as they happen.
Why Streaming Matters in Production
- Perceived performance: Time-to-first-token is often more important than total generation time. Users tolerate longer responses if they see progress immediately.
- User engagement: Streaming creates a conversational feel that keeps users engaged, similar to watching someone type in a chat window.
- Early cancellation: When users see output in real time, they can stop generation early if the response is going in the wrong direction, saving compute costs.
- Better UX for long outputs: Code generation, detailed explanations, and multi-step reasoning benefit enormously from incremental delivery.
- Observability: Streaming gives you visibility into the generation process, making it easier to debug agent behavior and monitor for issues like hallucination loops.
How to Enable Streaming in AutoGen
AutoGen exposes streaming through the stream parameter on configured LLM clients and through callback hooks that fire as chunks arrive. Let's start with a basic example.
import autogen
from autogen import ConversableAgent, UserProxyAgent
# Configure the LLM with streaming enabled
config_list = [
{
"model": "gpt-4o",
"api_key": "your-api-key",
}
]
llm_config = {
"config_list": config_list,
"stream": True, # Enable streaming at the config level
}
# Create agents
user_proxy = UserProxyAgent(
name="User",
human_input_mode="NEVER",
is_termination_msg=lambda msg: "TERMINATE" in msg.get("content", ""),
)
assistant = ConversableAgent(
name="Assistant",
llm_config=llm_config,
system_message="You are a helpful coding assistant. "
"Provide clear, concise answers.",
)
Setting "stream": True in the LLM config tells AutoGen to request a streaming response from the model API. However, by default, AutoGen collects the streamed chunks internally and only surfaces the complete message. To actually process chunks as they arrive, you need to use a streaming callback.
Using Streaming Callbacks
AutoGen provides a register_streaming_callback mechanism that lets you hook into the token stream. This is the core building block for any production streaming implementation.
import autogen
from autogen import ConversableAgent
from typing import Dict, Optional, Any
# Track which agent is currently generating
current_agent_name = None
def streaming_callback(
recipient: ConversableAgent,
messages: Optional[list[Dict]] = None,
sender: Optional[ConversableAgent] = None,
config: Optional[Any] = None,
) -> tuple[bool, str]:
"""
Called for each chunk received from the LLM.
The chunk text is in config if available, or you can
access the partial content through the recipient's internal state.
"""
chunk = config.get("content", "") if isinstance(config, dict) else ""
if chunk:
# In production, you'd forward this to your WebSocket/SSE client
print(chunk, end="", flush=True)
return False, "" # Return (is_termination, response)
# Register the callback on the assistant agent
assistant = ConversableAgent(
name="Assistant",
llm_config={
"config_list": [{"model": "gpt-4o", "api_key": "your-api-key"}],
"stream": True,
},
)
assistant.register_streaming_callback(streaming_callback)
The callback receives information about the recipient agent, the message history, the sender, and a config object that contains the current chunk. Returning (False, "") tells AutoGen to continue processing; returning (True, response) would terminate the conversation.
Building a Production Streaming Server
In production, you typically serve AutoGen behind a web server and stream responses to clients over Server-Sent Events (SSE) or WebSockets. Here's a complete FastAPI example using SSE.
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import autogen
from autogen import ConversableAgent, UserProxyAgent
import json
import asyncio
import uvicorn
app = FastAPI()
class ChatRequest(BaseModel):
message: str
session_id: str = "default"
# Store active agents per session (in production, use Redis or a DB)
sessions: dict[str, dict] = {}
def get_or_create_session(session_id: str) -> dict:
if session_id not in sessions:
config_list = [{"model": "gpt-4o", "api_key": "your-api-key"}]
user_proxy = UserProxyAgent(
name="User",
human_input_mode="NEVER",
is_termination_msg=lambda msg: "TERMINATE" in msg.get("content", ""),
)
assistant = ConversableAgent(
name="Assistant",
llm_config={"config_list": config_list, "stream": True},
system_message="You are a helpful assistant. Be concise.",
)
sessions[session_id] = {
"user_proxy": user_proxy,
"assistant": assistant,
}
return sessions[session_id]
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
session = get_or_create_session(request.session_id)
user_proxy = session["user_proxy"]
assistant = session["assistant"]
# Use an asyncio Queue to bridge sync callbacks and async streaming
queue: asyncio.Queue = asyncio.Queue()
loop = asyncio.get_event_loop()
def streaming_callback(recipient, messages=None, sender=None, config=None):
chunk = ""
if isinstance(config, dict):
chunk = config.get("content", "")
if chunk:
# Thread-safe put into the async queue
asyncio.run_coroutine_threadsafe(
queue.put(("chunk", chunk)), loop
)
return False, ""
assistant.register_streaming_callback(streaming_callback)
async def event_generator():
# Run the agent conversation in a thread since AutoGen is synchronous
def run_conversation():
try:
user_proxy.initiate_chat(
assistant,
message=request.message,
clear_history=False,
)
finally:
asyncio.run_coroutine_threadsafe(
queue.put(("done", None)), loop
)
import threading
thread = threading.Thread(target=run_conversation)
thread.start()
while True:
event_type, data = await queue.get()
if event_type == "done":
yield f"data: {json.dumps({'type': 'done'})}\n\n"
break
elif event_type == "chunk":
yield f"data: {json.dumps({'type': 'chunk', 'content': data})}\n\n"
thread.join()
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no", # Disable Nginx buffering
},
)
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
This example demonstrates several production-critical patterns: session management for stateful conversations, bridging AutoGen's synchronous execution with async SSE streaming using a queue and background thread, and proper SSE formatting with event types so the client can distinguish chunks from completion signals.
Handling Multi-Agent Streaming
When multiple agents are involved, you need to track which agent is speaking so the client can label the output correctly. Here's how to handle a two-agent conversation with labeled streams.
import autogen
from autogen import ConversableAgent, UserProxyAgent, GroupChat, GroupChatManager
import json
import asyncio
async def run_multi_agent_stream(user_message: str, queue: asyncio.Queue, loop):
config_list = [{"model": "gpt-4o", "api_key": "your-api-key"}]
llm_config = {"config_list": config_list, "stream": True}
# Track the currently speaking agent
state = {"current_speaker": None}
def make_callback(agent_name: str):
def callback(recipient, messages=None, sender=None, config=None):
chunk = ""
if isinstance(config, dict):
chunk = config.get("content", "")
if chunk:
asyncio.run_coroutine_threadsafe(
queue.put({
"type": "chunk",
"agent": agent_name,
"content": chunk,
}),
loop,
)
return False, ""
return callback
coder = ConversableAgent(
name="Coder",
llm_config=llm_config,
system_message="You are a senior developer. Write clean code.",
)
reviewer = ConversableAgent(
name="Reviewer",
llm_config=llm_config,
system_message="You review code for bugs and suggest improvements.",
)
user_proxy = UserProxyAgent(
name="User",
human_input_mode="NEVER",
is_termination_msg=lambda msg: "APPROVED" in msg.get("content", ""),
)
# Register streaming callbacks per agent
coder.register_streaming_callback(make_callback("Coder"))
reviewer.register_streaming_callback(make_callback("Reviewer"))
group_chat = GroupChat(
agents=[user_proxy, coder, reviewer],
messages=[],
max_round=6,
)
manager = GroupChatManager(
groupchat=group_chat,
llm_config=llm_config,
)
user_proxy.initiate_chat(manager, message=user_message)
asyncio.run_coroutine_threadsafe(
queue.put({"type": "done"}), loop
)
Each agent gets its own callback closure that tags chunks with the agent's name. The client can then render each agent's output in a different color or panel, giving users visibility into the collaborative process.
Client-Side Consumption
Here's a minimal JavaScript client that consumes the SSE stream and renders it in the browser.
const eventSource = new EventSource('/chat/stream');
// For POST-based SSE, use fetch instead:
async function streamChat(message, sessionId) {
const response = await fetch('/chat/stream', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ message, session_id: sessionId }),
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = '';
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split('\n');
buffer = lines.pop(); // Keep incomplete line in buffer
for (const line of lines) {
if (line.startsWith('data: ')) {
const data = JSON.parse(line.slice(6));
if (data.type === 'chunk') {
// Append chunk to the UI
appendToUI(data.agent, data.content);
} else if (data.type === 'done') {
markConversationComplete();
}
}
}
}
}
Best Practices for Production Streaming
1. Implement proper error handling and reconnection. Network connections drop. Your client should handle reconnection gracefully, and your server should be able to resume or restart a conversation. Include a session ID in every request so the server can reconstruct state.
from contextlib import asynccontextmanager
@asynccontextmanager
async def error_handling_stream(session_id: str):
try:
yield
except Exception as e:
# Log the error with full context
print(f"Stream error for session {session_id}: {e}")
# Emit an error event to the client
yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
finally:
# Clean up resources
if session_id in sessions:
# Optionally persist conversation history
pass
2. Set appropriate timeouts. LLM calls can hang. Configure timeouts at multiple levels: the HTTP client, the model API call, and the SSE connection. AutoGen doesn't expose timeout configuration directly, but you can wrap calls.
import signal
class TimeoutError(Exception):
pass
def run_with_timeout(func, args=(), kwargs=None, timeout=120):
kwargs = kwargs or {}
def handler(signum, frame):
raise TimeoutError("Agent call timed out")
old_handler = signal.signal(signal.SIGALRM, handler)
signal.alarm(timeout)
try:
result = func(*args, **kwargs)
finally:
signal.alarm(0)
signal.signal(signal.SIGALRM, old_handler)
return result
# Usage
run_with_timeout(
user_proxy.initiate_chat,
kwargs={"recipient": assistant, "message": user_msg},
timeout=60,
)
3. Rate-limit and authenticate your endpoints. Streaming endpoints are expensive. Implement per-user rate limiting and require authentication. Since SSE connections are long-lived, consider connection limits per user.
from slowapi import Limiter
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
@app.post("/chat/stream")
@limiter.limit("10/minute")
async def chat_stream(request: ChatRequest, raw_request: Request):
# Verify auth token
token = raw_request.headers.get("Authorization", "")
if not verify_token(token):
raise HTTPException(status_code=401, detail="Unauthorized")
# ... rest of the handler
4. Buffer and batch small chunks. Some models emit very small chunks (single characters). Sending each one as a separate SSE event can overwhelm the network. Consider batching chunks on a time interval.
import time
class ChunkBatcher:
def __init__(self, flush_interval=0.05):
self.buffer = []
self.last_flush = time.time()
self.flush_interval = flush_interval
def add(self, chunk: str) -> str | None:
self.buffer.append(chunk)
now = time.time()
if now - self.last_flush >= self.flush_interval:
return self.flush()
return None
def flush(self) -> str | None:
if not self.buffer:
return None
combined = "".join(self.buffer)
self.buffer = []
self.last_flush = time.time()
return combined
5. Log streamed content for observability. In production, you need to know what your agents said. Log the full assembled messages after streaming completes, not just the chunks, so you have a clean audit trail.
import logging
logger = logging.getLogger("autogen_streaming")
def streaming_callback_with_logging(recipient, messages=None, sender=None, config=None):
chunk = config.get("content", "") if isinstance(config, dict) else ""
if chunk:
# Accumulate for logging
if not hasattr(recipient, "_stream_buffer"):
recipient._stream_buffer = ""
recipient._stream_buffer += chunk
# Forward to client
print(chunk, end="", flush=True)
return False, ""
# After conversation completes, log the full messages
def log_conversation(agents: list[ConversibleAgent]):
for agent in agents:
for msg in agent.chat_messages.get(agent, []):
logger.info(
"Agent=%s Role=%s Content=%s",
agent.name,
msg.get("role"),
msg.get("content", "")[:500],
)
6. Handle backpressure. If the client is slow to consume the stream, your server can build up memory. Use bounded queues and drop or pause generation if the queue fills up.
# Use a bounded queue to apply backpressure
queue = asyncio.Queue(maxsize=100)
# In the callback, check queue size
def streaming_callback(recipient, messages=None, sender=None, config=None):
chunk = config.get("content", "") if isinstance(config, dict) else ""
if chunk:
try:
# Use put_nowait to avoid blocking the agent thread
queue.put_nowait(("chunk", chunk))
except asyncio.QueueFull:
# Client is too slow — log and drop, or terminate
logger.warning("Queue full, dropping chunk for slow client")
return False, ""
7. Test streaming behavior explicitly. Streaming introduces new failure modes: partial messages, interrupted connections, and ordering issues. Write integration tests that verify your streaming pipeline end to end.
import pytest
from fastapi.testclient import TestClient
def test_streaming_response(client: TestClient):
with client.stream(
"POST",
"/chat/stream",
json={"message": "Say hello", "session_id": "test-1"},
) as response:
assert response.status_code == 200
chunks = []
for line in response.iter_lines():
if line.startswith("data: "):
data = json.loads(line[6:])
if data["type"] == "chunk":
chunks.append(data["content"])
elif data["type"] == "done":
break
full_response = "".join(chunks)
assert len(full_response) > 0
assert "hello" in full_response.lower()
Deployment Considerations
When deploying streaming AutoGen applications, your infrastructure must support long-lived connections. Standard load balancers and reverse proxies often buffer responses by default, which defeats streaming. Configure Nginx with proxy_buffering off and proxy_read_timeout set high enough for long conversations. If you're using a CDN or API gateway, ensure it supports SSE passthrough. Serverless platforms like AWS Lambda have execution time limits that may be too short for extended multi-agent conversations — consider container-based deployments on ECS, Cloud Run, or Kubernetes for production workloads.
Memory management is also critical. Each active streaming session holds agent state, conversation history, and a queue in memory. For high concurrency, implement session eviction policies, persist conversation history to a database between turns, and monitor memory usage closely. A single AutoGen session with a long conversation history can consume significant memory, especially with GPT-4o's large context window.
Conclusion
Streaming responses transform AutoGen from a powerful prototyping tool into a production-ready conversational AI platform. By enabling streaming at the config level, registering callbacks to intercept chunks, bridging synchronous agent execution with async web frameworks, and following best practices around error handling, rate limiting, backpressure, and observability, you can deliver a responsive, engaging user experience that scales. The key is to treat streaming as a first-class concern in your architecture — not an afterthought — and to test the full pipeline from model API to browser rendering under realistic load conditions before shipping to production.