assembler

package
v0.260806.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 6, 2026 License: MPL-2.0 Imports: 11 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func OpenAIChatChoiceCount added in v0.260806.1 

func OpenAIChatChoiceCount(request any) int

OpenAIChatChoiceCount returns the request's expected number of choices, defaulting invalid, omitted, or non-Chat values to one.

Types

type AnthropicBetaSDKAssembler 

type AnthropicBetaSDKAssembler struct {
	// contains filtered or unexported fields
}

AnthropicBetaSDKAssembler wraps the SDK's BetaMessage.Accumulate method for v1 beta API streams.

func NewAnthropicBetaSDKAssembler 

func NewAnthropicBetaSDKAssembler() *AnthropicBetaSDKAssembler

NewAnthropicBetaSDKAssembler creates a new assembler for v1 beta streams.

func (*AnthropicBetaSDKAssembler) Accumulate 

func (a *AnthropicBetaSDKAssembler) Accumulate(event anthropic.BetaRawMessageStreamEventUnion) error

Accumulate processes a v1 beta stream event using the SDK's accumulation logic.

func (*AnthropicBetaSDKAssembler) Finish 

func (a *AnthropicBetaSDKAssembler) Finish() *anthropic.BetaMessage

Finish returns the accumulated BetaMessage.

func (*AnthropicBetaSDKAssembler) Result 

func (a *AnthropicBetaSDKAssembler) Result() *anthropic.BetaMessage

Result returns the internal message for direct access if needed.

type AnthropicSDKAssembler 

type AnthropicSDKAssembler struct {
	// contains filtered or unexported fields
}

AnthropicSDKAssembler wraps the SDK's Message.Accumulate method providing a unified interface for stream accumulation.

func NewAnthropicSDKAssembler 

func NewAnthropicSDKAssembler() *AnthropicSDKAssembler

NewAnthropicSDKAssembler creates a new assembler using the SDK's accumulate pattern.

func (*AnthropicSDKAssembler) Accumulate 

func (a *AnthropicSDKAssembler) Accumulate(event anthropic.MessageStreamEventUnion) error

Accumulate processes a stream event using the SDK's built-in accumulation logic. Returns an error if accumulation fails.

func (*AnthropicSDKAssembler) Finish 

func (a *AnthropicSDKAssembler) Finish() *anthropic.Message

Finish returns the accumulated Message.

func (*AnthropicSDKAssembler) Result 

func (a *AnthropicSDKAssembler) Result() *anthropic.Message

Result returns the internal message for direct access if needed.

type AnthropicStreamAssembler 

type AnthropicStreamAssembler struct {
	// contains filtered or unexported fields
}

AnthropicStreamAssembler assembles Anthropic streaming responses It is a pure assembler that doesn't depend on recording logic

func NewAnthropicStreamAssembler 

func NewAnthropicStreamAssembler() *AnthropicStreamAssembler

NewAnthropicStreamAssembler creates a new assembler for Anthropic streams

func (*AnthropicStreamAssembler) Finish 

func (a *AnthropicStreamAssembler) Finish(model string, inputTokens, outputTokens int) *anthropic.Message

Finish assembles the final response and returns it as anthropic.Message

func (*AnthropicStreamAssembler) RecordV1BetaEvent 

func (a *AnthropicStreamAssembler) RecordV1BetaEvent(event *anthropic.BetaRawMessageStreamEventUnion)

RecordV1BetaEvent processes a v1 beta stream event

func (*AnthropicStreamAssembler) RecordV1Event 

func (a *AnthropicStreamAssembler) RecordV1Event(event *anthropic.MessageStreamEventUnion)

RecordV1Event processes a v1 stream event

func (*AnthropicStreamAssembler) SetUsage 

func (a *AnthropicStreamAssembler) SetUsage(inputTokens, outputTokens int)

SetUsage sets the usage data from raw input/output counts. Prefer SetUsageFromTokenUsage when you have an *ai.TokenUsage in hand — it carries cache and reasoning fields the bare counts cannot.

func (*AnthropicStreamAssembler) SetUsageFromTokenUsage added in v0.260604.1 

func (a *AnthropicStreamAssembler) SetUsageFromTokenUsage(u *ai.TokenUsage)

SetUsageFromTokenUsage sets the assembled response usage from the canonical ai.TokenUsage type. Cache-read input tokens are mapped to anthropic.Usage.CacheReadInputTokens and cache writes to CacheCreationInputTokens; reasoning_tokens has no Anthropic analogue (extended-thinking tokens are billed inside output_tokens).

InputTokens goes through UncachedInputTokens() because canonical usage folds the cache-write cost into InputTokens while Anthropic's wire field excludes it.

type OpenAIChatStreamAssembler 

type OpenAIChatStreamAssembler struct {
	// contains filtered or unexported fields
}

OpenAIChatStreamAssembler wraps the SDK's ChatCompletionAccumulator providing a unified interface for stream accumulation.

func NewOpenAIStreamAssembler 

func NewOpenAIStreamAssembler() *OpenAIChatStreamAssembler

NewOpenAIStreamAssembler creates a new assembler for OpenAI streams

func (*OpenAIChatStreamAssembler) AddChunk 

func (a *OpenAIChatStreamAssembler) AddChunk(chunk openai.ChatCompletionChunk) bool

AddChunk incorporates a chunk into the accumulation. Chunks must be added in order. Returns false if accumulation failed.

func (*OpenAIChatStreamAssembler) Finish 

func (a *OpenAIChatStreamAssembler) Finish() *openai.ChatCompletion

Finish returns the accumulated ChatCompletion.

func (*OpenAIChatStreamAssembler) JustFinishedContent 

func (a *OpenAIChatStreamAssembler) JustFinishedContent() (string, bool)

JustFinishedContent returns the content when it was just completed. If the content is just completed, returns (content, true), otherwise ("", false).

func (*OpenAIChatStreamAssembler) JustFinishedRefusal 

func (a *OpenAIChatStreamAssembler) JustFinishedRefusal() (string, bool)

JustFinishedRefusal returns the refusal when it was just completed.

func (*OpenAIChatStreamAssembler) JustFinishedToolCall 

func (a *OpenAIChatStreamAssembler) JustFinishedToolCall() (openai.FinishedChatCompletionToolCall, bool)

JustFinishedToolCall returns a tool call when it was just completed. Note: Not reliable with ParallelToolCalls enabled.

func (*OpenAIChatStreamAssembler) Result 

func (a *OpenAIChatStreamAssembler) Result() *openai.ChatCompletionAccumulator

Result returns the internal accumulator for direct access if needed.

type ResponsesAssembler 

type ResponsesAssembler struct {
	// contains filtered or unexported fields
}

ResponsesAssembler accumulates OpenAI Responses API streaming events. It focuses on core functionality: text accumulation, tool calls, image generation, and final response construction. Inspired by internal/protocol/stream/anthropic_to_openai_responses.go

func NewResponsesAssembler 

func NewResponsesAssembler() *ResponsesAssembler

NewResponsesAssembler creates a new Responses API stream assembler.

func NewResponsesAssemblerWithID 

func NewResponsesAssemblerWithID(responseID, itemID string) *ResponsesAssembler

NewResponsesAssemblerWithID creates a new assembler with specific IDs (useful for testing)

func (*ResponsesAssembler) Accumulate 

func (a *ResponsesAssembler) Accumulate(event responses.ResponseStreamEventUnion) bool

Accumulate processes a Responses API stream event. Returns true if the event was handled, false otherwise.

func (*ResponsesAssembler) CurrentRefusal 

func (a *ResponsesAssembler) CurrentRefusal() string

CurrentRefusal returns the refusal text if the model refused to respond.

func (*ResponsesAssembler) CurrentText 

func (a *ResponsesAssembler) CurrentText() string

CurrentText returns the text being accumulated for the current part.

func (*ResponsesAssembler) Finish 

func (a *ResponsesAssembler) Finish() *responses.Response

Finish returns the accumulated result. If the stream completed successfully, returns the Response. Otherwise, returns nil - check Status() to determine the outcome.

func (*ResponsesAssembler) GetOrCreateItemID 

func (a *ResponsesAssembler) GetOrCreateItemID() string

GetOrCreateItemID returns the item ID, generating one if not set.

func (*ResponsesAssembler) GetOrCreateResponseID 

func (a *ResponsesAssembler) GetOrCreateResponseID() string

GetOrCreateResponseID returns the response ID, generating one if not set.

func (*ResponsesAssembler) HasImage added in v0.260531.1 

func (a *ResponsesAssembler) HasImage() bool

HasImage returns true if any image data was accumulated.

func (*ResponsesAssembler) HasImageAt added in v0.260531.1 

func (a *ResponsesAssembler) HasImageAt(idx int) bool

HasImageAt returns true if image data was accumulated at the specific output index.

func (*ResponsesAssembler) ImageCallIDAt added in v0.260531.1 

func (a *ResponsesAssembler) ImageCallIDAt(idx int) string

ImageCallIDAt returns the image call ID at the specific output index.

func (*ResponsesAssembler) ImageCallIDs added in v0.260531.1 

func (a *ResponsesAssembler) ImageCallIDs() map[int]string

ImageCallIDs returns all image generation call IDs as a map of output index to call ID.

func (*ResponsesAssembler) ImageCount added in v0.260531.1 

func (a *ResponsesAssembler) ImageCount() int

ImageCount returns the number of images accumulated.

func (*ResponsesAssembler) ImageDataAt added in v0.260531.1 

func (a *ResponsesAssembler) ImageDataAt(idx int) string

ImageDataAt returns the image data at the specific output index.

func (*ResponsesAssembler) Images added in v0.260531.1 

func (a *ResponsesAssembler) Images() map[int]string

Images returns all accumulated images as a map of output index to image data.

func (*ResponsesAssembler) IsCompleted 

func (a *ResponsesAssembler) IsCompleted() bool

IsCompleted returns true if the response is completed.

func (*ResponsesAssembler) IsFailed 

func (a *ResponsesAssembler) IsFailed() bool

IsFailed returns true if the response failed.

func (*ResponsesAssembler) IsFinished 

func (a *ResponsesAssembler) IsFinished() bool

IsFinished returns true if the stream has finished (completed, failed, incomplete, or error).

func (*ResponsesAssembler) IsIncomplete 

func (a *ResponsesAssembler) IsIncomplete() bool

IsIncomplete returns true if the response was incomplete.

func (*ResponsesAssembler) OutputText 

func (a *ResponsesAssembler) OutputText() string

OutputText returns the accumulated output text.

func (*ResponsesAssembler) Response 

func (a *ResponsesAssembler) Response() *responses.Response

Response returns the final Response object when completed. Returns nil if the response is not yet completed or failed.

func (*ResponsesAssembler) ResponseID 

func (a *ResponsesAssembler) ResponseID() string

ResponseID returns the response ID.

func (*ResponsesAssembler) SetItemID 

func (a *ResponsesAssembler) SetItemID(id string)

SetItemID sets a custom item ID.

func (*ResponsesAssembler) SetResponseID 

func (a *ResponsesAssembler) SetResponseID(id string)

SetResponseID sets a custom response ID.

func (*ResponsesAssembler) Status 

func (a *ResponsesAssembler) Status() string

Status returns the current response status.

func (*ResponsesAssembler) ToolCalls 

func (a *ResponsesAssembler) ToolCalls() map[int]ToolCallInfo

ToolCalls returns the accumulated tool calls. Returns a map of output index to tool call details.

type StreamAssembler added in v0.260806.1 

type StreamAssembler interface {
	Add(value any) error
	Finish() (any, error)
	Terminal() bool
	TerminalError() error
}

StreamAssembler is the protocol-owned common surface for reconstructing one complete response from native stream events. It accepts SDK events, Wire DTOs, or json.RawMessage values; protocol-specific handling remains here rather than in observers such as Recording.

func NewStreamAssembler added in v0.260806.1 

func NewStreamAssembler(api protocol.APIType) (StreamAssembler, error)

NewStreamAssembler adapts the existing protocol assemblers to one common interface. Protocol conversion is intentionally out of scope: the caller must select the protocol already spoken at its observation boundary.

func NewStreamAssemblerForRequest added in v0.260806.1 

func NewStreamAssemblerForRequest(api protocol.APIType, request any) (StreamAssembler, error)

NewStreamAssemblerForRequest configures protocol-specific expectations from the provider-bound request. Today this is used for OpenAI Chat's n choices.

type ToolCallInfo 

type ToolCallInfo struct {
	ItemID    string
	Name      string
	Arguments string
}

ToolCallInfo contains information about a tool call.

Source Files

View all Source files

Directories

Path Synopsis
Package streamemit provides a decoupled emission layer on top of the Anthropic stream assemblers in internal/protocol/assembler.
 Package streamemit provides a decoupled emission layer on top of the Anthropic stream assemblers in internal/protocol/assembler.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.