# How to Implement Streaming Responses in AgentScope Agents: A Complete Guide

> Implement streaming responses in AgentScope agents easily. Set stream=True and use stream_printing_messages for real-time output. Get started with this complete guide.

- Repository: [AgentScope-AI/agentscope](https://github.com/agentscope-ai/agentscope)
- Tags: how-to-guide
- Published: 2026-03-09

---

**Enable streaming by setting `stream=True` when instantiating your model, then consume partial responses using `stream_printing_messages` from the AgentScope pipeline utilities.**

AgentScope is a multi-agent platform that supports real-time, chunk-by-chunk output for both text and audio. This guide explains how to implement streaming responses in AgentScope agents using the `agentscope-ai/agentscope` repository architecture, covering model-level streaming APIs and agent-level message queues.

## Understanding Streaming Architecture in AgentScope

AgentScope implements streaming at two distinct levels to maximize flexibility for developers building interactive applications.

### Model-Level Streaming

All chat models inherit from `ChatModelBase` in [`src/agentscope/model/_model_base.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/model/_model_base.py) (lines 13-21). This base class defines a `self.stream` boolean flag set during instantiation. When `stream=True`, model implementations switch from synchronous completion APIs to streaming endpoints that yield partial `ChatResponse` objects as tokens arrive.

### Agent-Level Streaming

The `AgentBase` class in [`src/agentscope/agent/_agent_base.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/agent/_agent_base.py) (lines 12-27) manages a message queue system. Agents can enable this queue using `set_msg_queue_enabled()`, allowing partial responses to be forwarded to callers immediately via `self.print()` calls, rather than waiting for the final complete message.

## Enabling Model-Level Streaming

To stream responses from an LLM, instantiate any supported model with the `stream=True` parameter. The `OpenAIChatModel` implementation in [`src/agentscope/model/_openai_model.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/model/_openai_model.py) (lines 260-292) demonstrates this pattern by checking `self.stream` and returning an `AsyncGenerator[ChatResponse, None]` when enabled.

```python
from agentscope.model import OpenAIChatModel
from agentscope.message import Msg
import asyncio

async def stream_llm():
    # Enable streaming at the model level

    model = OpenAIChatModel(
        model_name="gpt-4o-mini",
        stream=True
    )
    
    # The model yields partial responses

    async for chunk in model([Msg("user", "Explain quantum computing.", "user")]):
        print(chunk.choices[0].message.content, end="", flush=True)

asyncio.run(stream_llm())

```

## Implementing Agent-Level Streaming with Message Queues

Agents consume the model's stream and forward chunks through an internal message queue. Enable this behavior using `set_msg_queue_enabled()` before running the agent.

```python
from agentscope.agent import ReActAgent
from agentscope.model import OpenAIChatModel

# Create streaming model

llm = OpenAIChatModel("gpt-4o-mini", stream=True)

# Initialize agent

agent = ReActAgent(
    name="streaming_agent",
    llm=llm,
    tools=[]
)

# Enable the message queue for streaming output

agent.set_msg_queue_enabled(True)

```

## Consuming Streams with Pipeline Helpers

The `stream_printing_messages` function in [`src/agentscope/pipeline/_functional.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/pipeline/_functional.py) (lines 7-33) manages the asynchronous consumption of the agent's message queue. It runs the agent's coroutine while yielding each queued message as it arrives, including a boolean flag indicating whether the chunk is the final one.

```python
from agentscope.pipeline import stream_printing_messages
from agentscope.message import Msg
import asyncio

async def consume_stream():
    async for msg, is_last in stream_printing_messages(
        agents=[agent],
        coroutine_task=agent(Msg("user", "Write a haiku about autumn.", "user")),
    ):
        print(msg.content, end="", flush=True)
        if is_last:
            print("\n[Stream complete]")

asyncio.run(consume_stream())

```

## Streaming with Text-to-Speech (TTS)

AgentScope supports simultaneous streaming of text and audio when using TTS models that support streaming. Set `yield_speech=True` in `stream_printing_messages` to receive `AudioBlock` objects alongside text chunks.

```python
from agentscope.tts import OpenAITTSModel
from agentscope.pipeline import stream_printing_messages

# Enable streaming for both LLM and TTS

llm = OpenAIChatModel("gpt-4o-mini", stream=True)
tts = OpenAITTSModel("tts-1", stream=True)

agent = ReActAgent(
    name="voice_agent",
    llm=llm,
    tts_model=tts,
)

async def stream_with_audio():
    async for msg, is_last, audio in stream_printing_messages(
        agents=[agent],
        coroutine_task=agent(Msg("user", "Read this aloud.", "user")),
        yield_speech=True,
    ):
        print(msg.content, end="")
        if audio:
            # audio is an AudioBlock containing PCM or MP3 data

            print(f" [Audio chunk: {len(audio.data)} bytes]")

```

## Summary

- **Model-level streaming** is controlled by the `stream` parameter in `ChatModelBase` subclasses like `OpenAIChatModel`, yielding `AsyncGenerator[ChatResponse, None]` objects.
- **Agent-level streaming** uses `AgentBase.set_msg_queue_enabled(True)` to queue partial outputs from `self.print()` calls.
- **Pipeline consumption** requires `stream_printing_messages` from [`src/agentscope/pipeline/_functional.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/pipeline/_functional.py) to asynchronously yield messages with completion flags.
- **Audio streaming** is supported by setting `yield_speech=True` when using TTS models that implement streaming interfaces.

## Frequently Asked Questions

### How do I enable streaming for OpenAI models in AgentScope?

Instantiate `OpenAIChatModel` with `stream=True`. According to the source code in [`src/agentscope/model/_openai_model.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/model/_openai_model.py) (lines 260-292), this switches the implementation to use the async streaming endpoint, returning an async generator that yields partial `ChatResponse` chunks instead of a single complete response.

### What is the difference between model-level and agent-level streaming?

**Model-level streaming** occurs in the LLM wrapper classes (e.g., `OpenAIChatModel`) where the underlying API returns tokens incrementally. **Agent-level streaming** occurs in `AgentBase` subclasses where the agent forwards these partial results through an internal message queue via `set_msg_queue_enabled()`, allowing real-time delivery to UIs or other agents while the model is still generating.

### Can I stream both text and audio simultaneously in AgentScope?

Yes. When using a TTS model that supports streaming (such as `OpenAITTSModel` with `stream=True`), pass `yield_speech=True` to `stream_printing_messages`. The generator yields a third element containing `AudioBlock` objects alongside text chunks and completion flags, as implemented in [`src/agentscope/pipeline/_functional.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/pipeline/_functional.py).

### Which AgentScope agents support streaming responses?

All agents inheriting from `AgentBase` support streaming when configured correctly. The `ReActAgent` and other built-in agents automatically handle streaming LLM responses by calling `self.print()` for each chunk. As long as the message queue is enabled via `set_msg_queue_enabled(True)` and the underlying model has `stream=True`, any AgentScope agent can produce real-time streaming output.