Documentation
¶
Overview ¶
Package capture holds the shared streaming-reduction library consumed by both the local tapes proxy and the cloud tapes-extproc sidecar. A Reducer turns the raw bytes of a single LLM turn — streaming (SSE, NDJSON) or one-shot (application/json) — into a canonical *llm.ChatResponse suitable for handing to tapes-ingest.
Callers construct reducers explicitly via NewXxxReducer constructors and dispatch them by provider name themselves; the package deliberately holds no global registry so import-order and init() side effects stay out of the call graph.
Index ¶
Constants ¶
const MaxDecompressedBytes = 32 << 20
MaxDecompressedBytes caps one decompressed body, per layer. It bounds a decompression bomb: the compressed bytes are already capped on the way in, but a few KiB of zstd expands to gigabytes if nothing stops it.
32 MiB matches extproc's cap exactly. It has to: a body extproc accepted and forwarded must not then be rejected here, or the raw lane would drop bytes the producer believed were safe to send.
const ProviderAnthropic = "anthropic"
ProviderAnthropic is the canonical provider-name string consumers use to key the Anthropic reducer in their dispatch maps.
const ProviderOpenAI = "openai"
ProviderOpenAI is the canonical provider-name string consumers use to key the OpenAI Responses reducer in their dispatch maps.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type DecodeStats ¶ added in v0.32.0
type DecodeStats struct {
// Truncated marks a body recovered from a stream that ended early —
// the decode succeeded only because partial output was accepted. The
// caller decides what that is worth; it is surfaced rather than
// swallowed so a salvaged reduction is never silently indistinguishable
// from a clean one.
Truncated bool
}
DecodeStats reports what had to be tolerated to produce the decoded bytes.
func DecodeContentEncoding ¶ added in v0.32.0
func DecodeContentEncoding(body []byte, encoding string) ([]byte, DecodeStats, error)
DecodeContentEncoding returns body decoded per encoding, for handing to a reducer. Callers store the ENCODED bytes and decode only for reduction: re-compression is not byte-identical, so a column that promises "verbatim" has to keep what arrived.
An unrecognized encoding is an error rather than a pass-through. Handing compressed bytes to a reducer that expects text yields a parse failure well away from the actual cause, and the bytes are still stored either way — so erroring here loses nothing and names the real problem.
Truncated streams are salvaged rather than refused when they yielded any output at all, and the salvage is reported in DecodeStats. A capture that lost its tail is still most of a turn; refusing it would discard everything the stream did deliver in exchange for nothing. This mirrors extproc's response-side behavior — and it must, because extproc forwards the truncated compressed bytes as they arrived, so these are exactly the bytes ingest is handed.
type Reducer ¶
type Reducer interface {
// Reduce consumes the raw request body and the raw response body and
// produces a canonical ChatResponse. The reqBody is supplied for context
// (the reducer may enrich the response with metadata the wire format
// doesn't carry); passing nil is valid when no request-side context is
// available. contentType is the upstream response's Content-Type, used to
// disambiguate streaming (text/event-stream, application/x-ndjson) from
// non-streaming (application/json) reduction paths.
//
// Errors are reserved for unrecoverable parse failures or malformed
// envelopes. Partial captures (mid-stream errors, EOF before terminal
// frame, per-block parse failures) should return a ChatResponse with
// diagnostic metadata in Extra rather than an error — callers want to
// see what happened instead of silence.
//
// Per-turn memory is not bounded inside the reducer. Content grows with
// the upstream's output, ceiling'd in practice by the caller's
// max_tokens. Callers that need a hard ceiling should impose one
// themselves; sidecar deployments track turn size via metrics rather
// than enforcing a cap so the full tape is preserved.
Reduce(ctx context.Context, reqBody, respBody io.Reader, contentType string) (*llm.ChatResponse, error)
}
Reducer produces a canonical *llm.ChatResponse from the raw request and response bodies for a single turn. Implementations must be stateless at the package level; per-turn state is held in local variables for the duration of a single Reduce call.
func NewAnthropicReducer ¶
func NewAnthropicReducer() Reducer
NewAnthropicReducer returns an Anthropic reducer. The value is stateless at the package level; per-turn state lives inside Reduce.
func NewOpenAIResponsesReducer ¶ added in v0.14.0
func NewOpenAIResponsesReducer() Reducer
NewOpenAIResponsesReducer returns an OpenAI Responses reducer. The value is stateless at the package level; per-turn state lives inside Reduce.
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
Package fixtures exposes the capture reducer test fixtures as an embed.FS so every consumer exercises the SAME bytes instead of hand-copied inline duplicates.
|
Package fixtures exposes the capture reducer test fixtures as an embed.FS so every consumer exercises the SAME bytes instead of hand-copied inline duplicates. |