BidRadar

Building the Embedding Generation Pipeline: Creating Semantic Representations of Organizational Knowledge

Introduction

In the previous article, we built the Knowledge Chunking Engine.

The platform can now transform large organizational documents into smaller, structured knowledge chunks that preserve:

  • Headings
  • Paragraph boundaries
  • Section hierarchy
  • Page references
  • Lists
  • Tables
  • Document metadata
  • Version relationships
  • Tenant ownership

These chunks are suitable for retrieval, but they are still stored as ordinary text.

Traditional keyword search can find exact terms, but tender requirements and organizational documents often describe the same concept using different language.

A tender may ask:

Describe the measures used to maintain service availability during major infrastructure failures.

The organization’s knowledge base may contain a section titled:

Business Continuity and Disaster Recovery

A simple keyword search may not recognize that these texts are closely related because the wording differs.

Semantic search solves this problem.

Instead of comparing exact words, semantic search compares the meaning of the tender requirement with the meaning of each knowledge chunk.

To make this possible, every chunk must be converted into a numerical representation known as an embedding.

An embedding is a vector—a sequence of numbers representing the semantic characteristics of text.

Conceptually:

"Business continuity and disaster recovery"
↓
Embedding Model
↓
[0.021, -0.143, 0.882, ...]

Texts with similar meanings produce vectors positioned close to one another in vector space.

The embedding pipeline therefore becomes the bridge between structured organizational knowledge and semantic retrieval.

In this article, we will build a production-ready embedding generation pipeline that:

  • Selects an embedding model through configuration
  • Supports multiple AI providers
  • Generates embeddings in batches
  • Stores vectors with each knowledge chunk
  • Tracks model and dimension versions
  • Retries temporary failures
  • Prevents duplicate processing
  • Respects tenant isolation
  • Tracks token usage and cost
  • Supports safe model migration
  • Monitors embedding quality
  • Queues work in background workers
  • Prepares the database for vector search

Development Progress

Proposal Generation Architecture     ████████████████████ 100%
Organization Knowledge Base          ████████████████████ 100%
Knowledge Ingestion Pipeline         ████████████████████ 100%
Knowledge Chunking Engine            ████████████████████ 100%
Embedding Generation Pipeline        ████████████████████ 100%
Vector Storage                       ░░░░░░░░░░░░░░░░░░░░   0%
Semantic Search                      ░░░░░░░░░░░░░░░░░░░░   0%
Hybrid Retrieval                     ░░░░░░░░░░░░░░░░░░░░   0%
Retrieval-Augmented Generation       ░░░░░░░░░░░░░░░░░░░░   0%
Proposal Templates                   ░░░░░░░░░░░░░░░░░░░░   0%
Requirement Extraction               ░░░░░░░░░░░░░░░░░░░░   0%
Compliance Matrix                    ░░░░░░░░░░░░░░░░░░░░   0%
Proposal Section Generation          ░░░░░░░░░░░░░░░░░░░░   0%
Proposal Review                      ░░░░░░░░░░░░░░░░░░░░   0%
Proposal Export                      ░░░░░░░░░░░░░░░░░░░░   0%

Why Embeddings Matter

Tender requirements and company knowledge rarely use identical wording.

Consider the following tender requirement:

The supplier must explain how critical services remain operational during regional outages.

The knowledge base may contain:

Our business continuity framework includes geographically separated recovery infrastructure, replicated data stores, and tested failover procedures.

These texts share few exact keywords.

However, they express closely related concepts:

  • Availability
  • Resilience
  • Failover
  • Disaster recovery
  • Business continuity

An embedding model converts both texts into vectors whose distance reflects their semantic relationship.

Tender Requirement Vector
≈
Knowledge Chunk Vector

This allows the platform to find relevant evidence even when terminology differs.

Keyword Search Versus Semantic Search

Keyword search asks:

Do these texts contain the same words?

Semantic search asks:

Do these texts express the same meaning?

Both approaches are valuable.

Later, the platform will combine:

Keyword Search
+
Vector Search
+
Metadata Filters
↓
Hybrid Retrieval

This article focuses on generating and managing the vectors needed for the vector-search component.

What Is an Embedding?

An embedding is an ordered list of floating-point numbers.

Example:

[
  0.0174,
  -0.2841,
  0.0482,
  0.7713,
  ...
]

The number of values is known as the embedding dimension.

Common vector sizes may include:

384 dimensions
768 dimensions
1024 dimensions
1536 dimensions
3072 dimensions

The dimension depends on the chosen embedding model.

Vectors from different models—or different dimensions—cannot normally be compared directly.

This means embedding model identity and version must be stored alongside every vector.

Embedding Pipeline Architecture

Knowledge Chunks
↓
Embedding Job Queue
↓
Embedding Worker
↓
Batch Builder
↓
Embedding Provider
↓
Vector Validation
↓
Embedding Storage
↓
Ready for Semantic Search

Complete Knowledge Processing Flow

The full knowledge pipeline now becomes:

Document Upload
↓
Malware Scan
↓
Text Extraction
↓
Cleaning and Normalization
↓
Chunk Generation
↓
Embedding Generation
↓
Vector Indexing
↓
Semantic Retrieval

Each stage must complete successfully before the next stage begins.

Design Principles

The embedding pipeline should follow several important principles.

Provider Independence

Application code should not depend directly on one embedding vendor.

Deterministic Input

The same chunk text and embedding configuration should produce a reusable result.

Idempotency

Retrying a job should not create duplicate active embeddings.

Version Awareness

Changing models must not silently mix incompatible vectors.

Tenant Isolation

Every embedding remains connected to an organization and source document.

Cost Visibility

The system must track token usage and embedding cost.

Batch Efficiency

Multiple chunks should be embedded together when supported.

Failure Recovery

Temporary provider failures should be retried safely.

Folder Structure

Add the following structure:

app/
├── ai/
│   └── embeddings/
│       ├── base.py
│       ├── registry.py
│       ├── schemas.py
│       ├── providers/
│       │   ├── openai_provider.py
│       │   ├── local_provider.py
│       │   └── mock_provider.py
│       └── exceptions.py
│
├── knowledge/
│   └── embeddings/
│       ├── batching.py
│       ├── service.py
│       ├── repository.py
│       ├── validator.py
│       ├── hashing.py
│       └── workers.py
│
└── models/
    └── knowledge_embedding.py

This separates generic AI provider integration from knowledge-specific orchestration.

Embedding Model Selection

The embedding model should be selected based on:

  • Retrieval quality
  • Supported languages
  • Vector dimension
  • Maximum input length
  • Latency
  • Batch capacity
  • Data-processing requirements
  • Deployment region
  • Cost
  • Provider reliability
  • Self-hosting requirements

The system should not hard-code a model name throughout the codebase.

Use configuration instead.

Embedding Configuration

Add settings such as:

EMBEDDING_PROVIDER
EMBEDDING_MODEL
EMBEDDING_DIMENSION
EMBEDDING_BATCH_SIZE
EMBEDDING_MAX_INPUT_TOKENS
EMBEDDING_TIMEOUT_SECONDS
EMBEDDING_MAX_RETRIES
EMBEDDING_NORMALIZE_VECTORS

Pydantic Settings

from pydantic import Field
from pydantic_settings import BaseSettings
class EmbeddingSettings(BaseSettings):
    provider: str = Field(
        default="openai",
        validation_alias="EMBEDDING_PROVIDER",
    )
    model: str = Field(
        validation_alias="EMBEDDING_MODEL"
    )
    dimension: int = Field(
        gt=0,
        validation_alias=(
            "EMBEDDING_DIMENSION"
        ),
    )
    batch_size: int = Field(
        default=64,
        ge=1,
        le=512,
        validation_alias=(
            "EMBEDDING_BATCH_SIZE"
        ),
    )
    max_input_tokens: int = Field(
        default=8191,
        ge=1,
        validation_alias=(
            "EMBEDDING_MAX_INPUT_TOKENS"
        ),
    )
    timeout_seconds: int = Field(
        default=60,
        ge=1,
        validation_alias=(
            "EMBEDDING_TIMEOUT_SECONDS"
        ),
    )
    max_retries: int = Field(
        default=5,
        ge=0,
        validation_alias=(
            "EMBEDDING_MAX_RETRIES"
        ),
    )
    normalize_vectors: bool = Field(
        default=True,
        validation_alias=(
            "EMBEDDING_NORMALIZE_VECTORS"
        ),
    )

Embedding Provider Interface

Create:

app/ai/embeddings/base.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
@dataclass(frozen=True)
class EmbeddingRequestItem:
    reference_id: str
    text: str
@dataclass(frozen=True)
class EmbeddingResultItem:
    reference_id: str
    vector: list[float]
    token_count: int | None
@dataclass(frozen=True)
class EmbeddingBatchResult:
    items: list[EmbeddingResultItem]
    provider: str
    model: str
    dimension: int
    total_tokens: int | None
class EmbeddingProvider(ABC):
    @abstractmethod
    def embed(
        self,
        *,
        items: list[EmbeddingRequestItem],
    ) -> EmbeddingBatchResult:
        raise NotImplementedError

The rest of the application should depend on this interface rather than a vendor SDK.

Why Use Reference IDs?

Providers may return results in positional order, but explicit reference identifiers make the application safer.

Example:

chunk:1001
chunk:1002
chunk:1003

When processing responses, the system can map each vector back to the correct chunk without relying only on array position.

Provider Registry

Create:

app/ai/embeddings/registry.py
class EmbeddingProviderRegistry:
    def __init__(
        self,
        providers: dict[
            str,
            EmbeddingProvider,
        ],
    ):
        self.providers = providers
    def get(
        self,
        provider_name: str,
    ) -> EmbeddingProvider:
        provider = self.providers.get(
            provider_name
        )
        if provider is None:
            raise UnsupportedEmbeddingProvider(
                provider_name
            )
        return provider

This allows the platform to support cloud and self-hosted models through the same service layer.

Preparing Text for Embedding

The text sent to the embedding model should include enough context to make the vector meaningful.

Embedding only the chunk body may lose important structural information.

A better input combines:

Document title
Category
Section hierarchy
Chunk heading
Chunk content

Example:

Document: Information Security Policy
Category: Security
Section: Incident Management > Incident Response
Content:
Our incident response team...

Embedding Input Builder

@dataclass(frozen=True)
class ChunkEmbeddingInput:
    chunk_id: int
    text: str
    input_hash: str
class ChunkEmbeddingInputBuilder:
    def build(
        self,
        *,
        chunk: KnowledgeChunk,
        document: KnowledgeDocument,
    ) -> ChunkEmbeddingInput:
        parts = [
            f"Document: {document.title}",
            f"Category: {document.category.value}",
        ]
        if chunk.section_path:
            parts.append(
                "Section: "
                + " > ".join(
                    chunk.section_path
                )
            )
        elif chunk.heading:
            parts.append(
                f"Section: {chunk.heading}"
            )
        parts.append(
            f"Content:\n{chunk.content}"
        )
        text = "\n".join(parts)
        return ChunkEmbeddingInput(
            chunk_id=chunk.id,
            text=text,
            input_hash=calculate_text_hash(
                text
            ),
        )

Why Hash the Embedding Input?

The chunk content may remain unchanged while metadata such as the heading or document title changes.

Because these values influence the embedding input, the system should hash the final constructed text rather than only the chunk body.

Embedding Input
↓
SHA-256
↓
Input Hash

If the input hash and model configuration have not changed, the existing embedding can be reused.

Hash Utility

import hashlib
def calculate_text_hash(
    text: str,
) -> str:
    normalized = text.strip().encode(
        "utf-8"
    )
    return hashlib.sha256(
        normalized
    ).hexdigest()

Embedding Database Design

There are two primary storage options.

Store the Vector on the Chunk

knowledge_chunks.embedding

This is simple but makes model migration and multiple embedding versions difficult.

Use a Separate Embedding Table

knowledge_embeddings

This supports:

  • Multiple models
  • Model migrations
  • Re-embedding
  • Historical records
  • Provider changes
  • A/B testing
  • Different retrieval indexes

For a production SaaS, use a separate table.

Knowledge Embedding Table

Create:

knowledge_embeddings

Recommended fields:

id
organization_id
knowledge_chunk_id
document_id
document_version_id
provider
model
model_revision
dimension
input_hash
vector
status
token_count
estimated_cost
error_code
error_message
generated_at
created_at
updated_at

Embedding Status

Use statuses such as:

PENDING
PROCESSING
READY
FAILED
SUPERSEDED
DELETED

Only READY embeddings should be included in semantic search.

SQLAlchemy Model

The exact vector column type will depend on the vector-storage implementation added in the next article.

For now, define the model conceptually with a vector-compatible column.

import enum
from datetime import datetime
from decimal import Decimal
from sqlalchemy import (
    DateTime,
    Enum,
    ForeignKey,
    Index,
    Integer,
    Numeric,
    String,
    Text,
    UniqueConstraint,
)
from sqlalchemy.orm import (
    Mapped,
    mapped_column,
)
from app.database.base import Base
class KnowledgeEmbeddingStatus(
    str,
    enum.Enum,
):
    PENDING = "pending"
    PROCESSING = "processing"
    READY = "ready"
    FAILED = "failed"
    SUPERSEDED = "superseded"
    DELETED = "deleted"
class KnowledgeEmbedding(Base):
    __tablename__ = (
        "knowledge_embeddings"
    )
    id: Mapped[int] = mapped_column(
        primary_key=True
    )
    organization_id: Mapped[int] = (
        mapped_column(
            ForeignKey(
                "organizations.id",
                ondelete="CASCADE",
            ),
            nullable=False,
            index=True,
        )
    )
    knowledge_chunk_id: Mapped[int] = (
        mapped_column(
            ForeignKey(
                "knowledge_chunks.id",
                ondelete="CASCADE",
            ),
            nullable=False,
            index=True,
        )
    )
    document_id: Mapped[int] = (
        mapped_column(
            ForeignKey(
                "knowledge_documents.id",
                ondelete="CASCADE",
            ),
            nullable=False,
            index=True,
        )
    )
    document_version_id: Mapped[int] = (
        mapped_column(
            ForeignKey(
                "knowledge_document_versions.id",
                ondelete="CASCADE",
            ),
            nullable=False,
            index=True,
        )
    )
    provider: Mapped[str] = mapped_column(
        String(100),
        nullable=False,
    )
    model: Mapped[str] = mapped_column(
        String(255),
        nullable=False,
    )
    model_revision: Mapped[
        str | None
    ] = mapped_column(
        String(255),
        nullable=True,
    )
    dimension: Mapped[int] = mapped_column(
        Integer,
        nullable=False,
    )
    input_hash: Mapped[str] = mapped_column(
        String(64),
        nullable=False,
    )
    status: Mapped[
        KnowledgeEmbeddingStatus
    ] = mapped_column(
        Enum(
            KnowledgeEmbeddingStatus,
            name=(
                "knowledge_embedding_status"
            ),
        ),
        default=(
            KnowledgeEmbeddingStatus.PENDING
        ),
        nullable=False,
        index=True,
    )
    token_count: Mapped[int | None] = (
        mapped_column(
            Integer,
            nullable=True,
        )
    )
    estimated_cost: Mapped[
        Decimal | None
    ] = mapped_column(
        Numeric(18, 8),
        nullable=True,
    )
    error_code: Mapped[
        str | None
    ] = mapped_column(
        String(100),
        nullable=True,
    )
    error_message: Mapped[
        str | None
    ] = mapped_column(
        Text,
        nullable=True,
    )
    generated_at: Mapped[
        datetime | None
    ] = mapped_column(
        DateTime(timezone=True),
        nullable=True,
    )
    created_at: Mapped[datetime] = (
        mapped_column(
            DateTime(timezone=True),
            nullable=False,
        )
    )
    updated_at: Mapped[datetime] = (
        mapped_column(
            DateTime(timezone=True),
            nullable=False,
        )
    )
    __table_args__ = (
        UniqueConstraint(
            "knowledge_chunk_id",
            "provider",
            "model",
            "input_hash",
            name=(
                "uq_knowledge_embedding_input"
            ),
        ),
        Index(
            "ix_embedding_org_model_status",
            "organization_id",
            "model",
            "status",
        ),
        Index(
            "ix_embedding_version_status",
            "document_version_id",
            "status",
        ),
    )

The vector column will be added when configuring pgvector or another vector backend.

Why Repeat Tenant and Document Identifiers?

The embedding links to a knowledge chunk, and the chunk already links to the organization and document.

However, storing these fields directly provides:

  • Faster tenant-scoped filtering
  • Simpler vector queries
  • Easier deletion
  • Better index design
  • Reduced join requirements
  • Stronger auditing

The service must verify that all identifiers belong to the same tenant and source hierarchy.

Vector Validation

Every provider response must be validated before storage.

Validate:

Result count equals request count
Reference IDs match
Vector is not empty
Dimension is correct
Values are numeric
Values are finite
Vector norm is valid
No missing chunks

Vector Validator

import math
class EmbeddingVectorValidator:
    def validate(
        self,
        *,
        vector: list[float],
        expected_dimension: int,
    ) -> None:
        if len(vector) != expected_dimension:
            raise EmbeddingDimensionMismatch(
                expected=expected_dimension,
                actual=len(vector),
            )
        if not vector:
            raise EmptyEmbeddingVector()
        for value in vector:
            if not isinstance(
                value,
                int | float,
            ):
                raise InvalidEmbeddingValue()
            if not math.isfinite(value):
                raise NonFiniteEmbeddingValue()

Vector Normalization

Some similarity functions work best with normalized vectors.

L2 normalization transforms a vector so its magnitude equals one.

vector
↓
divide by vector magnitude
↓
unit vector

Normalization Utility

import math
def l2_normalize(
    vector: list[float],
) -> list[float]:
    magnitude = math.sqrt(
        sum(
            value * value
            for value in vector
        )
    )
    if magnitude == 0:
        raise ZeroMagnitudeEmbedding()
    return [
        value / magnitude
        for value in vector
    ]

Whether application-side normalization is required depends on the model and similarity metric.

The decision must remain consistent across indexing and querying.

Similarity Metrics

Common vector similarity methods include:

Cosine Similarity
Inner Product
Euclidean Distance

The choice influences:

  • Vector normalization
  • Index configuration
  • Ranking behavior
  • Threshold calibration

A common starting point for normalized text embeddings is cosine similarity.

The next article will configure the vector storage and indexes.

Batch Generation

Sending one API request per chunk is inefficient.

Instead:

Chunk 1
Chunk 2
Chunk 3
...
Chunk 64
↓
One Embedding Request

Benefits include:

  • Lower network overhead
  • Better throughput
  • Lower latency per chunk
  • Fewer provider requests
  • More efficient worker execution

Batch Size Constraints

Batch size may be constrained by:

  • Maximum number of inputs
  • Maximum total tokens
  • Maximum request payload
  • Provider rate limits
  • Worker memory
  • Timeout limits

The batch builder must consider both item count and token count.

Batch Builder

@dataclass(frozen=True)
class EmbeddingBatch:
    items: list[ChunkEmbeddingInput]
    estimated_tokens: int
class EmbeddingBatchBuilder:
    def __init__(
        self,
        *,
        max_items: int,
        max_total_tokens: int,
    ):
        self.max_items = max_items
        self.max_total_tokens = (
            max_total_tokens
        )
    def build(
        self,
        inputs: list[
            ChunkEmbeddingInput
        ],
    ) -> list[EmbeddingBatch]:
        batches: list[EmbeddingBatch] = []
        current: list[
            ChunkEmbeddingInput
        ] = []
        current_tokens = 0
        for item in inputs:
            item_tokens = estimate_tokens(
                item.text
            )
            exceeds_items = (
                len(current)
                >= self.max_items
            )
            exceeds_tokens = (
                current
                and current_tokens
                + item_tokens
                > self.max_total_tokens
            )
            if (
                exceeds_items
                or exceeds_tokens
            ):
                batches.append(
                    EmbeddingBatch(
                        items=current,
                        estimated_tokens=(
                            current_tokens
                        ),
                    )
                )
                current = []
                current_tokens = 0
            current.append(item)
            current_tokens += item_tokens
        if current:
            batches.append(
                EmbeddingBatch(
                    items=current,
                    estimated_tokens=(
                        current_tokens
                    ),
                )
            )
        return batches

Oversized Chunk Handling

A chunk should already fall within configured token limits because the chunking engine controls size.

However, the embedding service must still defend against oversized inputs.

Possible actions:

Reject input
Re-chunk content
Truncate with warning
Mark processing failure

For knowledge quality, do not silently truncate.

The preferred behavior is:

Oversized chunk
↓
Mark chunk for reprocessing
↓
Generate smaller chunks
↓
Retry embedding

Embedding Repository

Create:

app/knowledge/embeddings/repository.py

The repository should support:

  • Finding reusable embeddings
  • Creating pending embedding records
  • Marking records as processing
  • Storing vectors
  • Marking failures
  • Superseding previous embeddings
  • Querying missing embeddings
  • Deleting embeddings by document version

Finding Reusable Embeddings

def find_ready_embedding(
    self,
    *,
    organization_id: int,
    chunk_id: int,
    provider: str,
    model: str,
    input_hash: str,
) -> KnowledgeEmbedding | None:
    statement = (
        select(KnowledgeEmbedding)
        .where(
            KnowledgeEmbedding
            .organization_id
            == organization_id,
            KnowledgeEmbedding
            .knowledge_chunk_id
            == chunk_id,
            KnowledgeEmbedding.provider
            == provider,
            KnowledgeEmbedding.model
            == model,
            KnowledgeEmbedding.input_hash
            == input_hash,
            KnowledgeEmbedding.status
            == KnowledgeEmbeddingStatus.READY,
        )
    )
    return self.session.scalar(
        statement
    )

This prevents unnecessary provider calls.

Idempotent Processing

The same background job may run more than once due to:

  • Worker restart
  • Queue redelivery
  • Timeout ambiguity
  • Manual retry
  • Deployment interruption

The embedding workflow must remain safe.

Job runs
↓
Ready embedding exists?
Yes → Skip
No → Generate

A unique database constraint provides an additional safety layer.

Concurrency Control

Two workers might attempt to embed the same chunk simultaneously.

Possible protection mechanisms include:

  • Database row locking
  • Unique constraints
  • Distributed locks
  • Job deduplication keys
  • Atomic status transitions

A practical approach is:

Create or lock embedding record
↓
Change PENDING to PROCESSING
↓
Only one worker succeeds

Atomic Claim Query

def claim_pending_embedding(
    self,
    *,
    embedding_id: int,
) -> bool:
    statement = (
        update(KnowledgeEmbedding)
        .where(
            KnowledgeEmbedding.id
            == embedding_id,
            KnowledgeEmbedding.status
            == KnowledgeEmbeddingStatus.PENDING,
        )
        .values(
            status=(
                KnowledgeEmbeddingStatus
                .PROCESSING
            ),
            updated_at=utc_now(),
        )
    )
    result = self.session.execute(
        statement
    )
    return result.rowcount == 1

Only the worker that successfully claims the record should process it.

Embedding Generation Service

Create:

app/knowledge/embeddings/service.py

The service coordinates:

  • Chunk loading
  • Tenant validation
  • Availability checks
  • Input construction
  • Hashing
  • Reuse detection
  • Batching
  • Provider calls
  • Validation
  • Cost calculation
  • Persistence
  • Usage tracking
  • Downstream indexing jobs

Service Workflow

Load Eligible Chunks
↓
Build Embedding Inputs
↓
Skip Existing Embeddings
↓
Create Pending Records
↓
Build Batches
↓
Call Provider
↓
Validate Vectors
↓
Normalize if Required
↓
Persist Results
↓
Track Usage
↓
Mark Chunks Ready

Embedding Service Example

class KnowledgeEmbeddingService:
    def generate_for_version(
        self,
        *,
        organization_id: int,
        document_version_id: int,
    ) -> EmbeddingGenerationSummary:
        chunks = (
            self.chunk_repository
            .list_for_embedding(
                organization_id=organization_id,
                document_version_id=(
                    document_version_id
                ),
            )
        )
        provider = self.provider_registry.get(
            self.settings.provider
        )
        pending_inputs = []
        for chunk in chunks:
            document = chunk.document
            embedding_input = (
                self.input_builder.build(
                    chunk=chunk,
                    document=document,
                )
            )
            existing = (
                self.embedding_repository
                .find_ready_embedding(
                    organization_id=(
                        organization_id
                    ),
                    chunk_id=chunk.id,
                    provider=(
                        self.settings.provider
                    ),
                    model=self.settings.model,
                    input_hash=(
                        embedding_input.input_hash
                    ),
                )
            )
            if existing:
                continue
            pending_inputs.append(
                embedding_input
            )
        batches = self.batch_builder.build(
            pending_inputs
        )
        generated_count = 0
        total_tokens = 0
        for batch in batches:
            result = provider.embed(
                items=[
                    EmbeddingRequestItem(
                        reference_id=(
                            f"chunk:{item.chunk_id}"
                        ),
                        text=item.text,
                    )
                    for item in batch.items
                ]
            )
            self._validate_batch_result(
                batch=batch,
                result=result,
            )
            self._persist_batch(
                organization_id=(
                    organization_id
                ),
                document_version_id=(
                    document_version_id
                ),
                batch=batch,
                result=result,
            )
            generated_count += len(
                result.items
            )
            total_tokens += (
                result.total_tokens or 0
            )
        return EmbeddingGenerationSummary(
            generated_count=generated_count,
            skipped_count=(
                len(chunks)
                - generated_count
            ),
            total_tokens=total_tokens,
        )

Production code should use explicit transactions and robust partial-batch failure handling.

Eligibility Rules

Not every chunk should be embedded.

The system should verify:

Document belongs to organization
Document version is processed
Chunk passed validation
Chunk is active
Document is not deleted
Document is eligible for indexing
Confidentiality policy permits provider use

Approval policy requires a deliberate decision.

Two possible strategies exist.

Embed Before Approval

Advantages:

  • Reviewers can test search
  • Approval becomes faster
  • Processing completes before publication

Disadvantages:

  • Unapproved content is sent to the embedding provider
  • Processing cost may be wasted

Embed After Approval

Advantages:

  • Only approved content is processed
  • Lower unnecessary AI cost
  • Stronger governance

Disadvantages:

  • Approval-to-availability delay

A practical design can generate embeddings after text review but keep them excluded from retrieval until approval.

Confidentiality-Aware Provider Routing

Restricted content may require a specific embedding environment.

Example:

Internal document
↓
Approved cloud provider
Confidential document
↓
Enterprise no-retention endpoint
Restricted document
↓
Self-hosted model

The provider router should evaluate:

  • Confidentiality level
  • Organization policy
  • Data residency
  • Subscription plan
  • Model availability
  • Regional requirements

Provider Routing Service

class EmbeddingProviderRouter:
    def select(
        self,
        *,
        organization: Organization,
        document: KnowledgeDocument,
    ) -> EmbeddingRoute:
        policy = (
            organization.ai_data_policy
        )
        if (
            document.confidentiality_level
            == ConfidentialityLevel.RESTRICTED
        ):
            return self._restricted_route(
                policy
            )
        return self._default_route(
            policy
        )

This keeps sensitive-data policy outside the generic embedding provider.

Background Worker

Embedding generation should run asynchronously.

Chunking Completed
↓
Embedding Job Queued
↓
Worker Claims Job
↓
Embeddings Generated
↓
Vector Indexing Ready

Worker Task Example

@task(
    name=(
        "knowledge.generate_embeddings"
    ),
    autoretry_for=(
        TemporaryEmbeddingError,
    ),
    retry_backoff=True,
    retry_jitter=True,
    max_retries=5,
)
def generate_embeddings_task(
    organization_id: int,
    document_version_id: int,
):
    service = (
        build_knowledge_embedding_service()
    )
    return service.generate_for_version(
        organization_id=organization_id,
        document_version_id=(
            document_version_id
        ),
    )

Do not pass document text or vectors through the queue payload.

Pass identifiers and load the records securely inside the worker.

Processing Job Integration

The existing knowledge_processing_jobs table already supports:

EMBEDDING

The worker should update the job lifecycle:

QUEUED
↓
RUNNING
↓
COMPLETED

or:

RUNNING
↓
RETRYING
↓
FAILED

Retry Strategy

Retry temporary failures such as:

  • Rate limits
  • Provider timeouts
  • Network errors
  • Temporary server errors
  • Connection resets
  • Short-lived database failures

Do not automatically retry permanent failures such as:

  • Unsupported model
  • Invalid API key
  • Dimension mismatch
  • Oversized input
  • Policy violation
  • Invalid chunk data

Retry Backoff

Use exponential backoff with jitter.

Example:

Attempt 1:
5 seconds
Attempt 2:
15 seconds
Attempt 3:
45 seconds
Attempt 4:
135 seconds

Jitter prevents many failed jobs from retrying simultaneously.

Partial Batch Failure

A batch request may fail entirely, or individual items may be invalid.

The pipeline should support:

Batch fails temporarily
↓
Retry entire batch

and:

One item permanently invalid
↓
Mark item failed
↓
Continue remaining items

This may require splitting a failed batch to isolate problematic inputs.

Batch Failure Isolation

Batch of 64 fails
↓
Split into two batches of 32
↓
Retry
↓
Continue splitting if required
↓
Identify invalid chunk

This prevents one bad item from blocking an entire document.

Embedding Cost Tracking

Embedding generation consumes tokens and may incur provider cost.

Track:

Organization
Document version
Provider
Model
Input tokens
Number of chunks
Number of requests
Estimated cost
Processing duration

Cost Calculation

from decimal import Decimal
def calculate_embedding_cost(
    *,
    token_count: int,
    price_per_million_tokens: Decimal,
) -> Decimal:
    return (
        Decimal(token_count)
        / Decimal(1_000_000)
        * price_per_million_tokens
    )

Pricing should come from a versioned model-pricing configuration rather than being hard-coded into business logic.

AI Usage Event

Record an event such as:

{
  "operation": "knowledge_embedding",
  "organization_id": 42,
  "document_version_id": 118,
  "provider": "configured_provider",
  "model": "configured_model",
  "input_tokens": 48120,
  "output_units": 612,
  "estimated_cost": "0.00384960"
}

Do not store sensitive chunk content in the AI usage record.

Quota Enforcement

Before generating embeddings, verify:

Monthly AI budget
Embedding token allowance
Knowledge storage allowance
Document processing entitlement
Provider access policy

Possible plan limits:

knowledge.embedding_tokens.monthly
knowledge.embedded_chunks.max
knowledge.embedding_models.allowed
knowledge.private_embeddings.enabled

Budget Failure Behavior

If the organization exceeds its embedding budget:

Mark job as BLOCKED
↓
Notify administrator
↓
Keep extracted chunks
↓
Resume after quota increase

Do not discard processed knowledge.

Embedding Versioning

Every embedding is tied to:

Provider
Model
Model revision
Dimension
Input hash
Generation configuration

This combination acts as the embedding version.

Why Model Versioning Matters

Suppose the platform changes from:

Embedding Model A
1536 dimensions

to:

Embedding Model B
1024 dimensions

Old and new vectors cannot safely share the same index.

The migration must be explicit.

Embedding Profile

Create a logical profile identifier.

knowledge-default-v1

The profile may point to:

Provider:
provider-a
Model:
embedding-model-a
Dimension:
1536
Normalization:
L2
Similarity:
Cosine

A later profile may be:

knowledge-default-v2

Application retrieval uses a profile rather than scattering raw model names across the codebase.

Embedding Profile Model

A future configuration table may contain:

embedding_profiles
id
name
provider
model
dimension
normalization
similarity_metric
status
created_at
activated_at
retired_at

For the first version, profiles can remain configuration-based.

Model Migration Strategy

A safe re-embedding migration follows:

Create New Embedding Profile
↓
Generate New Embeddings
↓
Build New Vector Index
↓
Run Retrieval Evaluation
↓
Switch Search Traffic
↓
Monitor
↓
Retire Old Profile

Never delete the old embeddings before the new index is validated.

Dual-Write Migration

During a gradual model migration:

New Chunk
↓
Generate v1 embedding
+
Generate v2 embedding

Retrieval continues using v1 until v2 is ready.

This reduces migration downtime.

Superseding Embeddings

When a chunk’s embedding input changes:

Existing READY embedding
↓
New input hash detected
↓
Generate new embedding
↓
Mark old embedding SUPERSEDED

Historical embeddings may be retained temporarily for audit or rollback.

Document Version Changes

Each document version has its own chunks and embeddings.

Policy v1
↓
Chunks v1
↓
Embeddings v1
Policy v2
↓
Chunks v2
↓
Embeddings v2

Approving version 2 should eventually remove version 1 from the active retrieval index while preserving its evidence history.

Deletion Workflow

When a knowledge version is deleted, remove or deactivate:

Chunks
Embeddings
Vector index entries
Cached retrieval results
Processing artifacts

A deletion event should be idempotent.

Alembic Migration

Create a migration:

alembic revision \
  --autogenerate \
  -m "create knowledge embeddings"

The migration should create:

knowledge_embeddings
Embedding status enum
Tenant indexes
Model indexes
Uniqueness constraints

The vector column and vector indexes will be finalized in the next article.

Migration Example

def upgrade() -> None:
    op.create_table(
        "knowledge_embeddings",
        sa.Column(
            "id",
            sa.BigInteger(),
            primary_key=True,
        ),
        sa.Column(
            "organization_id",
            sa.BigInteger(),
            nullable=False,
        ),
        sa.Column(
            "knowledge_chunk_id",
            sa.BigInteger(),
            nullable=False,
        ),
        sa.Column(
            "document_id",
            sa.BigInteger(),
            nullable=False,
        ),
        sa.Column(
            "document_version_id",
            sa.BigInteger(),
            nullable=False,
        ),
        sa.Column(
            "provider",
            sa.String(length=100),
            nullable=False,
        ),
        sa.Column(
            "model",
            sa.String(length=255),
            nullable=False,
        ),
        sa.Column(
            "dimension",
            sa.Integer(),
            nullable=False,
        ),
        sa.Column(
            "input_hash",
            sa.String(length=64),
            nullable=False,
        ),
        sa.Column(
            "status",
            knowledge_embedding_status,
            nullable=False,
        ),
        # Vector column added with
        # selected vector extension.
    )

Always review generated migrations manually.

API Design

Most embedding operations should not be exposed as ordinary user-facing APIs.

The normal workflow is automatic:

Chunking completes
↓
Embedding job starts

However, administrative endpoints are useful.

Recommended endpoints:

GET  /knowledge/documents/{document_id}/embedding-status
POST /knowledge/versions/{version_id}/embeddings/retry
POST /knowledge/versions/{version_id}/embeddings/rebuild
GET  /admin/embedding-jobs
GET  /admin/embedding-models

Embedding Status Response

class EmbeddingStatusResponse(
    BaseModel
):
    document_version_id: int
    total_chunks: int
    ready_embeddings: int
    pending_embeddings: int
    processing_embeddings: int
    failed_embeddings: int
    model: str
    dimension: int
    completed: bool

Retry Endpoint

@router.post(
    "/knowledge/versions/"
    "{version_id}/embeddings/retry",
    status_code=status.HTTP_202_ACCEPTED,
)
def retry_failed_embeddings(
    version_id: int,
    current_user: User = Depends(
        get_current_user
    ),
    organization: Organization = Depends(
        get_current_organization
    ),
    service: KnowledgeEmbeddingService = (
        Depends(
            get_knowledge_embedding_service
        )
    ),
):
    require_permission(
        user=current_user,
        organization=organization,
        permission=(
            "knowledge:reprocess"
        ),
    )
    return service.retry_failed(
        organization_id=organization.id,
        document_version_id=version_id,
        actor_user_id=current_user.id,
    )

Rebuild Security

A full embedding rebuild can consume substantial AI resources.

Restrict it to:

Knowledge Manager
Organization Administrator
Platform Administrator

The request should also require:

  • Budget check
  • Audit log
  • Idempotency key
  • Rate limiting
  • Confirmation in the frontend

React Processing Status

The knowledge document page can display:

Text extraction       Complete
Chunk generation      Complete
Embedding generation  482 / 612
Vector indexing       Waiting

Embedding Progress Component

export function EmbeddingProgress({
  status,
}) {
  const percentage =
    status.total_chunks === 0
      ? 0
      : Math.round(
          (status.ready_embeddings /
            status.total_chunks) *
            100
        );
  return (
    <section>
      <h3>Semantic indexing</h3>
      <progress
        value={status.ready_embeddings}
        max={status.total_chunks}
      />
      <p>
        {status.ready_embeddings} of{" "}
        {status.total_chunks} chunks
        embedded ({percentage}%)
      </p>
      {status.failed_embeddings > 0 && (
        <p role="alert">
          {status.failed_embeddings} chunks
          failed and require attention.
        </p>
      )}
    </section>
  );
}

Frontend Retry Action

Users with permission can see:

Retry failed embeddings

The interface should not expose provider secrets, raw vectors, or internal stack traces.

Audit Logging

Record events such as:

embedding_job_queued
embedding_generation_started
embedding_batch_completed
embedding_generation_failed
embedding_generation_retried
embedding_model_changed
embedding_rebuild_requested
embedding_profile_activated

Audit metadata should include:

organization_id
actor_user_id
document_version_id
embedding_profile
chunk_count
result
timestamp

Structured Operational Logs

Example:

{
  "event": "embedding_batch_completed",
  "organization_id": 42,
  "document_version_id": 118,
  "provider": "configured_provider",
  "model": "configured_model",
  "batch_size": 64,
  "token_count": 3120,
  "duration_ms": 842,
  "correlation_id": "job_..."
}

Do not log vectors or full source text.

Metrics

Add metrics such as:

knowledge_embedding_jobs_total
knowledge_embedding_jobs_failed_total
knowledge_embedding_chunks_total
knowledge_embedding_tokens_total
knowledge_embedding_cost_total
knowledge_embedding_batch_size
knowledge_embedding_duration_seconds
knowledge_embedding_provider_requests_total
knowledge_embedding_provider_errors_total
knowledge_embedding_dimension_mismatch_total
knowledge_embedding_queue_depth

Metric Labels

Reasonable labels include:

provider
model
status
error_type

Avoid high-cardinality labels such as:

document_id
chunk_id
organization_id

Tenant-level usage should be stored in database records or analytics events rather than infrastructure metric labels.

Monitoring Dashboard

The operations dashboard should show:

Embedding queue depth
Jobs running
Jobs failed
Chunks embedded per minute
Provider latency
Rate-limit events
Token usage
Daily cost
Average batch size
Current embedding profile

Alerts

Create alerts for:

Queue depth above threshold
Failure rate above threshold
Provider unavailable
Rate-limit surge
Dimension mismatch
Unexpected cost increase
No worker activity
Embedding latency degradation

Quality Monitoring

A technically valid embedding is not automatically a useful embedding.

Quality should be evaluated through retrieval tests.

Create a benchmark dataset containing:

Search query
Expected document
Expected chunk
Relevant category
Language
Minimum ranking target

Example:

Query:
How do we recover critical systems after a major outage?
Expected source:
Business Continuity Policy
Expected section:
Disaster Recovery

Retrieval Quality Metrics

Later semantic-search tests should measure:

Recall@K
Precision@K
Mean Reciprocal Rank
Normalized Discounted Cumulative Gain
Relevant Source Coverage

Embedding quality must be evaluated in the context of actual tender and proposal queries.

Multilingual Embeddings

Organizations may store knowledge in multiple languages.

Example:

Tender:
Dutch
Knowledge:
English
Proposal:
Dutch

A multilingual embedding model may retrieve semantically related content across languages.

The platform should store:

Chunk language
Query language
Embedding model language support

This allows retrieval evaluation by language pair.

Language-Specific Routing

A provider router may choose:

English-optimized model
Multilingual model
Domain-specific model

based on the document language and organization’s configuration.

For the initial version, using one multilingual model may reduce operational complexity.

Embedding Security

Embedding vectors are derived data, but they should still be treated as sensitive.

Vectors may reveal structural or semantic information about confidential content.

Protect them with:

  • Tenant isolation
  • Database encryption
  • Private networking
  • Access controls
  • Secure backups
  • Deletion workflows
  • Restricted administrative access

Do not expose raw vectors through ordinary APIs.

External Provider Controls

Before sending text to an external embedding provider, enforce:

Approved provider
Approved processing region
Organization consent
Data-retention policy
Confidentiality compatibility
Transport encryption
Provider authentication

For highly sensitive deployments, support a self-hosted embedding service.

Secret Management

Provider credentials must be stored in:

Secret manager
Encrypted runtime environment
Managed identity

Never store credentials in:

Source code
Git repository
Database logs
Frontend configuration
Job payloads

Rate Limiting

The platform must respect both:

  • Provider rate limits
  • Internal tenant quotas

Use a rate limiter based on:

Requests per minute
Tokens per minute
Concurrent batches
Organization budget

A centralized limiter prevents multiple workers from overwhelming the provider.

Caching

Embedding results can be reused when the complete embedding input and configuration are identical.

Cache key:

provider
model
model revision
dimension
normalization
input hash

Cross-tenant reuse requires careful privacy analysis.

The safest initial implementation limits cache reuse to the same organization and chunk record.

Backpressure

When the embedding queue grows too quickly:

Reduce worker intake
Pause low-priority rebuilds
Prioritize newly approved documents
Notify operations

Priority classes may include:

High:
Newly approved knowledge
Medium:
New document processing
Low:
Historical re-embedding

Job Prioritization

Proposal generation may require a document urgently.

An organization could request:

Priority indexing

The queue can support:

proposal_blocking
interactive
standard
maintenance

This should remain constrained by plan and budget.

Failure States in the User Interface

Users should see actionable messages.

Bad:

Embedding error

Better:

Semantic indexing failed for 3 chunks.
The original document remains available.
A knowledge manager can retry processing.

Do not display provider stack traces or secret-bearing messages.

Testing Strategy

Create:

tests/
└── knowledge/
    └── embeddings/
        ├── test_input_builder.py
        ├── test_hashing.py
        ├── test_batch_builder.py
        ├── test_vector_validator.py
        ├── test_normalization.py
        ├── test_embedding_repository.py
        ├── test_embedding_service.py
        ├── test_embedding_worker.py
        ├── test_idempotency.py
        ├── test_retry_behavior.py
        ├── test_tenant_isolation.py
        ├── test_cost_tracking.py
        └── test_model_migration.py

Input Builder Test

def test_embedding_input_contains_context(
    knowledge_chunk,
    knowledge_document,
    input_builder,
):
    result = input_builder.build(
        chunk=knowledge_chunk,
        document=knowledge_document,
    )
    assert knowledge_document.title in (
        result.text
    )
    assert knowledge_chunk.heading in (
        result.text
    )
    assert knowledge_chunk.content in (
        result.text
    )

Batch Size Test

def test_batch_builder_respects_item_limit():
    builder = EmbeddingBatchBuilder(
        max_items=2,
        max_total_tokens=10_000,
    )
    inputs = [
        make_embedding_input(1),
        make_embedding_input(2),
        make_embedding_input(3),
    ]
    batches = builder.build(inputs)
    assert len(batches) == 2
    assert len(batches[0].items) == 2
    assert len(batches[1].items) == 1

Token Limit Test

def test_batch_builder_respects_token_limit():
    builder = EmbeddingBatchBuilder(
        max_items=100,
        max_total_tokens=100,
    )
    inputs = [
        make_embedding_input(
            1,
            estimated_tokens=70,
        ),
        make_embedding_input(
            2,
            estimated_tokens=60,
        ),
    ]
    batches = builder.build(inputs)
    assert len(batches) == 2

Dimension Validation Test

def test_vector_dimension_must_match():
    validator = EmbeddingVectorValidator()
    with pytest.raises(
        EmbeddingDimensionMismatch
    ):
        validator.validate(
            vector=[0.1, 0.2],
            expected_dimension=3,
        )

Vector Normalization Test

def test_l2_normalization():
    result = l2_normalize(
        [3.0, 4.0]
    )
    magnitude = math.sqrt(
        sum(
            value * value
            for value in result
        )
    )
    assert magnitude == pytest.approx(
        1.0
    )

Idempotency Test

def test_existing_embedding_is_reused(
    service,
    embedding_repository,
    ready_embedding,
):
    summary = service.generate_for_version(
        organization_id=(
            ready_embedding.organization_id
        ),
        document_version_id=(
            ready_embedding
            .document_version_id
        ),
    )
    assert summary.generated_count == 0
    assert summary.skipped_count == 1

Tenant Isolation Test

def test_embedding_query_is_tenant_scoped(
    repository,
    embedding_factory,
    organization_a,
    organization_b,
):
    embedding = embedding_factory(
        organization_id=organization_a.id
    )
    result = repository.get_by_id(
        organization_id=organization_b.id,
        embedding_id=embedding.id,
    )
    assert result is None

Temporary Failure Test

def test_temporary_provider_failure_is_retried(
    worker,
    provider,
):
    provider.embed.side_effect = [
        TemporaryEmbeddingError(),
        successful_embedding_result(),
    ]
    result = worker.run(
        organization_id=42,
        document_version_id=118,
    )
    assert result.completed is True
    assert provider.embed.call_count == 2

Permanent Failure Test

def test_dimension_mismatch_is_not_retried(
    worker,
    provider,
):
    provider.embed.return_value = (
        result_with_wrong_dimension()
    )
    with pytest.raises(
        EmbeddingDimensionMismatch
    ):
        worker.run(
            organization_id=42,
            document_version_id=118,
        )
    assert provider.embed.call_count == 1

Cost Tracking Test

def test_embedding_usage_is_recorded(
    service,
    usage_repository,
):
    service.generate_for_version(
        organization_id=42,
        document_version_id=118,
    )
    event = usage_repository.latest()
    assert event.operation == (
        "knowledge_embedding"
    )
    assert event.input_tokens > 0
    assert event.estimated_cost >= 0

Model Migration Test

Verify that:

Old embedding profile remains searchable
New profile generates separate embeddings
Dimensions are not mixed
Traffic switches only after validation
Rollback returns to old profile

Integration Test

A complete integration test should execute:

Create knowledge document
↓
Create approved version
↓
Store extracted text
↓
Generate chunks
↓
Queue embedding job
↓
Mock provider returns vectors
↓
Persist embeddings
↓
Mark processing complete

End-to-End Processing Scenario

1. A knowledge manager uploads a policy
2. The file passes malware scanning
3. Text is extracted and normalized
4. The document is divided into 84 chunks
5. An embedding job is queued
6. The worker loads eligible chunks
7. Embedding inputs include headings and document context
8. Existing vectors are skipped
9. Remaining inputs are grouped into batches
10. The provider generates vectors
11. Every vector is validated
12. Vectors are normalized when configured
13. Embeddings are stored with model metadata
14. Token usage and cost are recorded
15. The processing job is marked complete
16. The interface displays 84 of 84 chunks embedded
17. The document becomes ready for vector indexing

Failure Scenario

1. A batch reaches the provider rate limit
2. The provider adapter classifies the error as temporary
3. The job enters RETRYING
4. Exponential backoff is applied
5. The batch succeeds on a later attempt
6. Existing completed batches are not regenerated
7. Usage is recorded only for successful provider work
8. The overall job completes

Model Upgrade Scenario

1. The platform creates embedding profile v2
2. Existing profile v1 remains active
3. Background workers generate v2 embeddings
4. A separate vector index is built
5. Retrieval benchmarks compare v1 and v2
6. Search traffic is switched to v2
7. Monitoring confirms quality and latency
8. v1 remains available for rollback
9. v1 is retired after the migration window

Deliverables

At the end of this article, the project now has:

✔ Provider-independent embedding interface
✔ Embedding provider registry
✔ Configurable embedding profiles
✔ Context-enriched embedding inputs
✔ Deterministic input hashing
✔ Separate embedding database model
✔ Embedding lifecycle statuses
✔ Batch generation
✔ Token-aware batching
✔ Vector validation
✔ Optional vector normalization
✔ Idempotent processing
✔ Concurrency protection
✔ Background embedding workers
✔ Retry and backoff strategy
✔ Partial-batch failure isolation
✔ Confidentiality-aware provider routing
✔ Tenant-scoped repositories
✔ Token and cost tracking
✔ Quota enforcement
✔ Model versioning
✔ Safe re-embedding strategy
✔ Administrative APIs
✔ React progress reporting
✔ Audit logging
✔ Operational metrics
✔ Quality-evaluation preparation
✔ Security controls
✔ Comprehensive tests

The platform can now transform structured organizational knowledge into numerical semantic representations.

Recommended Git Commit

git add .
git commit -m "Implement knowledge embedding generation pipeline"
git push origin main

What We Will Build Next

The embedding pipeline can now generate validated vectors for every eligible knowledge chunk.

However, generating vectors is only part of the solution.

The platform still needs an efficient way to:

  • Store high-dimensional vectors
  • Index millions of embeddings
  • Filter searches by organization
  • Calculate vector distance
  • Retrieve the nearest knowledge chunks
  • Replace indexes during model migration
  • Balance search quality and performance

The next article in the series is:

“Building the Vector Storage Layer with PostgreSQL and pgvector: Indexing Organizational Knowledge for Semantic Search.”

In that article, we will implement:

  • The pgvector PostgreSQL extension
  • Vector columns
  • Dimension-safe schema design
  • Cosine, inner-product, and Euclidean operators
  • Exact nearest-neighbor search
  • HNSW indexes
  • IVFFlat indexes
  • Tenant-aware vector filtering
  • Embedding-profile separation
  • Index tuning
  • Database migrations
  • Repository queries
  • Re-indexing workflows
  • Performance benchmarking
  • Backup and recovery considerations
  • Vector storage tests

This vector storage layer will make organizational knowledge efficiently searchable by semantic meaning.

Conclusion

The Embedding Generation Pipeline transforms structured knowledge chunks into semantic vectors that allow the Tender Opportunity Scanner to compare meaning instead of relying only on exact words.

By introducing a provider-independent interface, the platform can use cloud-hosted, enterprise, or self-hosted embedding models without coupling the knowledge system to one vendor. Context-enriched embedding inputs ensure that document titles, categories, section paths, headings, and chunk content all contribute to the vector representation.

Batch processing, deterministic input hashes, idempotent jobs, concurrency protection, retry strategies, quota enforcement, and cost tracking make embedding generation reliable at production scale. Model and dimension metadata prevent incompatible vectors from being mixed, while versioned embedding profiles support controlled upgrades and rollback.

Security remains part of the design. Tenant boundaries, confidentiality-aware provider routing, private storage, restricted administrative actions, and safe deletion workflows protect both source content and derived vectors.

The platform can now create semantic representations of organizational knowledge. The next step is to store and index those vectors efficiently with PostgreSQL and pgvector, creating the retrieval foundation required for semantic search, hybrid search, and evidence-grounded proposal generation.

Ben Kemp

Discover more from BidRadar

Subscribe now to keep reading and get access to the full archive.

Continue reading