Vector Store / Local Dynamic Embedding

Use

Module for generating text embeddings locally with a fluent (chainable) API, dynamic provider switching and similarity retrieval.

On this page

    Module for generating text embeddings locally with a fluent API (chainable), dynamic provider switching and similarity retrieval.

    The typical flow is: configure → process text → access chunks → retrieve by similarity → export.

    #How to run

    The module uses relative imports (EmbeddingFactory), so it must be run as a module, with the -m flag:

    python -m src.embedding.modules.local_embedding

    Running the file directly (python arquivo.py) breaks because of the relative import.

    #Quick start

    from src.embedding.modules.local_embedding.module import (
    LocalDynamicEmbedding,
    EmbeddingFactory,
)

# Monta o pipeline (encadeando as configurações)
pipeline = (
    LocalDynamicEmbedding()
    .with_fake_embeddings(size=384)          # embeddings falsos p/ testes
    .with_splitter(chunk_size=120, chunk_overlap=20)
    .with_top_k(3)
)

# Processa um texto e recebe a quantidade de chunks gerados
qtd = pipeline.process_text("seu texto...", metadata={"fonte": "exemplo"})

# Recupera os trechos mais similares a uma consulta
resultados = pipeline.retrieve("sua pergunta?", include_embedding=True)

    #Available providers

    EmbeddingFactory.available()   # -> lista os provedores registrados

    Use it to find out which embedding backends are installed/registered in the environment.

    #Building the pipeline (fluent API)

    Each with_* method returns the instance itself, allowing chaining. If you do not set an embeddings provider, the class falls back to fake automatically.

    MethodEffectParameters
    .with_fake_embeddings(size=384)Uses fake embeddings (vectors of dimension size) — ideal for tests without external dependenciessize: vector dimension
    .with_splitter(chunk_size, chunk_overlap)Configures how the text is split into chunkschunk_size: chunk size; chunk_overlap: overlap between chunks
    .with_top_k(k)Sets the default number of results returned in retrievalk: number of results

    #Factory constructors (switch provider in 1 line)

    To use a real provider, there are alternative constructors, for example:

    pipeline_openai = LocalDynamicEmbedding.from_openai_embeddings(
    model="text-embedding-3-large",
    chunk_size=3000,
    chunk_overlap=300,
    top_k=5,
)

    Requires the provider's dependencies (e.g. OpenAI) to be installed and configured.

    #Processing text

    qtd = pipeline.process_text(texto, metadata={"fonte": "apostila_energia"})
    • Returns: number of chunks generated in this call (int).
    • metadata (optional): dictionary attached to each chunk generated from this text.
    • It can be called several times; the chunks accumulate.
    pipeline.total_chunks   # total acumulado de chunks em todas as chamadas

    #Accessing the chunks

    pipeline.chunks is the list of chunk objects. Each chunk exposes:

    AttributeDescription
    chunk.indexChunk index
    chunk.contentChunk text
    chunk.lengthText length
    chunk.dimEmbedding vector dimension
    chunk.metadataAssociated metadata
    chunk.embeddingEmbedding vector (list of floats)

    Example:

    for chunk in pipeline.chunks:
    preview = chunk.content[:60].replace("\n", " ")
    print(f"[{chunk.index}] len={chunk.length} dim={chunk.dim} meta={chunk.metadata}")
    print(f"     texto: {preview}...")
    print(f"     embedding[:3]: {chunk.embedding[:3]}")

    #Similarity retrieval

    resultados = pipeline.retrieve("como gerar eletricidade a partir do vento?",
                               include_embedding=True)
    • Returns: list of dictionaries sorted by relevance. Each item contains:
      • score — similarity score (float)
      • content — text of the retrieved passage
      • embedding — vector of the passage (only if include_embedding=True)
    • The number of results follows the configured top_k.
    for i, r in enumerate(pipeline.retrieve(consulta, include_embedding=True), 1):
    print(f"#{i} score={r['score']:.4f} | dim={len(r['embedding'])}")
    print(f"   {r['content'][:70]}...")

    #Exporting the chunks (JSON-ready)

    dados = pipeline.get_chunks(include_embedding=False)
    • Returns: list of serializable dictionaries (ideal for json.dumps).
    • Use include_embedding=False to omit the vectors and produce a lean output.
    import json
exemplo = pipeline.get_chunks(include_embedding=False)[0]
print(json.dumps(exemplo, ensure_ascii=False, indent=2))

    #Full example

    from src.embedding.modules.local_embedding.module import (
    LocalDynamicEmbedding, EmbeddingFactory,
)
import json

print("Provedores disponíveis:", EmbeddingFactory.available())

texto = (
    "A energia solar é uma fonte renovável...\n\n"
    "A energia eólica aproveita a força dos ventos...\n\n"
    "Já os combustíveis fósseis são fontes não renováveis...\n\n"
    "O uso de baterias é essencial para armazenar energia..."
)

# 1) Configuração (sem embeddings -> usa fake automaticamente)
pipeline = (
    LocalDynamicEmbedding()
    .with_fake_embeddings(size=384)
    .with_splitter(chunk_size=120, chunk_overlap=20)
    .with_top_k(3)
)

# 2) Processamento
qtd = pipeline.process_text(texto, metadata={"fonte": "apostila_energia"})
print(f"Chunks gerados: {qtd} | total: {pipeline.total_chunks}")

# 3) Inspeção dos chunks
for chunk in pipeline.chunks:
    print(f"[{chunk.index}] dim={chunk.dim} meta={chunk.metadata}")

# 4) Recuperação
for i, r in enumerate(pipeline.retrieve("energia do vento?", include_embedding=True), 1):
    print(f"#{i} score={r['score']:.4f} -> {r['content'][:60]}...")

# 5) Exportação JSON
print(json.dumps(pipeline.get_chunks(include_embedding=False)[0],
                 ensure_ascii=False, indent=2))

    #Quick API reference

    EmbeddingFactory

    • EmbeddingFactory.available() → list of providers

    LocalDynamicEmbedding

    • LocalDynamicEmbedding() → empty instance (fallback: fake)
    • .with_fake_embeddings(size=384) → self
    • .with_splitter(chunk_size, chunk_overlap) → self
    • .with_top_k(k) → self
    • .from_openai_embeddings(model, chunk_size, chunk_overlap, top_k) → new instance (classmethod)
    • .process_text(texto, metadata=None) → int (chunks generated)
    • .total_chunks → int (accumulated)
    • .chunks → list of chunk objects (.index, .content, .length, .dim, .metadata, .embedding)
    • .retrieve(consulta, include_embedding=False) → list of dicts (score, content, [embedding])
    • .get_chunks(include_embedding=False) → list of serializable dicts

    #Notes

    • Without a configured provider, the pipeline uses fake embeddings — perfect for testing the flow without depending on external APIs.
    • Switching provider is a single line (via a factory constructor), without changing the rest of the code.
    • Always run as a module (python -m ...) because of the relative imports.

    This documentation describes the observable public API from the demo file. Internal implementation details (similarity algorithm, exact splitter format, registered providers) depend on the module's source code. If you want, send me module.py and I will complete it with those details.

    Source: src/embedding/modules/local_embedding/test/use.py

    Esc
    ↑↓navigate Enteropen