embedding

package
v2.4.19 Latest Latest
Warning

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

 Go to latest Published: Oct 9, 2026 License: MIT Imports: 22 Imported by: 0

Documentation

Overview

Package embedding provides text-to-vector embedding for Cortex.

Embeddings enable semantic search: instead of matching keywords, search by meaning. The service supports multiple backends:

  • ollama: Local Ollama server (default: nomic-embed-text, 768 dims)
  • openai: OpenAI API (text-embedding-3-small, 1536 dims)
  • openai-compatible: any OpenAI-compatible /embeddings endpoint, built only through NewSecure (server mode) with an approved outbound policy
  • none: Disabled (default)

HTTP client lifecycle: each backend holds a single reusable *http.Client (constructed once in New) so HTTP keepalive connections are pooled across all Embed() calls rather than re-created per call. The concrete types (*ollamaService, *openAIService) implement io.Closer: Close() calls CloseIdleConnections on the shared client, reaping the Transport's persistConn read/write goroutines. The composition root (app.Close, bench Close) type-asserts to io.Closer to invoke it — the Service interface itself is NOT bloated with Close(), so test fakes and stubs need no changes.

Package embedding also provides the durable embedding worker (ADR-04, W4, REQ-EMB-001).

The worker drains the transactional outbox asynchronously: it leases pending embed+upsert intents, hydrates the observation text, embeds it with the configured model (versioned namespace), upserts the vector, records index namespace coverage, and marks the intent complete. Failures retry with capped exponential backoff up to max_attempts; terminal (non-retryable) failures go to dead-letter. The worker uses a bounded pool, drains on shutdown BEFORE DB.Close, and propagates cancellation — no detached fire-and-forget goroutines.

Index

Constants

View Source 
const (
	ClassAuth             = "auth-failure"
	ClassModelNotFound    = "model-not-found"
	ClassEndpointNotFound = "endpoint-not-found"
	ClassRateLimited      = "rate-limited"
	ClassServer           = "provider-server-error"
	ClassBadRequest       = "bad-request"
	ClassUnknown          = "unknown"
)

Error classes returned by APIError.Class. The model-not-found class exists because OpenAI-compatible gateways disagree on unknown-model signaling: first-party OpenAI returns 404 model_not_found, Nan returns 401 auth_error with a body that names the model.

View Source 
const (
	// MaxSingleEmbeddingLength is ~2000 words / 8000 chars, well within the 8191 token limit of modern models.
	MaxSingleEmbeddingLength = 8000
	// ChunkOverlapChars is the sliding window overlap for paragraph continuity.
	ChunkOverlapChars = 400
)
View Source 
const (
	// EnvIngestDedup enables the hook when set to "true" or "1".
	EnvIngestDedup = "CORTEX_INGEST_DEDUP"
	// EnvIngestDedupThreshold overrides the similarity threshold (0 < t <= 1).
	EnvIngestDedupThreshold = "CORTEX_INGEST_DEDUP_THRESHOLD"
	// DefaultDedupThreshold is the minimum cosine similarity for a
	// near-duplicate verdict.
	DefaultDedupThreshold = 0.95
	// RelationDuplicatesOf is the graph relation recorded from the duplicate
	// (new) observation to the pre-existing (canonical) observation.
	RelationDuplicatesOf = "duplicates_of"
)

Near-duplicate detection at ingest (default OFF).

After an observation is successfully embedded and upserted (the same seam the embedding worker uses), the optional dedup hook queries the vector index for the nearest neighbor above a similarity threshold and, if found on a DIFFERENT observation id, records a `duplicates_of` graph relation from the new observation to the existing one. Nothing is deleted or skipped: provenance is preserved, and stats can expose duplicate counts through the edge store.

Exact-scan cost: the vector search is an exact O(N·d) cosine scan over the tenant's vectors (at 4096 dims ≈ 8K multiply-adds per candidate row). At the current corpus scale this is acceptable; it runs once per embedded observation inside the worker, never on the interactive save path. Once dims ≤ 2000 and an ANN index exists (pgvector HNSW), the same query is index-backed.

View Source 
const DefaultCacheCapacity = 1000

DefaultCacheCapacity is the default number of embeddings cached in memory.

View Source 
const EnvLeaseBatch = "CORTEX_EMBEDDING_LEASE_BATCH"

EnvLeaseBatch is the composition-level override for WorkerConfig.LeaseBatch (REQ-RET-105). When the config field is unset (<= 0) and this environment variable holds a positive integer, the worker leases that many intents per poll, which makes the batch-embed path (processBatch) live: the leased intents hydrate and embed in ONE EmbedBatch call. Unset, invalid, or non-positive values keep defaultLeaseBatch (1), so the local-mode default is unchanged. Fail-closed saturation is independent of this knob: the save path still fails when PendingCount exceeds MaxBacklog (REQ-EMB-001) regardless of the configured LeaseBatch.

Variables

This section is empty.

Functions

func ChunkText added in v2.3.0 

func ChunkText(text string, maxChars, overlapChars int) []string

ChunkText splits large content into overlapping chunks if it exceeds maxChars.

func LeaseBatchFromEnv added in v2.4.15 

func LeaseBatchFromEnv(getenv func(string) string) int

LeaseBatchFromEnv parses EnvLeaseBatch through the injected getenv function. It returns 0 when the variable is unset, empty, non-numeric, or <= 0 — the caller then keeps its own default. Exported so composition paths (server background worker) can share the same knob semantics.

func PrepareObservationText added in v2.3.0 

func PrepareObservationText(obs *domain.Observation) string

PrepareObservationText builds the dense text payload for embedding. It enriches the text with contextual metadata (Anthropic Contextual Retrieval technique) so dense vector search retrieves relevant memories even with brief query keywords.

Types

type APIError added in v2.4.18 

type APIError struct {
	Provider string
	Status   int
	// Type is the provider error type/code (e.g. auth_error, model_not_found).
	Type string
	// Message is the bounded provider message; never contains credentials.
	Message string
}

APIError is a typed, secret-free provider error.

func ClassifyError added in v2.4.18 

func ClassifyError(err error) (*APIError, bool)

ClassifyError extracts a typed provider error from err when one is present.

func NewAPIError added in v2.4.18 

func NewAPIError(provider string, status int, body io.Reader) *APIError

NewAPIError parses a bounded provider error body into an APIError. It never fails: unparsable bodies degrade to a status-only error. The reader is NOT closed — the caller owns the body lifecycle.

func (*APIError) Class added in v2.4.18 

func (e *APIError) Class() string

Class buckets the error into an actionable category. The Nan quirk is codified here: a 401/403 whose bounded body names an unknown model classifies as model-not-found, NOT auth-failure.

func (*APIError) Error added in v2.4.18 

func (e *APIError) Error() string

type BatchEmbedder added in v2.3.10 

type BatchEmbedder interface {
	EmbedBatch(ctx context.Context, texts []string) ([][]float32, error)
}

BatchEmbedder optionally generates vector embeddings for multiple texts in a single batch request.

type CachedService added in v2.3.3 

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

CachedService wraps any Service with a high-performance, thread-safe LRU cache.

func NewCachedService added in v2.3.3 

func NewCachedService(inner Service, capacity int) *CachedService

NewCachedService creates a new CachedService wrapping inner with the given capacity. If capacity <= 0, DefaultCacheCapacity (1000) is used.

func (*CachedService) Close added in v2.3.3 

func (c *CachedService) Close() error

Close closes the underlying service if it implements io.Closer.

func (*CachedService) Dimensions added in v2.3.3 

func (c *CachedService) Dimensions() int

Dimensions returns the embedding dimension size.

func (*CachedService) Embed added in v2.3.3 

func (c *CachedService) Embed(ctx context.Context, text string) ([]float32, error)

Embed returns a vector embedding for the given text, serving from LRU cache if available.

func (*CachedService) EmbedBatch added in v2.3.10 

func (c *CachedService) EmbedBatch(ctx context.Context, texts []string) ([][]float32, error)

EmbedBatch returns embeddings for multiple texts, serving from LRU cache when possible and batching cache misses through inner's BatchEmbedder if supported.

func (*CachedService) Len added in v2.3.3 

func (c *CachedService) Len() int

Len returns current number of cached entries.

func (*CachedService) Model added in v2.3.3 

func (c *CachedService) Model() string

Model returns the model identifier.

type Config 

type Config struct {
	Provider string // "ollama", "openai", "openai-compatible", "none"
	APIKey   string // API key (OpenAI only; defaults to env var)
	Model    string // Model name override
	BaseURL  string // Base URL override (Ollama: default http://localhost:11434)
	// Dimensions requests a provider-native embedding dimension via the
	// OpenAI `dimensions` request parameter (MRL/Matryoshka models, e.g.
	// qwen3-embedding). 0 = provider default (4096 for qwen3-embedding).
	// Only honored by the openai/openai-compatible backends. Validated
	// upstream (config) to the 8..4096 range; providers may further
	// restrict to their matryoshka dimension list.
	Dimensions int
	// Budget is the shared provider rate budget handed out by the server
	// composition for this provider base URL, so embedding requests pace
	// against the same bucket as rerank and agent chat. nil = unlimited.
	Budget *ratelimit.Limiter
}

Config configures the embedding service.

type DedupConfig added in v2.4.19 

type DedupConfig struct {
	Enabled   bool
	Threshold float64
}

DedupConfig carries the resolved ingest-dedup settings.

func DedupConfigFromEnv added in v2.4.19 

func DedupConfigFromEnv() *DedupConfig

DedupConfigFromEnv resolves the CORTEX_INGEST_DEDUP* environment variables. Returns nil when the feature is OFF (the default). An unparsable or out-of-range threshold falls back to the documented default rather than disabling the feature.

type OutboundPolicy added in v2.3.0 

type OutboundPolicy struct {
	AllowedHosts              []string
	AllowedPorts              []int
	AllowLoopback             bool
	AllowInsecureLoopbackHTTP bool
	// RailwayInternalEmbeddingHost is one exact, administrator-configured
	// *.railway.internal hostname permitted to resolve to a private IP for
	// the embedding provider. All other private HTTP destinations stay denied.
	RailwayInternalEmbeddingHost string
	MaxRedirects                 int
	MaxResponseBodyBytes         int64
	MaxConcurrent                int
	Timeout                      time.Duration
}

func (*OutboundPolicy) ApproveDestination added in v2.3.0 

func (p *OutboundPolicy) ApproveDestination(raw string) error

type Service 

type Service interface {
	// Embed returns a vector embedding for the given text.
	Embed(ctx context.Context, text string) ([]float32, error)

	// Dimensions returns the embedding dimension size.
	Dimensions() int

	// Model returns the model identifier.
	Model() string
}

Service generates embeddings from text.

func New 

func New(cfg Config) Service

New creates an embedding service from config with a transparent in-memory LRU cache. Returns nil if provider is "none" or empty.

func NewSecure added in v2.3.0 

func NewSecure(cfg Config, policy OutboundPolicy) (Service, error)

type Worker 

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

Worker is the durable embedding worker. It drains the outbox with a bounded pool of goroutines, retrying failures with capped backoff and dead-lettering terminal failures. Start returns a cancel func that drains on shutdown.

func NewWorker 

func NewWorker(outbox outboxQueue, obs observationReader, embeddings Service, vectors vectorWriter, cfg WorkerConfig) *Worker

NewWorker creates an embedding worker. The concrete *sqlitestore.OutboxStore and *sqlitestore.Store satisfy the interface parameters structurally; the vectors parameter accepts any domain.VectorIndex implementation (the sqlite_blob adapter is the W8 default, ADR-05). The worker depends on the domain port, not the concrete vector store.

func (*Worker) EnableDedup added in v2.4.19 

func (w *Worker) EnableDedup(edges edgeCreator, cfg *DedupConfig)

EnableDedup wires the optional near-duplicate hook. Called by the composition root when the feature is configured; a nil cfg disables it. Call before Start; not goroutine-safe with concurrent processing.

func (*Worker) IsSaturated 

func (w *Worker) IsSaturated(ctx context.Context) (bool, error)

IsSaturated reports whether the outbox backlog exceeds the configured MaxBacklog threshold. The save path calls this before enqueuing to fail-closed under overload (REQ-EMB-001 saturation/overload behavior).

func (*Worker) Start 

func (w *Worker) Start(ctx context.Context) context.CancelFunc

Start begins processing outbox intents in a bounded pool of goroutines. It first recovers any intents left 'leased' by a crashed worker (crash recovery, REQ-EMB-001). The returned cancel func is the worker's STOP function and implements explicit cancel/stop/join semantics with TWO bounded contexts:

  • runCtx bounds leasing, hydration, embedding, and upsert work.
  • finalizeCtx bounds outcome-recording operations (MarkComplete/MarkFailed/ DeadLetter/UpdateIndexState).

The stop func executes three phases:

  1. STOP — cancel runCtx: no new leases; in-flight embed/upsert abort and the intent is left leased for crash-recovery on next startup.
  2. JOIN — wait for all worker goroutines to exit, bounded by DrainTimeout. finalizeCtx is still alive, so outcomes of in-flight intents are recorded normally (no silent loss on a graceful drain).
  3. CANCEL FINALIZE — unconditionally cancel finalizeCtx. After this returns, no goroutine can complete a finalize DB write: • joined goroutines have already exited; • a goroutine still unwinding an in-flight finalize observes ctx.Done and the ExecContext aborts BEFORE touching the DB; • a goroutine stuck in an embed/upsert (runCtx already cancelled) cannot reach finalize; its intent remains leased → recovery.

Therefore DB.Close MUST be called only after this stop func returns — and when it does, no goroutine is touching or can touch the DB (no write-after-close, no leaked DB-accessing goroutine).

type WorkerConfig 

type WorkerConfig struct {
	// Concurrency is the bounded worker pool size (goroutines). Default 2.
	Concurrency int
	// PollInterval is how long to wait before re-checking for new work when the
	// outbox is empty. Default 100ms.
	PollInterval time.Duration
	// LeaseBatch is how many intents a single worker leases per poll. Default 1.
	LeaseBatch int
	// MaxBacklog is the saturation threshold: when PendingCount exceeds this,
	// the save path fails-closed (REQ-EMB-001 saturation). Default 1000.
	MaxBacklog int
	// DrainTimeout bounds how long Start's cancel func waits for in-flight work
	// on shutdown. Default 30s.
	DrainTimeout time.Duration
}

WorkerConfig configures the embedding worker.

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.