Documentation
¶
Overview ¶
Package pipeline defines the core contract of tokipe: the Request that flows through every optimization stage, the Response returned to the caller, the Stage extension point, and the Pipeline that runs them in order.
Nothing in this package performs I/O and it has zero third-party dependencies. See docs/spec.md §2.3.
Index ¶
Constants ¶
const MetaRouterError = "router.error"
MetaRouterError holds the error from a Router that panicked. The pipeline package reports nothing itself — it has no Recorder, by design — so it leaves the evidence in Metadata for a caller or a wrapping stage to surface.
const MetaShortCircuit = "_short_circuit_response"
MetaShortCircuit is the reserved Metadata key a stage sets to hand a finished *Response back to Pipeline.Run without making an LLM call.
const UnnamedClient = "unnamed_client"
UnnamedClient is reported in Metadata when a ModelClient's Name method panics or returns "".
const UnnamedStage = "unnamed_stage"
UnnamedStage names a StageError when the stage's own Name method panics or returns "".
Variables ¶
This section is empty.
Functions ¶
Types ¶
type CacheBreakpoint ¶
CacheBreakpoint marks where provider-side prompt caching should be anchored in the outgoing request. AfterMessageIndex is an index into Request.Messages.
type Delta ¶
type Delta struct {
Text string
// ModelUsed names the client producing the stream. Set on every delta so a
// consumer never has to wait for the end to know what answered.
ModelUsed string
// Usage is non-nil only on the delta that carries the provider's final
// accounting, if the provider reports one at all. A CLI backend or a
// non-streaming client wrapped by StreamOne may never set it.
Usage *Usage
}
Delta is one incremental piece of a streamed response.
Text is the increment, not the accumulated answer — concatenating every Delta's Text yields the full content.
type Message ¶
type Message struct {
Role string // "user" | "assistant" | "system"
Content string
// Static marks content the caller guarantees will not change for the
// lifetime of the session (system prompt, tool definitions). cache.Aligner
// anchors provider-side cache breakpoints only after static content.
//
// NOTE: additive extension to the spec's Message type — §2.4.6 requires
// "caller-marked messages" for the static segment but the spec's struct
// had no field to carry that mark.
Static bool
}
type ModelClient ¶
type ModelClient interface {
Send(ctx context.Context, req *Request) (*Response, error)
Name() string
}
ModelClient abstracts a single LLM provider/model endpoint.
NOTE: the spec placed this interface in package providers, but providers imports pipeline for Request/Response, which would be an import cycle. The interface is therefore declared here and re-exported as a type alias by package providers, so `providers.ModelClient` remains a valid, identical name for callers exactly as the spec's signatures describe.
type Pipeline ¶
type Pipeline struct {
// contains filtered or unexported fields
}
Pipeline runs an ordered list of stages, then performs the final model call.
func New ¶
func New(client ModelClient, stages ...Stage) *Pipeline
New builds a Pipeline that always calls client for the final Send.
func NewWithRouter ¶
func NewWithRouter(fallback ModelClient, router Router, stages ...Stage) *Pipeline
NewWithRouter builds a Pipeline that asks router which client to Send to after all stages have run. fallback is used when router is nil or returns a decision with no Client (fail-open).
func (*Pipeline) Run ¶
Run executes every stage in order, then performs the final model call.
The stage loop, short-circuit handling and routing all live in prepare, which RunStream shares. Keeping one copy is deliberate: duplicated, the streaming path could silently drift out of step with this one, and that is a bug class better designed out than tested for.
NOTE: a panic from Stage.Process is deliberately NOT recovered; see the Stage docs. Name() is guarded, because a broken name must never destroy an error Process already returned.
func (*Pipeline) RunStream ¶
RunStream is Run with an incremental result. Every stage runs first, exactly as in Run and in the same order — streaming is purely a property of the final model call, which is why no Stage needed to change to support it.
The returned sequence must be consumed to completion or abandoned; abandoning it (by breaking out of the range loop) cancels nothing on its own, so pass a cancellable ctx if that matters.
A short-circuiting preprocess rule yields exactly one delta and stops, so callers need no special case for "the answer arrived without a model".
type Request ¶
type Request struct {
// Query is the user-facing question/instruction for this turn.
Query string
// Messages is the full conversation history, oldest first.
Messages []Message
// ToolCalls, if non-nil, means this Request represents a tool-call
// resolution step rather than a fresh LLM turn.
ToolCalls []ToolCall
// NeedsRetrieval signals to the RAG stage whether retrieval should run at
// all. Default false — callers must opt in explicitly.
NeedsRetrieval bool
// RetrievedChunks is populated by the RAG stage; empty until then.
// It is *dynamic* content: it changes every turn, so no cache breakpoint
// may ever be anchored inside it (see cache.Aligner).
RetrievedChunks []Chunk
// TurnType classifies this turn for the budget package.
TurnType TurnType
// CacheBreakpoints is populated by cache.Aligner.
CacheBreakpoints []CacheBreakpoint
// Metadata is an open bag for stage-specific or caller-specific data that
// doesn't warrant a first-class field. Stages must namespace keys
// (e.g. "preprocess.matched_rule") to avoid collisions.
Metadata map[string]any
}
Request flows through every stage. Stages read and mutate fields relevant to their job and pass the (possibly modified) Request to the next stage.
type Response ¶
type Response struct {
Content string
ModelUsed string
ShortCircuited bool // true if a preprocess rule handled it without an LLM call
Usage Usage
}
Response is the final result returned to the caller after the pipeline (or a short-circuiting stage) has produced an answer.
type RouteDecision ¶
type RouteDecision struct {
Client ModelClient
Reason string
Confidence float64
}
type Router ¶
type Router interface {
Route(ctx context.Context, req *Request) RouteDecision
}
Router selects the ModelClient used for the final Send. It is deliberately NOT a Stage: it runs after every context-shaping stage so it can see the final prompt size, and wiring it as a special final step keeps that ordering un-misconfigurable (docs/spec.md §2.4.7).
Declared here rather than in package router for the same import-cycle reason as ModelClient; package router re-exports it.
type Stage ¶
type Stage interface {
// Process receives the current Request state and returns the next state.
// Returning a non-nil error aborts the pipeline UNLESS the stage's own
// contract says otherwise (see the fail-open rule in docs/spec.md §2.5).
Process(ctx context.Context, req *Request) (*Request, error)
// Name is used in logs/metrics to identify which stage did what.
Name() string
}
Stage is the single extension point of the pipeline. Every optimization technique (compression, caching, routing, etc.) is a Stage implementation.
type StageError ¶
func (*StageError) Error ¶
func (e *StageError) Error() string
func (*StageError) Unwrap ¶
func (e *StageError) Unwrap() error
type StreamingClient ¶
type StreamingClient interface {
ModelClient
// SendStream returns a sequence of deltas. The error return is for
// failures that happen before streaming starts (a rejected request, a
// refused connection); failures mid-stream are yielded as the error half
// of a pair, after which the sequence must stop.
SendStream(ctx context.Context, req *Request) (iter.Seq2[Delta, error], error)
}
StreamingClient is an OPTIONAL interface a ModelClient may also implement.
It is a separate interface rather than a method on ModelClient because ModelClient froze at v1.0.0 (see README §Stability). Adding a method there would have broken every existing implementation; adding an interface breaks nothing, and Pipeline.RunStream works with clients that implement neither.
Source Files