How LMForge Integrates the Vector Embedding Pipeline with Weaviate for Semantic Search

LMForge connects OpenAI embeddings cached in Redis to Weaviate via a LangChain wrapper, enabling filtered semantic search across dataset collections.

The LMForge end-to-end LLMOps platform implements a four-stage pipeline that transforms raw text into queryable semantic vectors stored in Weaviate. This integration leverages Redis for embedding caching, OpenAI for vector generation, and Weaviate as the cloud-native vector database to power multi-model agent retrieval.

Architecture Overview

The vector embedding pipeline operates through four distinct stages that bridge document ingestion to semantic retrieval.

• Embedding Generation: Raw documents flow through the EmbeddingsService class in api/internal/service/embeddings_service.py, which generates vectors using OpenAI's text-embedding-3-small model and caches them in Redis to eliminate redundant API calls.

• Vector Store Construction: The VectorDatabaseService in api/internal/service/vector_database_service.py instantiates a WeaviateVectorStore (LangChain wrapper) using the injected Weaviate client, targeting the Dataset collection with the cached embedding function.

• Persistence: Vectors are up-serted into the Weaviate collection via the collection property of VectorDatabaseService, storing original text in the text field alongside metadata fields like dataset_id and document_enabled.

• Retrieval: The SemanticRetriever class in api/internal/core/retrievers/semantic.py executes similarity_search_with_relevance_scores against the WeaviateVectorStore, applying Weaviate filters to restrict results by dataset permissions and enabled flags.

Core Components

Configuration Management

Default Weaviate connection parameters reside in api/config/default_config.py. These settings define the host, ports, and API key used during application startup.

# api/config/default_config.py

"WEAVIATE_HTTP_HOST": "localhost",
"WEAVIATE_HTTP_PORT": 8080,
"WEAVIATE_API_KEY": "ftBC9hKkjfdbdi0W3T6kEtMh5BZFpGa1DF8",

The Flask-Weaviate extension reads these values to establish the client connection.

FlaskWeaviate Extension

The singleton weaviate object created in api/internal/extension/weaviate_extension.py wraps the Weaviate client and injects it throughout the application via the dependency injection framework.

# api/internal/extension/weaviate_extension.py

from flask_weaviate import FlaskWeaviate
weaviate = FlaskWeaviate()

This pattern ensures consistent client reuse across services without recreating connections.

Embedding Service with Redis Caching

EmbeddingsService builds a CacheBackedEmbeddings instance that checks Redis before calling OpenAI. If the vector exists in cache, it returns immediately; otherwise, it generates and stores the embedding.

# api/internal/service/embeddings_service.py

self._store = RedisStore(client=redis)
self._embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
self._cache_backed_embeddings = CacheBackedEmbeddings.from_bytes_store(
    self._embeddings, self._store, namespace="embeddings"
)

This caching layer significantly reduces latency and API costs for frequently accessed content.

Vector Database Service

Using the injected FlaskWeaviate client, this service constructs a WeaviateVectorStore configured for the Dataset collection. The vector_store property exposes the LangChain-compatible interface.

# api/internal/service/vector_database_service.py

@property
def vector_store(self) -> WeaviateVectorStore:
    return WeaviateVectorStore(
        client=self.weaviate.client,
        index_name=COLLECTION_NAME,
        text_key="text",
        embedding=self.embeddings_service.cache_backed_embeddings,
    )

Semantic Retrieval with Filtering

SemanticRetriever receives query parameters and dataset permissions, then executes similarity search with Weaviate filters. The Filter object ensures only enabled documents from authorized datasets return results.

# api/internal/core/retrievers/semantic.py

search_result = self.vector_store.similarity_search_with_relevance_scores(
    query=query,
    k=k,
    **{
        "filters": Filter.all_of([
            Filter.by_property("dataset_id").contains_any([str(dataset_id) for dataset_id in self.dataset_ids]),
            Filter.by_property("document_enabled").equal(True),
            Filter.by_property("segment_enabled").equal(True),
        ]),
        **self.search_kwargs,
    }
)

Implementation Examples

Generating and Caching Embeddings

Instantiate the service with a Redis client to create cached vectors for downstream storage.

from api.internal.service.embeddings_service import EmbeddingsService
from redis import Redis

redis_client = Redis(host="localhost", port=6379)
emb_service = EmbeddingsService(redis=redis_client)

text = "What is the capital of France?"
vector = emb_service.cache_backed_embeddings.embed_query(text)  # cached via Redis

Storing Documents in Weaviate

Use the VectorDatabaseService to persist documents with metadata. The service handles embedding generation internally through the configured EmbeddingsService.

from api.internal.service.vector_database_service import VectorDatabaseService
from api.internal.extension.weaviate_extension import weaviate

# Injector creates the service with the FlaskWeaviate instance automatically

vector_service = VectorDatabaseService(weaviate=weaviate, embeddings_service=emb_service)

# Upsert a document (the VectorStore handles embedding internally)

doc = {
    "text": "Paris is the capital of France.",
    "dataset_id": "1234",
    "document_enabled": True,
    "segment_enabled": True
}
vector_service.collection.data.insert(doc)   # uses Weaviate collection API

Performing Semantic Search

Execute filtered retrieval using SemanticRetriever with specific dataset permissions and relevance scoring.

from api.internal.core.retrievers.semantic import SemanticRetriever
from api.internal.service.vector_database_service import VectorDatabaseService
from api.internal.service.embeddings_service import EmbeddingsService
from api.internal.extension.weaviate_extension import weaviate
from uuid import UUID

emb_service = EmbeddingsService(redis=Redis())
vec_service = VectorDatabaseService(weaviate=weaviate, embeddings_service=emb_service)

retriever = SemanticRetriever(
    dataset_ids=[UUID("1234")],
    vector_store=vec_service.vector_store,
    search_kwargs={"k": 5}
)

results = retriever.get_relevant_documents("Capital of France?")
for doc in results:
    print(doc.page_content, doc.metadata["score"])

Summary

• LMForge uses a layered architecture to integrate OpenAI embeddings with Weaviate for semantic search.
• Redis caching in EmbeddingsService eliminates redundant embedding generation and reduces API costs.
• FlaskWeaviate provides a singleton client pattern for consistent database connections across the application.
• WeaviateVectorStore acts as the bridge between LangChain interfaces and Weaviate's native vector storage.
• SemanticRetriever applies permission-based filtering using Weaviate's Filter API to enforce dataset-level access control.

Frequently Asked Questions

How does LMForge handle embedding caching to reduce OpenAI API costs?

LMForge implements CacheBackedEmbeddings in api/internal/service/embeddings_service.py, which wraps the OpenAI embedding model with a Redis-backed store. Before calling the API, the service checks if the text hash exists in Redis; if found, it returns the cached vector immediately, avoiding redundant API charges.

What Weaviate collection schema does LMForge use for vector storage?

According to api/internal/service/vector_database_service.py, LMForge stores vectors in a collection named Dataset (referenced via COLLECTION_NAME). The schema includes a text field for the original content and metadata fields such as dataset_id, document_enabled, and segment_enabled to support filtered retrieval.

How does the semantic retriever enforce dataset-level permissions?

The SemanticRetriever class in api/internal/core/retrievers/semantic.py accepts a list of authorized dataset_ids during instantiation. It constructs a Weaviate Filter object using Filter.by_property("dataset_id").contains_any() to restrict similarity searches to only those datasets the user is permitted to access, combined with enabled flags for document and segment status.

Can the vector embedding pipeline work with embedding models other than OpenAI?

While the current implementation in embeddings_service.py specifically configures OpenAIEmbeddings with the text-embedding-3-small model, the architecture uses LangChain's embedding abstractions. Replacing the OpenAI instance with another LangChain-compatible embedding provider (such as HuggingFace or Cohere) would require only modifying the EmbeddingsService initialization while preserving the Redis caching and Weaviate storage layers.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}