engine

package
v1.9.3 Latest Latest
Warning

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

 Go to latest Published: Nov 4, 2026 License: MIT Imports: 28 Imported by: 3

Documentation

Overview

Package engine provides the central Weave RAG pipeline coordinator.

Index

Constants

View Source 
const DefaultStalledAfter = 15 * time.Minute

DefaultStalledAfter is the StalledAfter an engine runs with when WithRecovery doesn't set one.

Variables

This section is empty.

Functions

This section is empty.

Types

type AssembleParams 

type AssembleParams struct {
	// MaxTokens is the token budget. Zero or less uses 4096.
	MaxTokens int
}

AssembleParams configures Assemble.

type AssembledContext 

type AssembledContext struct {
	Context     string `json:"context"`
	TotalTokens int    `json:"total_tokens"`
	MaxTokens   int    `json:"max_tokens"`
	// Included lists the 0-based positions of the hits that made it in, in
	// order. Marker [n] in Context is hit Included[n-1].
	Included []int `json:"included"`
	// FirstExcluded is the first hit position that was not included, whether
	// because the budget ran out or because the hit had no chunk to assemble;
	// -1 when every hit was included. Hits after it may still be included.
	FirstExcluded int `json:"first_excluded"`
	// TokenCounter names how tokens were estimated.
	TokenCounter string `json:"token_counter"`
}

AssembledContext is what Weave's default assembler builds from a ranking. It is not necessarily what an application sends: an app may assemble its own way.

type ChunkRef 

type ChunkRef struct {
	ChunkID id.ChunkID `json:"chunk_id"`
	Score   float64    `json:"score"`
}

ChunkRef names a chunk and the score it had in some earlier ranking.

type CollectionStatsResult 

type CollectionStatsResult struct {
	CollectionID   id.CollectionID `json:"collection_id"`
	CollectionName string          `json:"collection_name"`
	DocumentCount  int64           `json:"document_count"`
	ChunkCount     int64           `json:"chunk_count"`
	EmbeddingModel string          `json:"embedding_model"`
	ChunkStrategy  string          `json:"chunk_strategy"`
}

CollectionStatsResult holds aggregate statistics for a collection.

type CollectionUpdate 

type CollectionUpdate struct {
	Name        *string
	Description *string
	Metadata    *map[string]string
}

CollectionUpdate changes the parts of a collection that are safe to change after ingestion. A nil field is left alone. Chunk size, overlap and the recorded embedding model, dimensions and strategy are deliberately absent: Weave cannot re-chunk existing documents, so changing them would leave every existing chunk built under different settings from new ones.

type CompareHit 

type CompareHit struct {
	ScoredChunk
	Rank        int     `json:"rank"`
	VectorRank  int     `json:"vector_rank"`
	VectorScore float64 `json:"vector_score"`
}

CompareHit is a hit with its place in both rankings. VectorRank is 1-based; 0 means the chunk was not in the raw window at all.

type CompareParams 

type CompareParams struct {
	// CollectionID restricts both searches. Zero means every collection.
	CollectionID id.CollectionID
	// Tenant names the tenant to search. With no tenant on the context a
	// non-empty value becomes the tenant; with one, a different value
	// (including "") is ErrInvalidArgument. Both searches always run within
	// one tenant.
	//
	// Deprecated: put the tenant on the context with weave.WithTenant.
	Tenant *string
	// TopK is how many final hits to return. Zero or less uses the config
	// default.
	TopK int
	// MinScore applies to the final hits only. The raw search ignores it, so
	// the result can say what it removed.
	MinScore float64
}

CompareParams configures RetrieveCompare.

type CompareResult 

type CompareResult struct {
	Hits []CompareHit `json:"hits"`
	// LeftOut are raw-window hits the final ranking skipped over: hits not in
	// Hits whose vector rank is better than the worst vector rank in Hits (or
	// every such hit when a final hit sits outside the raw window or has no
	// chunk ID). In vector order, at most TopK of them.
	LeftOut []CompareHit `json:"left_out"`
	// Window is how many raw hits were asked for.
	Window int `json:"window"`
	// VectorMatches is how many raw hits came back, before MinScore.
	VectorMatches int `json:"vector_matches"`
	// BestVectorScore is the top raw score, or 0 with no matches.
	BestVectorScore float64 `json:"best_vector_score"`
	// Reordered is true when this query's final ranking differs from the raw
	// ranking: some hit's Rank is not its VectorRank, or LeftOut is not empty.
	// It says nothing about the retriever kind; read that from Components().
	Reordered bool `json:"reordered"`
	// SameSearch is true when no retriever is configured, so both sides are
	// derived from one vector search and cannot disagree.
	SameSearch bool `json:"same_search"`
	// Score is what Hits[i].Score means.
	Score           weave.ScoreKind `json:"score"`
	RetrieverMillis float64         `json:"retriever_ms"`
	VectorMillis    float64         `json:"vector_ms"`
}

CompareResult is the configured ranking beside the raw vector ranking.

type Component 

type Component struct {
	weave.ComponentInfo
	Type         string   `json:"type,omitempty"`
	Configured   bool     `json:"configured"`
	ContentTypes []string `json:"content_types,omitempty"`
	Dimensions   int      `json:"dimensions,omitempty"`
}

Component is one wired pipeline stage.

type Components 

type Components struct {
	Loader      Component `json:"loader"`
	Chunker     Component `json:"chunker"`
	Embedder    Component `json:"embedder"`
	VectorStore Component `json:"vector_store"`
	// Retriever is unconfigured when the engine falls back to a plain vector
	// search.
	Retriever Component `json:"retriever"`
	// Score is what a retrieval score means on this engine.
	Score weave.ScoreKind `json:"score"`
	// TenantFilter is "verified" when the vector store's exact tenant filter
	// is known to work, otherwise "unverified".
	TenantFilter string `json:"tenant_filter"`
}

Components reports what the engine actually runs.

type Engine 

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

Engine is the central coordinator for the Weave RAG pipeline.

func New 

func New(opts ...Option) (*Engine, error)

New creates a new Engine with the given options.

func (*Engine) Assemble 

func (e *Engine) Assemble(ctx context.Context, hits []ScoredChunk, p AssembleParams) (*AssembledContext, error)

Assemble runs the default assembler over hits.

func (*Engine) AssembleRefs 

func (e *Engine) AssembleRefs(ctx context.Context, refs []ChunkRef, p AssembleParams) (*AssembledContext, []ScoredChunk, error)

AssembleRefs reads each chunk back by ID within the context tenant and assembles them in the given order, without embedding anything. A missing chunk is an error: the ranking it came from is stale.

func (*Engine) CollectionStats 

func (e *Engine) CollectionStats(ctx context.Context, colID id.CollectionID) (*CollectionStatsResult, error)

CollectionStats returns aggregate statistics for a collection.

func (*Engine) Components 

func (e *Engine) Components() Components

Components describes every stage. A stage that is not configured reports Configured false, so a page can say so rather than call it active.

func (*Engine) Config 

func (e *Engine) Config() weave.Config

Config returns a copy of the engine's configuration.

func (*Engine) CountChunks 

func (e *Engine) CountChunks(ctx context.Context, filter *chunk.CountFilter) (int64, error)

CountChunks counts the context tenant's chunks matching the filter.

func (*Engine) CountCollections 

func (e *Engine) CountCollections(ctx context.Context, filter *collection.CountFilter) (int64, error)

CountCollections counts the context tenant's collections matching the filter.

func (*Engine) CountDocuments 

func (e *Engine) CountDocuments(ctx context.Context, filter *document.CountFilter) (int64, error)

CountDocuments counts the context tenant's documents matching the filter.

func (*Engine) CreateCollection 

func (e *Engine) CreateCollection(ctx context.Context, col *collection.Collection) error

CreateCollection creates a new collection.

func (*Engine) DeleteBySource 

func (e *Engine) DeleteBySource(ctx context.Context, colID id.CollectionID, sourceKey string) error

DeleteBySource removes every document of sourceKey in the collection, with their chunks and vectors. Nothing to delete is not an error.

It runs within the context tenant: another tenant's collection is ErrCollectionNotFound, and a source key that is empty, blank or longer than 1024 bytes is ErrInvalidArgument, both before anything is deleted. The source's vectors go first, filtered by collection, source and tenant, then each document's vectors, chunks and row. A version still being ingested when the documents are listed loses its row too, so that ingest fails at the swap instead of finishing; one that starts after the list is a new version and is kept.

func (*Engine) DeleteCollection 

func (e *Engine) DeleteCollection(ctx context.Context, colID id.CollectionID) error

DeleteCollection removes a collection and all its documents, chunks and vectors. Another tenant's collection is ErrCollectionNotFound, and nothing is deleted.

func (*Engine) DeleteDocument 

func (e *Engine) DeleteDocument(ctx context.Context, docID id.DocumentID) error

DeleteDocument removes a document and its chunks from both stores. Another tenant's document is ErrDocumentNotFound, and nothing is deleted.

func (*Engine) DescribeExtensions 

func (e *Engine) DescribeExtensions() []ExtensionInfo

DescribeExtensions lists registered extensions in registration order.

func (*Engine) Extensions 

func (e *Engine) Extensions() *plugins.Registry

Extensions returns the extension registry.

func (*Engine) GetChunk 

func (e *Engine) GetChunk(ctx context.Context, chunkID id.ChunkID) (*chunk.Chunk, error)

GetChunk reads one chunk within the context tenant.

func (*Engine) GetCollection 

func (e *Engine) GetCollection(ctx context.Context, colID id.CollectionID) (*collection.Collection, error)

GetCollection retrieves a collection by ID within the context tenant.

func (*Engine) GetCollectionByName 

func (e *Engine) GetCollectionByName(ctx context.Context, name string) (*collection.Collection, error)

GetCollectionByName retrieves a collection by name within the context tenant.

func (*Engine) GetDocument 

func (e *Engine) GetDocument(ctx context.Context, docID id.DocumentID) (*document.Document, error)

GetDocument retrieves a document by ID within the context tenant.

func (*Engine) Health 

func (e *Engine) Health(ctx context.Context) error

Health checks the health of the engine by pinging its store.

func (*Engine) HybridSearch 

func (e *Engine) HybridSearch(ctx context.Context, query string, params *HybridSearchParams) ([]ScoredChunk, error)

HybridSearch performs retrieval across one or more of the context tenant's collections, or all of them when none is named. A named collection the tenant doesn't have is ErrCollectionNotFound, before anything is embedded.

func (*Engine) Ingest 

func (e *Engine) Ingest(ctx context.Context, input *IngestInput) (*IngestResult, error)

Ingest processes a single document: load, chunk, embed, and store. With a SourceKey it is an upsert of that source (see IngestInput.SourceKey): the same content again embeds nothing and answers Unchanged, and new content replaces the source's current document, whose ID is then in Replaced. Content another ingest is still working on is answered with that document in its current state, pending or processing, and nothing embedded.

The ingest sends a heartbeat as it goes (see RecoveryConfig). When its document stops being pending or processing under it, because recovery failed it or it was deleted, Ingest deletes what it wrote and returns weave.ErrIngestInterrupted.

func (*Engine) IngestBatch 

func (e *Engine) IngestBatch(ctx context.Context, inputs []*IngestInput) ([]*IngestResult, error)

IngestBatch ingests multiple documents into a collection.

func (*Engine) ListChunks 

func (e *Engine) ListChunks(ctx context.Context, filter *chunk.ListFilter) ([]*chunk.Chunk, error)

ListChunks pages the context tenant's chunks by document or by collection. A filter naming neither is refused: an unscoped chunk listing is a full table scan nobody asked for.

func (*Engine) ListCollections 

func (e *Engine) ListCollections(ctx context.Context, filter *collection.ListFilter) ([]*collection.Collection, error)

ListCollections returns the context tenant's collections matching the given filter.

func (*Engine) ListDocuments 

func (e *Engine) ListDocuments(ctx context.Context, filter *document.ListFilter) ([]*document.Document, error)

ListDocuments returns the context tenant's documents matching the given filter.

func (*Engine) Logger 

func (e *Engine) Logger() log.Logger

Logger returns the engine's logger.

func (*Engine) RecoverStalled 

func (e *Engine) RecoverStalled(ctx context.Context, stalledAfter time.Duration) (RecoveryResult, error)

RecoverStalled runs one sweep over every tenant. Each stalled document is handled under its own tenant. A pending or processing row is marked failed with weave.ErrIngestInterrupted's message, and only then are its vectors and chunks deleted. A superseded row loses its vectors and chunks, then the row. Safe to run on every replica.

A document is stalled when it was last updated more than stalledAfter ago; zero means anything updated before now, and a negative value counts as zero, so a sweep never reaches past the present. The sweep lists at most BatchSize of them, oldest first, and needs no tenant on ctx. Marking failed comes first and is a guarded write: it only succeeds while the document is still pending or processing and has sent no heartbeat since the cutoff. When it loses (an ingest that came back to life sent a heartbeat, or made the document ready), the sweep leaves the document and all its data alone. Of two sweeps that list one document, only one counts it and fires the IngestFailed hook. Rows without a tenant are never listed; SkippedNoTenant is their count. A document whose data can't be deleted, or a superseded row that fails to go, returns the first such error after the rest of the batch is done.

This is the reverse of the order in the plan, which deleted first: that order could delete the data of an ingest that sent a heartbeat between the listing and the guarded write, and let it turn ready with chunks missing.

func (*Engine) RecoveryConfig 

func (e *Engine) RecoveryConfig() RecoveryConfig

RecoveryConfig returns the recovery settings the engine runs with: the ones WithRecovery gave, with a StalledAfter of zero or less replaced by DefaultStalledAfter, or no background loop and DefaultStalledAfter.

func (*Engine) ReindexCollection 

func (e *Engine) ReindexCollection(ctx context.Context, colID id.CollectionID) error

ReindexCollection re-embeds the chunks of every ready document in a collection and replaces each document's vectors in turn. Documents that aren't ready (an ingest still in flight, a failed one) and their vectors are left alone.

func (*Engine) Retrieve 

func (e *Engine) Retrieve(ctx context.Context, query string, opts ...RetrieveOption) ([]ScoredChunk, error)

Retrieve performs a semantic retrieval query within the context tenant, and never returns another tenant's chunks. A collection named with WithCollection that the tenant doesn't have is ErrCollectionNotFound, before the query is embedded. Each hit is read back from the metadata store by chunk ID; a hit whose row is missing is kept and marked Orphaned, as long as its vector is labelled with the tenant and carries no source_key. A source-keyed hit without a row is dropped, and the hits of one source all come from one version of it (see oneVersionPerSource).

func (*Engine) RetrieveCompare 

func (e *Engine) RetrieveCompare(ctx context.Context, query string, p CompareParams) (*CompareResult, error)

RetrieveCompare runs the configured retriever and a raw vector search over a wider window, within the context tenant, and reports where each final hit sat in the raw ranking. With no retriever configured it embeds the query once and derives both sides from the same search.

It needs the engine's own embedder and vector store, so a retriever-only engine is refused. The raw side searches the engine's vector store, so a retriever wired to a different store will not line up with it.

func (*Engine) Start 

func (e *Engine) Start(ctx context.Context) error

Start starts the engine's background work: with a RecoveryConfig Interval above zero and a store, the recovery loop, which runs RecoverStalled every Interval until Stop. Calling it again while the loop runs does nothing.

func (*Engine) Stop 

func (e *Engine) Stop(ctx context.Context) error

Stop gracefully shuts down the engine. It stops the recovery loop and waits for a sweep in progress to return before it closes the store.

func (*Engine) Store 

func (e *Engine) Store() store.Store

Store returns the engine's metadata store.

func (*Engine) UpdateCollection 

func (e *Engine) UpdateCollection(ctx context.Context, colID id.CollectionID, upd CollectionUpdate) (*collection.Collection, error)

UpdateCollection applies upd to one of the context tenant's collections and returns the stored collection.

type ExtensionInfo 

type ExtensionInfo struct {
	Name  string   `json:"name"`
	Hooks []string `json:"hooks"`
}

ExtensionInfo names a registered extension and the lifecycle hooks it implements. Extensions expose only a name, so hooks are found by type assertion against each hook interface.

type HybridSearchParams 

type HybridSearchParams struct {
	// Collections restricts search to these collection IDs.
	Collections []id.CollectionID `json:"collections,omitempty"`
	// TopK is the maximum number of results.
	TopK int `json:"top_k"`
	// Strategy is the search strategy name.
	Strategy string `json:"strategy,omitempty"`
	// MinScore is the minimum relevance score threshold.
	MinScore float64 `json:"min_score"`
}

HybridSearchParams configures a hybrid search across multiple collections.

type IngestInput 

type IngestInput struct {
	// CollectionID is the target collection.
	CollectionID id.CollectionID `json:"collection_id"`
	// Title is an optional document title.
	Title string `json:"title,omitempty"`
	// Source is the document source identifier (URL, path, etc.).
	Source string `json:"source,omitempty"`
	// SourceType is the MIME type or format hint.
	SourceType string `json:"source_type,omitempty"`
	// Content is the raw document content.
	Content string `json:"content"`
	// Metadata is optional document metadata.
	Metadata map[string]string `json:"metadata,omitempty"`
	// SourceKey makes the ingest an upsert: the new content replaces this
	// source's current document in the collection. Up to 1024 bytes, not blank.
	SourceKey string `json:"source_key,omitempty"`
}

IngestInput describes a document ingestion request.

type IngestResult 

type IngestResult struct {
	DocumentID id.DocumentID   `json:"document_id"`
	ChunkCount int             `json:"chunk_count"`
	State      document.State  `json:"state"`
	Unchanged  bool            `json:"unchanged,omitempty"` // same source, same content: nothing was embedded
	Replaced   []id.DocumentID `json:"replaced,omitempty"`  // previous versions this ingest superseded
}

IngestResult contains the outcome of a document ingestion.

type Option 

type Option func(*Engine) error

Option configures the Engine.

func WithChunker 

func WithChunker(c chunker.Chunker) Option

WithChunker sets the chunker.

func WithConfig 

func WithConfig(cfg weave.Config) Option

WithConfig sets the engine configuration.

func WithEmbedder 

func WithEmbedder(emb embedder.Embedder) Option

WithEmbedder sets the embedder.

func WithExtension 

func WithExtension(extension ext.Extension) Option

WithExtension registers an extension with the engine.

func WithLoader 

func WithLoader(l loader.Loader) Option

WithLoader sets the default document loader.

func WithLogger 

func WithLogger(l log.Logger) Option

WithLogger sets the structured logger.

func WithMetrics 

func WithMetrics(m gometrics.Metrics) Option

WithMetrics records the engine's metrics on m: ingest, retrieve and vector store durations, orphaned retrieve hits, and the OpenTelemetry GenAI client metrics for every embedding call the engine makes. A nil m records nothing, which is the default. The Forge extension passes the app's metrics. Spans need no option: they go to the global tracer provider.

func WithRecovery 

func WithRecovery(cfg RecoveryConfig) Option

WithRecovery sets how the engine recovers documents a crash left pending, processing or superseded; see RecoveryConfig. With an Interval above zero, Start runs RecoverStalled every Interval until Stop. StalledAfter is also how old a keyed ingest's in-progress document of the same content must be before another ingest claims it. A zero or negative StalledAfter is DefaultStalledAfter, as is leaving this option out, which also runs no loop.

func WithRetriever 

func WithRetriever(r retriever.Retriever) Option

WithRetriever sets the retriever.

func WithStore 

func WithStore(s store.Store) Option

WithStore sets the metadata store.

func WithVectorStore 

func WithVectorStore(vs vectorstore.VectorStore) Option

WithVectorStore sets the vector store.

type RecoveryConfig 

type RecoveryConfig struct {
	Interval     time.Duration // 0 or less: no background loop
	StalledAfter time.Duration
	BatchSize    int // per sweep; <= 0 means 100
}

RecoveryConfig sets how the engine recovers documents an ingest left behind when its process died: pending or processing documents that never finished, and superseded ones a replace never purged.

StalledAfter is how long a pending or processing document may go without a heartbeat before it counts as stalled. An ingest sends one when it starts, after chunking, after embedding and right before it marks the document ready, so StalledAfter must stay well above the longest single chunk or embed step. Recovery fails a document older than that, and a keyed ingest of the same content claims it instead of waiting.

Every zero means "the default": Interval 0 or less runs no loop, StalledAfter 0 or less is DefaultStalledAfter, and BatchSize 0 or less is 100. So RecoveryConfig{Interval: 5 * time.Minute} sweeps every 5 minutes with a 15 minute cutoff. A test that wants every document stalled at once sets StalledAfter to a nanosecond.

type RecoveryResult 

type RecoveryResult struct {
	Failed          int // pending or processing documents marked failed
	Purged          int // superseded documents removed
	SkippedNoTenant int // rows without a tenant (written before v1.9), left alone
}

RecoveryResult counts what one RecoverStalled sweep did.

type RetrieveOption 

type RetrieveOption func(*RetrieveParams)

RetrieveOption configures a retrieval query.

func WithCollection 

func WithCollection(colID id.CollectionID) RetrieveOption

WithCollection restricts retrieval to a specific collection.

func WithMinScore 

func WithMinScore(score float64) RetrieveOption

WithMinScore sets the minimum relevance score threshold.

func WithStrategy 

func WithStrategy(strategy string) RetrieveOption

WithStrategy sets the retrieval strategy (e.g. "similarity", "mmr", "hybrid").

func WithTenantFilter deprecated 

func WithTenantFilter(tenant string) RetrieveOption

WithTenantFilter names the tenant to retrieve for, as WithTenantID does, except that "" is a value: with a tenant on the context it is ErrInvalidArgument, and without one it is ErrTenantRequired.

Deprecated: put the tenant on the context with weave.WithTenant.

func WithTenantID deprecated 

func WithTenantID(tenantID string) RetrieveOption

WithTenantID names the tenant to retrieve for. With no tenant on the context it becomes the tenant; with one, it must equal it, or Retrieve returns ErrInvalidArgument. "" names nothing.

Deprecated: put the tenant on the context with weave.WithTenant.

func WithTopK 

func WithTopK(k int) RetrieveOption

WithTopK sets the maximum number of results.

type RetrieveParams 

type RetrieveParams struct {
	CollectionID string `json:"collection_id,omitempty"`
	// TenantID is set by the deprecated WithTenantID. Retrieval always runs
	// within the context tenant; see WithTenantID.
	TenantID string  `json:"tenant_id,omitempty"`
	TopK     int     `json:"top_k"`
	MinScore float64 `json:"min_score"`
	Strategy string  `json:"strategy,omitempty"`
	// TenantFilter is set by the deprecated WithTenantFilter; see there.
	TenantFilter *string `json:"-"`
}

RetrieveParams holds parsed retrieval parameters.

type ScoredChunk 

type ScoredChunk struct {
	Chunk *chunk.Chunk `json:"chunk"`
	Score float64      `json:"score"`
	// Hydrated is true when Chunk was read back from the metadata store by
	// ID, so every field on it is real.
	Hydrated bool `json:"hydrated"`
	// Orphaned is true when the vector store returned a chunk ID that has no
	// row in the metadata store. The hit keeps its rank and its text. A
	// source-keyed document's vector with no row is dropped instead, since
	// it belongs to a version not yet swapped in or already replaced.
	Orphaned bool `json:"orphaned,omitempty"`
}

ScoredChunk is a chunk with its relevance score.

Source Files

View all Source files

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.