vendor_fabric.meshy.persistence.vector_store

SQLite vector store for idempotency tracking and RAG embeddings.

This module provides a vector-enabled SQLite database for:

  1. Idempotency tracking - deduplicate asset generation requests

  2. RAG embeddings - semantic search over prompts, species, assets

Uses sqlite-vec for vector similarity search, falling back to standard SQLite if vectors aren’t needed.

Usage: from vendor_fabric.meshy.persistence.vector_store import VectorStore

store = VectorStore("assets.db")

# Track asset generation
store.record_generation(
    spec_hash="abc123",
    prompt="cute otter character",
    task_id="meshy-task-id",
    embedding=get_embedding("cute otter character"),
)

# Find similar prompts (RAG)
similar = store.search_similar("river otter", limit=5)

# Check idempotency
existing = store.get_by_spec_hash("abc123")

Requirements: pip install “vendor-fabric[meshy,vector]”

The vector extra includes:
- sqlite-vec (vector similarity extension)

get_embedding() uses sentence-transformers only when users install it
independently after reviewing its torch dependency tree.

Module Contents

Classes

GenerationRecord

Record of a 3D asset generation.

SimilarityResult

Result from similarity search.

VectorStore

SQLite vector store for asset generation tracking and RAG.

Functions

get_embedding

Get embedding for text using sentence-transformers.

API

class vendor_fabric.meshy.persistence.vector_store.GenerationRecord

Record of a 3D asset generation.

id: int | None = None
spec_hash: str = <Multiline-String>
project: str = <Multiline-String>
prompt: str = <Multiline-String>
art_style: str = 'sculpture'
task_id: str | None = None
status: str = 'pending'
model_url: str | None = None
embedding: list[float] | None = None
metadata: dict[str, Any] = 'field(...)'
created_at: datetime.datetime = 'field(...)'
updated_at: datetime.datetime = 'field(...)'
class vendor_fabric.meshy.persistence.vector_store.SimilarityResult

Result from similarity search.

record: vendor_fabric.meshy.persistence.vector_store.GenerationRecord = None
distance: float = None
score: float = None
class vendor_fabric.meshy.persistence.vector_store.VectorStore(db_path: str | pathlib.Path = 'vendor_fabric.meshy.db', embedding_dim: int = DEFAULT_EMBEDDING_DIM)

SQLite vector store for asset generation tracking and RAG.

Features:

  • Idempotent task tracking by spec_hash

  • Vector similarity search for RAG

  • Full-text search fallback when vectors unavailable

  • Atomic transactions

Args: db_path: Path to SQLite database file embedding_dim: Dimension of embedding vectors (default: 384 for MiniLM)

Initialization

DEFAULT_EMBEDDING_DIM = 384
record_generation(spec_hash: str, prompt: str, project: str = 'default', art_style: str = 'sculpture', task_id: str | None = None, embedding: list[float] | None = None, metadata: dict[str, Any] | None = None) extended_data.containers.ExtendedDict

Record a new generation (idempotent by spec_hash).

If a record with the same spec_hash exists, returns existing.

Args: spec_hash: Unique hash of the generation spec prompt: Text prompt for generation project: Project identifier art_style: Art style (realistic, sculpture, etc.) task_id: Meshy task ID if already submitted embedding: Optional embedding vector for RAG metadata: Additional metadata dict

Returns: Extended generation record payload (existing or newly created)

update_status(spec_hash: str, status: str, task_id: str | None = None, model_url: str | None = None) bool

Update generation status.

Args: spec_hash: Generation spec hash status: New status task_id: Optional task ID to update model_url: Optional model URL to update

Returns: True if record was updated, False if not found

get_by_spec_hash(spec_hash: str) extended_data.containers.ExtendedDict | None

Get generation record by spec hash.

Args: spec_hash: Generation spec hash

Returns: Extended generation record payload or None

get_by_task_id(task_id: str) extended_data.containers.ExtendedDict | None

Get generation record by Meshy task ID.

Args: task_id: Meshy task ID

Returns: Extended generation record payload or None

search_similar(query_embedding: list[float], limit: int = 10, project: str | None = None) extended_data.containers.ExtendedList[extended_data.containers.ExtendedDict]

Search for similar generations using vector similarity.

Args: query_embedding: Query embedding vector limit: Maximum results to return project: Optional project filter

Returns: Extended similarity result payloads ordered by similarity (highest first)

search_text(query: str, limit: int = 10, project: str | None = None) extended_data.containers.ExtendedList[extended_data.containers.ExtendedDict]

Full-text search for prompts.

Falls back to this when vector search is unavailable.

Args: query: Search query limit: Maximum results project: Optional project filter

Returns: Extended generation record payloads

list_pending(project: str | None = None) extended_data.containers.ExtendedList[extended_data.containers.ExtendedDict]

List all pending/in-progress generations.

Args: project: Optional project filter

Returns: Extended pending generation record payloads

compute_spec_hash(spec: dict[str, Any]) str

Compute deterministic hash for a generation spec.

Args: spec: Generation specification dict

Returns: SHA256 hex digest

close() None

Close database connection.

vendor_fabric.meshy.persistence.vector_store.get_embedding(text: str, model: str = 'all-MiniLM-L6-v2') extended_data.containers.ExtendedList[float] | None

Get embedding for text using sentence-transformers.

Args: text: Text to embed model: Model name (default: all-MiniLM-L6-v2)

Returns: Extended embedding vector or None if sentence-transformers not available