nimgent/providers/stream

This page is generated from the module’s exported API and ## documentation comments.

Shared OpenAI-family runtime: SSE, wake-fd, stream accumulation, usage JSON.

Types

SseAction

Action returned by an SSE event handler.

SseAction = enum
  sseContinue, sseStop, sseCancel

View source

SseDrive

Terminal state returned by an SSE stream driver.

SseDrive = enum
  sdEnded,                  ## [DONE] or handle sseStop
  sdCancelled, sdClosed      ## body ended without a terminal

View source

WakeWatch

Optional file-descriptor watcher used during streaming.

WakeWatch = object

View source

PendingTool

Tool-call fragments accumulated from a stream.

PendingTool = object
  id*: string
  name*: string
  args*: string
  itemId*: string

View source

StreamAcc

Text, reasoning, and tool fragments accumulated from a stream.

StreamAcc = object
  text*: string
  think*: string
  details*: JsonNode
  tools*: seq[PendingTool]
  parsedFinal*: bool

View source

Procedures

initStreamAcc

Create an empty stream accumulator.

proc initStreamAcc(): StreamAcc {.raises: [], tags: [], forbids: [].}

Returns: StreamAcc.

View source

parseOpenAiUsage

Chat Completions (prompt_tokens) and Responses (input_tokens) usage.

proc parseOpenAiUsage(usage: JsonNode; result: var Usage) {.raises: [],
    tags: [], forbids: [].}

Arguments

Name	Type	Default
usage	JsonNode
result	var Usage

View source

popLine

Remove and return one newline-terminated line from a buffer.

proc popLine(buf: var string): tuple[ok: bool, line: string] {.raises: [],
    tags: [], forbids: [].}

Returns: tuple[ok: bool, line: string].

Arguments

Name	Type	Default
buf	var string

View source

register

Register a wake file descriptor when one is supplied.

proc register(w: var WakeWatch; wakeFd: cint) {.raises: [], tags: [],
    forbids: [].}

Arguments

Name	Type	Default
w	var WakeWatch
wakeFd	cint

View source

unregister

Unregister a previously watched wake file descriptor.

proc unregister(w: var WakeWatch) {.raises: [], tags: [], forbids: [].}

Arguments

Name	Type	Default
w	var WakeWatch

View source

awaitWithWakeAsync

Check cancellation even when another input poller has consumed wakeFd.

proc awaitWithWakeAsync[T](fut: Future[T]; watch: ptr WakeWatch; wakeFd: cint;
                           onEvent: StreamCallback): Future[bool] {.
    stackTrace: false.}

Type parameters: T.

Returns: Future[bool].

Arguments

Name	Type	Default
fut	Future[T]
watch	ptr WakeWatch
wakeFd	cint
onEvent	StreamCallback

View source

drainBodyStreamAsync

Read all remaining chunks from an asynchronous response body.

proc drainBodyStreamAsync(bodyStream: FutureStream[string]): Future[string] {.
    stackTrace: false, raises: [Exception, ValueError], tags: [RootEffect],
    forbids: [].}

Returns: Future[string].

Arguments

Name	Type	Default
bodyStream	FutureStream[string]

View source

forEachSseAsync

Parse SSE data lines and pass each JSON event to a handler.

proc forEachSseAsync(bodyStream: FutureStream[string]; watch: ptr WakeWatch;
                     wakeFd: cint; onEvent: StreamCallback;
                     handle: proc (data: JsonNode): SseAction {.closure.}): Future[
    SseDrive] {.stackTrace: false, raises: [Exception, ValueError, OSError],
                tags: [RootEffect, TimeEffect, ReadIOEffect, WriteIOEffect],
                forbids: [].}

Returns: Future[SseDrive].

Arguments

Name	Type	Default
bodyStream	FutureStream[string]
watch	ptr WakeWatch
wakeFd	cint
onEvent	StreamCallback
handle	proc (data: JsonNode): SseAction {.closure.}

View source

assembleStream

Turn accumulated stream fragments into response content once.

proc assembleStream(acc: StreamAcc; response: var ProviderResponse) {.
    raises: [], tags: [ReadIOEffect, WriteIOEffect], forbids: [].}

Arguments

Name	Type	Default
acc	StreamAcc
response	var ProviderResponse

View source