/ 
Langertha-Knarr-1.102
/ Langertha::Knarr::Stream

NAME

Langertha::Knarr::Stream - Async chunk iterator returned by streaming Knarr handlers

VERSION

version 1.102

SYNOPSIS

use Langertha::Knarr::Stream;

# From a fixed list of strings
my $stream = Langertha::Knarr::Stream->from_list('hel', 'lo');

# From a sync generator
my @parts = ('hel', 'lo');
my $stream = Langertha::Knarr::Stream->new(
    generator => sub { @parts ? shift @parts : undef },
);

# From a future-yielding source (real async)
my $stream = Langertha::Knarr::Stream->new(
    source => sub { $next_chunk_future },
);

# Drain it
while ( defined( my $chunk = $stream->next_chunk_f->get ) ) {
    print $chunk;
}

DESCRIPTION

The chunk iterator that streaming Knarr handlers return. Supports two construction modes: a sync generator coderef that returns the next chunk string each call (or undef for end), or a source coderef that returns a Future resolving to the next chunk string. The Future form is the one real async backends like Net::Async::HTTP use; the generator form is for tests and simple cases.

generator

Optional. CodeRef returning the next chunk synchronously.

source

Optional. CodeRef returning a Future that resolves to the next chunk.

finish_reason

Optional. The backend's terminal finish reason, verbatim (stop, length, tool_calls, end_turn, MAX_TOKENS, ...), or undef when the backend reported none. Known once the stream is exhausted; the protocol maps it into its own vocabulary when it closes the stream. Settable, since a producer learns it only at the end. A stream that wraps another (see "upstream") and has none of its own answers with its upstream's.

tool_calls

ArrayRef of complete Langertha::ToolCall objects the backend emitted, empty when it emitted none. Like "finish_reason" it is known once the stream is exhausted: the protocol emits the calls in its own framing when it closes the stream. Settable; a stream that wraps another and has none of its own answers with its upstream's.

has_tool_calls

True when "tool_calls" holds at least one call.

usage

The backend's token usage for the whole stream as a Langertha::Usage, or undef when it reported none. Like "finish_reason" it is known once the stream is exhausted: the protocol puts it on its terminal frames and the tracing and request-log decorators record it. Settable; a provider usage HashRef (OpenAI prompt_tokens, Anthropic input_tokens, Ollama prompt_eval_count, ...) is upgraded through "from_hash" in Langertha::Usage, an empty one is no usage. A stream that wraps another and has none of its own answers with its upstream's.

model

Optional. The model name a handler routes the stream to, as the non-streaming "model" in Langertha::Knarr::Response: Langertha::Knarr::Handler::Router sets the model of the config entry. Settable; a stream that wraps another and has none of its own answers with its upstream's.

upstream_model

The model the backend reported on its stream chunks, or undef when it reported none. Known once the stream is exhausted; the tracing decorator records it as the generation's model. Settable; a stream that wraps another and has none of its own answers with its upstream's.

upstream

Optional. The stream this one wraps, as the tracing and request-log decorators do. Only consulted by "finish_reason", "tool_calls", "usage", "model" and "upstream_model".

next_chunk_f

Returns a Future resolving to the next chunk string, or undef when the stream is exhausted.

from_list

my $stream = Langertha::Knarr::Stream->from_list(@chunks);

Convenience constructor that builds a stream from a fixed list of chunk strings.

from_callback

my $stream = Langertha::Knarr::Stream->from_callback( sub {
    my ($emit, $done, $fail, $finish, $tool_call, $usage, $model) = @_;
    my $f = $engine->simple_chat_stream_realtime_f(
        sub {
            $emit->( $_[0]->content );
            $finish->( $_[0]->finish_reason ) if $_[0]->has_finish_reason;
            $tool_call->( @{ $_[0]->tool_calls } ) if $_[0]->has_tool_calls;
            $usage->( $_[0]->usage ) if $_[0]->has_usage;
            $model->( $_[0]->model ) if $_[0]->has_model;
        },
        @messages,
    );
    $f->on_done( $done );
    $f->on_fail( $fail );
    $f->retain;
});

Builds a stream backed by a callback-driven producer. The setup sub receives seven callbacks — $emit->($chunk), $done->(), $fail->($err), $finish->($finish_reason), $tool_call->(@tool_calls), $usage->($usage), $model->($model) — and is expected to wire them to the underlying async source. $finish sets "finish_reason"; an undef reason is ignored, a later one wins. $tool_call appends complete Langertha::ToolCall objects to "tool_calls". $usage sets "usage"; streamed usage is cumulative, so a later report wins, and an undef or empty one is ignored. $model sets "upstream_model"; an undef or empty name is ignored, a later one wins. Internally maintains a queue and pending Future so the consumer side can sit on next_chunk_f without polling.

This is the canonical replacement for the queue/pending/finished/error pump that engine-backed handlers used to inline.

SUPPORT

Issues

Please report bugs and feature requests on GitHub at https://github.com/Getty/langertha-knarr/issues.

IRC

Join #langertha on irc.perl.org or message Getty directly.

CONTRIBUTING

Contributions are welcome! Please fork the repository and submit a pull request.

AUTHOR

Torsten Raudssus <torsten@raudssus.de> https://raudssus.de/

COPYRIGHT AND LICENSE

This software is copyright (c) 2026 by Torsten Raudssus.

This is free software; you can redistribute it and/or modify it under the same terms as the Perl 5 programming language system itself.