Documentation
¶
Index ¶
- type FinishReason
- type StreamFrame
- type StreamOperation
- type StreamReader
- func FramesToChunks(frames *StreamReader[StreamFrame], ...) *StreamReader[content.Chunk]
- func FramesToChunksWithResult(frames *StreamReader[StreamFrame], ...) *StreamReader[content.Chunk]
- func NewStreamReader[T any](next func() (T, error), closer func() error) *StreamReader[T]
- func NewStreamReaderWithResult[T any](next func() (T, error), closer func() error, producer StreamResultProducer) *StreamReader[T]
- type StreamReaderError
- type StreamReaderFailure
- type StreamResult
- type StreamResultError
- type StreamResultProducer
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type FinishReason ¶
type FinishReason string
FinishReason is the provider-neutral reason a model stopped producing output. The zero value is explicit: the provider did not report a recognized reason.
const ( FinishReasonUnknown FinishReason = "" FinishReasonStop FinishReason = "stop" FinishReasonLength FinishReason = "length" FinishReasonToolUse FinishReason = "tool_use" FinishReasonContentFilter FinishReason = "content_filter" )
type StreamFrame ¶
StreamFrame is a raw stream event: an optional event name, optional metadata, and the data payload. It is wire-level rather than semantic.
type StreamOperation ¶
type StreamOperation string
StreamOperation identifies the public StreamReader operation that failed.
const ( StreamOperationNext StreamOperation = "Next" StreamOperationClose StreamOperation = "Close" )
type StreamReader ¶
type StreamReader[T any] struct { // contains filtered or unexported fields }
StreamReader is a pull-based iterator over streaming values of type T. Call Next to advance; it returns (zero, io.EOF) when the stream is exhausted. Always call Close when done — even after io.EOF — to release the underlying connection.
func FramesToChunks ¶
func FramesToChunks(frames *StreamReader[StreamFrame], mapFrame func(StreamFrame) ([]content.Chunk, error)) *StreamReader[content.Chunk]
FramesToChunks adapts a raw StreamFrame reader plus a per-frame semantic mapper into a content.Chunk stream. A StreamDecoder built on a wire framer (sse, ndjson, …) uses it to keep the frame-draining, multi-chunk buffering, and body-close plumbing in one place while supplying only its codec's per-frame decode logic.
mapFrame maps one raw frame to zero or more chunks:
- returning (chunks, nil) yields those chunks (buffered so a frame that decodes to several chunks drops none);
- returning (nil, nil) is a tolerant skip (malformed or uninteresting frame);
- returning a non-nil error ends the stream with that error, after any chunks returned alongside it have been delivered. Returning io.EOF is how a codec signals a terminal sentinel such as OpenAI's [DONE].
The returned reader's Close delegates to frames.Close, so the wire framer keeps ownership of the underlying body.
func FramesToChunksWithResult ¶
func FramesToChunksWithResult(frames *StreamReader[StreamFrame], mapFrame func(StreamFrame) ([]content.Chunk, error), producer StreamResultProducer) *StreamReader[content.Chunk]
FramesToChunksWithResult adds a semantic result producer for provider codecs that accumulate metadata while mapping frames. A present semantic result is authoritative; when it is absent, a result carried by the frame reader is propagated instead.
Mapper io.EOF is a clean semantic end. Chunks returned with io.EOF are fully drained before the downstream reader observes EOF.
Any other mapper error is drain-then-fail: chunks already decoded — buffered from an earlier frame or handed back alongside the error itself — are delivered first, and the error is then the stream's permanent terminal outcome. A malformed frame is still an error; only its timing relative to decoded content changes, so a caller sees partial content followed by an explicit failure instead of silently losing the generation. Once latched, the failure is returned by every later Next, no further frame is read, and the terminal-metadata producer is never consulted, so Result stays unavailable and a truncated stream can never present itself as complete.
func NewStreamReader ¶
func NewStreamReader[T any](next func() (T, error), closer func() error) *StreamReader[T]
NewStreamReader constructs a StreamReader from a next function and an optional closer. If closer is nil, Close is a no-op. next must return (zero, io.EOF) when the stream is exhausted.
func NewStreamReaderWithResult ¶
func NewStreamReaderWithResult[T any](next func() (T, error), closer func() error, producer StreamResultProducer) *StreamReader[T]
NewStreamReaderWithResult constructs a reader with a narrow terminal-result producer. The producer is not consulted until Next observes clean EOF. Its state may therefore accumulate while the stream is active; the reader takes an immutable snapshot at EOF and returns a fresh Usage copy on every Result.
func (*StreamReader[T]) Close ¶
func (r *StreamReader[T]) Close() error
Close releases the underlying connection. It is idempotent: the wrapped close func runs at most once (guarded by a sync.Once), so a double Close never runs the closer twice; every call returns the first call's result.
func (*StreamReader[T]) Next ¶
func (r *StreamReader[T]) Next() (T, error)
Next returns the next value. Returns (zero, io.EOF) when exhausted. Calls to Next are serialized; the underlying next function is never invoked concurrently by StreamReader. Close is deliberately not serialized behind a blocking Next so it can interrupt I/O. An underlying next/closer pair used concurrently must honor that contract itself.
func (*StreamReader[T]) Result ¶
func (r *StreamReader[T]) Result() (StreamResult, bool)
Result returns a fresh copy of authoritative terminal metadata. It is false before clean EOF, after any non-EOF failure, or when no producer result exists.
type StreamReaderError ¶
type StreamReaderError struct {
Operation StreamOperation
Failure StreamReaderFailure
}
StreamReaderError reports an invalid public StreamReader boundary.
func (*StreamReaderError) Error ¶
func (e *StreamReaderError) Error() string
type StreamReaderFailure ¶
type StreamReaderFailure string
StreamReaderFailure identifies a structurally invalid StreamReader boundary.
const ( StreamReaderFailureNilReceiver StreamReaderFailure = "nil receiver" StreamReaderFailureMissingNext StreamReaderFailure = "missing next function" StreamReaderFailureMissingFrameMapper StreamReaderFailure = "missing frame mapper" )
type StreamResult ¶
type StreamResult struct {
Usage *content.Usage
Model string
FinishReason FinishReason
// Attempts is how many establishment attempts a retrying decorator made
// before this stream opened; 0 when the producer does not count.
Attempts int
}
StreamResult is authoritative terminal metadata for one cleanly completed provider stream. Usage is absent when the provider did not report it.
type StreamResultError ¶
type StreamResultError struct {
// Cause remains directly inspectable. Unwrap exposes ordinary causes while
// suppressing EOF-bearing chains: metadata failure is never clean exhaustion.
Cause error
}
StreamResultError reports terminal metadata that could not be authorized.
func (*StreamResultError) Error ¶
func (e *StreamResultError) Error() string
func (*StreamResultError) Unwrap ¶
func (e *StreamResultError) Unwrap() error
Unwrap preserves normal typed error inspection unless the cause contains io.EOF, which must not escape as the stream's clean terminal sentinel.
type StreamResultProducer ¶
type StreamResultProducer func() (StreamResult, bool, error)
StreamResultProducer supplies the terminal metadata accumulated while a stream is read. It is called once, and only after Next observes clean EOF. A false bool means the producer has no authoritative terminal metadata.
Source Files