Compare commits

..
Author SHA1 Message Date
Sameer Kadam 6b972ca683 Merge branch 'main' into feat/vector-store-migrate 2026-08-30 17:01:02 +05:30
Zohaib HassnainandSameer Kadam ea0dd17bff feat(llms): add Gemini, Ollama, DeepSeek, Novita provider wrappers (#1262)
* feat(llms): add Gemini, Ollama, DeepSeek, Novita provider wrappers

* fix(llms): address Qodo review findings on provider wrappers PR

* fix(llms): address review findings

---------

Co-authored-by: Sameer Kadam <sskadam6305@gmail.com>
2026-08-30 17:00:14 +05:30
Zohaib Hassnain 24b0b4a208 fix(vector_store): make FAISS add_vectors idempotent for retried migrations 2026-08-30 16:17:18 +05:00
Sameer Kadam c3bf29a298 Merge branch 'main' into feat/vector-store-migrate 2026-08-30 15:56:13 +05:30
Shubham Srivastava f6cd62411b test(integrations): make crewai and langchain test dirs packages (#1252)
Both directories contain a test_degradation.py. Neither had an __init__.py,
so under pytest's default prepend import mode both modules were imported as
plain 'test_degradation' and the second collided with the first:

    import file mismatch:
    imported module 'test_degradation' has this __file__ attribute:
      tests/integrations/crewai/test_degradation.py
    which is not the same as the test file we want to collect:
      tests/integrations/langchain/test_degradation.py

That aborted collection for tests/integrations/, so the langchain
graceful-degradation tests never ran. tests/integrations/__init__.py
already exists, and most directories under tests/ carry one; these two
subpackages were simply missed.

Collection goes from 335 collected, 1 error to 337 collected.

Closes #1251
2026-08-30 15:23:43 +05:00
王林 cac6dfbe45 fix(explorer): surface server error detail in graph loading failures (#1260)
The nodes/edges fetch loops were throwing away the response body whenever the request returned a non-OK status.

Because of that, errors like a `503` caused by a missing `SEMANTICA_API_KEY` only showed up as:

`Fetch failed: 503`

even though the backend was already returning a more useful message in the response `detail`.

This change reads the JSON error body and includes `detail` in the thrown error when it's a string, so `GraphLoadingOverlay` can show the actual backend error to the user.

Closes #1256
2026-08-30 15:16:15 +05:00
Zohaib Hassnain 34519e8911 fix(vector_store): address Qodo finds 2026-08-30 15:03:50 +05:00
Zohaib Hassnain 1da779276a feat(vector_store): add scan_vectors enumeration and wire up store migrate 2026-08-30 14:53:00 +05:00
Mohd Kaif 80e9737542 Merge pull request #1254 from dex0shubham/fix/1134-progress-stream-stderr
fix(utils): write console progress to stderr instead of stdout
2026-08-30 14:22:17 +05:30
Mohd Kaif 14fb975fa1 Merge branch 'main' into fix/1134-progress-stream-stderr 2026-08-30 14:12:18 +05:30
Mohd Kaif 8b0ac61afd Merge pull request #1255 from semantica-agi/feat/claude-wrapper
feat(llms): add first class Anthropic provider wrapper
2026-08-30 13:43:58 +05:30
dex0shubham 3448ac0689 test(utils): restore both module bindings in the progress fixture
Importing a submodule rebinds it as an attribute of its parent package,
so restoring only the sys.modules entry left
semantica.utils.progress_tracker and
sys.modules['semantica.utils.progress_tracker'] pointing at different
objects for every test that ran afterwards.

Addresses review feedback on #1254.
2026-08-29 15:00:04 +01:00
dex0shubham 70dfbf151c fix(utils): write console progress to stderr instead of stdout
ConsoleProgressDisplay wrote every progress frame to sys.stdout. Progress
is diagnostic output, so stderr is the correct stream for it — tqdm and
most progress renderers default there for the same reason — and stdout
must stay clean for programs that carry a machine-readable protocol on
it. The stdio MCP servers put newline-delimited JSON-RPC on stdout, where
an interleaved progress bar makes a response body unparseable (#1134).

ConsoleProgressDisplay now takes an optional stream, defaulting to
stderr. The stream is resolved per write rather than captured at
construction, so a later rebinding of sys.stderr (pytest capture, for
instance) is honoured. All writes and the four bare flushes route through
it, and the emoji-capability probe now inspects that stream rather than
stdout, so a cp1252 stderr still degrades correctly.

The existing cp1252 tests in tests/deduplication/test_deduplication.py
patched sys.stdout to assert emoji auto-disabling; they now patch the
stream progress is actually written to. Their intent is unchanged.

Closes #1134 (point 1 only; the SEMANTICA_KG_PATH persistence and README
items remain with @akaszubski)
2026-08-29 13:53:48 +01:00
28 changed files with 1872 additions and 83 deletions
+97 -24
View File
@@ -62,7 +62,7 @@ Four factors drive provider selection, each optimized for different use cases:
**Accuracy** matters most in high-stakes decisions: clinical contraindication checks, credit committee reasoning, and legal document analysis. Frontier models like Claude or GPT-4 available through `LiteLLM` provide the strongest reasoning capabilities.
**Data residency** constraints eliminate cloud providers for classified or HIPAA-regulated workloads. `HuggingFaceLLM` with local model paths enables fully air-gapped deployments without network calls.
**Data residency** constraints eliminate cloud providers for classified or HIPAA-regulated workloads. `HuggingFaceLLM` with local model paths, or `Ollama` pointed at a local server, both enable fully air-gapped deployments without network calls.
**Cost at scale** favors high-throughput providers like Novita AI for bulk extraction pipelines processing thousands of documents per hour where per-token costs accumulate quickly.
@@ -198,6 +198,79 @@ print(risk.risk_level, risk.days_to_deadline)
Model selection follows the same tier structure as the other providers: a Haiku model for high-volume classification where cost matters more than depth, a Sonnet model as the default for most extraction and reasoning tasks, an Opus model when a task genuinely needs the deepest reasoning available and latency/cost are secondary. Check Anthropic's docs for the current model identifiers, since they're versioned and change over time.
## Gemini — Long Context and Multimodal Input
**Gemini** is Google's model family, with a context window large enough to hold entire codebases or long regulatory filings in a single call, and native support for image and document input alongside text. Reach for it when a task needs to reference a large amount of source material at once, or when the input isn't plain text.
The `Gemini` provider tries the newer `google-genai` SDK first and falls back to the older `google-generativeai` package if that's what's installed. Install with `pip install "semantica[llm-gemini]"` (or `pip install google-genai`) before using this provider.
```python
from semantica.llms import Gemini
gemini = Gemini(model="gemini-pro", api_key="YOUR_GEMINI_KEY")
# api_key falls back to the GEMINI_API_KEY environment variable
if not gemini.is_available():
raise RuntimeError("Gemini provider not configured - set GEMINI_API_KEY")
response = gemini.generate(
"Summarize the key obligations in a standard NDA in three bullet points."
)
print(response)
data = gemini.generate_structured(
"Extract the party names and effective date from: "
"This Agreement is entered into between Acme Corp and Globex LLC, "
"effective January 1, 2026."
)
print(data)
```
## Ollama — Local, Air-Gapped Inference
**Ollama** runs models entirely on your own machine, with no API key and no outbound network call. It's the right choice for air-gapped environments, offline development, or any workload where the source data can't leave the local network.
Unlike the other providers here, `Ollama` takes a `base_url` instead of an `api_key`. It talks to a local Ollama server over HTTP. Start the server with `ollama serve` and pull a model with `ollama pull llama2` before using this provider. Install the Python client with `pip install "semantica[llm-ollama]"` (or `pip install ollama`).
```python
from semantica.llms import Ollama
llm = Ollama(model="llama2", base_url="http://localhost:11434")
if not llm.is_available():
raise RuntimeError("Ollama provider not configured - is 'ollama serve' running?")
response = llm.generate("Explain the difference between a hash map and a tree map.")
print(response)
```
`is_available()` for Ollama does a real connectivity check (it calls the server's `list()` endpoint), unlike the API-key-based providers above, so a `False` here usually means the server isn't running rather than a missing credential.
## DeepSeek — Budget Reasoning at Scale
**DeepSeek** exposes an OpenAI-compatible API at a fraction of the cost of the larger US providers, with reasoning quality that holds up well for extraction and classification work. It's a reasonable default when you're processing a large volume of documents and don't need the deepest reasoning tier.
Install with `pip install "semantica[llm-deepseek]"` (or `pip install openai`, since DeepSeek is accessed through the OpenAI client pointed at a different base URL).
```python
from semantica.llms import DeepSeek
llm = DeepSeek(model="deepseek-chat", api_key="YOUR_DEEPSEEK_KEY")
# api_key falls back to the DEEPSEEK_API_KEY environment variable
if not llm.is_available():
raise RuntimeError("DeepSeek provider not configured - set DEEPSEEK_API_KEY")
response = llm.generate("List three risks of using a floating IP in a Kubernetes ingress.")
print(response)
data = llm.generate_structured(
"Extract the CVE ID and affected product from: "
"CVE-2024-3400 affects PAN-OS GlobalProtect gateways."
)
print(data)
```
## LiteLLM — One Interface, 100+ Providers
**LiteLLM** is a universal adapter that provides a single interface to over 100 different LLM providers, including Anthropic Claude, Azure OpenAI, AWS Bedrock, Google Vertex AI, and local Ollama instances. It acts as a translation layer, converting your unified API calls into provider-specific requests, enabling easy switching between providers without code changes.
@@ -361,30 +434,32 @@ for t in triplets:
## Novita AI — Cost-Efficient Bulk Extraction
Novita AI exposes an OpenAI-compatible API and is available as a built-in provider for the extraction layer. It is accessed differently from the `semantica.llms` classes — through `create_provider` from `semantica.semantic_extract.providers` — making it the right choice for high-volume NER pipelines where per-call cost matters.
**Novita AI** exposes an OpenAI-compatible API at low per-call cost, making it a reasonable choice for high-volume NER pipelines where cost matters more than getting the single best answer.
Install with `pip install "semantica[llm-novita]"` (or `pip install openai`, since Novita is accessed through the OpenAI client pointed at a different base URL).
```python
from semantica.llms import Novita
llm = Novita(model="deepseek/deepseek-v3.2", api_key="YOUR_NOVITA_KEY")
# api_key falls back to the NOVITA_API_KEY environment variable
if not llm.is_available():
raise RuntimeError("Novita provider not configured - set NOVITA_API_KEY")
response = llm.generate("Summarize the Basel III leverage ratio requirement.")
data = llm.generate_structured(
"Extract drug names and dosages from: "
"Patient received warfarin 5mg daily, aspirin 75mg daily, metformin 500mg twice daily."
)
```
Novita is also reachable as a provider name string for the NER interface, without going through the `Novita` class directly:
```python
from semantica.semantic_extract.providers import create_provider
from semantica.semantic_extract import NamedEntityRecognizer
# create_provider pools instances — same key reuses the same object
provider = create_provider(
"novita",
api_key="YOUR_NOVITA_KEY", # or set NOVITA_API_KEY env var
model="deepseek/deepseek-v3.2", # default model
)
if provider.is_available():
# Plain generation
response = provider.generate("Summarise the Basel III leverage ratio requirement.")
# Structured extraction — returns parsed dict
data = provider.generate_structured(
"Extract drug names and dosages from: "
"Patient received warfarin 5mg daily, aspirin 75mg daily, metformin 500mg twice daily."
)
# Use Novita through the NER interface — provider name as string
ner = NamedEntityRecognizer(
methods=["llm"],
provider="novita",
@@ -394,11 +469,9 @@ entities = ner.extract_entities(
"CVE-2024-3400 is exploited by UNC3886 targeting PAN-OS GlobalProtect."
)
for e in entities:
print("{} ({}) conf={:.2f}".format(e.text, e.label, e.confidence))
print("{} ({}) conf={:.2f}".format(e.text, e.label, e.confidence))
```
Novita requires the `openai` Python client under the hood — install with `pip install "semantica[llm-openai]"` or `pip install openai`.
## Domain Examples
<Tabs>
@@ -318,6 +318,20 @@ interface EdgeListResponse {
const PAGE_LIMIT = 1000;
/** Surface the server's `detail` message (e.g. auth/setup guidance) on non-OK responses. */
async function fetchErrorDetail(response: Response): Promise<string> {
try {
const body: unknown = await response.json();
const detail = (body as { detail?: unknown } | null)?.detail;
if (typeof detail === "string" && detail.trim()) {
return `${detail.trim()}`;
}
} catch {
// Non-JSON or unreadable body: fall back to the status-only message.
}
return "";
}
async function fetchAllNodes(
signal: AbortSignal,
onProgress?: (progress: GraphLoadProgress) => void,
@@ -335,7 +349,7 @@ async function fetchAllNodes(
const response = await fetch(url.toString(), { signal });
if (!response.ok) {
throw new Error(`Fetch failed: ${response.status}`);
throw new Error(`Fetch failed: ${response.status}${await fetchErrorDetail(response)}`);
}
const data: NodeListResponse = await response.json();
@@ -390,7 +404,7 @@ async function fetchAllEdges(
const response = await fetch(url.toString(), { signal });
if (!response.ok) {
throw new Error(`Fetch failed: ${response.status}`);
throw new Error(`Fetch failed: ${response.status}${await fetchErrorDetail(response)}`);
}
const data: EdgeListResponse = await response.json();
+2 -1
View File
@@ -107,11 +107,12 @@ llm-gemini = ["google-genai>=0.1.0"]
llm-anthropic = ["anthropic>=0.122.0"]
llm-ollama = ["ollama>=0.1.0"]
llm-deepseek = ["openai>=1.0.0"]
llm-novita = ["openai>=1.0.0"]
llm-litellm = ["litellm>=1.83.9"]
llm-instructor = ["instructor>=1.15.3"]
llm-all = [
"semantica[llm-openai,llm-groq,llm-gemini,llm-anthropic,llm-ollama,llm-deepseek,llm-litellm,llm-instructor]"
"semantica[llm-openai,llm-groq,llm-gemini,llm-anthropic,llm-ollama,llm-deepseek,llm-novita,llm-litellm,llm-instructor]"
]
# ---- Document Parsing ----
+114 -9
View File
@@ -3714,19 +3714,61 @@ def store_stats(cli_ctx: CLIContext, backend: str, fmt: str, local_json: bool) -
_run_with_error_handling(_action)
_MIGRATE_SUPPORTED_BACKENDS = {"faiss", "sqlite", "pgvector"}
_MIGRATE_BATCH_SIZE = 500
def _migrate_backend_config(vs_cfg: Dict[str, Any], backend: str) -> Dict[str, Any]:
"""Resolve per-backend config out of the vector_store config section.
Supports both a per-backend nested shape (``vector_store.faiss.dimension``)
and the common flat single-backend shape (``vector_store.backend`` +
sibling keys), since either can appear depending on how many backends a
user has configured.
"""
nested = vs_cfg.get(backend)
if isinstance(nested, dict):
return dict(nested)
if vs_cfg.get("backend") == backend:
return {k: v for k, v in vs_cfg.items() if k != "backend"}
return {}
def _require_faiss_index_path(cfg: Dict[str, Any], role: str) -> str:
"""FAISS has no server to hold state between commands: a fresh FAISSStore
starts empty and nothing outside the process persists it, so migration
needs an explicit on-disk index to read from or write to."""
index_path = cfg.get("index_path")
if not index_path:
raise click.ClickException(
f"faiss as migration {role} requires 'index_path' in the vector_store "
f"config (vector_store.faiss.index_path or vector_store.index_path "
f"when faiss is the configured backend)."
)
return index_path
@store.command("migrate")
@click.option("--from", "from_backend", required=True)
@click.option("--to", "to_backend", required=True)
@click.option("--namespace", default=None)
@click.option("--dry-run", "local_dry", is_flag=True, default=False)
@click.option("--json", "local_json", is_flag=True, default=False)
@click.pass_obj
def store_migrate(cli_ctx: CLIContext, from_backend: str, to_backend: str,
namespace: Optional[str], local_dry: bool) -> None:
namespace: Optional[str], local_dry: bool, local_json: bool) -> None:
"""Migrate data between backends.
Direct migration is only wired up between faiss, sqlite, and pgvector -
these are the backends whose storage contract supports paging through
every stored vector. Migrating to or from qdrant, pinecone, milvus, or
weaviate still needs the export/reindex workaround below, since each of
those needs its own enumeration design (Qdrant scroll, Pinecone list,
etc.) that hasn't been built yet.
\b
Example:
semantica store migrate --from faiss --to qdrant --namespace production --dry-run
semantica store migrate --from faiss --to sqlite --namespace production --dry-run
"""
cli_ctx = _require_ctx(cli_ctx)
@@ -3734,13 +3776,76 @@ def store_migrate(cli_ctx: CLIContext, from_backend: str, to_backend: str,
if _is_dry(cli_ctx, local_dry):
_dry(cli_ctx, "migrate", from_backend=from_backend, to_backend=to_backend)
return
raise click.ClickException(
f"Direct backend migration ({from_backend}{to_backend}) is not yet supported "
"by the vector store layer. To migrate, export your data first:\n"
" semantica export --format parquet --output dump.parquet\n"
f" semantica embed index dump.parquet --store {to_backend}"
+ (f" --namespace {namespace}" if namespace else "")
)
if from_backend not in _MIGRATE_SUPPORTED_BACKENDS or to_backend not in _MIGRATE_SUPPORTED_BACKENDS:
raise click.ClickException(
f"Direct backend migration ({from_backend}{to_backend}) is only supported "
f"between {', '.join(sorted(_MIGRATE_SUPPORTED_BACKENDS))}. To migrate involving "
"another backend, export your data first:\n"
" semantica export --format parquet --output dump.parquet\n"
f" semantica embed index dump.parquet --store {to_backend}"
+ (f" --namespace {namespace}" if namespace else "")
)
from .vector_store import VectorStore
vs_cfg = cli_ctx.config.to_dict().get("vector_store", {}) or {}
source_cfg = _migrate_backend_config(vs_cfg, from_backend)
dest_cfg = _migrate_backend_config(vs_cfg, to_backend)
source_index_path = None
if from_backend == "faiss":
source_index_path = _require_faiss_index_path(source_cfg, "source")
dest_index_path = None
if to_backend == "faiss":
dest_index_path = _require_faiss_index_path(dest_cfg, "destination")
source = VectorStore(backend=from_backend, config=source_cfg)
if source_index_path:
source._backend_store.load_index(source_index_path)
source_dimension = getattr(source._backend_store, "dimension", None)
if source_dimension and "dimension" not in dest_cfg:
dest_cfg["dimension"] = source_dimension
dest = VectorStore(backend=to_backend, config=dest_cfg)
if dest_index_path and Path(dest_index_path).exists():
dest._backend_store.load_index(dest_index_path)
migrated = 0
vectors_batch: List[Any] = []
metadata_batch: List[Dict[str, Any]] = []
ids_batch: List[str] = []
def _flush() -> None:
nonlocal migrated
if not vectors_batch:
return
dest.store_vectors(list(vectors_batch), list(metadata_batch), ids=list(ids_batch))
migrated += len(vectors_batch)
vectors_batch.clear()
metadata_batch.clear()
ids_batch.clear()
for item in source.iter_vectors(batch_size=_MIGRATE_BATCH_SIZE):
meta = dict(item.get("metadata") or {})
if namespace and "namespace" not in meta:
meta["namespace"] = namespace
vectors_batch.append(item["vector"])
metadata_batch.append(meta)
ids_batch.append(item["id"])
if len(vectors_batch) >= _MIGRATE_BATCH_SIZE:
_flush()
_flush()
if dest_index_path and migrated:
dest._backend_store.save_index(dest_index_path)
result = {"from": from_backend, "to": to_backend, "migrated": migrated}
if _is_json(cli_ctx, local_json):
_jecho(result)
else:
_ok(cli_ctx, f"Migrated {migrated} vectors from {from_backend} to {to_backend}")
_run_with_error_handling(_action)
+39 -6
View File
@@ -11,22 +11,26 @@ Supported Providers:
- HuggingFaceLLM: HuggingFace Transformers for local LLM inference
- LiteLLM: Unified interface to 100+ LLM providers (OpenAI, Anthropic, Groq, Azure, Bedrock, Vertex AI, etc.)
- Anthropic: Anthropic Claude API (Claude sonnet, Opus, Haiku, etc.)
- Gemini: Google Gemini API
- Ollama: Local models served through Ollama
- DeepSeek: DeepSeek's OpenAI-compatible API
- Novita: Novita AI's OpenAI-compatible API
Example Usage:
>>> from semantica.llms import Groq, OpenAI, HuggingFaceLLM, LiteLLM, Anthropic
>>>
>>>
>>> # Groq provider
>>> groq = Groq(model="llama-3.1-8b-instant", api_key="your-key")
>>> response = groq.generate("Hello, world!")
>>>
>>>
>>> # OpenAI provider
>>> openai = OpenAI(model="gpt-4", api_key="your-key")
>>> response = openai.generate("Hello, world!")
>>>
>>>
>>> # HuggingFace LLM provider
>>> hf = HuggingFaceLLM(model_name="gpt2")
>>> response = hf.generate("Hello, world!")
>>>
>>>
>>> # LiteLLM provider (supports 100+ LLMs)
>>> llm = LiteLLM(model="openai/gpt-4o", api_key="your-key")
>>> response = llm.generate("Hello, world!")
@@ -37,6 +41,22 @@ Example Usage:
>>> # Anthropic provider
>>> claude = Anthropic(model="claude-sonnet-4-6", api_key="the-key")
>>> response = claude.generate("Hello, world!")
>>>
>>> # Gemini provider
>>> gemini = Gemini(model="gemini-pro", api_key="your-key")
>>> response = gemini.generate("Hello, world!")
>>>
>>> # Ollama provider (local, no api_key)
>>> ollama = Ollama(model="llama2")
>>> response = ollama.generate("Hello, world!")
>>>
>>> # DeepSeek provider
>>> deepseek = DeepSeek(model="deepseek-chat", api_key="your-key")
>>> response = deepseek.generate("Hello, world!")
>>>
>>> # Novita provider
>>> novita = Novita(model="deepseek/deepseek-v3.2", api_key="your-key")
>>> response = novita.generate("Hello, world!")
Author: Semantica Contributors
License: MIT
@@ -47,6 +67,19 @@ from .openai import OpenAI
from .huggingface import HuggingFaceLLM
from .litellm import LiteLLM
from .anthropic import Anthropic
from .gemini import Gemini
from .ollama import Ollama
from .deepseek import DeepSeek
from .novita import Novita
__all__ = ["Groq", "OpenAI", "HuggingFaceLLM", "LiteLLM", "Anthropic"]
__all__ = [
"Groq",
"OpenAI",
"HuggingFaceLLM",
"LiteLLM",
"Anthropic",
"Gemini",
"Ollama",
"DeepSeek",
"Novita",
]
+111
View File
@@ -0,0 +1,111 @@
"""
DeepSeek LLM Provider
Wrapper for DeepSeek API provider with clean interface.
"""
from typing import Any, Dict, List, Optional, Union
from ..semantic_extract.providers import DeepSeekProvider
from ..utils.exceptions import ProcessingError
from ..utils.logging import get_logger
logger = get_logger("llms.deepseek")
class DeepSeek:
"""
DeepSeek LLM provider wrapper.
Provides clean interface to DeepSeek's OpenAI-compatible API.
Example:
>>> from semantica.llms import DeepSeek
>>> llm = DeepSeek(model="deepseek-chat", api_key="your-key")
>>> response = llm.generate("What is AI?")
"""
def __init__(
self,
model: str = "deepseek-chat",
api_key: Optional[str] = None,
**kwargs
):
"""
Initialize DeepSeek provider.
Args:
model: Model name (default: "deepseek-chat")
api_key: DeepSeek API key (default: from DEEPSEEK_API_KEY env var)
**kwargs: Additional provider options
"""
self.provider = DeepSeekProvider(api_key=api_key, model=model, **kwargs)
self.model = model
self.api_key = api_key
def is_available(self) -> bool:
"""Check if DeepSeek provider is available."""
return self.provider.is_available()
def generate(self, prompt: str, **kwargs) -> str:
"""
Generate text from prompt.
Args:
prompt: Input prompt text
**kwargs: Generation options (temperature, max_tokens, etc.)
Returns:
Generated text response
Raises:
ProcessingError: If provider is not available or generation fails
"""
if not self.is_available():
raise ProcessingError(
"DeepSeek provider not available. Set DEEPSEEK_API_KEY or pass api_key."
)
return self.provider.generate(prompt, **kwargs)
def generate_structured(self, prompt: str, **kwargs) -> Union[Dict[str, Any], List[Any]]:
"""
Generate structured JSON output.
Args:
prompt: Input prompt text
**kwargs: Generation options
Returns:
Parsed JSON response. A dict for a top-level JSON object, or a
list if the model returns a top-level JSON array.
Raises:
ProcessingError: If provider is not available or generation fails
"""
if not self.is_available():
raise ProcessingError(
"DeepSeek provider not available. Set DEEPSEEK_API_KEY or pass api_key."
)
return self.provider.generate_structured(prompt, **kwargs)
def generate_typed(self, prompt: str, schema: Any, max_retries: int = 3, **kwargs) -> Any:
"""
Generate output validated against a Pydantic schema.
Args:
prompt: Input prompt text
schema: Pydantic model class to validate the output against
max_retries: Number of retries if validation fails (default: 3)
**kwargs: Generation options
Returns:
An instance of `schema`, populated from the model's response
Raises:
ProcessingError: If provider is not available or generation fails
"""
if not self.is_available():
raise ProcessingError(
"DeepSeek provider not available. Set DEEPSEEK_API_KEY or pass api_key."
)
return self.provider.generate_typed(prompt, schema, max_retries=max_retries, **kwargs)
+111
View File
@@ -0,0 +1,111 @@
"""
Gemini LLM Provider
Wrapper for Google Gemini API provider with clean interface.
"""
from typing import Any, Dict, List, Optional, Union
from ..semantic_extract.providers import GeminiProvider
from ..utils.exceptions import ProcessingError
from ..utils.logging import get_logger
logger = get_logger("llms.gemini")
class Gemini:
"""
Google Gemini LLM provider wrapper.
Provides clean interface to Google's Gemini API.
Example:
>>> from semantica.llms import Gemini
>>> gemini = Gemini(model="gemini-pro", api_key="your-key")
>>> response = gemini.generate("What is AI?")
"""
def __init__(
self,
model: str = "gemini-pro",
api_key: Optional[str] = None,
**kwargs
):
"""
Initialize Gemini provider.
Args:
model: Model name (default: "gemini-pro")
api_key: Gemini API key (default: from GEMINI_API_KEY env var)
**kwargs: Additional provider options
"""
self.provider = GeminiProvider(api_key=api_key, model=model, **kwargs)
self.model = model
self.api_key = api_key
def is_available(self) -> bool:
"""Check if Gemini provider is available."""
return self.provider.is_available()
def generate(self, prompt: str, **kwargs) -> str:
"""
Generate text from prompt.
Args:
prompt: Input prompt text
**kwargs: Generation options (temperature, max_tokens, etc.)
Returns:
Generated text response
Raises:
ProcessingError: If provider is not available or generation fails
"""
if not self.is_available():
raise ProcessingError(
"Gemini provider not available. Set GEMINI_API_KEY or pass api_key."
)
return self.provider.generate(prompt, **kwargs)
def generate_structured(self, prompt: str, **kwargs) -> Union[Dict[str, Any], List[Any]]:
"""
Generate structured JSON output.
Args:
prompt: Input prompt text
**kwargs: Generation options
Returns:
Parsed JSON response. A dict for a top-level JSON object, or a
list if the model returns a top-level JSON array.
Raises:
ProcessingError: If provider is not available or parsing fails
"""
if not self.is_available():
raise ProcessingError(
"Gemini provider not available. Set GEMINI_API_KEY or pass api_key."
)
return self.provider.generate_structured(prompt, **kwargs)
def generate_typed(self, prompt: str, schema: Any, max_retries: int = 3, **kwargs) -> Any:
"""
Generate output validated against a Pydantic schema.
Args:
prompt: Input prompt text
schema: Pydantic model class to validate the output against
max_retries: Number of retries if validation fails (default: 3)
**kwargs: Generation options
Returns:
An instance of `schema`, populated from the model's response
Raises:
ProcessingError: If provider is not available or generation fails
"""
if not self.is_available():
raise ProcessingError(
"Gemini provider not available. Set GEMINI_API_KEY or pass api_key."
)
return self.provider.generate_typed(prompt, schema, max_retries=max_retries, **kwargs)
+111
View File
@@ -0,0 +1,111 @@
"""
Novita LLM Provider
Wrapper for Novita AI's OpenAI-compatible API with clean interface.
"""
from typing import Any, Dict, List, Optional, Union
from ..semantic_extract.providers import NovitaProvider
from ..utils.exceptions import ProcessingError
from ..utils.logging import get_logger
logger = get_logger("llms.novita")
class Novita:
"""
Novita AI LLM provider wrapper.
Provides clean interface to Novita's OpenAI-compatible API.
Example:
>>> from semantica.llms import Novita
>>> llm = Novita(model="deepseek/deepseek-v3.2", api_key="your-key")
>>> response = llm.generate("What is AI?")
"""
def __init__(
self,
model: str = "deepseek/deepseek-v3.2",
api_key: Optional[str] = None,
**kwargs
):
"""
Initialize Novita provider.
Args:
model: Model name (default: "deepseek/deepseek-v3.2")
api_key: Novita API key (default: from NOVITA_API_KEY env var)
**kwargs: Additional provider options
"""
self.provider = NovitaProvider(api_key=api_key, model=model, **kwargs)
self.model = model
self.api_key = api_key
def is_available(self) -> bool:
"""Check if Novita provider is available."""
return self.provider.is_available()
def generate(self, prompt: str, **kwargs) -> str:
"""
Generate text from prompt.
Args:
prompt: Input prompt text
**kwargs: Generation options (temperature, max_tokens, etc.)
Returns:
Generated text response
Raises:
ProcessingError: If provider is not available or generation fails
"""
if not self.is_available():
raise ProcessingError(
"Novita provider not available. Set NOVITA_API_KEY or pass api_key."
)
return self.provider.generate(prompt, **kwargs)
def generate_structured(self, prompt: str, **kwargs) -> Union[Dict[str, Any], List[Any]]:
"""
Generate structured JSON output.
Args:
prompt: Input prompt text
**kwargs: Generation options
Returns:
Parsed JSON response. A dict for a top-level JSON object, or a
list if the model returns a top-level JSON array.
Raises:
ProcessingError: If provider is not available or generation fails
"""
if not self.is_available():
raise ProcessingError(
"Novita provider not available. Set NOVITA_API_KEY or pass api_key."
)
return self.provider.generate_structured(prompt, **kwargs)
def generate_typed(self, prompt: str, schema: Any, max_retries: int = 3, **kwargs) -> Any:
"""
Generate output validated against a Pydantic schema.
Args:
prompt: Input prompt text
schema: Pydantic model class to validate the output against
max_retries: Number of retries if validation fails (default: 3)
**kwargs: Generation options
Returns:
An instance of `schema`, populated from the model's response
Raises:
ProcessingError: If provider is not available or generation fails
"""
if not self.is_available():
raise ProcessingError(
"Novita provider not available. Set NOVITA_API_KEY or pass api_key."
)
return self.provider.generate_typed(prompt, schema, max_retries=max_retries, **kwargs)
+116
View File
@@ -0,0 +1,116 @@
"""
Ollama LLM Provider
Wrapper for local Ollama models with clean interface.
"""
from typing import Any, Dict, List, Union
from ..semantic_extract.providers import OllamaProvider
from ..utils.exceptions import ProcessingError
from ..utils.logging import get_logger
logger = get_logger("llms.ollama")
class Ollama:
"""
Ollama LLM provider wrapper.
Provides clean interface to a local Ollama server. Unlike the other
providers here, this one has no API key. It talks to an Ollama
instance over HTTP, so make sure `ollama serve` is running first.
Example:
>>> from semantica.llms import Ollama
>>> llm = Ollama(model="llama2")
>>> response = llm.generate("What is AI?")
"""
def __init__(
self,
model: str = "llama2",
base_url: str = "http://localhost:11434",
**kwargs
):
"""
Initialize Ollama provider.
Args:
model: Model name (default: "llama2")
base_url: Ollama server URL (default: "http://localhost:11434")
**kwargs: Additional provider options
"""
self.provider = OllamaProvider(base_url=base_url, model=model, **kwargs)
self.model = model
self.base_url = base_url
def is_available(self) -> bool:
"""Check if Ollama provider is available."""
return self.provider.is_available()
def generate(self, prompt: str, **kwargs) -> str:
"""
Generate text from prompt.
Args:
prompt: Input prompt text
**kwargs: Generation options (temperature, max_tokens, etc.)
Returns:
Generated text response
Raises:
ProcessingError: If provider is not available or generation fails
"""
if not self.is_available():
raise ProcessingError(
"Ollama provider not available. Make sure Ollama is running "
"and reachable at the configured base_url."
)
return self.provider.generate(prompt, **kwargs)
def generate_structured(self, prompt: str, **kwargs) -> Union[Dict[str, Any], List[Any]]:
"""
Generate structured JSON output.
Args:
prompt: Input prompt text
**kwargs: Generation options
Returns:
Parsed JSON response. A dict for a top-level JSON object, or a
list if the model returns a top-level JSON array.
Raises:
ProcessingError: If provider is not available or parsing fails
"""
if not self.is_available():
raise ProcessingError(
"Ollama provider not available. Make sure Ollama is running "
"and reachable at the configured base_url."
)
return self.provider.generate_structured(prompt, **kwargs)
def generate_typed(self, prompt: str, schema: Any, max_retries: int = 3, **kwargs) -> Any:
"""
Generate output validated against a Pydantic schema.
Args:
prompt: Input prompt text
schema: Pydantic model class to validate the output against
max_retries: Number of retries if validation fails (default: 3)
**kwargs: Generation options
Returns:
An instance of `schema`, populated from the model's response
Raises:
ProcessingError: If provider is not available or generation fails
"""
if not self.is_available():
raise ProcessingError(
"Ollama provider not available. Make sure Ollama is running "
"and reachable at the configured base_url."
)
return self.provider.generate_typed(prompt, schema, max_retries=max_retries, **kwargs)
+53 -8
View File
@@ -673,6 +673,7 @@ class GeminiProvider(BaseProvider):
self.model = model
self.client = None
self._use_new_genai = False
self._legacy_model_cache: Dict[str, Any] = {}
self._init_client()
def _init_client(self):
@@ -694,6 +695,38 @@ class GeminiProvider(BaseProvider):
self.client = None
self.logger.warning("Gemini SDK not installed. Install with: pip install semantica[llm-gemini]")
def _legacy_client_for(self, requested_model: str):
"""Return a legacy-SDK GenerativeModel bound to this instance's own
API key, for the given model name.
The legacy google-generativeai package keeps its API key as
module-level state (genai.configure()), so any GenerativeModel built
by a different GeminiProvider instance in the same process can leave
that state pointing at a different key. Re-asserting configure()
with this instance's key right before use, instead of only once at
construction, keeps sequential calls across instances from reading
each other's credentials. A cache keyed by model name avoids
rebuilding a GenerativeModel on every call for the common case of
one model being reused.
"""
try:
import google.generativeai as old_genai
old_genai.configure(api_key=self.api_key)
except Exception:
# _init_client() already required this import to reach the
# legacy path in the first place, so this only happens when
# self.client was injected directly (tests). Fall back to it
# without reasserting credentials rather than failing calls
# that never needed the real SDK.
return self.client
if requested_model == self.model:
return self.client
cached = self._legacy_model_cache.get(requested_model)
if cached is None:
cached = old_genai.GenerativeModel(requested_model)
self._legacy_model_cache[requested_model] = cached
return cached
def is_available(self) -> bool:
"""Check if provider is available."""
return self.client is not None
@@ -724,7 +757,8 @@ class GeminiProvider(BaseProvider):
)
return self._resp_text(resp)
else:
response = self.client.generate_content(prompt, generation_config=config or None)
legacy_client = self._legacy_client_for(kwargs.get("model", self.model))
response = legacy_client.generate_content(prompt, generation_config=config or None)
return self._resp_text(response)
def generate_structured(self, prompt: str, **kwargs) -> dict:
@@ -733,15 +767,24 @@ class GeminiProvider(BaseProvider):
raise ProcessingError("Gemini client not initialized.")
json_prompt = f"{prompt}\n\nReturn the response as valid JSON only."
config = {}
self._add_if_set(config, kwargs, "temperature", "top_p", "top_k", "stop_sequences", "candidate_count")
if "max_tokens" in kwargs:
config["max_output_tokens"] = kwargs["max_tokens"]
if self._use_new_genai:
model = kwargs.get("model", self.model)
resp = self.client.models.generate_content(model=model, contents=json_prompt)
resp = self.client.models.generate_content(
model=model, contents=json_prompt, config=config or None
)
try:
return self._parse_json(self._resp_text(resp))
except Exception as e:
raise ProcessingError(f"Failed to parse JSON from Gemini response: {e}")
else:
response = self.client.generate_content(json_prompt)
legacy_client = self._legacy_client_for(kwargs.get("model", self.model))
response = legacy_client.generate_content(json_prompt, generation_config=config or None)
try:
return self._parse_json(self._resp_text(response))
except Exception as e:
@@ -967,6 +1010,8 @@ class OllamaProvider(BaseProvider):
def is_available(self) -> bool:
"""Check if provider is available."""
if self.client is None:
self._init_client()
return self.client is not None
def _build_options(self, kwargs: dict) -> Optional[dict]:
@@ -1018,7 +1063,6 @@ class DeepSeekProvider(BaseProvider):
self.api_key = api_key or config.get_api_key("deepseek")
self.base_url = "https://api.deepseek.com/v1"
self.model = model
self.base_url = "https://api.deepseek.com/v1"
self.client = None
self._init_client()
@@ -1045,7 +1089,7 @@ class DeepSeekProvider(BaseProvider):
"model": kwargs.get("model", self.model),
"messages": [{"role": "user", "content": prompt}],
}
self._add_if_set(create_kwargs, kwargs, "temperature", "max_tokens")
self._add_if_set(create_kwargs, kwargs, "temperature", "max_tokens", "top_p", "frequency_penalty", "presence_penalty", "seed", "stop", "user")
response = self.client.chat.completions.create(**create_kwargs)
return response.choices[0].message.content
@@ -1058,8 +1102,9 @@ class DeepSeekProvider(BaseProvider):
create_kwargs = {
"model": kwargs.get("model", self.model),
"messages": [{"role": "user", "content": prompt}],
"response_format": {"type": "json_object"},
}
self._add_if_set(create_kwargs, kwargs, "temperature", "max_tokens")
self._add_if_set(create_kwargs, kwargs, "temperature", "max_tokens", "top_p", "frequency_penalty", "presence_penalty", "seed", "stop", "user")
response = self.client.chat.completions.create(**create_kwargs)
try:
@@ -1103,7 +1148,7 @@ class NovitaProvider(BaseProvider):
"model": kwargs.get("model", self.model),
"messages": [{"role": "user", "content": prompt}],
}
self._add_if_set(create_kwargs, kwargs, "temperature", "max_tokens")
self._add_if_set(create_kwargs, kwargs, "temperature", "max_tokens", "top_p", "frequency_penalty", "presence_penalty", "seed", "stop", "user")
response = self.client.chat.completions.create(**create_kwargs)
return response.choices[0].message.content
@@ -1118,7 +1163,7 @@ class NovitaProvider(BaseProvider):
"messages": [{"role": "user", "content": prompt}],
"response_format": {"type": "json_object"},
}
self._add_if_set(create_kwargs, kwargs, "temperature", "max_tokens")
self._add_if_set(create_kwargs, kwargs, "temperature", "max_tokens", "top_p", "frequency_penalty", "presence_penalty", "seed", "stop", "user")
response = self.client.chat.completions.create(**create_kwargs)
try:
+43 -14
View File
@@ -43,7 +43,7 @@ from contextlib import contextmanager
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Tuple, Union
from typing import Any, Callable, Dict, List, Optional, TextIO, Tuple, Union
from .logging import get_logger
@@ -144,14 +144,37 @@ class ProgressDisplay(ABC):
class ConsoleProgressDisplay(ProgressDisplay):
"""Console progress display with real-time updates."""
def __init__(self, use_emoji: bool = True, update_interval: float = 0.1):
def __init__(
self,
use_emoji: bool = True,
update_interval: float = 0.1,
stream: Optional[TextIO] = None,
):
"""Initialize the console display.
Args:
use_emoji: Whether to decorate output with emoji.
update_interval: Minimum seconds between redraws.
stream: Where progress is written. Defaults to ``sys.stderr``.
Progress is diagnostic output, so stderr is the correct stream
for it, and stdout must stay clean for programs that carry a
machine-readable protocol on it — the stdio MCP servers put
newline-delimited JSON-RPC there, and a progress bar on stdout
corrupts that stream.
Left as ``None``, the stream is resolved on each write rather
than captured here, so a later rebinding of ``sys.stderr``
(pytest capture, for instance) is honoured.
"""
self.use_emoji = use_emoji
# Check if stdout supports emojis (especially on Windows)
self._stream = stream
# Check if the target stream supports emojis (especially on Windows)
if self.use_emoji:
try:
# Try encoding a test emoji with stdout's encoding
encoding = getattr(sys.stdout, "encoding", None)
# Try encoding a test emoji with the stream's encoding
encoding = getattr(self.stream, "encoding", None)
if encoding:
"🧠".encode(encoding)
except (UnicodeEncodeError, LookupError, AttributeError):
@@ -162,6 +185,11 @@ class ConsoleProgressDisplay(ProgressDisplay):
self.current_lines: Dict[str, str] = {}
self.lock = threading.Lock()
@property
def stream(self) -> TextIO:
"""The stream progress is written to; ``sys.stderr`` unless overridden."""
return self._stream if self._stream is not None else sys.stderr
def _should_update(self) -> bool:
"""Check if enough time has passed for update."""
now = time.time()
@@ -259,15 +287,16 @@ class ConsoleProgressDisplay(ProgressDisplay):
return f"{base_msg}: {message}"
def _safe_write(self, text: str) -> None:
"""Safely write text to stdout handling encoding errors."""
"""Safely write text to the progress stream handling encoding errors."""
stream = self.stream
try:
sys.stdout.write(text)
stream.write(text)
except UnicodeEncodeError:
# Fallback: encode with replacement and write decoded
# Use ascii as safe fallback if encoding is unknown or caused error
encoding = getattr(sys.stdout, "encoding", None) or "ascii"
encoding = getattr(stream, "encoding", None) or "ascii"
safe_text = text.encode(encoding, errors="replace").decode(encoding)
sys.stdout.write(safe_text)
stream.write(safe_text)
def update(self, item: ProgressItem) -> None:
"""Update console progress display."""
@@ -302,11 +331,11 @@ class ConsoleProgressDisplay(ProgressDisplay):
self._display_item_line(pipeline_item)
self._safe_write("\n")
sys.stdout.flush()
self.stream.flush()
else:
# Original single-item display
self._display_item_line(item)
sys.stdout.flush()
self.stream.flush()
def _display_item_line(self, item: ProgressItem) -> None:
"""Display a single progress item line."""
@@ -468,13 +497,13 @@ class ConsoleProgressDisplay(ProgressDisplay):
f"Completed: {completed} | Failed: {failed} | Total Time: {total_time:.2f}s\n"
)
self._safe_write("=" * 80 + "\n")
sys.stdout.flush()
self.stream.flush()
def clear(self) -> None:
"""Clear console display."""
with self.lock:
self._safe_write("\r" + " " * 100 + "\r")
sys.stdout.flush()
self.stream.flush()
self.current_lines.clear()
+55 -4
View File
@@ -66,12 +66,32 @@ class FAISSIndex:
self.metadata: Dict[str, Dict[str, Any]] = {}
def add_vectors(self, vectors: np.ndarray, ids: Optional[List[str]] = None):
"""Add vectors to index."""
"""
Add vectors to index.
Skips any id already present in vector_ids rather than appending a
second physical vector under the same id. FAISS indices here don't
support removing or replacing a single vector in place, so an
"update" isn't possible; without this check, re-running an add for
ids that already exist (e.g. retrying an interrupted migration)
would silently duplicate vectors under the same id on every retry.
"""
if ids is None:
ids = [f"vec_{i}" for i in range(len(vectors))]
self.index.add(vectors.astype(np.float32))
self.vector_ids.extend(ids)
new_rows = []
new_ids = []
existing = set(self.vector_ids)
for row, vec_id in zip(vectors, ids):
if vec_id in existing:
continue
new_rows.append(row)
new_ids.append(vec_id)
existing.add(vec_id)
if new_rows:
self.index.add(np.array(new_rows, dtype=np.float32))
self.vector_ids.extend(new_ids)
def search(
self, query_vectors: np.ndarray, k: int = 10
@@ -305,6 +325,12 @@ class FAISSStore:
"""
Add vectors to index.
Any id that already exists in the index is skipped rather than
stored as a second physical vector under the same id (see
FAISSIndex.add_vectors), so calling this again with ids from a
previous call is safe and doesn't accumulate duplicates. Metadata
for those ids is still updated.
Args:
vectors: List of vectors or numpy array
ids: Vector IDs
@@ -312,7 +338,8 @@ class FAISSStore:
**options: Additional options
Returns:
List of vector IDs
List of vector IDs (including ids that were already present
and therefore not re-added as new vectors)
"""
num_vectors = len(vectors) if isinstance(vectors, (list, np.ndarray)) else 1
tracking_id = self.progress_tracker.start_tracking(
@@ -526,6 +553,30 @@ class FAISSStore:
return results
def scan_vectors(self, offset: int = 0, limit: int = 100) -> List[Dict[str, Any]]:
"""
Page through stored vectors in insertion order.
Args:
offset: Number of vectors to skip
limit: Maximum number of vectors to return
Returns:
List of result dicts with 'id', 'metadata', and 'vector'
"""
if self.index is None or limit <= 0:
return []
ids_page = self.index.vector_ids[offset:offset + limit]
return [
{
"id": vector_id,
"metadata": self.get_metadata(vector_id) or {},
"vector": self.get_vector(vector_id),
}
for vector_id in ids_page
]
def get_stats(self) -> Dict[str, Any]:
"""Get index statistics."""
if self.index is None:
+46
View File
@@ -656,6 +656,52 @@ class PgVectorStore:
self.logger.warning(f"Failed to get metadata for {vector_id}: {e}")
return None
def scan_vectors(self, offset: int = 0, limit: int = 100) -> List[Dict[str, Any]]:
"""
Page through stored vectors ordered by id.
Args:
offset: Number of rows to skip
limit: Maximum number of rows to return
Returns:
List of result dicts with 'id', 'metadata', and 'vector'
"""
if not PSYCOPG3_AVAILABLE and not PSYCOPG2_AVAILABLE:
raise ProcessingError(
"Neither psycopg3 nor psycopg2 is available. "
"Install with: pip install psycopg[binary] or psycopg2-binary"
)
if limit <= 0:
return []
scan_sql = psycopg_sql.SQL("""
SELECT id, vector, metadata
FROM {}
ORDER BY id
LIMIT %s OFFSET %s
""").format(psycopg_sql.Identifier(self.table_name))
with self._get_connection() as conn:
try:
cur = conn.cursor()
cur.execute(scan_sql, (limit, offset))
rows = cur.fetchall()
cur.close()
results = []
for row in rows:
vec_id, vec, meta = row
results.append({
"id": vec_id,
"metadata": meta if isinstance(meta, dict) else json.loads(meta) if meta else {},
"vector": np.array(vec) if vec is not None else None,
})
return results
except Exception as e:
raise ProcessingError(f"Failed to scan vectors: {str(e)}") from e
def filter_by_metadata(
self, filters: Dict[str, Any], limit: int = 10
) -> List[Dict[str, Any]]:
@@ -616,6 +616,49 @@ class SQLiteVecStore:
self.logger.warning(f"Failed to get metadata for {vector_id}: {e}")
return None
def scan_vectors(self, offset: int = 0, limit: int = 100) -> List[Dict[str, Any]]:
"""
Page through stored vectors ordered by id.
Args:
offset: Number of rows to skip
limit: Maximum number of rows to return
Returns:
List of result dicts with 'id', 'metadata', and 'vector'
"""
if limit <= 0:
return []
query_sql = f"""
SELECT id, embedding, metadata
FROM {self.table_name}
ORDER BY id
LIMIT ? OFFSET ?
"""
with self._lock, self._get_connection() as conn:
try:
cur = conn.cursor()
cur.execute(query_sql, (limit, offset))
rows = cur.fetchall()
cur.close()
results = []
for row in rows:
vec_id, embedding_blob, meta_json = row
vec = None
if embedding_blob:
vec = np.frombuffer(embedding_blob, dtype=np.float32).copy()
results.append({
"id": vec_id,
"metadata": json.loads(meta_json) if meta_json else {},
"vector": vec,
})
return results
except Exception as e:
raise ProcessingError(f"Failed to scan vectors: {str(e)}") from e
def filter_by_metadata(
self, filters: Dict[str, Any], limit: int = 10
) -> List[Dict[str, Any]]:
+58
View File
@@ -824,6 +824,64 @@ class VectorStore:
else:
raise NotImplementedError(f"Backend store {type(self._backend_store).__name__} does not implement get_metadata")
def scan_vectors(self, offset: int = 0, limit: int = 100) -> List[Dict[str, Any]]:
"""
Page through stored vectors, backend-agnostic.
Follows the get_vector()/get_metadata() precedent (#843): the inmemory
backend pages its local dict directly, a persistent backend delegates
to a scan_vectors() on the wrapped store when available, and one that
cannot enumerate its contents raises NotImplementedError rather than
silently returning an empty page.
Args:
offset: Number of vectors to skip
limit: Maximum number of vectors to return
Returns:
List of result dicts with 'id', 'metadata', and 'vector'
"""
if limit <= 0:
return []
if self.backend == "inmemory":
ids_page = list(self.vectors.keys())[offset:offset + limit]
return [
{
"id": vec_id,
"metadata": self.metadata.get(vec_id, {}),
"vector": self.vectors.get(vec_id),
}
for vec_id in ids_page
]
elif self._backend_store and hasattr(self._backend_store, "scan_vectors"):
return self._backend_store.scan_vectors(offset=offset, limit=limit)
else:
raise NotImplementedError(
f"Backend store {type(self._backend_store).__name__} does not "
"implement scan_vectors(). Add a scan_vectors() method to the "
"backend store adapter to enable enumeration for this backend."
)
def iter_vectors(self, batch_size: int = 500):
"""
Iterate over every stored vector, one page at a time.
Args:
batch_size: Number of vectors to fetch per underlying scan_vectors() call
Yields:
Result dicts with 'id', 'metadata', and 'vector', in scan order
"""
offset = 0
while True:
page = self.scan_vectors(offset=offset, limit=batch_size)
if not page:
return
for item in page:
yield item
offset += len(page)
def count(self) -> int:
"""Return the number of vectors in the store, backend-agnostic.
+14 -14
View File
@@ -291,8 +291,8 @@ class TestDeduplication(unittest.TestCase):
class TestProgressTrackerEncoding(unittest.TestCase):
"""Regression tests for issue #531 — Unicode crash on cp1252 Windows consoles."""
def _make_cp1252_stdout(self):
"""Return a stdout-like object that raises UnicodeEncodeError for non-cp1252 chars."""
def _make_cp1252_stream(self):
"""Return a stream that raises UnicodeEncodeError for non-cp1252 chars."""
class CP1252Writer:
encoding = "cp1252"
def write(self, text):
@@ -304,39 +304,39 @@ class TestProgressTrackerEncoding(unittest.TestCase):
def test_safe_write_does_not_crash_on_cp1252(self):
"""_safe_write must not raise UnicodeEncodeError on a cp1252 console."""
display = ConsoleProgressDisplay()
orig = sys.stdout
sys.stdout = self._make_cp1252_stdout()
orig = sys.stderr
sys.stderr = self._make_cp1252_stream()
try:
display._safe_write("🧠 Semantica - 📊 Current Progress\n")
except UnicodeEncodeError:
self.fail("_safe_write raised UnicodeEncodeError on cp1252 stdout")
self.fail("_safe_write raised UnicodeEncodeError on a cp1252 stream")
finally:
sys.stdout = orig
sys.stderr = orig
def test_update_pipeline_header_does_not_crash_on_cp1252(self):
"""update() pipeline header write must not crash on a cp1252 console (issue #531)."""
from semantica.utils.progress_tracker import ProgressItem
display = ConsoleProgressDisplay()
display.use_emoji = True # force emoji path to exercise the fixed branch
orig = sys.stdout
sys.stdout = self._make_cp1252_stdout()
orig = sys.stderr
sys.stderr = self._make_cp1252_stream()
try:
display._safe_write("🧠 Semantica - 📊 Current Progress\n")
display._safe_write("=" * 150 + "\n")
except UnicodeEncodeError:
self.fail("Pipeline header write raised UnicodeEncodeError on cp1252 stdout")
finally:
sys.stdout = orig
sys.stderr = orig
def test_emoji_detection_disables_on_cp1252(self):
"""ConsoleProgressDisplay should auto-disable emoji when stdout is cp1252."""
orig = sys.stdout
sys.stdout = self._make_cp1252_stdout()
"""ConsoleProgressDisplay should auto-disable emoji when the progress stream is cp1252."""
orig = sys.stderr
sys.stderr = self._make_cp1252_stream()
try:
display = ConsoleProgressDisplay()
self.assertFalse(display.use_emoji, "use_emoji should be False on cp1252 stdout")
self.assertFalse(display.use_emoji, "use_emoji should be False on a cp1252 progress stream")
finally:
sys.stdout = orig
sys.stderr = orig
class TestResultLimiting(unittest.TestCase):
+1
View File
@@ -0,0 +1 @@
# tests/integrations/crewai package
+1
View File
@@ -0,0 +1 @@
# tests/integrations/langchain package
+95
View File
@@ -1359,6 +1359,101 @@ class TestStore:
result = runner.invoke(cli_module.main, ["store", "migrate", "--from", "faiss"])
assert result.exit_code != 0
def test_migrate_refuses_unsupported_backend_pair(self, runner):
result = runner.invoke(cli_module.main, ["store", "migrate",
"--from", "faiss", "--to", "qdrant"])
assert result.exit_code != 0
assert "faiss, pgvector, sqlite" in result.output
def _fake_migrate_store_module(self, source_items, stored, dest_configs=None):
class _FakeBackendStore:
def __init__(self, dimension=None):
self.dimension = dimension
class _FakeStore:
def __init__(self, backend, config=None, **kw):
self.backend = backend
self._config = config or {}
dim = self._config.get("dimension")
self._backend_store = _FakeBackendStore(dimension=dim)
if dest_configs is not None:
dest_configs[backend] = dict(self._config)
def iter_vectors(self, batch_size=500):
if self.backend == "sqlite":
yield from source_items
return
return
yield # pragma: no cover - makes this a generator for other backends
def store_vectors(self, vectors, metadata, ids=None):
for vec_id, meta in zip(ids, metadata):
stored[vec_id] = meta
return _fake_module(VectorStore=_FakeStore)
def test_migrate_runs_between_supported_backends(self, runner, monkeypatch):
source_items = [
{"id": "a", "vector": [0.1, 0.2], "metadata": {"tag": "x"}},
{"id": "b", "vector": [0.3, 0.4], "metadata": {}},
]
stored = {}
fake_vs = self._fake_migrate_store_module(source_items, stored)
monkeypatch.setitem(__import__("sys").modules, "semantica.vector_store", fake_vs)
result = runner.invoke(cli_module.main, ["store", "migrate",
"--from", "sqlite", "--to", "pgvector",
"--namespace", "prod", "--json"])
_ok(result)
data = _json_output(result)
assert data == {"from": "sqlite", "to": "pgvector", "migrated": 2}
assert stored == {"a": {"tag": "x", "namespace": "prod"}, "b": {"namespace": "prod"}}
def test_migrate_reports_zero_for_empty_source(self, runner, monkeypatch):
stored = {}
fake_vs = self._fake_migrate_store_module([], stored)
monkeypatch.setitem(__import__("sys").modules, "semantica.vector_store", fake_vs)
result = runner.invoke(cli_module.main, ["store", "migrate",
"--from", "sqlite", "--to", "pgvector", "--json"])
_ok(result)
assert _json_output(result)["migrated"] == 0
assert stored == {}
def test_migrate_inherits_source_dimension_into_dest(self, runner, monkeypatch):
source_items = [{"id": "a", "vector": [0.1, 0.2, 0.3], "metadata": {}}]
stored = {}
dest_configs: dict = {}
fake_vs = self._fake_migrate_store_module(source_items, stored, dest_configs)
monkeypatch.setitem(__import__("sys").modules, "semantica.vector_store", fake_vs)
monkeypatch.setattr(
cli_module.Config, "to_dict",
lambda self: {"vector_store": {"sqlite": {"dimension": 3}, "pgvector": {}}},
)
result = runner.invoke(cli_module.main, ["store", "migrate",
"--from", "sqlite", "--to", "pgvector", "--json"])
_ok(result)
assert dest_configs["pgvector"].get("dimension") == 3
def test_migrate_faiss_source_requires_index_path(self, runner, monkeypatch):
fake_vs = _fake_module(VectorStore=lambda **kw: MagicMock())
monkeypatch.setitem(__import__("sys").modules, "semantica.vector_store", fake_vs)
result = runner.invoke(cli_module.main, ["store", "migrate",
"--from", "faiss", "--to", "sqlite"])
assert result.exit_code != 0
assert "index_path" in result.output
def test_migrate_faiss_dest_requires_index_path(self, runner, monkeypatch):
fake_vs = _fake_module(VectorStore=lambda **kw: MagicMock())
monkeypatch.setitem(__import__("sys").modules, "semantica.vector_store", fake_vs)
result = runner.invoke(cli_module.main, ["store", "migrate",
"--from", "sqlite", "--to", "faiss"])
assert result.exit_code != 0
assert "index_path" in result.output
def test_flush_requires_confirm(self, runner):
result = runner.invoke(cli_module.main, ["store", "flush"])
assert result.exit_code != 0
+80
View File
@@ -0,0 +1,80 @@
"""Tests for the DeepSeek LLM provider wrapper (semantica.llms.DeepSeek)."""
from unittest.mock import MagicMock
import pytest
from semantica.llms import DeepSeek
from semantica.utils.exceptions import ProcessingError
def test_construction_stores_model_and_api_key():
llm = DeepSeek(model="deepseek-chat", api_key="fake-key")
assert llm.model == "deepseek-chat"
assert llm.api_key == "fake-key"
def test_is_available_false_with_no_key(monkeypatch):
monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False)
llm = DeepSeek(api_key=None)
assert llm.is_available() is False
def test_generate_raises_clear_error_when_unavailable(monkeypatch):
monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False)
llm = DeepSeek(api_key=None)
with pytest.raises(ProcessingError, match="DeepSeek provider not available"):
llm.generate("hello")
def test_generate_forwards_to_the_real_provider_when_available():
llm = DeepSeek(api_key="fake-key")
llm.provider = MagicMock()
llm.provider.is_available.return_value = True
llm.provider.generate.return_value = "a fake response"
result = llm.generate("hello", temperature=0.5)
assert result == "a fake response"
llm.provider.generate.assert_called_once_with("hello", temperature=0.5)
def test_generate_structured_forwards_to_the_real_provider():
llm = DeepSeek(api_key="fake-key")
llm.provider = MagicMock()
llm.provider.is_available.return_value = True
llm.provider.generate_structured.return_value = {"key": "value"}
result = llm.generate_structured("hello")
assert result == {"key": "value"}
llm.provider.generate_structured.assert_called_once_with("hello")
def test_generate_typed_forwards_schema_and_max_retries():
llm = DeepSeek(api_key="fake-key")
llm.provider = MagicMock()
llm.provider.is_available.return_value = True
fake_schema = object()
llm.provider.generate_typed.return_value = "typed result"
result = llm.generate_typed("hello", fake_schema, max_retries=5)
assert result == "typed result"
llm.provider.generate_typed.assert_called_once_with(
"hello", fake_schema, max_retries=5
)
def test_generate_structured_raises_clear_error_when_unavailable(monkeypatch):
monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False)
llm = DeepSeek(api_key=None)
with pytest.raises(ProcessingError, match="DeepSeek provider not available"):
llm.generate_structured("hello")
def test_generate_typed_raises_clear_error_when_unavailable(monkeypatch):
monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False)
llm = DeepSeek(api_key=None)
with pytest.raises(ProcessingError, match="DeepSeek provider not available"):
llm.generate_typed("hello", object())
+80
View File
@@ -0,0 +1,80 @@
"""Tests for the Gemini LLM provider wrapper (semantica.llms.Gemini)."""
from unittest.mock import MagicMock
import pytest
from semantica.llms import Gemini
from semantica.utils.exceptions import ProcessingError
def test_construction_stores_model_and_api_key():
gemini = Gemini(model="gemini-pro", api_key="fake-key")
assert gemini.model == "gemini-pro"
assert gemini.api_key == "fake-key"
def test_is_available_false_with_no_key(monkeypatch):
monkeypatch.delenv("GEMINI_API_KEY", raising=False)
gemini = Gemini(api_key=None)
assert gemini.is_available() is False
def test_generate_raises_clear_error_when_unavailable(monkeypatch):
monkeypatch.delenv("GEMINI_API_KEY", raising=False)
gemini = Gemini(api_key=None)
with pytest.raises(ProcessingError, match="Gemini provider not available"):
gemini.generate("hello")
def test_generate_forwards_to_the_real_provider_when_available():
gemini = Gemini(api_key="fake-key")
gemini.provider = MagicMock()
gemini.provider.is_available.return_value = True
gemini.provider.generate.return_value = "a fake response"
result = gemini.generate("hello", temperature=0.5)
assert result == "a fake response"
gemini.provider.generate.assert_called_once_with("hello", temperature=0.5)
def test_generate_structured_forwards_to_the_real_provider():
gemini = Gemini(api_key="fake-key")
gemini.provider = MagicMock()
gemini.provider.is_available.return_value = True
gemini.provider.generate_structured.return_value = {"key": "value"}
result = gemini.generate_structured("hello")
assert result == {"key": "value"}
gemini.provider.generate_structured.assert_called_once_with("hello")
def test_generate_typed_forwards_schema_and_max_retries():
gemini = Gemini(api_key="fake-key")
gemini.provider = MagicMock()
gemini.provider.is_available.return_value = True
fake_schema = object()
gemini.provider.generate_typed.return_value = "typed result"
result = gemini.generate_typed("hello", fake_schema, max_retries=5)
assert result == "typed result"
gemini.provider.generate_typed.assert_called_once_with(
"hello", fake_schema, max_retries=5
)
def test_generate_structured_raises_clear_error_when_unavailable(monkeypatch):
monkeypatch.delenv("GEMINI_API_KEY", raising=False)
gemini = Gemini(api_key=None)
with pytest.raises(ProcessingError, match="Gemini provider not available"):
gemini.generate_structured("hello")
def test_generate_typed_raises_clear_error_when_unavailable(monkeypatch):
monkeypatch.delenv("GEMINI_API_KEY", raising=False)
gemini = Gemini(api_key=None)
with pytest.raises(ProcessingError, match="Gemini provider not available"):
gemini.generate_typed("hello", object())
+80
View File
@@ -0,0 +1,80 @@
"""Tests for the Novita LLM provider wrapper (semantica.llms.Novita)."""
from unittest.mock import MagicMock
import pytest
from semantica.llms import Novita
from semantica.utils.exceptions import ProcessingError
def test_construction_stores_model_and_api_key():
llm = Novita(model="deepseek/deepseek-v3.2", api_key="fake-key")
assert llm.model == "deepseek/deepseek-v3.2"
assert llm.api_key == "fake-key"
def test_is_available_false_with_no_key(monkeypatch):
monkeypatch.delenv("NOVITA_API_KEY", raising=False)
llm = Novita(api_key=None)
assert llm.is_available() is False
def test_generate_raises_clear_error_when_unavailable(monkeypatch):
monkeypatch.delenv("NOVITA_API_KEY", raising=False)
llm = Novita(api_key=None)
with pytest.raises(ProcessingError, match="Novita provider not available"):
llm.generate("hello")
def test_generate_forwards_to_the_real_provider_when_available():
llm = Novita(api_key="fake-key")
llm.provider = MagicMock()
llm.provider.is_available.return_value = True
llm.provider.generate.return_value = "a fake response"
result = llm.generate("hello", temperature=0.5)
assert result == "a fake response"
llm.provider.generate.assert_called_once_with("hello", temperature=0.5)
def test_generate_structured_forwards_to_the_real_provider():
llm = Novita(api_key="fake-key")
llm.provider = MagicMock()
llm.provider.is_available.return_value = True
llm.provider.generate_structured.return_value = {"key": "value"}
result = llm.generate_structured("hello")
assert result == {"key": "value"}
llm.provider.generate_structured.assert_called_once_with("hello")
def test_generate_typed_forwards_schema_and_max_retries():
llm = Novita(api_key="fake-key")
llm.provider = MagicMock()
llm.provider.is_available.return_value = True
fake_schema = object()
llm.provider.generate_typed.return_value = "typed result"
result = llm.generate_typed("hello", fake_schema, max_retries=5)
assert result == "typed result"
llm.provider.generate_typed.assert_called_once_with(
"hello", fake_schema, max_retries=5
)
def test_generate_structured_raises_clear_error_when_unavailable(monkeypatch):
monkeypatch.delenv("NOVITA_API_KEY", raising=False)
llm = Novita(api_key=None)
with pytest.raises(ProcessingError, match="Novita provider not available"):
llm.generate_structured("hello")
def test_generate_typed_raises_clear_error_when_unavailable(monkeypatch):
monkeypatch.delenv("NOVITA_API_KEY", raising=False)
llm = Novita(api_key=None)
with pytest.raises(ProcessingError, match="Novita provider not available"):
llm.generate_typed("hello", object())
+78
View File
@@ -0,0 +1,78 @@
"""Tests for the Ollama LLM provider wrapper (semantica.llms.Ollama)."""
from unittest.mock import MagicMock
import pytest
from semantica.llms import Ollama
from semantica.utils.exceptions import ProcessingError
def test_construction_stores_model_and_base_url():
llm = Ollama(model="llama2", base_url="http://localhost:11434")
assert llm.model == "llama2"
assert llm.base_url == "http://localhost:11434"
def test_is_available_false_without_a_running_server():
"""No api_key here, Ollama has none. Without a real server (or the ollama
package) reachable at base_url, this must be a real False."""
llm = Ollama(base_url="http://localhost:1")
assert llm.is_available() is False
def test_generate_raises_clear_error_when_unavailable():
llm = Ollama(base_url="http://localhost:1")
with pytest.raises(ProcessingError, match="Ollama provider not available"):
llm.generate("hello")
def test_generate_forwards_to_the_real_provider_when_available():
llm = Ollama()
llm.provider = MagicMock()
llm.provider.is_available.return_value = True
llm.provider.generate.return_value = "a fake response"
result = llm.generate("hello", temperature=0.5)
assert result == "a fake response"
llm.provider.generate.assert_called_once_with("hello", temperature=0.5)
def test_generate_structured_forwards_to_the_real_provider():
llm = Ollama()
llm.provider = MagicMock()
llm.provider.is_available.return_value = True
llm.provider.generate_structured.return_value = {"key": "value"}
result = llm.generate_structured("hello")
assert result == {"key": "value"}
llm.provider.generate_structured.assert_called_once_with("hello")
def test_generate_typed_forwards_schema_and_max_retries():
llm = Ollama()
llm.provider = MagicMock()
llm.provider.is_available.return_value = True
fake_schema = object()
llm.provider.generate_typed.return_value = "typed result"
result = llm.generate_typed("hello", fake_schema, max_retries=5)
assert result == "typed result"
llm.provider.generate_typed.assert_called_once_with(
"hello", fake_schema, max_retries=5
)
def test_generate_structured_raises_clear_error_when_unavailable():
llm = Ollama(base_url="http://localhost:1")
with pytest.raises(ProcessingError, match="Ollama provider not available"):
llm.generate_structured("hello")
def test_generate_typed_raises_clear_error_when_unavailable():
llm = Ollama(base_url="http://localhost:1")
with pytest.raises(ProcessingError, match="Ollama provider not available"):
llm.generate_typed("hello", object())
+184
View File
@@ -0,0 +1,184 @@
"""Console progress must be written to stderr, never to stdout.
Progress is diagnostic output. Writing it to stdout corrupts any program that
carries a machine-readable protocol there the stdio MCP servers put
newline-delimited JSON-RPC on stdout, and a progress bar interleaved with a
response body makes that response unparseable (#1134).
Both servers currently defend against this by setting
SEMANTICA_DISABLE_PROGRESS, and a console display is only attached when the
console is interactive. Those are containments, not the fix: they give up
progress output entirely, and every future entry point has to remember them.
Writing to the correct stream in the first place is what these tests pin.
"""
import importlib
import io
import sys
import pytest
@pytest.fixture
def progress_module():
"""Import the real progress_tracker, bypassing a mocked sys.modules entry.
tests/test_extractors_dispatch.py assigns a MagicMock over
'semantica.utils.progress_tracker' at import time and never restores it, so
a plain module-level import here returns mocks when that file has already
run. Dropping the cached entry re-imports the real module.
Both bindings are restored afterwards: importing a submodule also rebinds it
as an attribute of its parent package, so restoring only the sys.modules
entry would leave `semantica.utils.progress_tracker` and
`sys.modules["semantica.utils.progress_tracker"]` pointing at different
objects for every test that follows.
"""
name = "semantica.utils.progress_tracker"
attr = name.rsplit(".", 1)[1]
parent = importlib.import_module("semantica.utils")
missing = object()
saved_entry = sys.modules.get(name, missing)
saved_attr = getattr(parent, attr, missing)
sys.modules.pop(name, None)
try:
module = importlib.import_module(name)
assert hasattr(module, "__file__"), "expected the real module, got a stand-in"
yield module
finally:
if saved_entry is missing:
sys.modules.pop(name, None)
else:
sys.modules[name] = saved_entry
if saved_attr is missing:
if hasattr(parent, attr):
delattr(parent, attr)
else:
setattr(parent, attr, saved_attr)
@pytest.fixture
def display_cls(progress_module):
return progress_module.ConsoleProgressDisplay
@pytest.fixture
def make_item(progress_module):
def _make(**overrides):
defaults = dict(
module="kg",
submodule="Reasoner",
message="Inferring facts",
status="running",
total_items=10,
processed_items=3,
)
defaults.update(overrides)
return progress_module.ProgressItem(**defaults)
return _make
class TestProgressStreamDefaults:
def test_defaults_to_stderr(self, display_cls):
assert display_cls().stream is sys.stderr
def test_default_is_not_stdout(self, display_cls):
"""The whole point: stdout stays clean for protocol traffic."""
assert display_cls().stream is not sys.stdout
def test_stream_follows_rebinding(self, display_cls, monkeypatch):
"""Resolved per write, so pytest capture and later rebinds are honoured."""
display = display_cls()
replacement = io.StringIO()
monkeypatch.setattr(sys, "stderr", replacement)
assert display.stream is replacement
def test_explicit_stream_overrides_the_default(self, display_cls):
buffer = io.StringIO()
assert display_cls(stream=buffer).stream is buffer
class TestProgressWritesGoToTheStream:
def test_update_writes_to_the_configured_stream(self, display_cls, make_item):
buffer = io.StringIO()
display = display_cls(stream=buffer, use_emoji=False, update_interval=0.0)
display.update(make_item())
assert buffer.getvalue(), "progress should have been rendered"
def test_update_writes_nothing_to_stdout(self, display_cls, make_item, monkeypatch):
"""Regression guard for #1134: stdout must stay untouched."""
fake_stdout = io.StringIO()
monkeypatch.setattr(sys, "stdout", fake_stdout)
buffer = io.StringIO()
display = display_cls(stream=buffer, use_emoji=False, update_interval=0.0)
display.update(make_item())
display.clear()
assert fake_stdout.getvalue() == "", (
f"console progress leaked to stdout: {fake_stdout.getvalue()!r}"
)
def test_default_display_writes_nothing_to_stdout(
self, display_cls, make_item, monkeypatch
):
"""Same guard without an explicit stream, i.e. the real default path."""
fake_stdout = io.StringIO()
fake_stderr = io.StringIO()
monkeypatch.setattr(sys, "stdout", fake_stdout)
monkeypatch.setattr(sys, "stderr", fake_stderr)
display = display_cls(use_emoji=False, update_interval=0.0)
display.update(make_item())
assert fake_stdout.getvalue() == ""
assert fake_stderr.getvalue(), "progress should have gone to stderr"
def test_clear_flushes_the_stream_not_stdout(self, display_cls, monkeypatch):
flushed = []
class RecordingStream(io.StringIO):
def flush(self):
flushed.append("stream")
class ExplodingStdout(io.StringIO):
def flush(self): # pragma: no cover - fails the test if reached
raise AssertionError("progress must not flush stdout")
monkeypatch.setattr(sys, "stdout", ExplodingStdout())
display = display_cls(
stream=RecordingStream(), use_emoji=False, update_interval=0.0
)
display.clear()
assert flushed == ["stream"]
class TestEncodingFallback:
def test_falls_back_when_the_stream_cannot_encode(self, display_cls):
"""A cp1252-style console must not raise on emoji; it degrades instead."""
class AsciiOnly(io.StringIO):
encoding = "ascii"
def write(self, text):
text.encode("ascii") # raises UnicodeEncodeError on emoji
return super().write(text)
buffer = AsciiOnly()
display = display_cls(stream=buffer, use_emoji=True, update_interval=0.0)
display._safe_write("progress \U0001f504 bar\n")
assert "progress" in buffer.getvalue()
if __name__ == "__main__":
pytest.main([__file__])
+88 -1
View File
@@ -3,7 +3,7 @@ from unittest.mock import MagicMock
import numpy as np
import pytest
from semantica.vector_store.faiss_store import FAISSIndex
from semantica.vector_store.faiss_store import FAISSIndex, FAISSStore
def test_get_vector_reconstructs_from_flat_l2_index():
@@ -120,3 +120,90 @@ def test_get_vector_reconstructs_from_real_ivfflat_index_without_prior_direct_ma
result = index.get_vector("vec_target")
np.testing.assert_allclose(result, vectors[3], atol=1e-6)
def _store_with_fake_index(ids, metadata_by_id=None):
backend_index = MagicMock()
backend_index.reconstruct.side_effect = lambda idx: [float(idx)] * 3
index = FAISSIndex(backend_index, dimension=3)
index.vector_ids = list(ids)
index.metadata = dict(metadata_by_id or {})
store = FAISSStore(dimension=3)
store.index = index
return store
def test_scan_vectors_returns_all_across_pages():
store = _store_with_fake_index(["a", "b", "c", "d", "e"])
seen_ids = []
offset = 0
while True:
page = store.scan_vectors(offset=offset, limit=2)
if not page:
break
seen_ids.extend(p["id"] for p in page)
offset += len(page)
assert seen_ids == ["a", "b", "c", "d", "e"]
def test_scan_vectors_includes_vector_and_metadata():
store = _store_with_fake_index(["a"], {"a": {"tag": "only"}})
page = store.scan_vectors(offset=0, limit=10)
assert len(page) == 1
assert page[0]["id"] == "a"
assert page[0]["metadata"] == {"tag": "only"}
np.testing.assert_array_equal(page[0]["vector"], np.array([0.0, 0.0, 0.0], dtype=np.float32))
def test_scan_vectors_no_index_returns_empty_list():
store = FAISSStore(dimension=3)
assert store.scan_vectors(offset=0, limit=10) == []
def test_scan_vectors_zero_limit_returns_empty_list():
store = _store_with_fake_index(["a"])
assert store.scan_vectors(offset=0, limit=0) == []
def test_scan_vectors_offset_past_end_returns_empty_list():
store = _store_with_fake_index(["a"])
assert store.scan_vectors(offset=100, limit=10) == []
def test_add_vectors_retry_with_same_ids_does_not_duplicate():
"""Re-running add_vectors with ids already in the index (e.g. retrying
an interrupted migration) must not create a second physical vector
under the same id."""
backend_index = MagicMock()
store = FAISSStore(dimension=3)
store.index = FAISSIndex(backend_index, dimension=3)
vectors = np.array([[1, 2, 3], [4, 5, 6], [7, 8, 9], [10, 11, 12]], dtype=np.float32)
ids = ["a", "b", "c", "d"]
store.add_vectors(vectors, ids=ids, metadata=[{"i": i} for i in range(4)])
assert store.count() == 4
store.add_vectors(vectors, ids=ids, metadata=[{"i": i} for i in range(4)])
assert store.count() == 4
assert store.index.vector_ids == ids
def test_add_vectors_retry_with_partial_overlap_only_adds_new_ids():
backend_index = MagicMock()
store = FAISSStore(dimension=3)
store.index = FAISSIndex(backend_index, dimension=3)
store.add_vectors(np.array([[1, 2, 3], [4, 5, 6]], dtype=np.float32), ids=["a", "b"])
store.add_vectors(np.array([[1, 2, 3], [7, 8, 9]], dtype=np.float32), ids=["a", "c"])
assert store.index.vector_ids == ["a", "b", "c"]
second_call_vectors = backend_index.add.call_args[0][0]
assert second_call_vectors.shape[0] == 1
np.testing.assert_array_equal(second_call_vectors[0], np.array([7, 8, 9], dtype=np.float32))
+42
View File
@@ -467,6 +467,48 @@ class TestPgVectorStoreDelete:
assert success is True
class TestPgVectorStoreScan:
"""Test scan_vectors pagination."""
def test_scan_returns_all_vectors_across_pages(self, store):
vectors = [np.random.rand(128).astype(np.float32) for _ in range(5)]
ids = store.add(vectors, [{"index": i} for i in range(5)])
seen_ids = []
offset = 0
while True:
page = store.scan_vectors(offset=offset, limit=2)
if not page:
break
seen_ids.extend(p["id"] for p in page)
offset += len(page)
assert set(seen_ids) == set(ids)
assert len(seen_ids) == 5
def test_scan_page_includes_vector_and_metadata(self, store):
vectors = [np.random.rand(128).astype(np.float32)]
ids = store.add(vectors, [{"tag": "only"}])
page = store.scan_vectors(offset=0, limit=10)
assert len(page) == 1
assert page[0]["id"] == ids[0]
assert page[0]["metadata"] == {"tag": "only"}
assert page[0]["vector"] is not None
def test_scan_empty_store_returns_empty_list(self, store):
assert store.scan_vectors(offset=0, limit=10) == []
def test_scan_zero_limit_returns_empty_list(self, store):
store.add([np.random.rand(128).astype(np.float32)])
assert store.scan_vectors(offset=0, limit=0) == []
def test_scan_offset_past_end_returns_empty_list(self, store):
store.add([np.random.rand(128).astype(np.float32)])
assert store.scan_vectors(offset=100, limit=10) == []
class TestPgVectorStoreIndex:
"""Test index creation operations."""
@@ -415,6 +415,48 @@ class TestSQLiteVecStoreStats:
assert stats["vector_count"] == 4
class TestSQLiteVecStoreScan:
"""Test scan_vectors pagination."""
def test_scan_returns_all_vectors_across_pages(self, store):
vectors = [np.random.rand(128).astype(np.float32) for _ in range(5)]
ids = store.add(vectors, [{"index": i} for i in range(5)])
seen_ids = []
offset = 0
while True:
page = store.scan_vectors(offset=offset, limit=2)
if not page:
break
seen_ids.extend(p["id"] for p in page)
offset += len(page)
assert set(seen_ids) == set(ids)
assert len(seen_ids) == 5
def test_scan_page_includes_vector_and_metadata(self, store):
vectors = [np.random.rand(128).astype(np.float32)]
ids = store.add(vectors, [{"tag": "only"}])
page = store.scan_vectors(offset=0, limit=10)
assert len(page) == 1
assert page[0]["id"] == ids[0]
assert page[0]["metadata"] == {"tag": "only"}
assert page[0]["vector"] is not None
def test_scan_empty_store_returns_empty_list(self, store):
assert store.scan_vectors(offset=0, limit=10) == []
def test_scan_zero_limit_returns_empty_list(self, store):
store.add([np.random.rand(128).astype(np.float32)])
assert store.scan_vectors(offset=0, limit=0) == []
def test_scan_offset_past_end_returns_empty_list(self, store):
store.add([np.random.rand(128).astype(np.float32)])
assert store.scan_vectors(offset=100, limit=10) == []
class TestSQLiteVecStoreFilterByMetadata:
"""Test filter_by_metadata, including list-valued metadata handling."""
@@ -120,6 +120,78 @@ class VectorStoreCountTests(unittest.TestCase):
self.assertIn("count()", msg)
# ---------------------------------------------------------------------------
# VectorStore.scan_vectors() / iter_vectors() dispatch tests
# ---------------------------------------------------------------------------
class _ScanningBackendStore:
"""Fake persistent backend store that supports scan_vectors()."""
def __init__(self, items):
self._items = items
def scan_vectors(self, offset=0, limit=100):
return self._items[offset:offset + limit]
class _NonScanningBackendStore:
"""Fake persistent backend store without any scan capability."""
class VectorStoreScanVectorsTests(unittest.TestCase):
"""VectorStore.scan_vectors() / iter_vectors() backend-agnostic accessors."""
def setUp(self):
self.vectors = [np.array([1.0, 0.0]), np.array([0.0, 1.0]), np.array([1.0, 1.0])]
self.metadata = [{"type": "a"}, {"type": "b"}, {"type": "c"}]
def test_scan_inmemory_pages_through_all_vectors(self):
store = VectorStore(backend="inmemory", dimension=2)
ids = store.store_vectors(self.vectors, self.metadata)
page1 = store.scan_vectors(offset=0, limit=2)
page2 = store.scan_vectors(offset=2, limit=2)
self.assertEqual([p["id"] for p in page1], ids[:2])
self.assertEqual([p["id"] for p in page2], ids[2:])
self.assertEqual(page2[0]["metadata"], {"type": "c"})
def test_scan_inmemory_empty_store(self):
store = VectorStore(backend="inmemory", dimension=2)
self.assertEqual(store.scan_vectors(offset=0, limit=10), [])
def test_scan_zero_limit_returns_empty_list(self):
store = VectorStore(backend="inmemory", dimension=2)
store.store_vectors(self.vectors, self.metadata)
self.assertEqual(store.scan_vectors(offset=0, limit=0), [])
def test_scan_delegates_to_backend_store(self):
items = [{"id": "a", "metadata": {}, "vector": None}]
store = VectorStore(backend="inmemory", dimension=2)
store.backend = "faiss"
store._backend_store = _ScanningBackendStore(items)
self.assertEqual(store.scan_vectors(offset=0, limit=10), items)
def test_scan_raises_not_implemented_without_backend_support(self):
store = VectorStore(backend="inmemory", dimension=2)
store.backend = "faiss"
store._backend_store = _NonScanningBackendStore()
with self.assertRaises(NotImplementedError):
store.scan_vectors(offset=0, limit=10)
def test_iter_vectors_walks_every_page(self):
store = VectorStore(backend="inmemory", dimension=2)
ids = store.store_vectors(self.vectors, self.metadata)
collected = list(store.iter_vectors(batch_size=2))
self.assertEqual([item["id"] for item in collected], ids)
def test_iter_vectors_empty_store_yields_nothing(self):
store = VectorStore(backend="inmemory", dimension=2)
self.assertEqual(list(store.iter_vectors(batch_size=2)), [])
# ---------------------------------------------------------------------------
# VectorManager tests — inmemory backend
# ---------------------------------------------------------------------------