diff --git a/AGENTS.md b/AGENTS.md index 5a2f200..f05b5c3 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -18,6 +18,7 @@ handoff all come from `openai-agents`. | Module | Role | |--------|------| | `quantmind/knowledge/` | Pydantic data standard (`FlattenKnowledge` / `TreeKnowledge` / `GraphKnowledge`) — dependency leaf | +| `quantmind/library/` | Local persistence and semantic retrieval for canonical knowledge — depends only on `knowledge` | | `quantmind/configs/` | Operation cfg + typed input models or unions (`BaseFlowCfg`, `NewsWindow`, `PaperInput`) — depends only on `knowledge` | | `quantmind/preprocess/` | Deterministic fetch / format / clean / time utilities — depends only on `utils` | | `quantmind/flows/` | Apex layer: public library operations (`paper_flow`, `collect_news`, `batch_run`) | diff --git a/CLAUDE.md b/CLAUDE.md index a599942..37109ba 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -19,6 +19,7 @@ handoff all come from `openai-agents`. | Module | Role | |--------|------| | `quantmind/knowledge/` | Pydantic data standard (`FlattenKnowledge` / `TreeKnowledge` / `GraphKnowledge`) — dependency leaf | +| `quantmind/library/` | Local persistence and semantic retrieval for canonical knowledge — depends only on `knowledge` | | `quantmind/configs/` | Operation cfg + typed input models or unions (`BaseFlowCfg`, `NewsWindow`, `PaperInput`) — depends only on `knowledge` | | `quantmind/preprocess/` | Deterministic fetch / format / clean / time utilities — depends only on `utils` | | `quantmind/flows/` | Apex layer: public library operations (`paper_flow`, `collect_news`, `batch_run`) | diff --git a/contexts/usage/README.md b/contexts/usage/README.md index eaea201..93289a3 100644 --- a/contexts/usage/README.md +++ b/contexts/usage/README.md @@ -9,6 +9,7 @@ current public API, focused examples, and component-specific guidance. | Installation and common usage | [Root README usage](../../README.md#-usage-examples) | | Paper extraction | [Paper guide](../../docs/papers.md) | | News collection | [News design and behavior](../../docs/design/en/news.md) | +| Local semantic search | [Library guide](../../docs/library.md) and [focused example](../../examples/library/README.md) | | Runnable operation examples | [`examples/flows/`](../../examples/flows/) | | Focused preprocessing examples | [`examples/preprocess/`](../../examples/preprocess/) | diff --git a/docs/README.md b/docs/README.md index 9b61f89..1a05f3e 100644 --- a/docs/README.md +++ b/docs/README.md @@ -17,6 +17,7 @@ harness. | Paper extraction | `quantmind.flows.paper_flow` | `PaperInput`, `PaperFlowCfg` | `Paper` | [README usage](../README.md#-usage-examples) | [Papers](papers.md) | | News collection | `quantmind.flows.collect_news` | `NewsWindow`, `NewsCollectionCfg` | `NewsBatch` from `quantmind.preprocess` | [Collect news](../examples/flows/collect_news.py) | [News collection design](design/en/news.md) | | Bounded fan-out | `quantmind.flows.batch_run` | Operation inputs and shared config | `BatchResult` | [README usage](../README.md#-usage-examples) | API docstrings | +| Local semantic search | `quantmind.library.LocalKnowledgeLibrary` | `BaseKnowledge`, `SemanticQuery` | `list[SemanticHit]` | [Library example](../examples/library/README.md) | [Library guide](library.md) | Import public inputs and configs from `quantmind.configs` and current public operations from `quantmind.flows`. Import result contracts from the canonical diff --git a/docs/library.md b/docs/library.md new file mode 100644 index 0000000..703a017 --- /dev/null +++ b/docs/library.md @@ -0,0 +1,128 @@ +# Local Semantic Knowledge Library + +`quantmind.library` persists canonical `BaseKnowledge` in SQLite and ranks +rebuildable NumPy indexes with exact cosine similarity. It is a financial +knowledge API, not a generic RAG framework: provider and storage details remain +private, and search returns typed QuantMind evidence without generating an +answer. + +## Retrieval grain + +- Each `FlattenKnowledge` item produces one target from its exact + `embedding_text()` projection. +- Each `TreeKnowledge` produces one item target for the root with + `node_id=None`, plus one target for every non-root `TreeNode`. +- Canonical typed records are the source of truth. Embeddings and filter + columns are derived data and can be replaced by re-putting the item. + +## Local storage model + +The default local database is SQLite, with separate canonical and derived +concerns: + +- `knowledge_items` stores one aggregate root per `BaseKnowledge`. The type + discriminator and schema version select the concrete Pydantic model. +- `knowledge_nodes` stores every canonical `TreeNode` separately with its item, + parent, position, payload, and content hash. A large tree is not hidden in the + aggregate-root JSON row. +- `semantic_records` stores rebuildable item/root/node projections and vectors. + +Concrete types do not get tables such as `news`, `earnings`, or `papers`. +Creating one table per Pydantic subtype would duplicate common metadata and +require a database migration whenever the knowledge standard adds a type. The +aggregate-root plus normalized-node model preserves typed validation while +giving future tree navigation a stable node-level storage boundary. + +SQLite is the default because canonical knowledge needs transactions, foreign +keys, typed reconstruction, and explicit corrupt/stale/not-found behavior. A +vector database such as Chroma optimizes the derived similarity-search layer; +it does not replace those canonical-storage responsibilities. The current local +scale therefore uses a rebuildable NumPy exact-cosine index backed by +`semantic_records`, without adding another service or dependency. If corpus +size later requires an approximate or remote vector index, that derived layer +can change privately without changing `LocalKnowledgeLibrary`, the canonical +tables, or user code. + +The durable vector metadata records the embedding model and dimension, exact +projection hash, source content hash, knowledge schema version, and projection +schema version. Re-putting an unchanged item ID reuses its vectors. A changed +node projection replaces only that node vector; model, dimension, source hash, +or schema changes replace the affected item's vectors. + +## Financial-time filters + +`as_of` is the information cutoff represented by the knowledge. `available_at` +is when its source became observable. They intentionally remain separate: + +- `as_of_before` includes records whose information cutoff is at or before the + query cutoff. +- `available_at_before` includes only records with a known availability time at + or before the query cutoff. Unknown availability is excluded. + +Consequently, `as_of_before` alone does not prevent look-ahead. Ingestion flows +should populate `available_at` from publication time. When publication time is +unknown, the caller can copy `SourceRef.fetched_at` to `available_at` as a +conservative upper bound. + +`item_types` and `source_kinds` use any-of matching. Every requested tag must be +present. `confidence`, `tree_id`, and both time cutoffs combine with those +filters before ranking. + +## Usage + +The bundled AI-infrastructure scenario contains primary-source-backed `News`, +`Earnings`, and a real research `Paper` tree. Its canonical source JSON is +precompiled into a SQLite database with six `text-embedding-3-small` targets, +so the user-facing example does not perform ingestion or document embedding. +Put `OPENAI_API_KEY` in `.env` and run: + +```bash +python examples/library/semantic_search.py +``` + +The example: + +1. Opens the ready-to-search SQLite bundle through `LocalKnowledgeLibrary`. +2. Embeds only the user's query with `text-embedding-3-small`. +3. Searches a concrete AI-infrastructure question with a no-look-ahead + availability cutoff. +4. Resolves every hit with `get()` and prints source, citation, financial time, + and the root-to-node path and content for tree evidence. + +Maintainers can regenerate the model-specific database from the auditable JSON +after changing the source data or storage schema: + +```bash +python scripts/examples/build_ai_infrastructure_bundle.py +``` + +The bundle's facts and short citations come directly from the +[Compute Trends Across Three Eras of Machine Learning paper](https://arxiv.org/abs/2202.05924), +[Microsoft's FY2025 AI-datacenter investment announcement](https://blogs.microsoft.com/on-the-issues/2025/01/03/the-golden-opportunity-for-american-ai/), +and [NVIDIA's FY2026 Q1 results](https://investor.nvidia.com/news/press-release-details/2025/NVIDIA-Announces-Financial-Results-for-First-Quarter-Fiscal-2026/default.aspx). + +Representative output has this shape; scores depend on the selected embedding +model: + +```text +Bundle: .../examples/library/data/ai_infrastructure.db +Query: What evidence shows demand for AI infrastructure is expanding? + +1. score=0.812 type=paper + path: ... > ... + matched: ... + source: https://... + citation: ... +``` + +`SemanticHit.matched_text` is the exact projection used in ranking. Use +`get(item_id)` to resolve the full canonical item and, for node hits, the node +identified by `node_id`. + +Missing IDs raise `KeyError`. Stored canonical payloads that no longer validate +and orphaned or mismatched derived records raise `RuntimeError` with stale-data +context. Invalid vector bytes and inconsistent stored dimensions raise a clear +corrupt-index `RuntimeError`; provider or query dimension mismatches raise +`ValueError`. `delete()` does not deserialize canonical JSON, so a stale +canonical payload can still be removed together with all derived targets in one +transaction. diff --git a/examples/library/README.md b/examples/library/README.md new file mode 100644 index 0000000..99c9a7c --- /dev/null +++ b/examples/library/README.md @@ -0,0 +1,39 @@ +# Local Knowledge Library Example + +This example searches an auditable AI-infrastructure knowledge bundle containing +`News`, `Earnings`, and a research `Paper` tree. + +## Run + +Put `OPENAI_API_KEY` in the repository `.env`, then run from the repository root: + +```bash +python examples/library/semantic_search.py +``` + +The bundled SQLite database already contains six 1536-dimensional +`text-embedding-3-small` document targets. Running the example embeds only the +query, then uses `LocalKnowledgeLibrary.open()`, `search()`, and `get()` to show +ranked evidence and tree paths. + +## Files + +- `semantic_search.py` is the user-facing query example. +- `data/ai_infrastructure.json` is the canonical, reviewable source bundle. +- `data/ai_infrastructure.db` is the generated SQLite library with canonical + records and model-specific embeddings. Do not edit it directly. + +## Rebuild the bundle + +Regenerate the database after changing the source JSON, library schema, +retrieval-target rules, embedding model, or dimension: + +```bash +python scripts/examples/build_ai_infrastructure_bundle.py +``` + +The build validates every source item as canonical QuantMind knowledge and +removes an incomplete database if embedding generation fails. + +See [the library guide](../../docs/library.md) for storage, retrieval-grain, and +financial-time semantics. diff --git a/examples/library/data/ai_infrastructure.db b/examples/library/data/ai_infrastructure.db new file mode 100644 index 0000000..10f4eed Binary files /dev/null and b/examples/library/data/ai_infrastructure.db differ diff --git a/examples/library/data/ai_infrastructure.json b/examples/library/data/ai_infrastructure.json new file mode 100644 index 0000000..f68cad3 --- /dev/null +++ b/examples/library/data/ai_infrastructure.json @@ -0,0 +1,222 @@ +{ + "scenario": "An investor tests whether the long-run growth of machine-learning compute is translating into hyperscaler AI-datacenter investment and accelerator-vendor revenue, while respecting when each source became available.", + "items": [ + { + "id": "b25838d6-c3cc-4bf5-98c6-6f2aa686daf1", + "item_type": "paper", + "schema_version": "1.0", + "as_of": "2022-03-09T00:00:00+00:00", + "available_at": "2022-03-09T00:00:00+00:00", + "created_at": "2026-07-16T15:59:16+08:00", + "source": { + "kind": "arxiv", + "uri": "https://arxiv.org/abs/2202.05924", + "fetched_at": "2026-07-16T15:59:16+08:00" + }, + "confidence": "high", + "citations": [ + { + "source_id": "arxiv:2202.05924v2", + "page": 1, + "quote": "fast-growing compute requirements", + "tree_id": "b25838d6-c3cc-4bf5-98c6-6f2aa686daf1", + "node_id": "4d1ab280-b5a1-46ae-8139-d805038cafe3" + } + ], + "tags": [ + "ai-infrastructure", + "training-compute", + "scaling" + ], + "disclaimers": [ + "The paper measures historical training-compute trends; it does not forecast public-company revenue or investment returns." + ], + "root_node_id": "4d1ab280-b5a1-46ae-8139-d805038cafe3", + "nodes": { + "4d1ab280-b5a1-46ae-8139-d805038cafe3": { + "node_id": "4d1ab280-b5a1-46ae-8139-d805038cafe3", + "parent_id": null, + "position": 0, + "title": "Compute Trends Across Three Eras of Machine Learning", + "summary": "Sevilla and co-authors study the training compute of milestone machine-learning systems and identify pre-deep-learning, deep-learning, and large-scale eras, documenting a sharp acceleration in compute demand after 2010.", + "content": null, + "citations": [ + { + "source_id": "arxiv:2202.05924v2", + "page": 1, + "quote": "three distinct eras", + "tree_id": "b25838d6-c3cc-4bf5-98c6-6f2aa686daf1", + "node_id": "4d1ab280-b5a1-46ae-8139-d805038cafe3" + } + ], + "children_ids": [ + "3e9bc0d2-d270-4f52-8c15-d6b514d6d1ef", + "7a221d20-2a81-4790-9635-4c21a139f727", + "f99e3258-12ab-455c-b717-d89d2ba7cb71" + ] + }, + "3e9bc0d2-d270-4f52-8c15-d6b514d6d1ef": { + "node_id": "3e9bc0d2-d270-4f52-8c15-d6b514d6d1ef", + "parent_id": "4d1ab280-b5a1-46ae-8139-d805038cafe3", + "position": 0, + "title": "Dataset and method", + "summary": "The authors curate training-compute estimates for 123 milestone systems, separate models into historical eras, and fit log-linear trends with bootstrap confidence intervals.", + "content": "The dataset records estimated compute for final training runs. The study compares trends across time and model scale; its reported confidence intervals are based on 1,000 bootstrap samples.", + "citations": [ + { + "source_id": "arxiv:2202.05924v2", + "page": 1, + "quote": "123 milestone Machine Learning systems", + "tree_id": "b25838d6-c3cc-4bf5-98c6-6f2aa686daf1", + "node_id": "3e9bc0d2-d270-4f52-8c15-d6b514d6d1ef" + }, + { + "source_id": "arxiv:2202.05924v2", + "page": 16, + "quote": "bootstrap with B = 1000 samples", + "tree_id": "b25838d6-c3cc-4bf5-98c6-6f2aa686daf1", + "node_id": "3e9bc0d2-d270-4f52-8c15-d6b514d6d1ef" + } + ], + "children_ids": [] + }, + "7a221d20-2a81-4790-9635-4c21a139f727": { + "node_id": "7a221d20-2a81-4790-9635-4c21a139f727", + "parent_id": "4d1ab280-b5a1-46ae-8139-d805038cafe3", + "position": 1, + "title": "Measured compute trends", + "summary": "The paper estimates roughly 20-month doubling before 2010, about six-month doubling for regular-scale deep-learning systems after 2010, and a distinct large-scale trend emerging around 2015.", + "content": "Large-scale systems begin roughly two orders of magnitude above the contemporary regular-scale trend. Their estimated training compute doubles in about ten months, although the authors caution that the category and slope are uncertain.", + "citations": [ + { + "source_id": "arxiv:2202.05924v2", + "page": 3, + "quote": "5.7 months", + "tree_id": "b25838d6-c3cc-4bf5-98c6-6f2aa686daf1", + "node_id": "7a221d20-2a81-4790-9635-4c21a139f727" + }, + { + "source_id": "arxiv:2202.05924v2", + "page": 5, + "quote": "doubling every 9 to 10 months", + "tree_id": "b25838d6-c3cc-4bf5-98c6-6f2aa686daf1", + "node_id": "7a221d20-2a81-4790-9635-4c21a139f727" + } + ], + "children_ids": [] + }, + "f99e3258-12ab-455c-b717-d89d2ba7cb71": { + "node_id": "f99e3258-12ab-455c-b717-d89d2ba7cb71", + "parent_id": "4d1ab280-b5a1-46ae-8139-d805038cafe3", + "position": 2, + "title": "Interpretation limits", + "summary": "The historical association between frontier systems and higher compute budgets supports infrastructure-demand research, but it is not a direct forecast and the paper documents material estimation and selection uncertainty.", + "content": "Only the final training run is counted, while hyperparameter searches are often undisclosed. Corporate labs may also withhold large models, and the boundary between regular-scale and large-scale systems is partly judgmental.", + "citations": [ + { + "source_id": "arxiv:2202.05924v2", + "page": 24, + "quote": "sources of uncertainty and potential weaknesses", + "tree_id": "b25838d6-c3cc-4bf5-98c6-6f2aa686daf1", + "node_id": "f99e3258-12ab-455c-b717-d89d2ba7cb71" + } + ], + "children_ids": [] + } + }, + "arxiv_id": "2202.05924", + "authors": [ + "Jaime Sevilla", + "Lennart Heim", + "Anson Ho", + "Tamay Besiroglu", + "Marius Hobbhahn", + "Pablo Villalobos" + ], + "asset_classes": [ + "equities" + ] + }, + { + "id": "07c75b73-4d11-49f3-a40a-f5b4327e2ec1", + "item_type": "news", + "schema_version": "1.0", + "as_of": "2025-01-03T00:00:00+00:00", + "available_at": "2025-01-03T00:00:00+00:00", + "created_at": "2026-07-16T15:59:16+08:00", + "source": { + "kind": "http", + "uri": "https://blogs.microsoft.com/on-the-issues/2025/01/03/the-golden-opportunity-for-american-ai/", + "fetched_at": "2026-07-16T15:59:16+08:00" + }, + "confidence": "high", + "citations": [ + { + "source_id": "microsoft:on-the-issues:2025-01-03", + "quote": "approximately $80 billion" + } + ], + "tags": [ + "MSFT", + "ai-infrastructure", + "datacenters", + "capex" + ], + "disclaimers": [ + "The source states an investment plan, not completed spending.", + "The source provides a publication date without a time; timestamp and available_at are normalized to midnight UTC." + ], + "headline": "Microsoft plans approximately $80 billion of FY2025 AI-datacenter investment, with more than half in the United States", + "event_type": "capital_investment_plan", + "timestamp": "2025-01-03T00:00:00+00:00", + "entities": [ + "Microsoft", + "AI datacenters", + "United States" + ], + "sentiment": "positive", + "materiality": "high" + }, + { + "id": "9204ab37-62dc-43d6-9e06-290abe2d1ff9", + "item_type": "earnings", + "schema_version": "1.0", + "as_of": "2025-04-27T23:59:59-07:00", + "available_at": "2025-05-28T00:00:00-07:00", + "created_at": "2026-07-16T15:59:16+08:00", + "source": { + "kind": "http", + "uri": "https://investor.nvidia.com/news/press-release-details/2025/NVIDIA-Announces-Financial-Results-for-First-Quarter-Fiscal-2026/default.aspx", + "fetched_at": "2026-07-16T15:59:16+08:00" + }, + "confidence": "high", + "citations": [ + { + "source_id": "nvidia:earnings:fy2026-q1", + "quote": "Data Center revenue of $39.1 billion" + }, + { + "source_id": "nvidia:earnings:fy2026-q1", + "quote": "Revenue is expected to be $45.0 billion" + } + ], + "tags": [ + "NVDA", + "ai-infrastructure", + "accelerators", + "data-center" + ], + "disclaimers": [ + "Revenue is represented in US dollars and EPS is GAAP diluted EPS.", + "The source provides a release date without a time; available_at is normalized to midnight Pacific time.", + "Q2 FY2026 guidance is forward-looking and includes an expected H20 revenue loss from export controls." + ], + "ticker": "NVDA", + "period": "FY2026Q1", + "revenue": 44062000000.0, + "eps": 0.76, + "guidance": "NVIDIA expected Q2 FY2026 revenue of $45.0 billion, plus or minus 2%, including an approximately $8.0 billion H20 revenue loss attributed to export-control limitations.", + "surprise_flags": [] + } + ] +} diff --git a/examples/library/semantic_search.py b/examples/library/semantic_search.py new file mode 100644 index 0000000..9cb6f1f --- /dev/null +++ b/examples/library/semantic_search.py @@ -0,0 +1,65 @@ +"""Search a ready-to-use local bundle of financial knowledge.""" + +import asyncio +import os +from datetime import datetime, timezone +from pathlib import Path + +from dotenv import load_dotenv + +from quantmind.knowledge import TreeKnowledge +from quantmind.library import LocalKnowledgeLibrary, SemanticQuery + +_BUNDLE_PATH = Path(__file__).parent / "data" / "ai_infrastructure.db" +_EMBEDDING_MODEL = "text-embedding-3-small" +_EMBEDDING_DIMENSIONS = 1536 + + +async def main() -> None: + """Search and resolve evidence from the prebuilt local bundle.""" + load_dotenv() + if not os.getenv("OPENAI_API_KEY"): + raise SystemExit("Set OPENAI_API_KEY before running this example.") + + library = await LocalKnowledgeLibrary.open( + _BUNDLE_PATH, + embedding_model=_EMBEDDING_MODEL, + embedding_dimensions=_EMBEDDING_DIMENSIONS, + ) + try: + query = SemanticQuery( + text="What evidence shows demand for AI infrastructure is expanding?", + source_kinds=["http", "arxiv"], + tags=["ai-infrastructure"], + available_at_before=datetime(2026, 1, 1, tzinfo=timezone.utc), + top_k=6, + ) + hits = await library.search(query) + print(f"Bundle: {_BUNDLE_PATH}") + print(f"Query: {query.text}\n") + for rank, hit in enumerate(hits, start=1): + item = await library.get(hit.item_id) + print(f"{rank}. score={hit.score:.3f} type={hit.item_type}") + if hit.node_id is not None and isinstance(item, TreeKnowledge): + path = " > ".join( + node.title for node in item.find_path(hit.node_id) + ) + print(f" path: {path}") + node = item.nodes[hit.node_id] + if node.content: + print(f" content: {node.content}") + print(f" matched: {hit.matched_text}") + print(f" source: {hit.source.uri}") + print( + f" as_of={hit.as_of.date()} available_at={hit.available_at}" + ) + for citation in hit.citations: + detail = f" — {citation.quote}" if citation.quote else "" + print(f" citation: {citation.source_id}{detail}") + print() + finally: + await library.close() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/pyproject.toml b/pyproject.toml index 50d6fe9..f15b0f3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -119,10 +119,9 @@ reportIncompatibleVariableOverride = "none" # import-linter: architectural boundary contracts # ---------------------------------------------------------------------------- # Encodes the target architecture: utils, knowledge, and preprocess are -# leaves; configs depends only on knowledge; flows + magic is the apex -# layer (PR5). The transitional packages (config/, flow/, llm/, models/) -# were deleted in PR5; they remain in the forbidden lists as a tripwire -# against accidental re-introduction during future refactors. +# leaves; configs depends only on knowledge; library depends only on +# knowledge; flows + magic is the apex layer. The transitional packages +# (config/, flow/, llm/, models/) remain forbidden as a tripwire. [tool.importlinter] root_packages = ["quantmind"] @@ -135,6 +134,7 @@ forbidden_modules = [ "quantmind.configs", "quantmind.flows", "quantmind.knowledge", + "quantmind.library", "quantmind.magic", "quantmind.preprocess", ] @@ -146,6 +146,7 @@ source_modules = ["quantmind.knowledge"] forbidden_modules = [ "quantmind.configs", "quantmind.flows", + "quantmind.library", "quantmind.magic", "quantmind.preprocess", "quantmind.utils", @@ -157,6 +158,7 @@ type = "forbidden" source_modules = ["quantmind.configs"] forbidden_modules = [ "quantmind.flows", + "quantmind.library", "quantmind.magic", "quantmind.preprocess", ] @@ -169,9 +171,27 @@ forbidden_modules = [ "quantmind.configs", "quantmind.flows", "quantmind.knowledge", + "quantmind.library", "quantmind.magic", ] +[[tool.importlinter.contracts]] +name = "library only depends on knowledge" +type = "forbidden" +source_modules = ["quantmind.library"] +forbidden_modules = [ + "quantmind.config", + "quantmind.configs", + "quantmind.flow", + "quantmind.flows", + "quantmind.llm", + "quantmind.magic", + "quantmind.mind", + "quantmind.models", + "quantmind.preprocess", + "quantmind.utils", +] + [[tool.importlinter.contracts]] name = "flows + magic is apex (no transitional package imports)" type = "forbidden" diff --git a/quantmind/knowledge/_base.py b/quantmind/knowledge/_base.py index d2165af..b6ae31c 100644 --- a/quantmind/knowledge/_base.py +++ b/quantmind/knowledge/_base.py @@ -7,9 +7,10 @@ - `GraphKnowledge` — cross-item edges (placeholder, future PR) The `as_of` field is mandatory by design: financial knowledge is only useful -when its time-of-validity is explicit. `source` and `extraction` are typed -references (not bare strings) so dedup, audit, and re-runs all have stable -keys. +when its information cutoff is explicit. Optional `available_at` records when +the source became observable so research can prevent look-ahead independently +from information time. `source` and `extraction` are typed references (not bare +strings) so dedup, audit, and re-runs all have stable keys. Subclasses MUST override `embedding_text()` to declare what string the store layer should embed for them. The contract is enforced at runtime, not at @@ -80,6 +81,10 @@ class BaseKnowledge(BaseModel): # Time (financial mandate) as_of: datetime = Field(..., description="Information cutoff time.") + available_at: datetime | None = Field( + default=None, + description="Time the source became observable to researchers.", + ) created_at: datetime = Field( default_factory=lambda: datetime.now(timezone.utc) ) diff --git a/quantmind/library/__init__.py b/quantmind/library/__init__.py new file mode 100644 index 0000000..edfde4b --- /dev/null +++ b/quantmind/library/__init__.py @@ -0,0 +1,6 @@ +"""Local semantic retrieval for canonical QuantMind knowledge.""" + +from quantmind.library._types import SemanticHit, SemanticQuery +from quantmind.library.local import LocalKnowledgeLibrary + +__all__ = ["LocalKnowledgeLibrary", "SemanticHit", "SemanticQuery"] diff --git a/quantmind/library/_internal/__init__.py b/quantmind/library/_internal/__init__.py new file mode 100644 index 0000000..9c39f0c --- /dev/null +++ b/quantmind/library/_internal/__init__.py @@ -0,0 +1 @@ +"""Private implementation details for ``quantmind.library``.""" diff --git a/quantmind/library/_internal/exact_cosine.py b/quantmind/library/_internal/exact_cosine.py new file mode 100644 index 0000000..0a6f0c1 --- /dev/null +++ b/quantmind/library/_internal/exact_cosine.py @@ -0,0 +1,236 @@ +"""Deterministic NumPy exact-cosine filtering and ranking.""" + +from collections.abc import Sequence +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any +from uuid import UUID + +import numpy as np +from numpy.typing import NDArray + +from quantmind.library._types import SemanticQuery + + +@dataclass(frozen=True) +class _IndexRecord: + """Durable metadata and vector bytes for one searchable target.""" + + target_id: str + item_id: UUID + node_id: UUID | None + item_type: str + matched_text: str + as_of: float + available_at: float | None + source_kind: str + confidence: str + tags: frozenset[str] + tree_id: UUID | None + dimension: int + embedding: bytes + + +@dataclass(frozen=True) +class _RankedRecord: + """One exact-cosine result before canonical hit resolution.""" + + record: _IndexRecord + score: float + + +def _coerce_provider_vectors( + values: Any, + *, + expected_count: int, + expected_dimensions: int | None, +) -> NDArray[np.float32]: + """Validate provider output before it can enter the durable index.""" + try: + vectors = np.asarray(values, dtype=np.float32) + except (TypeError, ValueError) as exc: + raise ValueError( + "Embedding provider returned non-rectangular data" + ) from exc + if vectors.ndim != 2 or vectors.shape[0] != expected_count: + raise ValueError( + "Embedding provider returned an unexpected number or shape of vectors" + ) + dimensions = int(vectors.shape[1]) + if dimensions < 1: + raise ValueError("Embedding provider returned zero-dimensional vectors") + if expected_dimensions is not None and dimensions != expected_dimensions: + raise ValueError( + "Embedding dimension mismatch: provider returned " + f"{dimensions}, expected {expected_dimensions}" + ) + if not np.isfinite(vectors).all(): + raise ValueError("Embedding provider returned non-finite values") + if np.any(np.linalg.norm(vectors, axis=1) == 0): + raise ValueError("Embedding provider returned a zero vector") + return np.ascontiguousarray(vectors, dtype=np.float32) + + +def _decode_stored_vector( + blob: bytes, dimension: int, target_id: str +) -> NDArray[np.float32]: + """Decode and validate a persisted vector as corrupt-index protection.""" + if dimension < 1 or len(blob) != dimension * np.dtype(" tuple[bytes, int]: + """Encode one validated provider vector for SQLite persistence.""" + return ( + np.asarray(vector, dtype=" float: + """Normalize an aware financial timestamp for deterministic filtering.""" + if value.tzinfo is None or value.utcoffset() is None: + raise ValueError(f"{field_name} must be timezone-aware") + return value.astimezone(timezone.utc).timestamp() + + +class _CosineCandidates: + """Filtered records and normalized vectors ready for one query.""" + + def __init__( + self, + records: tuple[_IndexRecord, ...], + matrix: NDArray[np.float32], + ) -> None: + self._records = records + self._matrix = matrix + + def rank( + self, + query_vector: NDArray[np.float32], + *, + top_k: int, + ) -> list[_RankedRecord]: + """Rank candidates best-first with stable target-ID tie breaking.""" + if query_vector.shape[0] != self._matrix.shape[1]: + raise ValueError( + "Embedding dimension mismatch: query vector has " + f"{query_vector.shape[0]} dimensions but the index has " + f"{self._matrix.shape[1]}" + ) + normalized = query_vector / np.linalg.norm(query_vector) + scores = self._matrix @ normalized + ranked = sorted( + zip(self._records, scores, strict=True), + key=lambda pair: (-float(pair[1]), pair[0].target_id), + )[:top_k] + return [ + _RankedRecord(record=record, score=float(score)) + for record, score in ranked + ] + + +class _ExactCosineIndex: + """Immutable in-memory exact-cosine index rebuilt from SQLite rows.""" + + def __init__( + self, + records: tuple[_IndexRecord, ...], + matrix: NDArray[np.float32], + ) -> None: + self._records = records + self._matrix = matrix + + @classmethod + def build(cls, records: Sequence[_IndexRecord]) -> "_ExactCosineIndex": + """Validate durable vectors and build a normalized matrix.""" + vectors: list[NDArray[np.float32]] = [] + index_dimension: int | None = None + for record in records: + if index_dimension is None: + index_dimension = record.dimension + elif record.dimension != index_dimension: + raise RuntimeError( + "Corrupt index data: stored targets have inconsistent " + "embedding dimensions" + ) + vector = _decode_stored_vector( + record.embedding, + record.dimension, + record.target_id, + ) + vectors.append( + np.asarray(vector / np.linalg.norm(vector), dtype=np.float32) + ) + matrix = ( + np.ascontiguousarray(np.vstack(vectors), dtype=np.float32) + if vectors + else np.empty((0, 0), dtype=np.float32) + ) + return cls(tuple(records), matrix) + + def filter(self, query: SemanticQuery) -> _CosineCandidates | None: + """Apply metadata and financial-time filters before query embedding.""" + as_of_before = ( + _timestamp(query.as_of_before, "SemanticQuery.as_of_before") + if query.as_of_before is not None + else None + ) + available_at_before = ( + _timestamp( + query.available_at_before, + "SemanticQuery.available_at_before", + ) + if query.available_at_before is not None + else None + ) + item_types = set(query.item_types) if query.item_types else None + source_kinds = set(query.source_kinds) if query.source_kinds else None + required_tags = set(query.tags) if query.tags else None + indices: list[int] = [] + for index, record in enumerate(self._records): + if item_types is not None and record.item_type not in item_types: + continue + if ( + source_kinds is not None + and record.source_kind not in source_kinds + ): + continue + if ( + query.confidence is not None + and record.confidence != query.confidence + ): + continue + if required_tags is not None and not required_tags.issubset( + record.tags + ): + continue + if query.tree_id is not None and record.tree_id != query.tree_id: + continue + if as_of_before is not None and record.as_of > as_of_before: + continue + if available_at_before is not None and ( + record.available_at is None + or record.available_at > available_at_before + ): + continue + indices.append(index) + if not indices: + return None + return _CosineCandidates( + tuple(self._records[index] for index in indices), + np.ascontiguousarray(self._matrix[indices], dtype=np.float32), + ) diff --git a/quantmind/library/_internal/index_embeddings.py b/quantmind/library/_internal/index_embeddings.py new file mode 100644 index 0000000..9f45ef8 --- /dev/null +++ b/quantmind/library/_internal/index_embeddings.py @@ -0,0 +1,58 @@ +"""Private embedding client used to build and query the local index.""" + +from collections.abc import Sequence +from typing import Any, Protocol + +from openai import AsyncOpenAI + + +class _EmbeddingProvider(Protocol): + """Embedding seam used by production code and deterministic tests.""" + + async def embed( + self, + texts: Sequence[str], + *, + model: str, + dimensions: int | None, + ) -> Sequence[Sequence[float]]: + """Embed texts in input order.""" + ... + + async def close(self) -> None: + """Release provider-owned resources.""" + ... + + +class _OpenAIEmbeddingProvider: + """Generate index embeddings without exposing provider response types.""" + + def __init__(self) -> None: + self._client: AsyncOpenAI | None = None + + async def embed( + self, + texts: Sequence[str], + *, + model: str, + dimensions: int | None, + ) -> list[list[float]]: + client = self._client + if client is None: + client = AsyncOpenAI() + self._client = client + kwargs: dict[str, Any] = { + "input": list(texts), + "model": model, + "encoding_format": "float", + } + if dimensions is not None: + kwargs["dimensions"] = dimensions + response = await client.embeddings.create(**kwargs) + ordered = sorted(response.data, key=lambda item: item.index) + return [item.embedding for item in ordered] + + async def close(self) -> None: + if self._client is not None: + await self._client.close() + self._client = None diff --git a/quantmind/library/_internal/retrieval_targets.py b/quantmind/library/_internal/retrieval_targets.py new file mode 100644 index 0000000..5cc1726 --- /dev/null +++ b/quantmind/library/_internal/retrieval_targets.py @@ -0,0 +1,70 @@ +"""Map canonical knowledge to stable searchable text targets.""" + +import hashlib +from dataclasses import dataclass +from uuid import UUID + +from quantmind.knowledge import BaseKnowledge, FlattenKnowledge, TreeKnowledge + +_PROJECTION_SCHEMA_VERSION = "1" + + +@dataclass(frozen=True) +class _RetrievalTarget: + """One stable item or non-root node target to embed.""" + + target_id: str + node_id: UUID | None + text: str + projection_hash: str + tree_id: UUID | None + + +def _project_knowledge(item: BaseKnowledge) -> list[_RetrievalTarget]: + """Project flat items and trees at the issue-defined retrieval grain.""" + if not isinstance(item, (FlattenKnowledge, TreeKnowledge)): + raise TypeError( + "LocalKnowledgeLibrary supports FlattenKnowledge and TreeKnowledge" + ) + + root_text = item.embedding_text() + if not root_text: + raise ValueError("knowledge embedding_text() must not be empty") + tree_id = item.id if isinstance(item, TreeKnowledge) else None + targets = [ + _RetrievalTarget( + target_id=f"item:{item.id}", + node_id=None, + text=root_text, + projection_hash=hashlib.sha256( + root_text.encode("utf-8") + ).hexdigest(), + tree_id=tree_id, + ) + ] + + if isinstance(item, TreeKnowledge): + if item.root_node_id not in item.nodes: + raise ValueError("tree root_node_id is not present in nodes") + for node_id, node in sorted( + item.nodes.items(), key=lambda pair: str(pair[0]) + ): + if node_id != node.node_id: + raise ValueError("tree node map key does not match node_id") + if node_id == item.root_node_id: + continue + text = node.embedding_text() + if not text: + raise ValueError("tree node embedding_text() must not be empty") + targets.append( + _RetrievalTarget( + target_id=f"node:{item.id}:{node_id}", + node_id=node_id, + text=text, + projection_hash=hashlib.sha256( + text.encode("utf-8") + ).hexdigest(), + tree_id=item.id, + ) + ) + return targets diff --git a/quantmind/library/_internal/sqlite_store.py b/quantmind/library/_internal/sqlite_store.py new file mode 100644 index 0000000..3a7eb12 --- /dev/null +++ b/quantmind/library/_internal/sqlite_store.py @@ -0,0 +1,789 @@ +"""Concrete SQLite persistence for canonical knowledge and index records.""" + +import hashlib +import json +import sqlite3 +from collections.abc import Sequence +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from uuid import UUID + +from pydantic import ValidationError + +from quantmind.knowledge import ( + BaseKnowledge, + Earnings, + Factor, + News, + Paper, + PaperKnowledgeCard, + Thesis, + TreeKnowledge, +) +from quantmind.library._internal.exact_cosine import _IndexRecord +from quantmind.library._internal.retrieval_targets import ( + _PROJECTION_SCHEMA_VERSION, + _RetrievalTarget, +) + +_DATABASE_SCHEMA_VERSION = 2 + +_KNOWLEDGE_CLASSES: dict[str, type[BaseKnowledge]] = { + f"{knowledge_type.__module__}:{knowledge_type.__qualname__}": knowledge_type + for knowledge_type in ( + Earnings, + Factor, + News, + Paper, + PaperKnowledgeCard, + Thesis, + ) +} + + +@dataclass(frozen=True) +class _CanonicalNode: + """One normalized canonical tree node.""" + + node_id: UUID + parent_id: UUID | None + position: int + payload: str + content_hash: str + + +@dataclass(frozen=True) +class _CanonicalDocument: + """Canonical aggregate root plus separately persisted tree nodes.""" + + knowledge_class: str + item_shape: str + payload: str + canonical_hash: str + nodes: tuple[_CanonicalNode, ...] + + +@dataclass(frozen=True) +class _StoredEmbedding: + """Existing vector and invalidation metadata for one target.""" + + target_id: str + embedding_model: str + dimension: int + projection_hash: str + source_content_hash: str | None + knowledge_schema_version: str + projection_schema_version: str + embedding: bytes + + +@dataclass(frozen=True) +class _PreparedPut: + """Validated canonical write plus the vectors it may retain.""" + + item: BaseKnowledge + canonical: _CanonicalDocument + as_of: float + available_at: float | None + tags_json: str + existing_embeddings: dict[str, _StoredEmbedding] + + +def _json_payload(value: object) -> str: + """Encode canonical JSON with stable ordering and separators.""" + return json.dumps( + value, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ) + + +def _canonical_payload(item: BaseKnowledge) -> _CanonicalDocument: + """Split a supported canonical aggregate into root and node records.""" + knowledge_class = f"{type(item).__module__}:{type(item).__qualname__}" + if knowledge_class not in _KNOWLEDGE_CLASSES: + raise TypeError( + f"Unsupported knowledge type '{type(item).__name__}'; " + "only canonical quantmind.knowledge types can be persisted" + ) + full_payload = _json_payload(item.model_dump(mode="json")) + canonical_hash = hashlib.sha256(full_payload.encode("utf-8")).hexdigest() + if not isinstance(item, TreeKnowledge): + return _CanonicalDocument( + knowledge_class=knowledge_class, + item_shape="flat", + payload=full_payload, + canonical_hash=canonical_hash, + nodes=(), + ) + + nodes: list[_CanonicalNode] = [] + for node_id, node in sorted( + item.nodes.items(), key=lambda pair: str(pair[0]) + ): + node_payload = _json_payload(node.model_dump(mode="json")) + nodes.append( + _CanonicalNode( + node_id=node_id, + parent_id=node.parent_id, + position=node.position, + payload=node_payload, + content_hash=hashlib.sha256( + node_payload.encode("utf-8") + ).hexdigest(), + ) + ) + return _CanonicalDocument( + knowledge_class=knowledge_class, + item_shape="tree", + payload=_json_payload(item.model_dump(mode="json", exclude={"nodes"})), + canonical_hash=canonical_hash, + nodes=tuple(nodes), + ) + + +def _assemble_canonical_payload( + *, + item_id: str, + item_shape: str, + item_payload: str, + expected_node_count: int, + node_records: Sequence[tuple[str, str | None, int, str, str]], +) -> str: + """Rehydrate normalized tree nodes into the canonical Pydantic payload.""" + if item_shape == "flat": + if expected_node_count or node_records: + raise RuntimeError( + f"Stale canonical knowledge for item '{item_id}': " + "flat knowledge unexpectedly has tree nodes" + ) + return item_payload + if item_shape != "tree": + raise RuntimeError( + f"Stale canonical knowledge for item '{item_id}': " + f"unsupported item shape '{item_shape}'" + ) + if len(node_records) != expected_node_count: + raise RuntimeError( + f"Stale canonical knowledge for item '{item_id}': expected " + f"{expected_node_count} canonical nodes, found {len(node_records)}" + ) + try: + root_payload = json.loads(item_payload) + except json.JSONDecodeError as exc: + raise RuntimeError( + f"Stale canonical knowledge for item '{item_id}': invalid root JSON" + ) from exc + if not isinstance(root_payload, dict): + raise RuntimeError( + f"Stale canonical knowledge for item '{item_id}': invalid root payload" + ) + nodes: dict[str, object] = {} + for node_id, parent_id, position, node_payload, node_hash in node_records: + actual_hash = hashlib.sha256(node_payload.encode("utf-8")).hexdigest() + if actual_hash != node_hash: + raise RuntimeError( + f"Stale canonical knowledge for item '{item_id}': " + f"node '{node_id}' content hash mismatch" + ) + try: + parsed_node = json.loads(node_payload) + except json.JSONDecodeError as exc: + raise RuntimeError( + f"Stale canonical knowledge for item '{item_id}': " + f"node '{node_id}' contains invalid JSON" + ) from exc + if ( + not isinstance(parsed_node, dict) + or parsed_node.get("node_id") != node_id + or parsed_node.get("parent_id") != parent_id + or parsed_node.get("position") != position + ): + raise RuntimeError( + f"Stale canonical knowledge for item '{item_id}': " + f"node '{node_id}' metadata does not match its payload" + ) + nodes[node_id] = parsed_node + root_payload["nodes"] = nodes + return _json_payload(root_payload) + + +def _load_canonical( + *, + item_id: str, + knowledge_class: str, + item_type: str, + schema_version: str, + payload: str, + canonical_hash: str, +) -> BaseKnowledge: + """Validate stored canonical bytes against identity and schema metadata.""" + actual_hash = hashlib.sha256(payload.encode("utf-8")).hexdigest() + if actual_hash != canonical_hash: + raise RuntimeError( + f"Stale canonical knowledge for item '{item_id}': content hash mismatch" + ) + model = _KNOWLEDGE_CLASSES.get(knowledge_class) + if model is None: + raise RuntimeError( + f"Stale canonical knowledge for item '{item_id}': " + f"unsupported stored type '{knowledge_class}'" + ) + try: + item = model.model_validate_json(payload) + except ValidationError as exc: + raise RuntimeError( + f"Stale canonical knowledge for item '{item_id}': " + "stored payload no longer validates" + ) from exc + if ( + str(item.id) != item_id + or item.item_type != item_type + or item.schema_version != schema_version + ): + raise RuntimeError( + f"Stale canonical knowledge for item '{item_id}': identity or schema " + "metadata does not match the payload" + ) + return item + + +def _timestamp(value: datetime, field_name: str) -> float: + """Normalize an aware canonical timestamp for SQLite filtering.""" + if value.tzinfo is None or value.utcoffset() is None: + raise ValueError(f"{field_name} must be timezone-aware") + return value.astimezone(timezone.utc).timestamp() + + +def _initialize_schema(db: sqlite3.Connection) -> None: + """Create the current schema or reject an incompatible local database.""" + version_row = db.execute("PRAGMA user_version").fetchone() + version = int(version_row[0]) + if version not in (0, _DATABASE_SCHEMA_VERSION): + raise RuntimeError( + "Stale knowledge library schema: database version " + f"{version}, expected {_DATABASE_SCHEMA_VERSION}" + ) + if version == _DATABASE_SCHEMA_VERSION: + return + db.executescript( + f""" + CREATE TABLE knowledge_items ( + item_id TEXT PRIMARY KEY, + knowledge_class TEXT NOT NULL, + item_type TEXT NOT NULL, + item_shape TEXT NOT NULL CHECK (item_shape IN ('flat', 'tree')), + schema_version TEXT NOT NULL, + payload_json TEXT NOT NULL, + canonical_hash TEXT NOT NULL, + node_count INTEGER NOT NULL CHECK (node_count >= 0), + target_count INTEGER NOT NULL CHECK (target_count > 0) + ); + + CREATE TABLE knowledge_nodes ( + item_id TEXT NOT NULL, + node_id TEXT NOT NULL, + parent_id TEXT, + position INTEGER NOT NULL, + payload_json TEXT NOT NULL, + content_hash TEXT NOT NULL, + PRIMARY KEY (item_id, node_id), + FOREIGN KEY (item_id) REFERENCES knowledge_items(item_id) + ON DELETE CASCADE + ); + + CREATE TABLE semantic_records ( + target_id TEXT PRIMARY KEY, + item_id TEXT NOT NULL, + node_id TEXT, + item_type TEXT NOT NULL, + matched_text TEXT NOT NULL, + as_of REAL NOT NULL, + available_at REAL, + source_kind TEXT NOT NULL, + confidence TEXT NOT NULL, + tags_json TEXT NOT NULL, + tree_id TEXT, + embedding_model TEXT NOT NULL, + dimension INTEGER NOT NULL, + projection_hash TEXT NOT NULL, + source_content_hash TEXT, + knowledge_schema_version TEXT NOT NULL, + projection_schema_version TEXT NOT NULL, + item_canonical_hash TEXT NOT NULL, + embedding BLOB NOT NULL, + FOREIGN KEY (item_id) REFERENCES knowledge_items(item_id) + ON DELETE CASCADE + ); + + CREATE INDEX semantic_records_item_id + ON semantic_records(item_id); + CREATE INDEX semantic_records_filters + ON semantic_records(item_type, source_kind, confidence, tree_id); + CREATE INDEX knowledge_nodes_parent + ON knowledge_nodes(item_id, parent_id, position); + + PRAGMA user_version = {_DATABASE_SCHEMA_VERSION}; + """ + ) + + +class _SQLiteStore: + """Own the concrete SQLite schema, transactions, and record validation.""" + + def __init__(self, db: sqlite3.Connection) -> None: + self._db = db + + @classmethod + def open(cls, path: str | Path) -> "_SQLiteStore": + """Open and initialize a concrete SQLite knowledge store.""" + supplied_path = str(path) + database_path = ( + supplied_path + if supplied_path == ":memory:" + else str(Path(supplied_path).expanduser()) + ) + if database_path != ":memory:": + Path(database_path).parent.mkdir(parents=True, exist_ok=True) + db: sqlite3.Connection | None = None + try: + db = sqlite3.connect(database_path, isolation_level=None) + db.row_factory = sqlite3.Row + db.execute("PRAGMA foreign_keys = ON") + db.execute("PRAGMA busy_timeout = 5000") + _initialize_schema(db) + except sqlite3.DatabaseError as exc: + if db is not None: + db.close() + raise RuntimeError( + f"Corrupt knowledge library database at '{database_path}'" + ) from exc + except Exception: + if db is not None: + db.close() + raise + return cls(db) + + def prepare_put(self, item: BaseKnowledge) -> _PreparedPut: + """Validate a canonical write and load vectors it may retain.""" + rows = self._db.execute( + "SELECT * FROM semantic_records WHERE item_id = ?", + (str(item.id),), + ).fetchall() + existing = { + str(row["target_id"]): _StoredEmbedding( + target_id=str(row["target_id"]), + embedding_model=str(row["embedding_model"]), + dimension=int(row["dimension"]), + projection_hash=str(row["projection_hash"]), + source_content_hash=( + str(row["source_content_hash"]) + if row["source_content_hash"] is not None + else None + ), + knowledge_schema_version=str(row["knowledge_schema_version"]), + projection_schema_version=str(row["projection_schema_version"]), + embedding=bytes(row["embedding"]), + ) + for row in rows + } + return _PreparedPut( + item=item, + canonical=_canonical_payload(item), + as_of=_timestamp(item.as_of, "BaseKnowledge.as_of"), + available_at=( + _timestamp(item.available_at, "BaseKnowledge.available_at") + if item.available_at is not None + else None + ), + tags_json=json.dumps( + item.tags, + ensure_ascii=False, + separators=(",", ":"), + ), + existing_embeddings=existing, + ) + + def put( + self, + prepared: _PreparedPut, + targets: Sequence[_RetrievalTarget], + vectors: dict[str, tuple[bytes, int]], + *, + embedding_model: str, + ) -> None: + """Atomically replace canonical and derived records for one item.""" + item = prepared.item + canonical = prepared.canonical + try: + self._db.execute("BEGIN IMMEDIATE") + self._db.execute( + """ + INSERT INTO knowledge_items ( + item_id, knowledge_class, item_type, item_shape, + schema_version, payload_json, canonical_hash, + node_count, target_count + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(item_id) DO UPDATE SET + knowledge_class = excluded.knowledge_class, + item_type = excluded.item_type, + item_shape = excluded.item_shape, + schema_version = excluded.schema_version, + payload_json = excluded.payload_json, + canonical_hash = excluded.canonical_hash, + node_count = excluded.node_count, + target_count = excluded.target_count + """, + ( + str(item.id), + canonical.knowledge_class, + item.item_type, + canonical.item_shape, + item.schema_version, + canonical.payload, + canonical.canonical_hash, + len(canonical.nodes), + len(targets), + ), + ) + self._db.execute( + "DELETE FROM semantic_records WHERE item_id = ?", + (str(item.id),), + ) + self._db.execute( + "DELETE FROM knowledge_nodes WHERE item_id = ?", + (str(item.id),), + ) + for node in canonical.nodes: + self._db.execute( + """ + INSERT INTO knowledge_nodes ( + item_id, node_id, parent_id, position, + payload_json, content_hash + ) VALUES (?, ?, ?, ?, ?, ?) + """, + ( + str(item.id), + str(node.node_id), + str(node.parent_id) if node.parent_id else None, + node.position, + node.payload, + node.content_hash, + ), + ) + for target in targets: + blob, dimension = vectors[target.target_id] + self._db.execute( + """ + INSERT INTO semantic_records ( + target_id, item_id, node_id, item_type, + matched_text, as_of, available_at, source_kind, + confidence, tags_json, tree_id, embedding_model, + dimension, projection_hash, source_content_hash, + knowledge_schema_version, + projection_schema_version, item_canonical_hash, + embedding + ) VALUES ( + ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, + ?, ?, ? + ) + """, + ( + target.target_id, + str(item.id), + str(target.node_id) + if target.node_id is not None + else None, + item.item_type, + target.text, + prepared.as_of, + prepared.available_at, + item.source.kind, + item.confidence, + prepared.tags_json, + str(target.tree_id) + if target.tree_id is not None + else None, + embedding_model, + dimension, + target.projection_hash, + item.source.content_hash, + item.schema_version, + _PROJECTION_SCHEMA_VERSION, + canonical.canonical_hash, + blob, + ), + ) + self._db.execute("COMMIT") + except Exception: + if self._db.in_transaction: + self._db.execute("ROLLBACK") + raise + + def get(self, item_id: UUID) -> BaseKnowledge: + """Return validated canonical knowledge or report not-found/stale data.""" + row = self._db.execute( + "SELECT * FROM knowledge_items WHERE item_id = ?", + (str(item_id),), + ).fetchone() + if row is None: + derived_count = int( + self._db.execute( + """ + SELECT + (SELECT COUNT(*) FROM semantic_records + WHERE item_id = ?) + + + (SELECT COUNT(*) FROM knowledge_nodes + WHERE item_id = ?) + """, + (str(item_id), str(item_id)), + ).fetchone()[0] + ) + if derived_count: + raise RuntimeError( + f"Stale data for item '{item_id}': child records exist " + "without canonical knowledge" + ) + raise KeyError(f"Knowledge item '{item_id}' not found") + item_key = str(row["item_id"]) + node_rows = self._db.execute( + """ + SELECT node_id, parent_id, position, payload_json, content_hash + FROM knowledge_nodes + WHERE item_id = ? + ORDER BY node_id + """, + (item_key,), + ).fetchall() + payload = _assemble_canonical_payload( + item_id=item_key, + item_shape=str(row["item_shape"]), + item_payload=str(row["payload_json"]), + expected_node_count=int(row["node_count"]), + node_records=[ + ( + str(node_row["node_id"]), + ( + str(node_row["parent_id"]) + if node_row["parent_id"] is not None + else None + ), + int(node_row["position"]), + str(node_row["payload_json"]), + str(node_row["content_hash"]), + ) + for node_row in node_rows + ], + ) + return _load_canonical( + item_id=item_key, + knowledge_class=str(row["knowledge_class"]), + item_type=str(row["item_type"]), + schema_version=str(row["schema_version"]), + payload=payload, + canonical_hash=str(row["canonical_hash"]), + ) + + def delete(self, item_id: UUID) -> None: + """Transactionally remove canonical knowledge and every child record.""" + exists = self._db.execute( + "SELECT 1 FROM knowledge_items WHERE item_id = ?", + (str(item_id),), + ).fetchone() + if exists is None: + derived_count = int( + self._db.execute( + """ + SELECT + (SELECT COUNT(*) FROM semantic_records + WHERE item_id = ?) + + + (SELECT COUNT(*) FROM knowledge_nodes + WHERE item_id = ?) + """, + (str(item_id), str(item_id)), + ).fetchone()[0] + ) + if derived_count: + raise RuntimeError( + f"Stale data for item '{item_id}': cannot delete child " + "records without canonical knowledge" + ) + raise KeyError(f"Knowledge item '{item_id}' not found") + try: + self._db.execute("BEGIN IMMEDIATE") + self._db.execute( + "DELETE FROM semantic_records WHERE item_id = ?", + (str(item_id),), + ) + self._db.execute( + "DELETE FROM knowledge_nodes WHERE item_id = ?", + (str(item_id),), + ) + self._db.execute( + "DELETE FROM knowledge_items WHERE item_id = ?", + (str(item_id),), + ) + self._db.execute("COMMIT") + except Exception: + if self._db.in_transaction: + self._db.execute("ROLLBACK") + raise + + def load_index_records( + self, + *, + embedding_model: str, + embedding_dimensions: int | None, + ) -> list[_IndexRecord]: + """Validate SQLite relationships and load typed exact-index records.""" + orphan = self._db.execute( + """ + SELECT r.item_id + FROM semantic_records AS r + LEFT JOIN knowledge_items AS i ON i.item_id = r.item_id + WHERE i.item_id IS NULL + LIMIT 1 + """ + ).fetchone() + if orphan is not None: + raise RuntimeError( + f"Stale index data for item '{orphan['item_id']}': " + "derived records exist without canonical knowledge" + ) + orphan_node = self._db.execute( + """ + SELECT n.item_id + FROM knowledge_nodes AS n + LEFT JOIN knowledge_items AS i ON i.item_id = n.item_id + WHERE i.item_id IS NULL + LIMIT 1 + """ + ).fetchone() + if orphan_node is not None: + raise RuntimeError( + f"Stale canonical knowledge for item '{orphan_node['item_id']}': " + "tree nodes exist without an aggregate root" + ) + incomplete_tree = self._db.execute( + """ + SELECT i.item_id, i.item_shape, i.node_count, + COUNT(n.node_id) AS actual_count + FROM knowledge_items AS i + LEFT JOIN knowledge_nodes AS n ON n.item_id = i.item_id + GROUP BY i.item_id, i.item_shape, i.node_count + HAVING COUNT(n.node_id) != i.node_count + OR (i.item_shape = 'flat' AND i.node_count != 0) + OR (i.item_shape = 'tree' AND i.node_count = 0) + LIMIT 1 + """ + ).fetchone() + if incomplete_tree is not None: + raise RuntimeError( + f"Stale canonical knowledge for item " + f"'{incomplete_tree['item_id']}': expected " + f"{incomplete_tree['node_count']} tree nodes, found " + f"{incomplete_tree['actual_count']}" + ) + incomplete = self._db.execute( + """ + SELECT i.item_id, i.target_count, COUNT(r.target_id) AS actual_count + FROM knowledge_items AS i + LEFT JOIN semantic_records AS r ON r.item_id = i.item_id + GROUP BY i.item_id, i.target_count + HAVING COUNT(r.target_id) != i.target_count + LIMIT 1 + """ + ).fetchone() + if incomplete is not None: + raise RuntimeError( + f"Stale index data for item '{incomplete['item_id']}': expected " + f"{incomplete['target_count']} targets, found " + f"{incomplete['actual_count']}" + ) + + rows = self._db.execute( + """ + SELECT r.*, i.canonical_hash AS current_canonical_hash, + i.schema_version AS current_schema_version + FROM semantic_records AS r + JOIN knowledge_items AS i ON i.item_id = r.item_id + ORDER BY r.target_id + """ + ).fetchall() + records: list[_IndexRecord] = [] + for row in rows: + target_id = str(row["target_id"]) + dimension = int(row["dimension"]) + if str(row["embedding_model"]) != embedding_model: + raise RuntimeError( + f"Stale index data for target '{target_id}': embedding model " + "changed; re-put the canonical item" + ) + if ( + embedding_dimensions is not None + and dimension != embedding_dimensions + ): + raise RuntimeError( + f"Stale index data for target '{target_id}': embedding " + "dimension changed; re-put the canonical item" + ) + if ( + str(row["projection_schema_version"]) + != _PROJECTION_SCHEMA_VERSION + or str(row["knowledge_schema_version"]) + != str(row["current_schema_version"]) + or str(row["item_canonical_hash"]) + != str(row["current_canonical_hash"]) + ): + raise RuntimeError( + f"Stale index data for target '{target_id}': projection or " + "canonical metadata changed; re-put the canonical item" + ) + try: + parsed_tags = json.loads(str(row["tags_json"])) + except json.JSONDecodeError as exc: + raise RuntimeError( + f"Corrupt index data for target '{target_id}': invalid tags" + ) from exc + if not isinstance(parsed_tags, list) or not all( + isinstance(tag, str) for tag in parsed_tags + ): + raise RuntimeError( + f"Corrupt index data for target '{target_id}': invalid tags" + ) + node_value = row["node_id"] + tree_value = row["tree_id"] + try: + record = _IndexRecord( + target_id=target_id, + item_id=UUID(str(row["item_id"])), + node_id=UUID(str(node_value)) if node_value else None, + item_type=str(row["item_type"]), + matched_text=str(row["matched_text"]), + as_of=float(row["as_of"]), + available_at=( + float(row["available_at"]) + if row["available_at"] is not None + else None + ), + source_kind=str(row["source_kind"]), + confidence=str(row["confidence"]), + tags=frozenset(parsed_tags), + tree_id=UUID(str(tree_value)) if tree_value else None, + dimension=dimension, + embedding=bytes(row["embedding"]), + ) + except (TypeError, ValueError) as exc: + raise RuntimeError( + f"Corrupt index data for target '{target_id}': invalid metadata" + ) from exc + records.append(record) + return records + + def close(self) -> None: + """Close the owned SQLite connection.""" + self._db.close() diff --git a/quantmind/library/_types.py b/quantmind/library/_types.py new file mode 100644 index 0000000..9fd970e --- /dev/null +++ b/quantmind/library/_types.py @@ -0,0 +1,71 @@ +"""Public domain types for semantic knowledge retrieval.""" + +from datetime import datetime +from typing import Literal +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +from quantmind.knowledge import Citation, SourceRef + + +class SemanticQuery(BaseModel): + """A financial-time-aware semantic query over canonical knowledge.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + text: str = Field(min_length=1) + item_types: list[str] | None = None + source_kinds: ( + list[ + Literal[ + "arxiv", + "http", + "doi", + "local", + "rss", + "transcript", + "manual", + ] + ] + | None + ) = None + confidence: Literal["low", "medium", "high"] | None = None + tags: list[str] | None = None + tree_id: UUID | None = None + as_of_before: datetime | None = None + available_at_before: datetime | None = None + top_k: int = Field(default=10, ge=1) + + @field_validator("text") + @classmethod + def _text_must_not_be_blank(cls, value: str) -> str: + stripped = value.strip() + if not stripped: + raise ValueError("query text must not be blank") + return stripped + + @field_validator("as_of_before", "available_at_before") + @classmethod + def _cutoffs_must_be_timezone_aware( + cls, value: datetime | None + ) -> datetime | None: + if value is not None and value.tzinfo is None: + raise ValueError("financial-time cutoffs must be timezone-aware") + return value + + +class SemanticHit(BaseModel): + """Auditable evidence returned by semantic ranking.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + item_id: UUID + node_id: UUID | None + item_type: str + score: float + matched_text: str + as_of: datetime + available_at: datetime | None + source: SourceRef + citations: list[Citation] diff --git a/quantmind/library/local.py b/quantmind/library/local.py new file mode 100644 index 0000000..6055d05 --- /dev/null +++ b/quantmind/library/local.py @@ -0,0 +1,248 @@ +"""Opinionated SQLite and NumPy semantic knowledge library.""" + +import asyncio +from pathlib import Path +from uuid import UUID + +from typing_extensions import Self + +from quantmind.knowledge import BaseKnowledge, TreeKnowledge +from quantmind.library._internal.exact_cosine import ( + _coerce_provider_vectors, + _decode_stored_vector, + _encode_vector, + _ExactCosineIndex, +) +from quantmind.library._internal.index_embeddings import ( + _EmbeddingProvider, + _OpenAIEmbeddingProvider, +) +from quantmind.library._internal.retrieval_targets import ( + _PROJECTION_SCHEMA_VERSION, + _project_knowledge, + _RetrievalTarget, +) +from quantmind.library._internal.sqlite_store import _SQLiteStore +from quantmind.library._types import SemanticHit, SemanticQuery + + +class LocalKnowledgeLibrary: + """Persist and semantically search canonical QuantMind knowledge locally.""" + + def __init__( + self, + store: _SQLiteStore, + *, + embedding_model: str, + embedding_dimensions: int | None, + embedding_provider: _EmbeddingProvider, + ) -> None: + self._store: _SQLiteStore | None = store + self._embedding_model = embedding_model + self._embedding_dimensions = embedding_dimensions + self._embedding_provider = embedding_provider + self._lock = asyncio.Lock() + self._index: _ExactCosineIndex | None = None + + @classmethod + async def open( + cls, + path: str | Path, + *, + embedding_model: str, + embedding_dimensions: int | None = None, + _embedding_provider: _EmbeddingProvider | None = None, + ) -> Self: + """Open a local library without performing network I/O. + + Args: + path: SQLite database path or ``":memory:"``. + embedding_model: Provider model identity recorded with every vector. + embedding_dimensions: Optional requested vector dimension. + + Returns: + An open local library. + """ + if not embedding_model.strip(): + raise ValueError("embedding_model must not be blank") + if embedding_dimensions is not None and embedding_dimensions < 1: + raise ValueError("embedding_dimensions must be positive") + return cls( + _SQLiteStore.open(path), + embedding_model=embedding_model, + embedding_dimensions=embedding_dimensions, + embedding_provider=( + _embedding_provider or _OpenAIEmbeddingProvider() + ), + ) + + async def put(self, item: BaseKnowledge) -> None: + """Atomically persist canonical knowledge and its affected targets.""" + async with self._lock: + store = self._store + if store is None: + raise RuntimeError("LocalKnowledgeLibrary is closed") + prepared = store.prepare_put(item) + targets = _project_knowledge(item) + existing = prepared.existing_embeddings + affected: list[_RetrievalTarget] = [] + vectors: dict[str, tuple[bytes, int]] = {} + for target in targets: + stored = existing.get(target.target_id) + needs_embedding = stored is None + if stored is not None: + needs_embedding = any( + ( + stored.embedding_model != self._embedding_model, + stored.projection_hash != target.projection_hash, + stored.source_content_hash + != item.source.content_hash, + stored.knowledge_schema_version + != item.schema_version, + stored.projection_schema_version + != _PROJECTION_SCHEMA_VERSION, + self._embedding_dimensions is not None + and stored.dimension != self._embedding_dimensions, + ) + ) + if needs_embedding: + affected.append(target) + continue + assert stored is not None + _decode_stored_vector( + stored.embedding, + stored.dimension, + target.target_id, + ) + vectors[target.target_id] = ( + stored.embedding, + stored.dimension, + ) + + if affected: + provider_values = await self._embedding_provider.embed( + [target.text for target in affected], + model=self._embedding_model, + dimensions=self._embedding_dimensions, + ) + generated = _coerce_provider_vectors( + provider_values, + expected_count=len(affected), + expected_dimensions=self._embedding_dimensions, + ) + for target, vector in zip(affected, generated, strict=True): + vectors[target.target_id] = _encode_vector(vector) + + store.put( + prepared, + targets, + vectors, + embedding_model=self._embedding_model, + ) + self._index = None + + async def get(self, item_id: UUID) -> BaseKnowledge: + """Return validated canonical knowledge or report not-found/stale data.""" + async with self._lock: + store = self._store + if store is None: + raise RuntimeError("LocalKnowledgeLibrary is closed") + return store.get(item_id) + + async def search(self, query: SemanticQuery) -> list[SemanticHit]: + """Rank filtered targets with deterministic exact cosine similarity.""" + async with self._lock: + store = self._store + if store is None: + raise RuntimeError("LocalKnowledgeLibrary is closed") + if self._index is None: + self._index = _ExactCosineIndex.build( + store.load_index_records( + embedding_model=self._embedding_model, + embedding_dimensions=self._embedding_dimensions, + ) + ) + candidates = self._index.filter(query) + if candidates is None: + return [] + + provider_values = await self._embedding_provider.embed( + [query.text], + model=self._embedding_model, + dimensions=self._embedding_dimensions, + ) + query_vectors = _coerce_provider_vectors( + provider_values, + expected_count=1, + expected_dimensions=self._embedding_dimensions, + ) + ranked = candidates.rank(query_vectors[0], top_k=query.top_k) + + canonical: dict[UUID, BaseKnowledge] = {} + hits: list[SemanticHit] = [] + for result in ranked: + record = result.record + item = canonical.get(record.item_id) + if item is None: + try: + item = store.get(record.item_id) + except KeyError as exc: + raise RuntimeError( + f"Stale index data for item '{record.item_id}': " + "canonical knowledge is missing" + ) from exc + canonical[record.item_id] = item + if isinstance(item, TreeKnowledge): + if record.node_id is None: + citations = item.root().citations + else: + node = item.nodes.get(record.node_id) + if node is None: + raise RuntimeError( + f"Stale index data for target " + f"'{record.target_id}': canonical node is missing" + ) + citations = node.citations + else: + if record.node_id is not None: + raise RuntimeError( + f"Stale index data for target '{record.target_id}': " + "node target belongs to non-tree knowledge" + ) + citations = item.citations + hits.append( + SemanticHit( + item_id=record.item_id, + node_id=record.node_id, + item_type=item.item_type, + score=result.score, + matched_text=record.matched_text, + as_of=item.as_of, + available_at=item.available_at, + source=item.source, + citations=citations, + ) + ) + return hits + + async def delete(self, item_id: UUID) -> None: + """Transactionally remove canonical knowledge and every child record.""" + async with self._lock: + store = self._store + if store is None: + raise RuntimeError("LocalKnowledgeLibrary is closed") + store.delete(item_id) + self._index = None + + async def close(self) -> None: + """Close SQLite and provider-owned resources; repeated calls are safe.""" + async with self._lock: + store = self._store + if store is None: + return + self._store = None + self._index = None + try: + await self._embedding_provider.close() + finally: + store.close() diff --git a/scripts/examples/build_ai_infrastructure_bundle.py b/scripts/examples/build_ai_infrastructure_bundle.py new file mode 100644 index 0000000..b68e0f3 --- /dev/null +++ b/scripts/examples/build_ai_infrastructure_bundle.py @@ -0,0 +1,70 @@ +"""Compile the library example source into a ready-to-search SQLite bundle.""" + +import asyncio +import json +import os +from pathlib import Path + +from dotenv import load_dotenv + +from quantmind.knowledge import BaseKnowledge, Earnings, News, Paper +from quantmind.library import LocalKnowledgeLibrary + +_ROOT = Path(__file__).parents[2] +_SOURCE_PATH = ( + _ROOT / "examples" / "library" / "data" / "ai_infrastructure.json" +) +_OUTPUT_PATH = _ROOT / "examples" / "library" / "data" / "ai_infrastructure.db" +_EMBEDDING_MODEL = "text-embedding-3-small" +_EMBEDDING_DIMENSIONS = 1536 +_KNOWLEDGE_TYPES: dict[str, type[BaseKnowledge]] = { + "earnings": Earnings, + "news": News, + "paper": Paper, +} + + +def _load_source_items() -> list[BaseKnowledge]: + """Validate the auditable source JSON as canonical knowledge.""" + bundle = json.loads(_SOURCE_PATH.read_text(encoding="utf-8")) + items: list[BaseKnowledge] = [] + for payload in bundle["items"]: + item_type = str(payload["item_type"]) + knowledge_type = _KNOWLEDGE_TYPES.get(item_type) + if knowledge_type is None: + raise ValueError(f"Unsupported source item_type: {item_type}") + items.append(knowledge_type.model_validate(payload)) + return items + + +async def main() -> None: + """Build a fresh model-specific database from the source bundle.""" + load_dotenv() + if not os.getenv("OPENAI_API_KEY"): + raise SystemExit("Set OPENAI_API_KEY before building the bundle.") + + items = _load_source_items() + _OUTPUT_PATH.unlink(missing_ok=True) + succeeded = False + library = await LocalKnowledgeLibrary.open( + _OUTPUT_PATH, + embedding_model=_EMBEDDING_MODEL, + embedding_dimensions=_EMBEDDING_DIMENSIONS, + ) + try: + for item in items: + await library.put(item) + succeeded = True + finally: + await library.close() + if not succeeded: + _OUTPUT_PATH.unlink(missing_ok=True) + + print( + f"Built {_OUTPUT_PATH.relative_to(_ROOT)} with {len(items)} items " + f"using {_EMBEDDING_MODEL}." + ) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tests/knowledge/test_base.py b/tests/knowledge/test_base.py index eae7380..f1fc840 100644 --- a/tests/knowledge/test_base.py +++ b/tests/knowledge/test_base.py @@ -102,6 +102,19 @@ def test_default_confidence_is_medium(self): item = _ConcreteKnowledge(as_of=_now(), source=_src()) self.assertEqual(item.confidence, "medium") + def test_available_at_is_optional_and_round_trips(self): + item = _ConcreteKnowledge( + as_of=_now(), + available_at=_now() + timedelta(hours=2), + source=_src(), + ) + revived = _ConcreteKnowledge.model_validate_json(item.model_dump_json()) + self.assertEqual(revived.available_at, _now() + timedelta(hours=2)) + + def test_available_at_defaults_to_unknown(self): + item = _ConcreteKnowledge(as_of=_now(), source=_src()) + self.assertIsNone(item.available_at) + def test_default_schema_version(self): item = _ConcreteKnowledge(as_of=_now(), source=_src()) self.assertEqual(item.schema_version, "1.0") diff --git a/tests/library/__init__.py b/tests/library/__init__.py new file mode 100644 index 0000000..56155a9 --- /dev/null +++ b/tests/library/__init__.py @@ -0,0 +1 @@ +"""Tests for the local semantic knowledge library.""" diff --git a/tests/library/test_example_bundle.py b/tests/library/test_example_bundle.py new file mode 100644 index 0000000..a6d0b8c --- /dev/null +++ b/tests/library/test_example_bundle.py @@ -0,0 +1,183 @@ +import json +import sqlite3 +import unittest +from collections.abc import Sequence +from datetime import datetime, timezone +from pathlib import Path +from uuid import UUID + +from quantmind.knowledge import ( + BaseKnowledge, + Earnings, + News, + Paper, + TreeKnowledge, +) +from quantmind.library import LocalKnowledgeLibrary, SemanticQuery + +_BUNDLE_PATH = ( + Path(__file__).parents[2] + / "examples" + / "library" + / "data" + / "ai_infrastructure.json" +) +_DATABASE_PATH = _BUNDLE_PATH.with_suffix(".db") +_KNOWLEDGE_TYPES: dict[str, type[BaseKnowledge]] = { + "earnings": Earnings, + "news": News, + "paper": Paper, +} + + +class _QueryEmbeddingProvider: + """Provide one local query vector for the prebuilt OpenAI index.""" + + def __init__(self) -> None: + self.calls: list[tuple[str, int | None, tuple[str, ...]]] = [] + + async def embed( + self, + texts: Sequence[str], + *, + model: str, + dimensions: int | None, + ) -> list[list[float]]: + self.calls.append((model, dimensions, tuple(texts))) + if model != "text-embedding-3-small" or dimensions != 1536: + raise ValueError("Unexpected prebuilt embedding metadata") + return [[1.0, *([0.0] * 1535)] for _ in texts] + + async def close(self) -> None: + """Release no resources.""" + + +class ExampleBundleTests(unittest.TestCase): + def setUp(self) -> None: + bundle = json.loads(_BUNDLE_PATH.read_text(encoding="utf-8")) + self.scenario = str(bundle["scenario"]) + self.items = [ + _KNOWLEDGE_TYPES[str(payload["item_type"])].model_validate(payload) + for payload in bundle["items"] + ] + + def test_bundle_is_a_concrete_cross_shape_scenario(self): + self.assertIn("AI", self.scenario) + self.assertEqual( + {type(item) for item in self.items}, + {News, Earnings, Paper}, + ) + self.assertTrue( + all("ai-infrastructure" in item.tags for item in self.items) + ) + + def test_sources_and_financial_times_are_auditable(self): + for item in self.items: + self.assertIsNotNone(item.source.uri) + assert item.source.uri is not None + self.assertTrue(item.source.uri.startswith("https://")) + self.assertIsNotNone(item.as_of.utcoffset()) + self.assertIsNotNone(item.available_at) + assert item.available_at is not None + self.assertIsNotNone(item.available_at.utcoffset()) + self.assertTrue(item.citations) + + def test_paper_tree_has_stable_valid_navigation(self): + paper = next(item for item in self.items if isinstance(item, Paper)) + self.assertEqual(len(paper.nodes), 4) + self.assertEqual( + [node.position for node in paper.children_of(paper.root_node_id)], + [0, 1, 2], + ) + self.assertEqual(len(list(paper.walk_dfs())), 4) + for node_id, node in paper.nodes.items(): + self.assertIsInstance(node_id, UUID) + self.assertEqual(node_id, node.node_id) + path = paper.find_path(node_id) + self.assertEqual(path[0].node_id, paper.root_node_id) + self.assertEqual(path[-1].node_id, node_id) + + def test_bundle_creates_six_documented_semantic_targets(self): + target_count = sum( + len(item.nodes) if isinstance(item, TreeKnowledge) else 1 + for item in self.items + ) + self.assertEqual(target_count, 6) + + +class ExampleBundleSearchTests(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + bundle = json.loads(_BUNDLE_PATH.read_text(encoding="utf-8")) + self.items = [ + _KNOWLEDGE_TYPES[str(payload["item_type"])].model_validate(payload) + for payload in bundle["items"] + ] + + async def test_prebuilt_bundle_searches_and_matches_source_json(self): + self.assertTrue(_DATABASE_PATH.is_file()) + with sqlite3.connect(_DATABASE_PATH) as db: + self.assertEqual( + db.execute("SELECT COUNT(*) FROM knowledge_items").fetchone()[ + 0 + ], + 3, + ) + self.assertEqual( + db.execute("SELECT COUNT(*) FROM knowledge_nodes").fetchone()[ + 0 + ], + 4, + ) + self.assertEqual( + db.execute("SELECT COUNT(*) FROM semantic_records").fetchone()[ + 0 + ], + 6, + ) + self.assertEqual( + set( + db.execute( + "SELECT DISTINCT embedding_model, dimension " + "FROM semantic_records" + ).fetchall() + ), + {("text-embedding-3-small", 1536)}, + ) + + provider = _QueryEmbeddingProvider() + library = await LocalKnowledgeLibrary.open( + _DATABASE_PATH, + embedding_model="text-embedding-3-small", + embedding_dimensions=1536, + _embedding_provider=provider, + ) + try: + hits = await library.search( + SemanticQuery( + text=( + "What evidence shows demand for AI infrastructure " + "is expanding?" + ), + source_kinds=["http", "arxiv"], + tags=["ai-infrastructure"], + available_at_before=datetime( + 2026, 1, 1, tzinfo=timezone.utc + ), + top_k=10, + ) + ) + self.assertEqual(len(hits), 6) + self.assertEqual( + {hit.item_type for hit in hits}, + {"news", "earnings", "paper"}, + ) + self.assertEqual(sum(hit.node_id is not None for hit in hits), 3) + for item in self.items: + self.assertEqual(await library.get(item.id), item) + self.assertEqual(len(provider.calls), 1) + finally: + await library.close() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/library/test_local.py b/tests/library/test_local.py new file mode 100644 index 0000000..6e1785b --- /dev/null +++ b/tests/library/test_local.py @@ -0,0 +1,693 @@ +import json +import sqlite3 +import tempfile +import unittest +from collections.abc import Sequence +from datetime import datetime, timezone +from pathlib import Path +from uuid import UUID, uuid4 + +from quantmind.knowledge import ( + Citation, + FlattenKnowledge, + News, + Paper, + SourceRef, + TreeNode, +) +from quantmind.library import LocalKnowledgeLibrary, SemanticQuery + + +class _FakeEmbeddingProvider: + def __init__( + self, + *, + dimension: int = 2, + vectors: dict[str, list[float]] | None = None, + ) -> None: + self.dimension = dimension + self.vectors = vectors or {} + self.calls: list[tuple[str, int | None, tuple[str, ...]]] = [] + self.closed = False + + async def embed( + self, + texts: Sequence[str], + *, + model: str, + dimensions: int | None, + ) -> list[list[float]]: + self.calls.append((model, dimensions, tuple(texts))) + output: list[list[float]] = [] + for text in texts: + configured = self.vectors.get(text) + if configured is not None: + output.append(configured) + continue + size = dimensions or self.dimension + seed = sum( + (index + 1) * ord(char) for index, char in enumerate(text) + ) + output.append( + [ + float(1 + ((seed >> (offset * 3)) % 17)) + for offset in range(size) + ] + ) + return output + + async def close(self) -> None: + self.closed = True + + +class _UnsupportedKnowledge(FlattenKnowledge): + item_type: str = "unsupported" + + def embedding_text(self) -> str: + return "unsupported canonical knowledge" + + +def _time(day: int, hour: int = 0) -> datetime: + return datetime(2026, 7, day, hour, tzinfo=timezone.utc) + + +def _news( + headline: str, + *, + item_id: UUID | None = None, + as_of: datetime | None = None, + available_at: datetime | None = None, + source_kind: str = "rss", + source_hash: str | None = None, + confidence: str = "medium", + tags: list[str] | None = None, + citations: list[Citation] | None = None, +) -> News: + return News( + id=item_id or uuid4(), + as_of=as_of or _time(1), + available_at=available_at, + source=SourceRef( + kind=source_kind, # type: ignore[arg-type] + uri=f"https://example.com/{headline}", + content_hash=source_hash, + ), + confidence=confidence, # type: ignore[arg-type] + citations=citations or [], + tags=tags or [], + headline=headline, + event_type="capital_expenditure", + entities=["Alpha Corp"], + timestamp=as_of or _time(1), + ) + + +def _paper() -> tuple[Paper, UUID, UUID]: + root_id = uuid4() + methods_id = uuid4() + results_id = uuid4() + root = TreeNode( + node_id=root_id, + title="Capital Spending Outlook", + summary="Company-wide investment outlook", + citations=[Citation(source_id="paper", page=1, node_id=root_id)], + children_ids=[methods_id, results_id], + ) + methods = TreeNode( + node_id=methods_id, + parent_id=root_id, + title="Methodology", + summary="Survey of management guidance", + citations=[Citation(source_id="paper", page=2, node_id=methods_id)], + ) + results = TreeNode( + node_id=results_id, + parent_id=root_id, + position=1, + title="Results", + summary="Capital expenditure is expected to rise", + ) + paper = Paper( + as_of=_time(1), + available_at=_time(2), + source=SourceRef(kind="arxiv", content_hash="paper-v1"), + confidence="high", + tags=["macro", "rates", "capex"], + citations=[Citation(source_id="paper", page=1)], + root_node_id=root_id, + nodes={root_id: root, methods_id: methods, results_id: results}, + arxiv_id="2607.00111", + ) + return paper, methods_id, results_id + + +class LocalKnowledgeLibraryTests(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self._temporary_directory = tempfile.TemporaryDirectory() + self.db_path = Path(self._temporary_directory.name) / "library.db" + + def tearDown(self) -> None: + self._temporary_directory.cleanup() + + async def test_flat_put_get_and_deterministic_best_first_search(self): + alpha = _news( + "Alpha expands capacity", + available_at=_time(2), + tags=["capex"], + citations=[Citation(source_id="rss:alpha", quote="capacity rises")], + ) + beta = _news("Beta pauses investment", available_at=_time(2)) + query_text = "management increases capital expenditure" + provider = _FakeEmbeddingProvider( + vectors={ + alpha.embedding_text(): [1.0, 0.0], + beta.embedding_text(): [0.0, 1.0], + query_text: [1.0, 0.0], + } + ) + library = await LocalKnowledgeLibrary.open( + self.db_path, + embedding_model="fake-2d", + embedding_dimensions=2, + _embedding_provider=provider, + ) + try: + await library.put(beta) + await library.put(alpha) + hits = await library.search(SemanticQuery(text=query_text, top_k=2)) + self.assertEqual([hit.item_id for hit in hits], [alpha.id, beta.id]) + self.assertAlmostEqual(hits[0].score, 1.0) + self.assertIsNone(hits[0].node_id) + self.assertEqual(hits[0].matched_text, alpha.embedding_text()) + self.assertEqual(hits[0].source, alpha.source) + self.assertEqual(hits[0].citations, alpha.citations) + self.assertEqual(hits[0].available_at, alpha.available_at) + self.assertEqual(await library.get(alpha.id), alpha) + finally: + await library.close() + + async def test_unsupported_knowledge_fails_before_embedding(self): + provider = _FakeEmbeddingProvider() + library = await LocalKnowledgeLibrary.open( + self.db_path, + embedding_model="fake-2d", + embedding_dimensions=2, + _embedding_provider=provider, + ) + item = _UnsupportedKnowledge( + as_of=_time(1), + source=SourceRef(kind="manual"), + ) + try: + with self.assertRaisesRegex( + TypeError, "Unsupported knowledge type" + ): + await library.put(item) + self.assertEqual(provider.calls, []) + finally: + await library.close() + + async def test_tree_root_and_non_root_nodes_use_exact_grain_and_filters( + self, + ): + paper, methods_id, results_id = _paper() + unrelated = _news("Unrelated event", tags=["macro", "rates"]) + query_text = "capital expenditure outlook" + provider = _FakeEmbeddingProvider( + vectors={ + paper.embedding_text(): [1.0, 0.0], + paper.nodes[methods_id].embedding_text(): [0.8, 0.2], + paper.nodes[results_id].embedding_text(): [0.9, 0.1], + unrelated.embedding_text(): [0.0, 1.0], + query_text: [1.0, 0.0], + } + ) + library = await LocalKnowledgeLibrary.open( + self.db_path, + embedding_model="fake-2d", + embedding_dimensions=2, + _embedding_provider=provider, + ) + try: + await library.put(unrelated) + await library.put(paper) + with sqlite3.connect(self.db_path) as db: + root_row = db.execute( + """ + SELECT item_shape, payload_json, node_count + FROM knowledge_items WHERE item_id = ? + """, + (str(paper.id),), + ).fetchone() + assert root_row is not None + self.assertEqual(root_row[0], "tree") + self.assertNotIn("nodes", json.loads(root_row[1])) + self.assertEqual(root_row[2], 3) + self.assertEqual( + db.execute( + "SELECT COUNT(*) FROM knowledge_nodes WHERE item_id = ?", + (str(paper.id),), + ).fetchone()[0], + 3, + ) + hits = await library.search( + SemanticQuery( + text=query_text, + item_types=["paper"], + source_kinds=["arxiv"], + confidence="high", + tags=["macro", "rates"], + tree_id=paper.id, + as_of_before=_time(1), + available_at_before=_time(2), + top_k=10, + ) + ) + self.assertEqual(len(hits), 3) + self.assertEqual(sum(hit.node_id is None for hit in hits), 1) + self.assertEqual( + {hit.node_id for hit in hits if hit.node_id is not None}, + {methods_id, results_id}, + ) + root_hit = next(hit for hit in hits if hit.node_id is None) + self.assertEqual(root_hit.matched_text, paper.embedding_text()) + self.assertEqual(root_hit.citations, paper.root().citations) + methods_hit = next(hit for hit in hits if hit.node_id == methods_id) + self.assertEqual( + methods_hit.matched_text, + paper.nodes[methods_id].embedding_text(), + ) + self.assertEqual( + methods_hit.citations, + paper.nodes[methods_id].citations, + ) + stored = await library.get(methods_hit.item_id) + self.assertIsInstance(stored, Paper) + assert isinstance(stored, Paper) + self.assertEqual( + stored.find_path(methods_id)[-1].node_id, methods_id + ) + finally: + await library.close() + + async def test_as_of_and_availability_cutoffs_are_distinct(self): + safe = _news( + "Known after cutoff", + as_of=_time(1), + available_at=_time(3), + ) + unknown = _news("Unknown availability", as_of=_time(1)) + later_information = _news( + "Later information available early", + as_of=_time(5), + available_at=_time(2), + ) + provider = _FakeEmbeddingProvider() + library = await LocalKnowledgeLibrary.open( + self.db_path, + embedding_model="fake-2d", + embedding_dimensions=2, + _embedding_provider=provider, + ) + try: + for item in (safe, unknown, later_information): + await library.put(item) + as_of_hits = await library.search( + SemanticQuery(text="cutoff", as_of_before=_time(2), top_k=10) + ) + self.assertEqual( + {hit.item_id for hit in as_of_hits}, + {safe.id, unknown.id}, + ) + availability_hits = await library.search( + SemanticQuery( + text="cutoff", + available_at_before=_time(2, 12), + top_k=10, + ) + ) + self.assertEqual( + {hit.item_id for hit in availability_hits}, + {later_information.id}, + ) + combined = await library.search( + SemanticQuery( + text="cutoff", + as_of_before=_time(2), + available_at_before=_time(4), + top_k=10, + ) + ) + self.assertEqual([hit.item_id for hit in combined], [safe.id]) + finally: + await library.close() + + async def test_reput_is_idempotent_and_invalidates_only_changed_metadata( + self, + ): + item = _news( + "Original headline", + source_hash="source-v1", + tags=["old"], + ) + provider = _FakeEmbeddingProvider() + library = await LocalKnowledgeLibrary.open( + self.db_path, + embedding_model="fake-2d", + embedding_dimensions=2, + _embedding_provider=provider, + ) + try: + await library.put(item) + self.assertEqual(len(provider.calls), 1) + await library.put(item) + self.assertEqual(len(provider.calls), 1) + + metadata_only = item.model_copy(update={"tags": ["new"]}) + await library.put(metadata_only) + self.assertEqual(len(provider.calls), 1) + + changed_text = metadata_only.model_copy( + update={"headline": "Changed headline"} + ) + await library.put(changed_text) + self.assertEqual( + provider.calls[-1][2], (changed_text.embedding_text(),) + ) + + changed_source = changed_text.model_copy( + update={ + "source": changed_text.source.model_copy( + update={"content_hash": "source-v2"} + ) + } + ) + await library.put(changed_source) + self.assertEqual(len(provider.calls), 3) + + changed_schema = changed_source.model_copy( + update={"schema_version": "1.1"} + ) + await library.put(changed_schema) + self.assertEqual(len(provider.calls), 4) + finally: + await library.close() + + async def test_tree_projection_change_invalidates_only_one_node(self): + paper, methods_id, _ = _paper() + provider = _FakeEmbeddingProvider() + library = await LocalKnowledgeLibrary.open( + self.db_path, + embedding_model="fake-2d", + embedding_dimensions=2, + _embedding_provider=provider, + ) + try: + await library.put(paper) + self.assertEqual(len(provider.calls[0][2]), 3) + changed_node = paper.nodes[methods_id].model_copy( + update={"summary": "A newly described survey method"} + ) + changed_nodes = dict(paper.nodes) + changed_nodes[methods_id] = changed_node + changed_paper = paper.model_copy(update={"nodes": changed_nodes}) + await library.put(changed_paper) + self.assertEqual( + provider.calls[-1][2], (changed_node.embedding_text(),) + ) + + schema_change = changed_paper.model_copy( + update={"schema_version": "1.1"} + ) + await library.put(schema_change) + self.assertEqual(len(provider.calls[-1][2]), 3) + finally: + await library.close() + + async def test_model_and_dimension_changes_have_explicit_stale_rebuild_path( + self, + ): + item = _news("Model metadata") + first = _FakeEmbeddingProvider(dimension=2) + library = await LocalKnowledgeLibrary.open( + self.db_path, + embedding_model="model-a", + embedding_dimensions=2, + _embedding_provider=first, + ) + await library.put(item) + await library.close() + + second = _FakeEmbeddingProvider(dimension=2) + library = await LocalKnowledgeLibrary.open( + self.db_path, + embedding_model="model-b", + embedding_dimensions=2, + _embedding_provider=second, + ) + with self.assertRaisesRegex(RuntimeError, "Stale index data.*model"): + await library.search(SemanticQuery(text="query")) + self.assertEqual(second.calls, []) + await library.put(item) + self.assertEqual(len(second.calls), 1) + await library.close() + + third = _FakeEmbeddingProvider(dimension=3) + library = await LocalKnowledgeLibrary.open( + self.db_path, + embedding_model="model-b", + embedding_dimensions=3, + _embedding_provider=third, + ) + try: + await library.put(item) + self.assertEqual(third.calls[0][1], 3) + hits = await library.search(SemanticQuery(text="query")) + self.assertEqual([hit.item_id for hit in hits], [item.id]) + finally: + await library.close() + + async def test_index_rebuilds_from_sqlite_without_reembedding_items(self): + item = _news("Persistent item", available_at=_time(2)) + first = _FakeEmbeddingProvider() + library = await LocalKnowledgeLibrary.open( + self.db_path, + embedding_model="fake-2d", + embedding_dimensions=2, + _embedding_provider=first, + ) + await library.put(item) + await library.close() + + second = _FakeEmbeddingProvider() + library = await LocalKnowledgeLibrary.open( + self.db_path, + embedding_model="fake-2d", + embedding_dimensions=2, + _embedding_provider=second, + ) + try: + self.assertEqual(await library.get(item.id), item) + self.assertEqual(second.calls, []) + hits = await library.search(SemanticQuery(text="persistent")) + self.assertEqual([hit.item_id for hit in hits], [item.id]) + self.assertEqual(second.calls[0][2], ("persistent",)) + finally: + await library.close() + + async def test_delete_removes_canonical_root_and_nodes_transactionally( + self, + ): + paper, _, _ = _paper() + provider = _FakeEmbeddingProvider() + library = await LocalKnowledgeLibrary.open( + self.db_path, + embedding_model="fake-2d", + embedding_dimensions=2, + _embedding_provider=provider, + ) + try: + await library.put(paper) + await library.delete(paper.id) + with self.assertRaises(KeyError): + await library.get(paper.id) + with self.assertRaises(KeyError): + await library.delete(paper.id) + calls_before_search = len(provider.calls) + self.assertEqual( + await library.search(SemanticQuery(text="query")), [] + ) + self.assertEqual(len(provider.calls), calls_before_search) + finally: + await library.close() + with sqlite3.connect(self.db_path) as db: + self.assertEqual( + db.execute("SELECT COUNT(*) FROM knowledge_items").fetchone()[ + 0 + ], + 0, + ) + self.assertEqual( + db.execute("SELECT COUNT(*) FROM semantic_records").fetchone()[ + 0 + ], + 0, + ) + self.assertEqual( + db.execute("SELECT COUNT(*) FROM knowledge_nodes").fetchone()[ + 0 + ], + 0, + ) + + async def test_stale_canonical_get_fails_but_delete_can_recover(self): + item = _news("Stale canonical") + provider = _FakeEmbeddingProvider() + library = await LocalKnowledgeLibrary.open( + self.db_path, + embedding_model="fake-2d", + embedding_dimensions=2, + _embedding_provider=provider, + ) + try: + await library.put(item) + with sqlite3.connect(self.db_path) as db: + db.execute( + "UPDATE knowledge_items SET payload_json = '{}' " + "WHERE item_id = ?", + (str(item.id),), + ) + with self.assertRaisesRegex( + RuntimeError, "Stale canonical knowledge" + ): + await library.get(item.id) + await library.delete(item.id) + with self.assertRaises(KeyError): + await library.get(item.id) + finally: + await library.close() + + async def test_orphan_and_missing_derived_data_are_reported_as_stale(self): + item = _news("Orphan target") + provider = _FakeEmbeddingProvider() + library = await LocalKnowledgeLibrary.open( + self.db_path, + embedding_model="fake-2d", + embedding_dimensions=2, + _embedding_provider=provider, + ) + await library.put(item) + await library.close() + with sqlite3.connect(self.db_path) as db: + db.execute("PRAGMA foreign_keys = OFF") + db.execute( + "DELETE FROM knowledge_items WHERE item_id = ?", (str(item.id),) + ) + + reopened_provider = _FakeEmbeddingProvider() + library = await LocalKnowledgeLibrary.open( + self.db_path, + embedding_model="fake-2d", + embedding_dimensions=2, + _embedding_provider=reopened_provider, + ) + try: + with self.assertRaisesRegex(RuntimeError, "Stale data"): + await library.get(item.id) + with self.assertRaisesRegex(RuntimeError, "Stale data"): + await library.delete(item.id) + with self.assertRaisesRegex(RuntimeError, "Stale index data"): + await library.search(SemanticQuery(text="query")) + finally: + await library.close() + + async def test_corrupt_canonical_tree_node_fails_rehydration(self): + paper, methods_id, _ = _paper() + library = await LocalKnowledgeLibrary.open( + self.db_path, + embedding_model="fake-2d", + embedding_dimensions=2, + _embedding_provider=_FakeEmbeddingProvider(), + ) + try: + await library.put(paper) + with sqlite3.connect(self.db_path) as db: + db.execute( + """ + UPDATE knowledge_nodes SET payload_json = '{}' + WHERE item_id = ? AND node_id = ? + """, + (str(paper.id), str(methods_id)), + ) + with self.assertRaisesRegex( + RuntimeError, "node.*content hash mismatch" + ): + await library.get(paper.id) + finally: + await library.close() + + async def test_corrupt_vector_and_query_dimension_mismatch_fail_clearly( + self, + ): + item = _news("Corrupt vector") + provider = _FakeEmbeddingProvider() + library = await LocalKnowledgeLibrary.open( + self.db_path, + embedding_model="fake-2d", + embedding_dimensions=2, + _embedding_provider=provider, + ) + await library.put(item) + await library.close() + with sqlite3.connect(self.db_path) as db: + db.execute( + "UPDATE semantic_records SET embedding = ? WHERE item_id = ?", + (b"bad", str(item.id)), + ) + + library = await LocalKnowledgeLibrary.open( + self.db_path, + embedding_model="fake-2d", + embedding_dimensions=2, + _embedding_provider=_FakeEmbeddingProvider(), + ) + try: + with self.assertRaisesRegex(RuntimeError, "Corrupt index data"): + await library.search(SemanticQuery(text="query")) + finally: + await library.close() + + mismatch_path = Path(self._temporary_directory.name) / "mismatch.db" + library = await LocalKnowledgeLibrary.open( + mismatch_path, + embedding_model="fake", + embedding_dimensions=2, + _embedding_provider=_FakeEmbeddingProvider(dimension=2), + ) + await library.put(item) + await library.close() + library = await LocalKnowledgeLibrary.open( + mismatch_path, + embedding_model="fake", + _embedding_provider=_FakeEmbeddingProvider(dimension=3), + ) + try: + with self.assertRaisesRegex(ValueError, "dimension mismatch"): + await library.search(SemanticQuery(text="query")) + finally: + await library.close() + + async def test_closed_library_rejects_operations(self): + library = await LocalKnowledgeLibrary.open( + self.db_path, + embedding_model="fake-2d", + _embedding_provider=_FakeEmbeddingProvider(), + ) + await library.close() + await library.close() + with self.assertRaisesRegex(RuntimeError, "closed"): + await library.get(uuid4()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/library/test_types.py b/tests/library/test_types.py new file mode 100644 index 0000000..2476fb0 --- /dev/null +++ b/tests/library/test_types.py @@ -0,0 +1,34 @@ +import unittest +from datetime import datetime + +from pydantic import ValidationError + +import quantmind.library as library +from quantmind.library import SemanticQuery + + +class PublicTypeTests(unittest.TestCase): + def test_package_exports_only_domain_retrieval_types(self): + self.assertEqual( + set(library.__all__), + {"LocalKnowledgeLibrary", "SemanticQuery", "SemanticHit"}, + ) + + def test_query_rejects_blank_text(self): + with self.assertRaises(ValidationError): + SemanticQuery(text=" ") + + def test_query_rejects_naive_financial_cutoff(self): + with self.assertRaisesRegex(ValidationError, "timezone-aware"): + SemanticQuery( + text="rates", + available_at_before=datetime(2026, 7, 16), + ) + + def test_query_rejects_non_positive_top_k(self): + with self.assertRaises(ValidationError): + SemanticQuery(text="rates", top_k=0) + + +if __name__ == "__main__": + unittest.main()