VectorGroundTruthStore¶

Semantic vector store for RAG-based factual grounding. Ingest documents, then pass to CoherenceScorer for fact-checked scoring. Supports pluggable backends via a registry pattern.

Usage¶

from director_ai.core.retrieval.vector_store import VectorGroundTruthStore

store = VectorGroundTruthStore()
store.ingest([
    "Refunds are available within 30 days of purchase.",
    "Standard shipping takes 5-7 business days.",
    "Pro plan costs $49/month.",
])

# Use with scorer
from director_ai import CoherenceScorer

scorer = CoherenceScorer(
    threshold=0.6,
    ground_truth_store=store,
    use_nli=True,
)

VectorGroundTruthStore Parameters¶

Parameter	Type	Default	Description
`backend`	`VectorBackend \\| None`	`None`	Backend instance (default: `InMemoryBackend`)
`tenant_id`	`str`	`""`	Default tenant ID for multi-tenant stores

Methods¶

add()¶

store.add(
    key="refund-policy",
    value="Refunds are available within 30 days.",
    metadata={"kb_version_bump": "patch"},  # patch, minor, or major
)

Facts start at 1.0.0. Replacing a fact with different content bumps the patch version by default. Pass kb_version_bump="minor" or "major" in metadata when the source change is a larger schema or policy change. The vector metadata is stamped with:

kb_version
kb_chunk_version
kb_content_hash
kb_previous_hash
kb_record_kind
kb_source_key
kb_chunk_index

Use fact_version(key), fact_version_record(key), or version_manifest() to inspect the in-process version ledger.

retract_fact()¶

store.retract_fact("refund-policy", reason="source withdrawn")

Retraction records mark a fact or derived chunk source as unusable for retrieval without deleting backend rows. retrieve_context() and retrieve_context_with_chunks() filter matching vector results and keyword fallback facts after retraction. Use retraction_records() to inspect the event log.

replace_fact()¶

store.replace_fact(
    "refund-policy",
    "Refunds are available within 45 days.",
    reason="policy update",
)

Replacement records preserve the superseded version and content hash while the new value is indexed under the same key. Use replacement_records() to inspect the event log.

kb_snapshot_root()¶

root = store.kb_snapshot_root(tenant_id="acme")
audit_payload = store.kb_snapshot_audit_record(tenant_id="acme")

kb_snapshot_root() returns a deterministic SHA-256 Merkle root over the tenant-visible KB version ledger. Leaves are sorted by tenant, key, record kind, and chunk index, so the same KB state produces the same root regardless of ingestion order. Retractions and replacements update the root because the snapshot includes status, current content hash, previous hash, and semantic version fields.

kb_snapshot_audit_record() returns a compact payload with:

event
tenant_id
revision
record_count
retraction_count
replacement_count
conflict_count
merkle_root

Pass this payload to AuditLogger.log_review(kb_snapshot=audit_payload) when a review decision must carry the KB state it was grounded against.

conflict_reports()¶

store.add_fact(
    "signed-dose",
    "Dose is 5 mg.",
    metadata={
        "claim_id": "dose-claim",
        "signed_fact_id": "signed-1",
        "claim_source": "signed_fact",
    },
)

store.add_fact(
    "incoming-dose",
    "Dose is 10 mg.",
    metadata={"claim_id": "dose-claim"},
)

reports = store.conflict_reports()

conflict_reports() returns tenant-scoped records created during fact writes when a new fact overlaps a retracted ledger entry, differs from a protected signed fact, differs from a passport claim, or declares an explicit contradicts relation. Reports are advisory: ingestion continues, retrieval still uses the active version ledger, and callers can route reports to review queues or audit sinks.

freshness_status_signals()¶

store.add_fact(
    "trial-paper",
    "Trial X reported a 12 percent response rate.",
    metadata={
        "external_id": "doi:10.example/trial-paper",
        "source_timestamp": "1710000000",
        "citation_status": "active",
        "status_source": "publisher-feed",
    },
)

signals = store.freshness_status_signals()

Use freshness_status_signals() to pass KB source age and external citation status metadata into score_temporal_freshness(citation_statuses=signals). The method emits tenant-scoped dictionaries with source_id, status, status_source, and any available timestamp fields.

ingest()¶

store.ingest(documents: list[str]) -> None

Add documents to the store. Each document is embedded and indexed as a derived vector chunk with kb_record_kind="derived_chunk" and a semantic chunk version.

retrieve_context()¶

context = store.retrieve_context(query: str, top_k: int = 3, tenant_id: str = "") -> str | None

Retrieve concatenated context string for a query (matching parent GroundTruthStore interface). Use retrieve_context_with_chunks() for structured EvidenceChunk results.

VectorBackend¶

Abstract protocol for vector storage backends. Implement add() and query() to create a custom backend.

from director_ai.core.retrieval.vector_store import VectorBackend

class MyBackend(VectorBackend):
    def add(self, texts: list[str], ids: list[str] | None = None) -> None:
        ...

    def query(self, text: str, top_k: int = 3) -> list[tuple[str, float]]:
        # Returns list of (text, distance) pairs
        ...

Built-in Backends¶

Backend	Install	Description
`InMemoryBackend`	included	TF-IDF cosine similarity. No deps, good for testing.
`SentenceTransformerBackend`	`pip install director-ai[embeddings]`	Dense embeddings via `sentence-transformers`. Production-quality.
`ChromaBackend`	`pip install director-ai[vector]`	ChromaDB persistent store. Scales to millions of documents.

ChromaBackend¶

from director_ai.core.retrieval.vector_store import ChromaBackend

backend = ChromaBackend(
    collection_name="legal_contracts",
    persist_directory="/data/chroma",
    embedding_model="BAAI/bge-large-en-v1.5",
)
store = VectorGroundTruthStore(backend=backend)

SentenceTransformerBackend¶

from director_ai.core.retrieval.vector_store import SentenceTransformerBackend

backend = SentenceTransformerBackend(
    model_name="BAAI/bge-large-en-v1.5",
)
store = VectorGroundTruthStore(backend=backend)

Backend Registry¶

Register custom backends for use with DirectorConfig.vector_backend:

from director_ai.core.retrieval.vector_store import register_vector_backend, get_vector_backend

register_vector_backend("qdrant", MyQdrantBackend)
BackendClass = get_vector_backend("qdrant")  # returns the class, not an instance
backend = BackendClass(**kwargs)

Function	Purpose
`register_vector_backend(name, cls)`	Register a backend class
`get_vector_backend(name)`	Look up a registered backend class
`list_vector_backends()`	List registered backend names

Reranking¶

Enable cross-encoder reranking for improved retrieval precision:

scorer = CoherenceScorer(
    ground_truth_store=store,
    reranker_enabled=True,
    reranker_model="cross-encoder/ms-marco-MiniLM-L-6-v2",
    reranker_top_k_multiplier=3,  # Retrieve 3x, rerank to top_k
)

Full API¶

director_ai.core.retrieval.vector_store.VectorGroundTruthStore ¶

VectorGroundTruthStore(backend: VectorBackend | None = None, tenant_id: str = '')

Bases: GroundTruthStore

Ground truth store with vector-based semantic retrieval.

Extends the keyword-based GroundTruthStore with embedding-based similarity search. Falls back to keyword matching when the vector backend returns no results.

Parameters:

Name	Type	Description	Default
`backend`	`VectorBackend — vector DB backend (default: InMemoryBackend).`		`None`

fact_version ¶

fact_version(key: str, tenant_id: str = '') -> str | None

Return the semantic version currently recorded for key.

fact_version_record ¶

fact_version_record(key: str, tenant_id: str = '') -> dict[str, str] | None

Return version metadata for key without exposing mutable state.

version_manifest ¶

version_manifest(tenant_id: str = '') -> dict[str, dict[str, str]]

Return version records visible to tenant_id.

retraction_records ¶

retraction_records(tenant_id: str = '') -> list[dict[str, str]]

Return fact and chunk retraction events visible to tenant_id.

replacement_records ¶

replacement_records(tenant_id: str = '') -> list[dict[str, str]]

Return fact replacement events visible to tenant_id.

conflict_reports ¶

conflict_reports(tenant_id: str = '', key: str | None = None) -> list[dict[str, str]]

Return KB conflict reports visible to tenant_id.

kb_snapshot_records ¶

kb_snapshot_records(tenant_id: str = '') -> list[dict[str, str]]

Return canonical KB snapshot records visible to tenant_id.

kb_snapshot_root ¶

kb_snapshot_root(tenant_id: str = '') -> str

Return a deterministic Merkle root for the current KB snapshot.

kb_snapshot_audit_record ¶

kb_snapshot_audit_record(tenant_id: str = '') -> dict[str, str | int]

Return a compact audit payload for the current KB snapshot.

freshness_status_signals ¶

freshness_status_signals(tenant_id: str = '', key: str | None = None) -> list[dict[str, str | float]]

Return citation status signals for temporal freshness scoring.

retract_fact ¶

retract_fact(key: str, *, tenant_id: str = '', reason: str = '') -> dict[str, str]

Mark a fact or derived chunk source as unusable for retrieval.

replace_fact ¶

replace_fact(key: str, value: str, *, tenant_id: str = '', reason: str = '', metadata: dict[str, Any] | None = None) -> dict[str, str]

Add a replacement value and record the superseded hash.

add_fact ¶

add_fact(key: str, value: str, tenant_id: str = '', metadata: dict[str, Any] | None = None) -> None

Alias for add() — also populates parent keyword store.

ingest ¶

ingest(texts: list[str], tenant_id: str = '') -> int

Bulk-add plain text documents into the vector backend.

retrieve_context ¶

retrieve_context(query: str, top_k: int = 3, tenant_id: str = '') -> str | None

Retrieve context as a string (matching parent interface).

Falls back to keyword-based parent if vector search returns nothing.

grounded `classmethod` ¶

grounded(embedding_model: str = RECOMMENDED_EMBEDDING_MODEL, use_hybrid: bool = True, rrf_k: int = 60, tenant_id: str = '') -> VectorGroundTruthStore

Factory for the recommended grounded retrieval recipe.

Sets up hybrid retrieval (BM25 + dense) with a sentence-transformer embedding model. This is the intended production path for domain profiles (medical, finance, legal) where NLI-only scoring has 100% FPR without KB grounding.

Usage::

store = VectorGroundTruthStore.grounded()
store.ingest(["Your product documentation...", ...])
scorer = CoherenceScorer(ground_truth_store=store, use_nli=True)

Parameters:

Name	Type	Description	Default
`embedding_model`	`str`	HuggingFace model ID for dense embeddings. Default: `BAAI/bge-large-en-v1.5`.	`RECOMMENDED_EMBEDDING_MODEL`
`use_hybrid`	`bool`	Wrap dense backend with BM25 + RRF fusion (default True).	`True`
`rrf_k`	`int`	Reciprocal Rank Fusion parameter (default 60).	`60`
`tenant_id`	`str`	Default tenant scope for multi-tenant deployments.	`''`

retrieve_context_with_chunks ¶

retrieve_context_with_chunks(query: str, top_k: int = 3, tenant_id: str = '') -> list[EvidenceChunk]

Retrieve context as EvidenceChunk objects.

director_ai.core.retrieval.vector_store.VectorBackend ¶

Bases: ABC

Protocol for vector database backends.

delete ¶

delete(doc_ids: list[str]) -> int

Delete documents by id when a backend supports mutation.

aadd `async` ¶

aadd(doc_id: str, text: str, metadata: dict[str, Any] | None = None) -> None

Async add — delegates to sync add via executor by default.

aquery `async` ¶

aquery(text: str, n_results: int = 3, tenant_id: str = '') -> list[dict[str, Any]]

Async query — delegates to sync query via executor by default.

director_ai.core.retrieval.vector_store.InMemoryBackend ¶

InMemoryBackend()

Bases: VectorBackend

Simple in-memory cosine-similarity backend (no external deps).

Uses TF-IDF-like word overlap for embedding approximation. Suitable for testing and small fact stores.

director_ai.core.retrieval.vector_store.ChromaBackend ¶

ChromaBackend(collection_name: str = 'director_ai_facts', persist_directory: str | None = None, embedding_model: str | None = None)

Bases: VectorBackend

ChromaDB backend for production vector search.

Requires pip install chromadb sentence-transformers.

director_ai.core.retrieval.vector_store.SentenceTransformerBackend ¶

SentenceTransformerBackend(model_name: str = 'BAAI/bge-large-en-v1.5')

Bases: VectorBackend

Embedding-based backend using sentence-transformers directly.

Recommended model: BAAI/bge-large-en-v1.5 (best quality/speed tradeoff). Alternative: Snowflake/snowflake-arctic-embed-l for multilingual.

Requires pip install sentence-transformers.

VectorGroundTruthStore¶

Usage¶