Streaming Responses

If the application LLM supports streaming, the NeMo Guardrails library can stream tokens as well. Streaming is automatically enabled when you use the stream_async() method - no configuration is required.

For information about configuring streaming with output guardrails, refer to the following:

• For configuration, refer to Output Rail Streaming.
• For sample Python client code, refer to Tutorials.

Usage

Chat CLI

You can enable streaming when launching the NeMo Guardrails library chat CLI by using the --streaming option:

$
nemoguardrails chat --config=examples/configs/streaming --streaming

Python API

You can use the streaming directly from the python API in two ways:

1. Simple: receive just the chunks (tokens).
2. Full: receive both the chunks as they are generated and the full response at the end.

For the simple usage, you need to call the stream_async method on the LLMRails instance:

1
from nemoguardrails import LLMRails
2
3
app = LLMRails(config)
4
5
history = [{"role": "user", "content": "What is the capital of France?"}]
6
7
async for chunk in app.stream_async(messages=history):
8
    print(f"CHUNK: {chunk}")
9
    # Or do something else with the token

For the full usage, you need to provide a StreamingHandler instance to the generate_async method on the LLMRails instance:

1
from nemoguardrails import LLMRails
2
from nemoguardrails.streaming import StreamingHandler
3
4
app = LLMRails(config)
5
6
history = [{"role": "user", "content": "What is the capital of France?"}]
7
8
streaming_handler = StreamingHandler()
9
10
async def process_tokens():
11
    async for chunk in streaming_handler:
12
        print(f"CHUNK: {chunk}")
13
        # Or do something else with the token
14
15
asyncio.create_task(process_tokens())
16
17
result = await app.generate_async(
18
    messages=history, streaming_handler=streaming_handler
19
)
20
print(result)

Warning: Using StreamingHandler directly is deprecated and will be removed in a future release. Use stream_async() instead.

Using External Async Token Generators

You can also provide your own async generator that yields tokens, which is useful when:

• You want to use a different LLM provider that has its own streaming API.
• You have pre-generated responses that you want to stream through guardrails.
• You want to implement custom token generation logic.
• You want to test your output rails or its config in streaming mode on predefined responses without actually relying on an actual LLM generation.

To use an external generator, pass it to the generator parameter of stream_async:

1
from nemoguardrails import LLMRails
2
from typing import AsyncIterator
3
4
app = LLMRails(config)
5
6
async def my_token_generator() -> AsyncIterator[str]:
7
    # This could be from OpenAI API, Anthropic API, or any other LLM API that already has a streaming token generator. Mocking the stream here, for a simple example.
8
    tokens = ["Hello", " ", "world", "!"]
9
    for token in tokens:
10
        yield token
11
12
messages = [{"role": "user", "content": "The most famous program ever written is"}]
13
14
# use the external generator with guardrails
15
async for chunk in app.stream_async(
16
    messages=messages,
17
    generator=my_token_generator()
18
):
19
    print(f"CHUNK: {chunk}")

When using an external generator:

• The internal LLM generation is completely bypassed.
• Output rails are still applied to the LLM responses returned by the external generator, if configured.
• The generator should yield string tokens.

Example with a real LLM API:

1
async def openai_streaming_generator(messages) -> AsyncIterator[str]:
2
    """Example using OpenAI's streaming API."""
3
    import openai
4
5
    stream = await openai.ChatCompletion.create(
6
        model="gpt-4o",
7
        messages=messages,
8
        stream=True
9
    )
10
11
    # Yield tokens as they arrive
12
    async for chunk in stream:
13
        if chunk.choices[0].delta.content:
14
            yield chunk.choices[0].delta.content
15
16
config = RailsConfig.from_path("config/with_output_rails")
17
app = LLMRails(config)
18
19
async for chunk in app.stream_async(
20
    messages=[{"role": "user", "content": "Tell me a story"}],
21
    generator=openai_streaming_generator(messages)
22
):
23
    # output rails will be applied to these chunks
24
    print(chunk, end="", flush=True)

This feature enables seamless integration of the NeMo Guardrails library with any streaming LLM or token source while maintaining all the safety features of output rails.

Streaming Metadata

When using stream_async(), you can receive per-chunk metadata (provider metadata and token usage) by setting include_metadata=True:

1
async for chunk in rails.stream_async(messages=messages, include_metadata=True):
2
    print(chunk)

With include_metadata=True, each chunk is a dict with a mandatory "text" key. A chunk that carries metadata also has a "metadata" key, which can hold the following:

• provider_metadata: nonstandard fields the provider returned with the chunk, such as response_headers and vendor extensions.
• usage: normalized token counts as input_tokens, output_tokens, and total_tokens.

For text responses, the stream ends with a single terminal frame that has empty text. If the provider attaches usage to a text chunk, the library includes usage on that text frame. If usage arrives in a chunk without text, IORails folds it into the terminal frame, while LLMRails emits it on a separate empty frame before the terminal frame.

For example, an IORails stream with a final usage-only provider chunk has this shape:

1
{"text": "Hello", "metadata": {"provider_metadata": {"response_headers": {"...": "..."}}}}
2
{"text": "!", "metadata": {"provider_metadata": {"response_headers": {"...": "..."}}}}
3
{"text": "", "metadata": {
4
    "provider_metadata": {"response_headers": {"...": "..."}},
5
    "usage": {"input_tokens": 75, "output_tokens": 9, "total_tokens": 84},
6
    "response_metadata": None,
7
    "usage_metadata": None
8
}}

The empty terminal frame always includes the response_metadata and usage_metadata keys. They are retained for backward compatibility, and the internal stream_async() paths leave both as None. Read token counts from usage instead. If you use StreamingHandler directly, the handler can still push these keys and accumulate them onto the terminal frame.

An IORails tool-call response adds an assembled tool-call JSON frame after this empty frame, so the tool-call frame is the final frame in that stream.

Without include_metadata, chunks are plain strings (default behavior).

Streaming Metadata on IORails

IORails uses the same provider_metadata and usage keys as LLMRails. The following IORails-specific behaviors apply:

• IORails sets stream_options to {"include_usage": True} on the main model request so the provider returns usage on the terminal chunk. Override it through llm_params if your provider needs different behavior.

  1
  async for chunk in rails.stream_async(
  2
      messages=messages,
  3
      options={"llm_params": {"stream_options": {"include_usage": False}}},
  4
      include_metadata=True,
  5
  ):
  6
      print(chunk)

• IORails raises ValueError when you combine include_metadata=True with output-rail streaming, because its buffer strategy operates on plain string chunks. Either set include_metadata=False or disable rails.output.streaming.enabled.

On LLMRails, include_metadata=True preserves metadata only when output-rail streaming is disabled. When output-rail streaming is enabled, the buffer yields plain strings and drops provider_metadata and usage.

The library does not emit a standalone empty frame for a chunk that carries only provider metadata and no text. The metadata remains visible when the provider also includes it on a content or usage chunk. This behavior prevents repeated response headers from flooding the stream.

Generation Options and Streaming

stream_async() accepts an options argument on both engines. Unlike generate() and generate_async(), passing it does not change what the iterator yields. The chunk shape is governed solely by include_metadata: plain strings when it is False, and {"text": ..., "metadata": ...} dictionaries when it is True. Neither engine yields or returns a GenerationResponse from a stream.

This behavior applies when the engine generates the stream. On LLMRails, passing an external generator bypasses internal generation, so options and include_metadata do not change the supplied iterator. The generator determines the chunk shape, and only configured streaming output rails can wrap it.

1
# Identical chunk shapes: `options` does not select a structured return here.
2
async for chunk in rails.stream_async(messages=messages):
3
    ...  # str
4
5
async for chunk in rails.stream_async(
6
    messages=messages,
7
    options={"llm_params": {"temperature": 0.2}},
8
):
9
    ...  # still str

The following table shows what each engine reads from options while streaming:

LLMRails runs generate_async() in a background task and forwards options to it. It then discards the GenerationResponse that the call returns. Anything you request through log or output_vars is therefore computed but unreachable from the stream. IORails reads only llm_params and the four rail toggles in its streaming path.

Two GenerationResponse fields have no streamed equivalent:

• reasoning_content: Neither engine emits provider reasoning deltas as chunk text. A model that embeds <think> tags in its content streams them as ordinary text, so output rails process the reasoning. IORails logs one warning per request when it detects this.
• tool_calls: IORails emits assembled tool calls as a terminal chunk whose text is a {"tool_calls": [...]} JSON string. The string uses the OpenAI wire shape with function.arguments as a JSON string, while the non-streaming GenerationResponse.tool_calls uses dictionary arguments. LLMRails does not push tool calls into the stream, so they reach only the discarded GenerationResponse.

To obtain a GenerationResponse for a request, use generate_async(). To obtain token usage while streaming, read the usage metadata frame described above. For the non-streaming option and field tables, refer to Generation Options.

Token Usage Tracking

Token usage statistics are available when streaming responses, depending on provider support. When the provider does not return token usage statistics, the terminal chunk’s metadata has no usage key.

Accessing Token Usage Information

You can access token usage statistics through the detailed logging capabilities of the NeMo Guardrails library. Use the log generation option to capture comprehensive information about LLM calls, including token usage:

1
response = rails.generate(messages=messages, options={
2
    "log": {
3
        "llm_calls": True,
4
        "activated_rails": True
5
    }
6
})
7
8
for llm_call in response.log.llm_calls:
9
    print(f"Task: {llm_call.task}")
10
    print(f"Total tokens: {llm_call.total_tokens}")
11
    print(f"Prompt tokens: {llm_call.prompt_tokens}")
12
    print(f"Completion tokens: {llm_call.completion_tokens}")

Alternatively, on LLMRails you can use the explain() method to get a summary of token usage:

1
info = rails.explain()
2
info.print_llm_calls_summary()

For more information about streaming token usage support across different providers, refer to the LangChain documentation on token usage tracking. For detailed information about accessing generation logs and token usage, see Generation Options: Detailed Logging Information and Logging.

Streaming for LLMs Deployed Using HuggingFacePipeline

We also support streaming for LLMs deployed using HuggingFacePipeline. One example is provided in the HF Pipeline Dolly configuration.

To use streaming for HF Pipeline LLMs, you need to create an nemoguardrails.integrations.langchain.providers.huggingface.AsyncTextIteratorStreamer streamer object, add it to the kwargs of the pipeline and to the model_kwargs of the HuggingFacePipelineCompatible object.

1
from nemoguardrails.integrations.langchain.providers.huggingface import AsyncTextIteratorStreamer
2
3
# instantiate tokenizer object required by LLM
4
streamer = AsyncTextIteratorStreamer(tokenizer, skip_prompt=True)
5
params = {"temperature": 0.01, "max_new_tokens": 100, "streamer": streamer}
6
7
pipe = pipeline(
8
    # all other parameters
9
    **params,
10
)
11
12
llm = HuggingFacePipelineCompatible(pipeline=pipe, model_kwargs=params)