LemonAi.
View Source
Async event stream for streaming LLM responses with OTP-compliant lifecycle management.
This module provides a producer/consumer pattern for streaming events from LLM providers with the following BEAM/OTP guarantees:
Features
- Owner Monitoring: Streams are linked to an owner process and automatically cancel when the owner dies.
- Task Linking: Streaming tasks can be attached and are properly shutdown when the stream is canceled.
- Bounded Queues: Configurable queue limits prevent unbounded memory growth.
Backpressure:
push/2returns:ok | {:error, :overflow}for flow control.- Cancellation: Explicit
cancel/2API for clean stream termination. - Timeouts: Configurable stream timeout with automatic cancellation.
Usage
# Start a stream with options
{:ok, stream} = EventStream.start_link(
owner: self(),
max_queue: 1000,
timeout: 300_000
)
# Attach a supervised task
{:ok, task_pid} = Task.Supervisor.start_child(LemonAi.StreamTaskSupervisor, fn ->
# ... streaming logic ...
end)
EventStream.attach_task(stream, task_pid)
# Producer pushes events (with backpressure)
case EventStream.push(stream, {:text_delta, 0, "Hello", partial}) do
:ok -> :continue
{:error, :overflow} -> :stop_producing
end
# Consumer reads events
stream
|> EventStream.events()
|> Enum.each(fn event -> IO.inspect(event) end)
# Or cancel explicitly
EventStream.cancel(stream, :user_requested)Summary
Functions
Attach a streaming task to this event stream.
Cancel the stream with a reason.
Returns a specification to start this module under a supervisor.
Collect all text from the stream into a single string.
Complete the stream with a final message.
Signal an error on the stream.
Get a lazy enumerable of events from the stream.
Push an event to the stream (synchronous with backpressure).
Push an event to the stream (asynchronous, fire-and-forget).
Get the final result of the stream, blocking until complete.
Start a new event stream.
Get current queue statistics.
Types
@type drop_strategy() :: :drop_oldest | :drop_newest | :error
@type event() :: {:start, LemonAi.Types.AssistantMessage.t()} | {:text_start, non_neg_integer(), LemonAi.Types.AssistantMessage.t()} | {:text_delta, non_neg_integer(), String.t(), LemonAi.Types.AssistantMessage.t()} | {:text_end, non_neg_integer(), String.t(), LemonAi.Types.AssistantMessage.t()} | {:thinking_start, non_neg_integer(), LemonAi.Types.AssistantMessage.t()} | {:thinking_delta, non_neg_integer(), String.t(), LemonAi.Types.AssistantMessage.t()} | {:thinking_end, non_neg_integer(), String.t(), LemonAi.Types.AssistantMessage.t()} | {:tool_call_start, non_neg_integer(), LemonAi.Types.AssistantMessage.t()} | {:tool_call_delta, non_neg_integer(), String.t(), LemonAi.Types.AssistantMessage.t()} | {:tool_call_end, non_neg_integer(), LemonAi.Types.ToolCall.t(), LemonAi.Types.AssistantMessage.t()} | {:done, LemonAi.Types.AssistantMessage.stop_reason(), LemonAi.Types.AssistantMessage.t()} | {:error, LemonAi.Types.AssistantMessage.stop_reason(), LemonAi.Types.AssistantMessage.t()}
@type option() :: {:owner, pid()} | {:max_queue, pos_integer()} | {:drop_strategy, drop_strategy()} | {:timeout, timeout()}
@type t() :: pid()
Functions
Attach a streaming task to this event stream.
When the stream is canceled or the owner dies, the attached task will be
shutdown using Task.shutdown/2.
Cancel the stream with a reason.
This will:
- Mark the stream as canceled
- Shutdown any attached streaming task
- Wake up all waiters with a terminal event
- Stop the GenServer
Returns a specification to start this module under a supervisor.
See Supervisor.
Collect all text from the stream into a single string.
@spec complete(t(), LemonAi.Types.AssistantMessage.t()) :: :ok
Complete the stream with a final message.
@spec error(t(), LemonAi.Types.AssistantMessage.t()) :: :ok
Signal an error on the stream.
@spec events(t()) :: Enumerable.t()
Get a lazy enumerable of events from the stream.
This returns a Stream that will block when no events are available and complete when the stream is done. Terminal events (done/error/canceled) are included in the stream before it halts.
Push an event to the stream (synchronous with backpressure).
Returns :ok on success or {:error, :overflow} if the queue is full
(when using :error drop strategy) or {:error, :canceled} if the
stream has been canceled.
Use push_async/2 if you don't need backpressure feedback.
Push an event to the stream (asynchronous, fire-and-forget).
This is a non-blocking push that ignores backpressure. Use push/2
if you need confirmation that the event was accepted.
If using :drop_oldest or :drop_newest strategies, events will be
dropped silently on overflow. With :error strategy, overflow events
are still dropped but a warning is logged.
@spec result(t(), timeout()) :: {:ok, LemonAi.Types.AssistantMessage.t()} | {:error, LemonAi.Types.AssistantMessage.t() | term()}
Get the final result of the stream, blocking until complete.
@spec start_link([option()]) :: GenServer.on_start()
Start a new event stream.
Options
:owner- Process to monitor. Stream cancels if owner dies. Default:self():max_queue- Maximum events to buffer. Default: 10000:drop_strategy- What to do on overflow::drop_oldest,:drop_newest, or:error. Default::error:timeout- Stream timeout in milliseconds. Default: 300000ms
@spec stats(t()) :: %{ queue_size: non_neg_integer(), max_queue: pos_integer(), dropped: non_neg_integer() }
Get current queue statistics.