vendor_fabric.meshy.persistence.vector_store¶
SQLite vector store for idempotency tracking and RAG embeddings.
This module provides a vector-enabled SQLite database for:
Idempotency tracking - deduplicate asset generation requests
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¶
Record of a 3D asset generation. |
|
Result from similarity search. |
|
SQLite vector store for asset generation tracking and RAG. |
Functions¶
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.
- 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