diff --git a/.gitignore b/.gitignore index 363455e06c0..0475aceee35 100644 --- a/.gitignore +++ b/.gitignore @@ -7,6 +7,9 @@ *.iws *.pyc *.pyo +.pytest_cache/ +.venv/ +*.egg-info/ .idea/ .idea_modules/ .settings diff --git a/Agent_module/QUICKSTART.md b/Agent_module/QUICKSTART.md index f58fbe03b1e..c59e441b651 100644 --- a/Agent_module/QUICKSTART.md +++ b/Agent_module/QUICKSTART.md @@ -1,10 +1,15 @@ # carbon_data Quickstart -`Agent_module.carbon_data` is the Agent Data Infra: one `.carbondata` file = -one self-contained knowledge base, simultaneously serving **RAG semantic -search / long-term memory / structured queries / knowledge-graph traversal**. +`Agent_module.carbon_data` is an Agent-native context store: one +`.carbondata` file persists context that multiple Agents can write, retrieve, +and hand off. It combines **semantic context retrieval / cross-agent memory / +structured state queries / context relationship traversal**. > Companion runnable demo: [`examples/carbondata_quickstart.py`](../examples/carbondata_quickstart.py) +> +> The Agent module stores SQLite data. It is not binary-compatible with +> classic Apache CarbonData columnar fact files, even though both currently +> use the `.carbondata` suffix. --- @@ -20,13 +25,16 @@ Optional dependencies are loaded on demand: | `hnswlib` | fast ANN for 10k+ vectors | optional (falls back to brute force when missing) | ```bash -pip install numpy -pip install hnswlib # optional +# Run from the repository root. +python3 -m pip install -e "./Agent_module[test]" +python3 -m pip install -e "./Agent_module[all]" # optional HNSW support too ``` A `.carbondata` file **is just a SQLite database**, so you can crack it open with any SQLite tool (`sqlite3 kb.carbondata`, DBeaver, DataGrip) for ad-hoc -debugging. +debugging. WAL mode can create temporary `-wal` and `-shm` sidecars while +the database is open, so copy or back up a live store using SQLite-aware +tools rather than copying only the primary file. --- @@ -193,7 +201,8 @@ acceleration whenever: - `mode="vector"` (including the vector leg of `mode="hybrid"`) - `metric="cosine"` - no JSON `filters=` are present -- `namespace=` is applied as a post-filter (with 3× over-sampling) +- `namespace=` is applied as an adaptive post-filter that expands until + `top_k` is satisfied or the full index has been considered Explicit override: ```python @@ -203,8 +212,8 @@ store.search(q, embedder=emb, use_hnsw="on") # force HNSW (raises if not elig ``` On the first search, HNSW is built from the entire `embedding` table and -persisted to a `vector_index` blob; later add/delete calls are detected via -a row-count comparison and trigger a lazy rebuild. +persisted to a `vector_index` blob. API mutations explicitly invalidate the +affected index; row-count checks also detect out-of-band adds and deletes. ### 4.3 Namespace isolation @@ -226,7 +235,9 @@ with store.transaction(): ``` `ingest_text` / `ingest_table` / `remember` are already wrapped in their own -transactions internally. +transactions internally. Nested `transaction()` calls use savepoints: an +inner exception rolls back the inner block, while an outer transaction may +continue if it catches that exception. ### 4.5 Admin & ops @@ -260,8 +271,7 @@ print(store.stats()) — each file covers one milestone end-to-end. - **Run the suite:** ```bash - cd Agent_module - .venv/bin/pytest tests/carbon_data/ -q + python3 -m pytest Agent_module/tests/carbon_data/ -q ``` --- @@ -272,8 +282,7 @@ print(store.stats()) Run the demo to see it all in action: ```bash -cd Agent_module -.venv/bin/python examples/carbondata_quickstart.py +python3 Agent_module/examples/carbondata_quickstart.py ``` For the full API surface, read the `class CarbonStore` docstring in diff --git a/Agent_module/README.md b/Agent_module/README.md index 9ff2bc8e765..957baa34b60 100644 --- a/Agent_module/README.md +++ b/Agent_module/README.md @@ -1,20 +1,31 @@ -# Agent Module — Agent Data Infra +# Agent Module — Agent-Native Context Data -`Agent_module` is a Python library that gives an LLM agent **a single -knowledge file** to read from. That file — with the `.carbondata` extension -— is a self-contained SQLite database that simultaneously serves the four -data-access patterns agents actually need: +`Agent_module` provides a data format and local store designed specifically +for Agent access. Its primary job is to persist and share context across +Agents — including memories, task artifacts, structured state, semantic +fragments, and the relationships among them — so another Agent can recover +the right context efficiently instead of replaying an entire conversation. + +The `.carbondata` context file is a self-contained SQLite database serving +four complementary access patterns: | Pattern | What an agent calls it for | |---|---| -| **RAG semantic search** | "Find chunks relevant to this question." | -| **Long-term memory** | "What have I learned about this user / session before?" | -| **Structured query** | "Give me all rows where `team = 'ml'`." | -| **Knowledge-graph traversal** | "Which documents does this one reference, two hops out?" | +| **Semantic context retrieval** | "Find the prior context relevant to this task." | +| **Cross-agent memory** | "What did another Agent learn or decide earlier?" | +| **Structured state query** | "Give me the state for this task/session/actor." | +| **Context relationship traversal** | "Which artifacts, decisions, and tasks led here?" | One file. One Python handle (`CarbonStore`). One schema. No separate vector DB, KV store, graph DB, or relational DB to wire together. +> **Format boundary:** the Agent module's `.carbondata` file is an +> application-identified SQLite database. It is not the same binary format +> as the classic Apache CarbonData columnar fact files that also use the +> `.carbondata` suffix, and the two are not interchangeable. Treat the Agent +> format as an experimental local-store format while its long-term naming +> and interoperability contract is defined. + ``` ┌─────────────────────────────┐ agent code ───► │ CarbonStore (kb.carbondata)│ @@ -43,7 +54,7 @@ Agent_module/ ├── QUICKSTART.md # full hands-on guide (read this next) ├── examples/ │ └── carbondata_quickstart.py # end-to-end runnable demo -└── tests/carbon_data/ # 249 tests across M1–M9 +└── tests/carbon_data/ # 250+ tests across M1–M9 ``` --- @@ -52,13 +63,15 @@ Agent_module/ - Python 3.10+ - `numpy` (required) -- `hnswlib` (optional — enables ANN acceleration when corpus > ~10k chunks; - falls back to brute force when missing) +- `hnswlib` (optional — enables ANN acceleration and is most useful for + larger corpora; falls back to brute force when missing) - `pytest` (optional — for running the test suite) +From the repository root, install the module and its test dependencies: + ```bash -pip install numpy -pip install hnswlib # optional but recommended +python3 -m pip install -e "./Agent_module[test]" +python3 -m pip install -e "./Agent_module[all]" # also install optional HNSW support ``` --- @@ -132,6 +145,11 @@ You now have a working `kb.carbondata` file on disk. Open it again in another script with `carbon_data.open("kb.carbondata")` — all your entities, embeddings, relations, and memories will still be there. +SQLite WAL mode may create temporary `-wal` and `-shm` sidecar files while +the store is open. A clean close normally checkpoints the database back to +the primary file; applications must not copy a live store without its WAL +state. + ### Step 3 — go deeper [**QUICKSTART.md**](QUICKSTART.md) covers each scenario in detail: @@ -147,7 +165,7 @@ For the full API surface, read the `class CarbonStore` docstring in ## Running the tests ```bash -pytest Agent_module/tests/ -v +python3 -m pytest Agent_module/tests/ -v ``` Tests are split by milestone (M1 schema → M9 admin); each file is a diff --git a/Agent_module/carbon_data/store.py b/Agent_module/carbon_data/store.py index 55cf4729abe..8f70ddfc502 100644 --- a/Agent_module/carbon_data/store.py +++ b/Agent_module/carbon_data/store.py @@ -110,7 +110,8 @@ def __init__( self._mode: Mode = mode self._tx_depth = 0 # 0 = autocommit-per-write; >0 = user transaction # Per-model HNSW index cache. Populated lazily on first vector - # search; invalidated whenever embedding count diverges. + # search; mutations explicitly invalidate affected models while + # count checks protect against out-of-band adds/deletes. self._hnsw_cache: dict[str, HnswIndex] = {} # ------------------------------------------------------------------ @@ -214,20 +215,33 @@ def transaction(self) -> Iterator[None]: """ Group writes atomically. - Nested calls are a no-op at the inner level: only the outermost - context commits/rolls back. On any exception, the outer frame - rolls back and re-raises. + Nested calls use SQLite savepoints. An exception rolls back the + current nesting level and is re-raised; if the caller catches an + inner exception, the outer transaction may continue safely. """ conn = self._require_rw() outer = self._tx_depth == 0 + savepoint = f"carbondata_tx_{self._tx_depth}" + if outer: + conn.execute("BEGIN") + else: + conn.execute(f"SAVEPOINT {savepoint}") self._tx_depth += 1 try: yield if outer: conn.commit() + else: + conn.execute(f"RELEASE SAVEPOINT {savepoint}") except Exception: if outer: conn.rollback() + else: + conn.execute(f"ROLLBACK TO SAVEPOINT {savepoint}") + conn.execute(f"RELEASE SAVEPOINT {savepoint}") + # A search inside the failed scope may have built an in-memory + # index from data that no longer exists after rollback. + self._hnsw_cache.clear() raise finally: self._tx_depth -= 1 @@ -336,9 +350,15 @@ def delete_entity(self, entity_id: str) -> bool: """Remove an entity and (via ON DELETE CASCADE) its chunks, embeddings, relations, and memory_ext row. Returns True iff something was deleted.""" + deleted = False with self._write_ctx() as conn: cur = conn.execute("DELETE FROM entity WHERE id=?", (entity_id,)) - return cur.rowcount > 0 + deleted = cur.rowcount > 0 + if deleted: + conn.execute("DELETE FROM vector_index") + if deleted: + self._hnsw_cache.clear() + return deleted # ------------------------------------------------------------------ # Chunk CRUD @@ -420,7 +440,12 @@ def delete_chunks(self, entity_id: str) -> int: """Delete all chunks for an entity. Returns the count removed.""" with self._write_ctx() as conn: cur = conn.execute("DELETE FROM chunk WHERE entity_id=?", (entity_id,)) - return cur.rowcount + deleted = cur.rowcount + if deleted: + conn.execute("DELETE FROM vector_index") + if deleted: + self._hnsw_cache.clear() + return deleted # ------------------------------------------------------------------ # Structured query @@ -505,6 +530,8 @@ def set_embedding( (chunk_id, model, dim, vector) VALUES (?, ?, ?, ?)""", (chunk_id, model, dim, blob), ) + conn.execute("DELETE FROM vector_index WHERE model=?", (model,)) + self._hnsw_cache.pop(model, None) def get_embedding(self, chunk_id: str, *, model: str) -> Optional[Any]: """Return the stored vector (numpy array) or None.""" @@ -585,6 +612,13 @@ def embed_chunks( payload, ) total += len(batch) + if total: + write_conn.execute( + "DELETE FROM vector_index WHERE model=?", + (embedder.model,), + ) + if total: + self._hnsw_cache.pop(embedder.model, None) return total # ------------------------------------------------------------------ @@ -631,6 +665,8 @@ def search( 'on' forces HNSW (errors if filters/non-cosine/no hnswlib); 'off' always uses brute force. """ + if top_k <= 0: + return [] if mode == "vector": ranked = self._run_vector( query, top_k=top_k, model=model, embedder=embedder, @@ -764,37 +800,46 @@ def _run_vector_hnsw( f"for model {model!r}" ) - # Over-fetch when post-filtering by namespace so we can survive - # candidates that get dropped. 3x is plenty for typical workloads. - oversample = 3 if namespace is not None else 1 - raw = idx.search(q_vec, top_k=top_k * oversample) - if not raw: - return [] - - rowids = [r[0] for r in raw] - placeholders = ",".join("?" * len(rowids)) - params: list[Any] = list(rowids) - sql = ( - f"SELECT chunk.rowid, chunk.id " - f"FROM chunk JOIN entity ON entity.id = chunk.entity_id " - f"WHERE chunk.rowid IN ({placeholders})" + conn = self._require_open() + fetch_k = min( + idx.count, + max(top_k, top_k * 3 if namespace is not None else top_k), ) - if namespace is not None: - sql += " AND entity.namespace = ?" - params.append(namespace) - conn = self._require_open() - rid_to_cid = {r[0]: r[1] for r in conn.execute(sql, params).fetchall()} + while fetch_k > 0: + raw = idx.search(q_vec, top_k=fetch_k) + if not raw: + return [] - out: list[tuple[str, float]] = [] - for rid, score in raw: - cid = rid_to_cid.get(rid) - if cid is None: - continue - out.append((cid, score)) - if len(out) >= top_k: - break - return out + rowids = [r[0] for r in raw] + placeholders = ",".join("?" * len(rowids)) + params: list[Any] = list(rowids) + sql = ( + f"SELECT chunk.rowid, chunk.id " + f"FROM chunk JOIN entity ON entity.id = chunk.entity_id " + f"WHERE chunk.rowid IN ({placeholders})" + ) + if namespace is not None: + sql += " AND entity.namespace = ?" + params.append(namespace) + + rid_to_cid = { + r[0]: r[1] for r in conn.execute(sql, params).fetchall() + } + out = [ + (rid_to_cid[rid], score) + for rid, score in raw + if rid in rid_to_cid + ] + if len(out) >= top_k or fetch_k >= idx.count: + return out[:top_k] + + # A sparse namespace may not occur in the first 3× candidates. + # Expand progressively until top_k is satisfied or the complete + # index has been considered. + fetch_k = min(idx.count, max(fetch_k * 2, top_k)) + + return [] # ---- HNSW cache + persistence ------------------------------------ @@ -810,8 +855,8 @@ def _get_hnsw(self, model: str) -> Optional[HnswIndex]: Return a usable HNSW for ``model``, building or reloading as needed. Cache invariant: ``HnswIndex.count`` equals the live embedding - count. Any divergence — adds, deletes, or first use after a - reopen — triggers a rebuild from the embedding table. + count. Adds/deletes are caught by the count check; in-process vector + replacements explicitly invalidate the affected model. """ cur_count = self._embedding_count(model) if cur_count == 0: @@ -1220,6 +1265,8 @@ def _delete_entities(self, ids: list[str]) -> int: conn.executemany( "DELETE FROM entity WHERE id=?", [(i,) for i in ids] ) + conn.execute("DELETE FROM vector_index") + self._hnsw_cache.clear() return len(ids) def _load_memory(self, entity_id: str) -> Optional[MemoryItem]: @@ -1617,13 +1664,13 @@ def ingest_table( def validate(self) -> ValidationReport: """ - Run consistency checks. Read-only; safe on a r-mode handle. + Run consistency checks. Safe on a read-only handle. Catches: - SQLite page-level corruption (PRAGMA integrity_check) - Foreign-key violations (PRAGMA foreign_key_check) - Bad application_id / unsupported schema_version - - chunk vs chunk_fts row-count divergence + - FTS5 index corruption (deep check on writable handles) - Stale per-model HNSW blob (count != live embedding count) """ conn = self._require_open() @@ -1660,16 +1707,30 @@ def validate(self) -> ValidationReport: f"SELECT COUNT(*) FROM {table}" ).fetchone()[0] - # FTS5 'integrity-check' compares index entries against the - # external content table (chunk). It surfaces missing index rows, - # stale tokens, and corruption — anything a row-count check - # cannot, since external-content COUNT(*) is served from chunk. - try: - conn.execute( - "INSERT INTO chunk_fts(chunk_fts) VALUES('integrity-check')" - ) - except sqlite3.DatabaseError as exc: - issues.append(f"chunk_fts integrity-check: {exc}") + if self._mode == "r": + # FTS5 exposes its deep integrity check as a special INSERT, + # which SQLite rejects on a read-only connection. Still verify + # that the virtual table is readable without producing a false + # corruption report for a healthy read-only store. + try: + conn.execute("SELECT rowid FROM chunk_fts LIMIT 1").fetchall() + except sqlite3.DatabaseError as exc: + issues.append(f"chunk_fts read-check: {exc}") + else: + # FTS5 'integrity-check' compares index entries against the + # external content table. The control command opens a SQLite + # transaction even though it does not change user data, so end + # that transaction when validate() started it. + was_in_transaction = conn.in_transaction + try: + conn.execute( + "INSERT INTO chunk_fts(chunk_fts) VALUES('integrity-check')" + ) + except sqlite3.DatabaseError as exc: + issues.append(f"chunk_fts integrity-check: {exc}") + finally: + if not was_in_transaction and conn.in_transaction: + conn.rollback() # Per-model HNSW staleness — only complain when a blob exists. idx_rows = conn.execute( diff --git a/Agent_module/pyproject.toml b/Agent_module/pyproject.toml new file mode 100644 index 00000000000..5940d6a5ae9 --- /dev/null +++ b/Agent_module/pyproject.toml @@ -0,0 +1,40 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "apache-carbondata-agent" +version = "0.1.0" +description = "Agent-native context data format and local store for cross-agent collaboration" +readme = "README.md" +requires-python = ">=3.10" +license = { text = "Apache-2.0" } +dependencies = [ + "numpy>=1.24", +] + +[project.optional-dependencies] +hnsw = [ + "hnswlib>=0.8.0", +] +test = [ + "pytest>=7.4", +] +all = [ + "hnswlib>=0.8.0", + "pytest>=7.4", +] + +[tool.setuptools] +packages = [ + "Agent_module", + "Agent_module.carbon_data", + "Agent_module.carbon_data.chunkers", + "Agent_module.carbon_data.embedders", + "Agent_module.carbon_data.index", +] +package-dir = { Agent_module = "." } + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "-ra" diff --git a/Agent_module/tests/carbon_data/test_m2_crud.py b/Agent_module/tests/carbon_data/test_m2_crud.py index d9ce8b5ea0a..82d0dd64731 100644 --- a/Agent_module/tests/carbon_data/test_m2_crud.py +++ b/Agent_module/tests/carbon_data/test_m2_crud.py @@ -309,6 +309,23 @@ def test_nested_inner_exception_rolls_back_all( assert store.get_entity("a") is None assert store.get_entity("b") is None + def test_caught_inner_exception_rolls_back_to_savepoint( + self, store: CarbonStore + ) -> None: + with store.transaction(): + store.put_entity(id="a", kind="doc") + try: + with store.transaction(): + store.put_entity(id="b", kind="doc") + raise RuntimeError("inner") + except RuntimeError: + pass + store.put_entity(id="c", kind="doc") + + assert store.get_entity("a") is not None + assert store.get_entity("b") is None + assert store.get_entity("c") is not None + def test_chunks_added_in_transaction_all_or_nothing( self, store: CarbonStore ) -> None: diff --git a/Agent_module/tests/carbon_data/test_m5_hnsw.py b/Agent_module/tests/carbon_data/test_m5_hnsw.py index f95788aa157..60bcb57030e 100644 --- a/Agent_module/tests/carbon_data/test_m5_hnsw.py +++ b/Agent_module/tests/carbon_data/test_m5_hnsw.py @@ -292,6 +292,63 @@ def test_delete_invalidates_cache(self, store: CarbonStore) -> None: model=emb.model, top_k=1) assert store._hnsw_cache[emb.model].count == 5 + def test_replace_embedding_invalidates_cache_and_blob( + self, store: CarbonStore + ) -> None: + emb, _, _ = _seed(store, n=5, dim=4, seed=13) + store.search(np.ones(4, dtype=np.float32), + model=emb.model, top_k=1) + assert emb.model in store._hnsw_cache + assert _vector_index_count(store) == 1 + + first_chunk = store.list_chunks("e-default")[0] + store.set_embedding( + first_chunk.id, + np.array([9.0, 0.0, 0.0, 0.0], dtype=np.float32), + model=emb.model, + ) + + assert emb.model not in store._hnsw_cache + assert _vector_index_count(store) == 0 + + def test_reembed_invalidates_cache_and_blob( + self, store: CarbonStore + ) -> None: + emb, _, _ = _seed(store, n=5, dim=4, seed=14) + store.search(np.ones(4, dtype=np.float32), + model=emb.model, top_k=1) + assert emb.model in store._hnsw_cache + assert _vector_index_count(store) == 1 + + store.embed_chunks(emb, missing_only=False) + + assert emb.model not in store._hnsw_cache + assert _vector_index_count(store) == 0 + + def test_rollback_clears_index_built_from_uncommitted_vectors( + self, store: CarbonStore + ) -> None: + emb, _, _ = _seed(store, n=5, dim=4, seed=15) + first_chunk = store.list_chunks("e-default")[0] + + with pytest.raises(RuntimeError): + with store.transaction(): + store.set_embedding( + first_chunk.id, + np.array([9.0, 0.0, 0.0, 0.0], dtype=np.float32), + model=emb.model, + ) + store.search( + np.ones(4, dtype=np.float32), + model=emb.model, + top_k=1, + use_hnsw="on", + ) + assert emb.model in store._hnsw_cache + raise RuntimeError("rollback") + + assert emb.model not in store._hnsw_cache + def test_full_cleanup_drops_cache_entry(self, store: CarbonStore) -> None: emb, _, _ = _seed(store, n=3, dim=4, seed=12) store.search(np.zeros(4, dtype=np.float32), @@ -329,3 +386,41 @@ def test_namespace_filter_applied(self, tmp_path: Path) -> None: assert all(h.entity.namespace == "ns-a" for h in hits) finally: s.close() + + def test_sparse_namespace_expands_until_top_k(self, tmp_path: Path) -> None: + p = tmp_path / "kb.carbondata" + s = create(p) + try: + s.put_entity(id="target", kind="document", namespace="target") + target_chunks = s.add_chunks( + "target", [f"target-{i}" for i in range(4)] + ) + s.put_entity(id="other", kind="document", namespace="other") + other_chunks = s.add_chunks( + "other", [f"other-{i}" for i in range(12)] + ) + for chunk_id in target_chunks: + s.set_embedding( + chunk_id, + np.array([-1.0, 0.0], dtype=np.float32), + model="sparse-ns", + ) + for chunk_id in other_chunks: + s.set_embedding( + chunk_id, + np.array([1.0, 0.0], dtype=np.float32), + model="sparse-ns", + ) + + hits = s.search( + np.array([1.0, 0.0], dtype=np.float32), + model="sparse-ns", + top_k=4, + namespace="target", + use_hnsw="on", + ) + + assert len(hits) == 4 + assert all(hit.entity.namespace == "target" for hit in hits) + finally: + s.close() diff --git a/Agent_module/tests/carbon_data/test_m9_admin.py b/Agent_module/tests/carbon_data/test_m9_admin.py index 1f894f24a69..33dae5266e9 100644 --- a/Agent_module/tests/carbon_data/test_m9_admin.py +++ b/Agent_module/tests/carbon_data/test_m9_admin.py @@ -64,6 +64,28 @@ def test_empty_db_passes(self, store: CarbonStore) -> None: assert r.ok is True assert r.counts["entity"] == 0 + def test_validate_does_not_leave_transaction_open( + self, store: CarbonStore + ) -> None: + _populate(store) + assert store._require_open().in_transaction is False + assert store.validate().ok is True + assert store._require_open().in_transaction is False + + def test_readonly_validate_passes(self, tmp_path: Path) -> None: + p = tmp_path / "kb.carbondata" + s = create(p) + _populate(s) + s.close() + + readonly = cd_open(p, mode="r") + try: + report = readonly.validate() + assert report.ok is True + assert report.issues == [] + finally: + readonly.close() + class TestValidateDetectsIssues: def test_detects_stale_hnsw_meta(self, tmp_path: Path) -> None: