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)
- 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 ¶
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 )
const DefaultCacheCapacity = 1000
DefaultCacheCapacity is the default number of embeddings cached in memory.
Variables ¶
This section is empty.
Functions ¶
func ChunkText ¶ added in v2.3.0
ChunkText splits large content into overlapping chunks if it exceeds maxChars.
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 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
Embed returns a vector embedding for the given text, serving from LRU cache if available.
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", "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)
}
Config configures the embedding service.
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.
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) IsSaturated ¶
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:
- STOP — cancel runCtx: no new leases; in-flight embed/upsert abort and the intent is left leased for crash-recovery on next startup.
- 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).
- 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