worker

package
v0.7.0 Latest Latest
Warning

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

Go to latest
Published: Mar 10, 2026 License: Apache-2.0 Imports: 15 Imported by: 0

Documentation

Overview

Package worker provides a background job processing pool.

Index

Constants

View Source
const (
	// DefaultChunkSize is the default chunk size in characters.
	DefaultChunkSize = 2048
	// DefaultChunkOverlap is the default overlap in characters.
	DefaultChunkOverlap = 200
)
View Source
const DefaultCompactionSystemPrompt = `` /* 429-byte string literal not displayed */

DefaultCompactionSystemPrompt instructs the LLM to merge multiple chunks into a coherent summary.

View Source
const DefaultCompactionUserPrompt = "Merge the following %d chunks into a single coherent summary:\n\n%s"

DefaultCompactionUserPrompt is the template for the user message in compaction. The first %d is replaced with the number of chunks; the second %s is replaced with their content.

View Source
const DefaultExtractionSystemPrompt = `` /* 1142-byte string literal not displayed */

DefaultExtractionSystemPrompt is the system prompt for extracting facts from conversation chunks.

View Source
const DefaultExtractionUserPrompt = "Extract facts from the following conversation:\n\n%s"

DefaultExtractionUserPrompt is the template for the user message in fact extraction. The placeholder %s is replaced with the chunk content.

View Source
const DefaultMaintenanceSystemPrompt = `` /* 986-byte string literal not displayed */

DefaultMaintenanceSystemPrompt is the system prompt for deciding how to handle a candidate fact relative to existing memories.

View Source
const DefaultMaintenanceUserPrompt = "Candidate fact: %s\n\nExisting memories:\n%s"

DefaultMaintenanceUserPrompt is the template for the user message in memory maintenance. The first %s is replaced with the candidate fact. The second %s is replaced with the formatted existing memories.

Variables

This section is empty.

Functions

func ExtractText

func ExtractText(data []byte, contentType string) (string, error)

ExtractText extracts text content from raw bytes based on the content type. Supported types: text/plain (passthrough), text/html (tag stripping). If contentType is empty, the type is detected from the data using http.DetectContentType.

func SplitText

func SplitText(text string, chunkSize, overlap int) []string

SplitText splits text into chunks with overlap. If the text is shorter than or equal to chunkSize, it returns a single chunk. The overlap ensures context continuity between adjacent chunks.

Types

type ChunkingWorker

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

ChunkingWorker splits extracted text into chunks and creates the next pipeline job.

func NewChunkingWorker

func NewChunkingWorker(docStore *store.DocumentStore, chunkStore *store.ChunkStore, jobStore *store.JobStore, topicStore *store.TopicStore, llmProvider llm.Provider) *ChunkingWorker

NewChunkingWorker creates a new chunking worker. The llmProvider is optional; it is only needed when a topic uses semantic chunking. Pass nil if semantic chunking is not required.

func (*ChunkingWorker) Process

func (w *ChunkingWorker) Process(ctx context.Context, job *store.ProcessingJob) error

Process splits the document's extracted text into chunks and enqueues an embedding job.

func (*ChunkingWorker) Type

func (w *ChunkingWorker) Type() string

Type returns the job type this worker handles.

type CompactionWorker added in v0.5.0

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

CompactionWorker merges multiple chunks into a single summary chunk using an LLM.

func NewCompactionWorker added in v0.5.0

func NewCompactionWorker(
	chunkStore *store.ChunkStore,
	linkStore *store.LinkStore,
	compactionStore *store.CompactionStore,
	docStore *store.DocumentStore,
	jobStore *store.JobStore,
	vectorBackend vector.Backend,
	embedProvider EmbeddingProvider,
	llmProvider llm.Provider,
) *CompactionWorker

NewCompactionWorker creates a new compaction worker.

func (*CompactionWorker) Process added in v0.5.0

func (w *CompactionWorker) Process(ctx context.Context, job *store.ProcessingJob) error

Process compacts chunks for a document using an LLM to synthesize a summary.

func (*CompactionWorker) Type added in v0.5.0

func (w *CompactionWorker) Type() string

Type returns the job type this worker handles.

type EmbeddingProvider

type EmbeddingProvider interface {
	// Embed computes embeddings for the given texts.
	Embed(ctx context.Context, texts []string) ([][]float64, error)
	// Dimensions returns the number of dimensions this provider produces.
	Dimensions() int
}

EmbeddingProvider computes vector embeddings for text.

type EmbeddingWorker

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

EmbeddingWorker computes embeddings for document chunks and stores them.

func NewEmbeddingWorker

func NewEmbeddingWorker(
	docStore *store.DocumentStore,
	chunkStore *store.ChunkStore,
	topicStore *store.TopicStore,
	jobStore *store.JobStore,
	vectorBackend vector.Backend,
	provider EmbeddingProvider,
) *EmbeddingWorker

NewEmbeddingWorker creates a new embedding worker.

func (*EmbeddingWorker) Process

func (w *EmbeddingWorker) Process(ctx context.Context, job *store.ProcessingJob) error

Process computes embeddings for all chunks without embeddings, stores them, and marks the document as ready.

func (*EmbeddingWorker) Type

func (w *EmbeddingWorker) Type() string

Type returns the job type this worker handles.

type ExtractionWorker

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

ExtractionWorker extracts text from uploaded document content.

func NewExtractionWorker

func NewExtractionWorker(docStore *store.DocumentStore, jobStore *store.JobStore) *ExtractionWorker

NewExtractionWorker creates a new extraction worker.

func (*ExtractionWorker) Process

func (w *ExtractionWorker) Process(ctx context.Context, job *store.ProcessingJob) error

Process extracts text from the document's raw content.

func (*ExtractionWorker) Type

func (w *ExtractionWorker) Type() string

Type returns the job type this worker handles.

type MemoryExtractionWorker

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

MemoryExtractionWorker processes memory_extraction jobs by extracting candidate facts from conversation chunks using an LLM.

func NewMemoryExtractionWorker

func NewMemoryExtractionWorker(
	docStore *store.DocumentStore,
	chunkStore *store.ChunkStore,
	topicStore *store.TopicStore,
	jobStore *store.JobStore,
	llmProvider llm.Provider,
) *MemoryExtractionWorker

NewMemoryExtractionWorker creates a new memory extraction worker.

func (*MemoryExtractionWorker) Process

Process extracts candidate facts from a document's chunks and creates memory_maintenance jobs for each extracted fact.

func (*MemoryExtractionWorker) Type

func (w *MemoryExtractionWorker) Type() string

Type returns the job type this worker handles.

type MemoryMaintenanceWorker

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

MemoryMaintenanceWorker processes memory_maintenance jobs by evaluating a candidate fact against existing memories and executing the appropriate action.

func NewMemoryMaintenanceWorker

func NewMemoryMaintenanceWorker(
	memStore *store.MemoryStore,
	jobStore *store.JobStore,
	vectorBackend vector.Backend,
	embedProvider EmbeddingProvider,
	llmProvider llm.Provider,
) *MemoryMaintenanceWorker

NewMemoryMaintenanceWorker creates a new memory maintenance worker.

func (*MemoryMaintenanceWorker) Process

Process evaluates a candidate fact against existing memories and executes the appropriate action (ADD, UPDATE, DELETE, or NOOP).

func (*MemoryMaintenanceWorker) Type

func (w *MemoryMaintenanceWorker) Type() string

Type returns the job type this worker handles.

type OpenAIEmbeddingProvider added in v0.2.1

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

OpenAIEmbeddingProvider computes embeddings via the OpenAI API.

func NewOpenAIEmbeddingProvider added in v0.2.1

func NewOpenAIEmbeddingProvider(apiKey, model string, dimensions int) *OpenAIEmbeddingProvider

NewOpenAIEmbeddingProvider creates an embedding provider that calls OpenAI.

func (*OpenAIEmbeddingProvider) Dimensions added in v0.2.1

func (p *OpenAIEmbeddingProvider) Dimensions() int

Dimensions returns the embedding dimension.

func (*OpenAIEmbeddingProvider) Embed added in v0.2.1

func (p *OpenAIEmbeddingProvider) Embed(ctx context.Context, texts []string) ([][]float64, error)

Embed computes embeddings for the given texts via the OpenAI API.

type Pool

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

Pool manages background workers that poll for and process jobs.

func NewPool

func NewPool(jobStore *store.JobStore, concurrency int, pollInterval time.Duration, logger *slog.Logger) *Pool

NewPool creates a new worker pool.

func (*Pool) Register

func (p *Pool) Register(w Worker)

Register adds a worker for a job type. Must be called before Start.

func (*Pool) Start

func (p *Pool) Start(ctx context.Context)

Start begins polling goroutines. Each goroutine polls for jobs across all registered worker types.

func (*Pool) Stop

func (p *Pool) Stop()

Stop gracefully shuts down the pool, waiting for in-flight jobs with a timeout.

type StubEmbeddingProvider

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

StubEmbeddingProvider returns deterministic fake embeddings for testing. It produces vectors of the configured dimension where each element is a simple hash of the input text, normalized to [0,1].

func NewStubEmbeddingProvider

func NewStubEmbeddingProvider(dim int) *StubEmbeddingProvider

NewStubEmbeddingProvider creates a stub provider with the given dimension.

func (*StubEmbeddingProvider) Dimensions

func (s *StubEmbeddingProvider) Dimensions() int

Dimensions returns the embedding dimension.

func (*StubEmbeddingProvider) Embed

func (s *StubEmbeddingProvider) Embed(_ context.Context, texts []string) ([][]float64, error)

Embed returns deterministic embeddings based on the input text.

type Worker

type Worker interface {
	// Type returns the job type string that this worker handles.
	// It must match the job_type column in processing_jobs.
	Type() string

	// Process executes the job. Return nil on success or an error on failure.
	Process(ctx context.Context, job *store.ProcessingJob) error
}

Worker processes a specific type of job.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL