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

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 (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 (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 (lines 260-292) demonstrates this pattern by checking self.stream and returning an AsyncGenerator[ChatResponse, None] when enabled.

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.

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 (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.

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.

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 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 (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.

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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}