feat(acms): add text, vector, and graph backend protocol implementations
CI / benchmark-publish (pull_request) Has been skipped
CI / lint (pull_request) Successful in 14s
CI / quality (pull_request) Successful in 17s
CI / build (pull_request) Successful in 15s
CI / typecheck (pull_request) Successful in 33s
CI / security (pull_request) Successful in 42s
CI / unit_tests (pull_request) Successful in 2m36s
CI / docker (pull_request) Successful in 40s
CI / integration_tests (pull_request) Successful in 3m17s
CI / coverage (pull_request) Successful in 4m21s
CI / lint (push) Successful in 12s
CI / build (push) Successful in 15s
CI / quality (push) Successful in 16s
CI / security (push) Successful in 30s
CI / typecheck (push) Successful in 33s
CI / benchmark-regression (push) Has been skipped
CI / unit_tests (push) Successful in 2m36s
CI / docker (push) Successful in 40s
CI / integration_tests (push) Successful in 3m57s
CI / coverage (push) Successful in 5m22s
CI / benchmark-publish (push) Has been cancelled
CI / benchmark-regression (pull_request) Successful in 25m12s

Implemented the Backend Abstraction Layer (BAL) for the Advanced Context
Management System, following the specification in docs/specification.md
Section ACMS > Backend Abstraction Layer and ADR-014.

Key additions:

- TextBackend protocol with search(query, scope, max_results) returning
  list[TextResult], and TextResult frozen dataclass (uko_uri, content,
  score, metadata fields)
- VectorBackend protocol with similarity_search(embedding, scope, top_k)
  returning list[VectorResult], and VectorResult frozen dataclass
- GraphBackend protocol with sparql_query(query, scope),
  get_triples(subject), and traverse(start, depth) methods returning
  GraphResult frozen dataclass (triples, metadata fields)
- In-memory stub backends (InMemoryTextBackend, InMemoryVectorBackend,
  InMemoryGraphBackend) that validate arguments and return empty results,
  serving as development placeholders and test doubles
- DI container registration as configurable Singletons with provider
  selection via override_providers()
- Behave BDD feature (35 scenarios / 83 steps) covering protocol
  compliance, argument validation, result immutability, and DI resolution
- Robot Framework smoke tests (6 tests) for integration verification
- ASV benchmarks for stub query overhead and instantiation time
- Reference documentation at docs/reference/acms_backends.md

Design decisions:
- Used @runtime_checkable Protocol for structural subtyping, consistent
  with existing ResourceHandler pattern
- Used frozen dataclasses (not Pydantic) for result types to minimize
  overhead in the hot path of context assembly
- scope parameter typed as frozenset[str] for immutability and hashability
- Stubs registered as default Singletons; production backends swap via DI

ISSUES CLOSED: #498
This commit was merged in pull request #519.
This commit is contained in:
2026-03-02 23:32:51 +00:00
committed by Forgejo
parent 279c6112ae
commit 025d379946
11 changed files with 1621 additions and 2 deletions
+147
View File
@@ -0,0 +1,147 @@
"""ASV benchmarks for ACMS Backend Abstraction Layer.
Measures the performance of:
- InMemoryTextBackend.search() stub query overhead
- InMemoryVectorBackend.similarity_search() stub query overhead
- InMemoryGraphBackend.sparql_query() / get_triples() / traverse() overhead
- Backend instantiation time
- Result dataclass construction overhead
"""
from __future__ import annotations
import importlib
import sys
from pathlib import Path
# Ensure the local *source* tree is importable even when ASV has an
# older build of the package installed.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
# Force-reload so ASV picks up the source tree version.
import cleveragents # noqa: E402
importlib.reload(cleveragents)
from cleveragents.domain.models.acms.backends import ( # noqa: E402
GraphResult,
TextResult,
VectorResult,
)
from cleveragents.domain.models.acms.stubs import ( # noqa: E402
InMemoryGraphBackend,
InMemoryTextBackend,
InMemoryVectorBackend,
)
# ---------------------------------------------------------------------------
# Backend instantiation benchmarks
# ---------------------------------------------------------------------------
class BackendInstantiationSuite:
"""Benchmark backend object creation overhead."""
timeout = 60
def time_create_text_backend(self) -> None:
InMemoryTextBackend()
def time_create_vector_backend(self) -> None:
InMemoryVectorBackend()
def time_create_graph_backend(self) -> None:
InMemoryGraphBackend()
# ---------------------------------------------------------------------------
# Stub query overhead benchmarks
# ---------------------------------------------------------------------------
class TextBackendQuerySuite:
"""Benchmark InMemoryTextBackend.search() stub overhead."""
timeout = 60
def setup(self) -> None:
self.backend = InMemoryTextBackend()
self.scope: frozenset[str] = frozenset({"RES01", "RES02"})
def time_search_simple(self) -> None:
self.backend.search("auth flow", scope=self.scope)
def time_search_with_max_results(self) -> None:
self.backend.search("auth flow", scope=self.scope, max_results=5)
def time_search_large_scope(self) -> None:
large_scope: frozenset[str] = frozenset(f"RES{i:04d}" for i in range(1000))
self.backend.search("query", scope=large_scope)
class VectorBackendQuerySuite:
"""Benchmark InMemoryVectorBackend.similarity_search() stub overhead."""
timeout = 60
def setup(self) -> None:
self.backend = InMemoryVectorBackend()
self.scope: frozenset[str] = frozenset({"RES01", "RES02"})
self.embedding: list[float] = [0.1] * 768
def time_similarity_search_simple(self) -> None:
self.backend.similarity_search(self.embedding, scope=self.scope)
def time_similarity_search_with_top_k(self) -> None:
self.backend.similarity_search(self.embedding, scope=self.scope, top_k=5)
class GraphBackendQuerySuite:
"""Benchmark InMemoryGraphBackend query and traversal stub overhead."""
timeout = 60
def setup(self) -> None:
self.backend = InMemoryGraphBackend()
self.scope: frozenset[str] = frozenset({"RES01"})
def time_sparql_query(self) -> None:
self.backend.sparql_query(
"SELECT ?s WHERE { ?s a uko:Container }", scope=self.scope
)
def time_get_triples(self) -> None:
self.backend.get_triples("uko-py:class/Auth")
def time_traverse(self) -> None:
self.backend.traverse("uko-py:class/Auth", depth=3)
# ---------------------------------------------------------------------------
# Result dataclass construction benchmarks
# ---------------------------------------------------------------------------
class ResultConstructionSuite:
"""Benchmark frozen dataclass construction overhead."""
timeout = 60
def time_text_result(self) -> None:
TextResult(uko_uri="uko:test", content="hello", score=0.5)
def time_vector_result(self) -> None:
VectorResult(uko_uri="uko:vec", content="embed", score=0.8)
def time_graph_result_empty(self) -> None:
GraphResult()
def time_graph_result_with_triples(self) -> None:
GraphResult(
triples=[
("uko:A", "uko:contains", "uko:B"),
("uko:B", "uko:references", "uko:C"),
]
)
+167
View File
@@ -0,0 +1,167 @@
# ACMS Backend Abstraction Layer
The Backend Abstraction Layer (BAL) provides a uniform interface to heterogeneous
data stores used by the Advanced Context Management System. Context strategies
query text, vector, and graph backends through protocol-defined APIs, enabling
backend implementations to be swapped without changing strategy code.
## Architecture
```
Strategy ──► ScopedView ──► Backend Protocol ──► Physical Store
└── resource_filter (scope)
```
Each backend type has a `Protocol` (structural subtyping) and a corresponding
frozen result `dataclass`. Strategies never instantiate backends directly —
they receive scoped views from the Context Assembly Pipeline via dependency
injection.
## Protocol Contracts
### TextBackend
Full-text search over indexed resource content.
| Method | Signature | Returns |
|--------|-----------|---------|
| `search` | `(query: str, *, scope: frozenset[str], max_results: int = 20)` | `list[TextResult]` |
**Preconditions:**
- `query` must be a non-empty string.
- `max_results` must be a positive integer.
**Postconditions:**
- Returns results ordered by descending `score`.
- Results are restricted to resources whose ULIDs appear in `scope`.
### VectorBackend
Embedding similarity search over chunked resource content.
| Method | Signature | Returns |
|--------|-----------|---------|
| `similarity_search` | `(embedding: list[float], *, scope: frozenset[str], top_k: int = 20)` | `list[VectorResult]` |
**Preconditions:**
- `embedding` must be a non-empty list of floats.
- `top_k` must be a positive integer.
**Postconditions:**
- Returns results ordered by descending `score` (cosine similarity).
- Results are restricted to resources whose ULIDs appear in `scope`.
### GraphBackend
Knowledge-graph queries and traversals over UKO triples.
| Method | Signature | Returns |
|--------|-----------|---------|
| `sparql_query` | `(query: str, *, scope: frozenset[str])` | `GraphResult` |
| `get_triples` | `(subject: str)` | `GraphResult` |
| `traverse` | `(start: str, *, depth: int = 2)` | `GraphResult` |
**Preconditions:**
- `query` and `subject` and `start` must be non-empty strings.
- `depth` must be a non-negative integer.
**Postconditions:**
- `sparql_query` restricts results to resources in `scope`.
- `traverse` returns triples discovered within `depth` hops of `start`.
## Result Types
All result types are frozen (immutable) dataclasses.
### TextResult
| Field | Type | Description |
|-------|------|-------------|
| `uko_uri` | `str` | UKO URI of the matching information unit (non-empty) |
| `content` | `str` | Matched text content |
| `score` | `float` | Relevance score, 0.0 -- 1.0 |
| `metadata` | `dict[str, str]` | Backend-specific metadata (default: empty) |
### VectorResult
| Field | Type | Description |
|-------|------|-------------|
| `uko_uri` | `str` | UKO URI of the matching information unit (non-empty) |
| `content` | `str` | Text content of matched chunk |
| `score` | `float` | Cosine similarity, 0.0 -- 1.0 |
| `metadata` | `dict[str, str]` | Backend-specific metadata (default: empty) |
### GraphResult
| Field | Type | Description |
|-------|------|-------------|
| `triples` | `list[tuple[str, str, str]]` | (subject, predicate, object) triples |
| `metadata` | `dict[str, str]` | Backend-specific metadata (default: empty) |
## Extension Points
### Implementing a Custom Backend
To add a new backend (e.g., Tantivy for text search):
1. Create a class that satisfies the corresponding `Protocol` (e.g., `TextBackend`).
2. Implement all protocol methods with proper argument validation.
3. Register the backend in the DI container by overriding the provider:
```python
from cleveragents.application.container import override_providers
override_providers(text_backend=TantivyTextBackend(index_path="/data/tantivy"))
```
### In-Memory Stubs
The following stubs are provided for development and testing:
| Stub | Protocol | Behaviour |
|------|----------|-----------|
| `InMemoryTextBackend` | `TextBackend` | Returns empty list; validates args |
| `InMemoryVectorBackend` | `VectorBackend` | Returns empty list; validates args |
| `InMemoryGraphBackend` | `GraphBackend` | Returns empty `GraphResult`; validates args |
Stubs are registered as the default providers in the DI container. They can
be overridden at runtime for production or per-test via `override_providers()`.
### DI Container Registration
Backends are registered as singletons in the DI container
(`cleveragents.application.container.Container`):
```python
text_backend = providers.Singleton(InMemoryTextBackend)
vector_backend = providers.Singleton(InMemoryVectorBackend)
graph_backend = providers.Singleton(InMemoryGraphBackend)
```
To swap to a production backend, override the provider before first use:
```python
container = get_container()
container.text_backend.override(providers.Singleton(TantivyTextBackend))
```
## Module Locations
| Module | Purpose |
|--------|---------|
| `cleveragents.domain.models.acms.backends` | Protocol definitions and result dataclasses |
| `cleveragents.domain.models.acms.stubs` | In-memory stub implementations |
| `cleveragents.application.container` | DI container backend registration |
## Related
- [CRP Reference](crp.md) — Context Request Protocol types
- [Specification: ACMS > Backend Abstraction Layer](../specification.md) — Authoritative design
- [ADR-014: Context Management](../adr/ADR-014-context-management-acms.md) — Architecture decision record
+171
View File
@@ -0,0 +1,171 @@
Feature: ACMS Backend Abstraction Layer
As a developer
I want text, vector, and graph backend protocols with in-memory stubs
So that context strategies can query heterogeneous data stores uniformly
# ---- TextResult ----
Scenario: Create a valid TextResult
Given a TextResult with uko_uri "uko-py:class/Auth" and score 0.9
Then the text result uko_uri should be "uko-py:class/Auth"
And the text result content should be "class Auth: pass"
And the text result score should be 0.9
And the text result metadata should be empty
Scenario: TextResult rejects empty uko_uri
Then creating a TextResult with empty uko_uri should raise ValueError
Scenario: TextResult rejects score above 1.0
Then creating a TextResult with score 1.5 should raise ValueError
Scenario: TextResult rejects score below 0.0
Then creating a TextResult with score -0.1 should raise ValueError
Scenario: TextResult is immutable
Given a TextResult with uko_uri "uko-py:func/login" and score 0.5
Then modifying the text result score should raise an error
# ---- VectorResult ----
Scenario: Create a valid VectorResult
Given a VectorResult with uko_uri "uko-py:class/VecStore" and score 0.85
Then the vector result uko_uri should be "uko-py:class/VecStore"
And the vector result content should be "vector store content"
And the vector result score should be 0.85
And the vector result metadata should be empty
Scenario: VectorResult rejects empty uko_uri
Then creating a VectorResult with empty uko_uri should raise ValueError
Scenario: VectorResult rejects score above 1.0
Then creating a VectorResult with score 1.5 should raise ValueError
Scenario: VectorResult rejects score below 0.0
Then creating a VectorResult with score -0.1 should raise ValueError
Scenario: VectorResult is immutable
Given a VectorResult with uko_uri "uko-py:func/embed" and score 0.7
Then modifying the vector result score should raise an error
# ---- GraphResult ----
Scenario: Create a valid GraphResult with no triples
Given a GraphResult with no triples
Then the graph result triples should be empty
And the graph result metadata should be empty
Scenario: Create a GraphResult with triples
Given a GraphResult with triple "uko:A" "uko:contains" "uko:B"
Then the graph result should have 1 triple
Scenario: GraphResult is immutable
Given a GraphResult with no triples
Then modifying the graph result metadata should raise an error
# ---- InMemoryTextBackend ----
Scenario: InMemoryTextBackend satisfies TextBackend protocol
Given an InMemoryTextBackend instance
Then it should be a TextBackend
Scenario: InMemoryTextBackend search returns empty list
Given an InMemoryTextBackend instance
When I search for "auth flow" with scope "RES01"
Then the text search results should be empty
Scenario: InMemoryTextBackend search with max_results
Given an InMemoryTextBackend instance
When I search for "auth" with scope "RES01" and max_results 5
Then the text search results should be empty
Scenario: InMemoryTextBackend rejects empty query
Given an InMemoryTextBackend instance
Then searching with empty query should raise ValueError
Scenario: InMemoryTextBackend rejects non-positive max_results
Given an InMemoryTextBackend instance
Then searching with max_results 0 should raise ValueError
# ---- InMemoryVectorBackend ----
Scenario: InMemoryVectorBackend satisfies VectorBackend protocol
Given an InMemoryVectorBackend instance
Then it should be a VectorBackend
Scenario: InMemoryVectorBackend search returns empty list
Given an InMemoryVectorBackend instance
When I similarity search with embedding and scope "RES01"
Then the vector search results should be empty
Scenario: InMemoryVectorBackend search with top_k
Given an InMemoryVectorBackend instance
When I similarity search with embedding and scope "RES01" and top_k 5
Then the vector search results should be empty
Scenario: InMemoryVectorBackend rejects empty embedding
Given an InMemoryVectorBackend instance
Then similarity searching with empty embedding should raise ValueError
Scenario: InMemoryVectorBackend rejects non-positive top_k
Given an InMemoryVectorBackend instance
Then similarity searching with top_k 0 should raise ValueError
# ---- InMemoryGraphBackend ----
Scenario: InMemoryGraphBackend satisfies GraphBackend protocol
Given an InMemoryGraphBackend instance
Then it should be a GraphBackend
Scenario: InMemoryGraphBackend sparql_query returns empty result
Given an InMemoryGraphBackend instance
When I run sparql query "SELECT ?s WHERE { ?s a uko:Container }" with scope "RES01"
Then the graph query result should have no triples
Scenario: InMemoryGraphBackend sparql_query rejects empty query
Given an InMemoryGraphBackend instance
Then sparql querying with empty query should raise ValueError
Scenario: InMemoryGraphBackend get_triples returns empty result
Given an InMemoryGraphBackend instance
When I get triples for subject "uko-py:class/Auth"
Then the graph query result should have no triples
Scenario: InMemoryGraphBackend get_triples rejects empty subject
Given an InMemoryGraphBackend instance
Then getting triples with empty subject should raise ValueError
Scenario: InMemoryGraphBackend traverse returns empty result
Given an InMemoryGraphBackend instance
When I traverse from "uko-py:class/Auth" with depth 3
Then the graph query result should have no triples
Scenario: InMemoryGraphBackend traverse rejects empty start
Given an InMemoryGraphBackend instance
Then traversing with empty start should raise ValueError
Scenario: InMemoryGraphBackend traverse rejects negative depth
Given an InMemoryGraphBackend instance
Then traversing with depth -1 should raise ValueError
# ---- DI Container Registration ----
Scenario: DI container provides a TextBackend
Given the DI container
Then the container should provide a text_backend
And the text_backend should be a TextBackend
Scenario: DI container provides a VectorBackend
Given the DI container
Then the container should provide a vector_backend
And the vector_backend should be a VectorBackend
Scenario: DI container provides a GraphBackend
Given the DI container
Then the container should provide a graph_backend
And the graph_backend should be a GraphBackend
Scenario: DI container backends are singletons
Given the DI container
Then resolving text_backend twice should return the same instance
And resolving vector_backend twice should return the same instance
And resolving graph_backend twice should return the same instance
+426
View File
@@ -0,0 +1,426 @@
"""Step definitions for the ACMS Backend Abstraction Layer feature."""
from __future__ import annotations
from dataclasses import FrozenInstanceError
from typing import Any
from behave import given, then, when
from cleveragents.application.container import get_container, reset_container
from cleveragents.domain.models.acms.backends import (
GraphBackend,
GraphResult,
TextBackend,
TextResult,
VectorBackend,
VectorResult,
)
from cleveragents.domain.models.acms.stubs import (
InMemoryGraphBackend,
InMemoryTextBackend,
InMemoryVectorBackend,
)
__all__: list[str] = []
# ---------------------------------------------------------------------------
# TextResult steps
# ---------------------------------------------------------------------------
@given('a TextResult with uko_uri "{uko_uri}" and score {score:g}')
def step_given_text_result(context: Any, uko_uri: str, score: float) -> None:
context.text_result = TextResult(
uko_uri=uko_uri,
content="class Auth: pass",
score=score,
)
@then('the text result uko_uri should be "{expected}"')
def step_then_text_result_uko_uri(context: Any, expected: str) -> None:
assert context.text_result.uko_uri == expected
@then('the text result content should be "{expected}"')
def step_then_text_result_content(context: Any, expected: str) -> None:
assert context.text_result.content == expected
@then("the text result score should be {expected:g}")
def step_then_text_result_score(context: Any, expected: float) -> None:
assert context.text_result.score == expected
@then("the text result metadata should be empty")
def step_then_text_result_metadata_empty(context: Any) -> None:
assert context.text_result.metadata == {}
@then("creating a TextResult with empty uko_uri should raise ValueError")
def step_then_text_result_empty_uko_uri(context: Any) -> None:
try:
TextResult(uko_uri="", content="x", score=0.5)
raise AssertionError("Expected ValueError")
except ValueError:
pass
@then("creating a TextResult with score {score:g} should raise ValueError")
def step_then_text_result_bad_score(context: Any, score: float) -> None:
try:
TextResult(uko_uri="uko:x", content="x", score=score)
raise AssertionError("Expected ValueError")
except ValueError:
pass
@then("modifying the text result score should raise an error")
def step_then_text_result_immutable(context: Any) -> None:
try:
context.text_result.score = 0.1 # type: ignore[misc]
raise AssertionError("Expected FrozenInstanceError")
except FrozenInstanceError:
pass
# ---------------------------------------------------------------------------
# VectorResult steps
# ---------------------------------------------------------------------------
@given('a VectorResult with uko_uri "{uko_uri}" and score {score:g}')
def step_given_vector_result(context: Any, uko_uri: str, score: float) -> None:
context.vector_result = VectorResult(
uko_uri=uko_uri,
content="vector store content",
score=score,
)
@then('the vector result uko_uri should be "{expected}"')
def step_then_vector_result_uko_uri(context: Any, expected: str) -> None:
assert context.vector_result.uko_uri == expected
@then('the vector result content should be "{expected}"')
def step_then_vector_result_content(context: Any, expected: str) -> None:
assert context.vector_result.content == expected
@then("the vector result score should be {expected:g}")
def step_then_vector_result_score(context: Any, expected: float) -> None:
assert context.vector_result.score == expected
@then("the vector result metadata should be empty")
def step_then_vector_result_metadata_empty(context: Any) -> None:
assert context.vector_result.metadata == {}
@then("creating a VectorResult with empty uko_uri should raise ValueError")
def step_then_vector_result_empty_uko_uri(context: Any) -> None:
try:
VectorResult(uko_uri="", content="x", score=0.5)
raise AssertionError("Expected ValueError")
except ValueError:
pass
@then("creating a VectorResult with score {score:g} should raise ValueError")
def step_then_vector_result_bad_score(context: Any, score: float) -> None:
try:
VectorResult(uko_uri="uko:x", content="x", score=score)
raise AssertionError("Expected ValueError")
except ValueError:
pass
@then("modifying the vector result score should raise an error")
def step_then_vector_result_immutable(context: Any) -> None:
try:
context.vector_result.score = 0.1 # type: ignore[misc]
raise AssertionError("Expected FrozenInstanceError")
except FrozenInstanceError:
pass
# ---------------------------------------------------------------------------
# GraphResult steps
# ---------------------------------------------------------------------------
@given("a GraphResult with no triples")
def step_given_graph_result_empty(context: Any) -> None:
context.graph_result = GraphResult()
@given('a GraphResult with triple "{s}" "{p}" "{o}"')
def step_given_graph_result_triple(context: Any, s: str, p: str, o: str) -> None:
context.graph_result = GraphResult(triples=[(s, p, o)])
@then("the graph result triples should be empty")
def step_then_graph_result_triples_empty(context: Any) -> None:
assert context.graph_result.triples == []
@then("the graph result metadata should be empty")
def step_then_graph_result_metadata_empty(context: Any) -> None:
assert context.graph_result.metadata == {}
@then("the graph result should have {count:d} triple")
def step_then_graph_result_triple_count(context: Any, count: int) -> None:
assert len(context.graph_result.triples) == count
@then("modifying the graph result metadata should raise an error")
def step_then_graph_result_immutable(context: Any) -> None:
try:
context.graph_result.metadata = {"x": "y"} # type: ignore[misc]
raise AssertionError("Expected FrozenInstanceError")
except FrozenInstanceError:
pass
# ---------------------------------------------------------------------------
# InMemoryTextBackend steps
# ---------------------------------------------------------------------------
@given("an InMemoryTextBackend instance")
def step_given_text_backend(context: Any) -> None:
context.backend = InMemoryTextBackend()
@then("it should be a TextBackend")
def step_then_is_text_backend(context: Any) -> None:
assert isinstance(context.backend, TextBackend)
@when('I search for "{query}" with scope "{scope}"')
def step_when_text_search(context: Any, query: str, scope: str) -> None:
context.search_results = context.backend.search(query, scope=frozenset({scope}))
@when('I search for "{query}" with scope "{scope}" and max_results {n:d}')
def step_when_text_search_max(context: Any, query: str, scope: str, n: int) -> None:
context.search_results = context.backend.search(
query, scope=frozenset({scope}), max_results=n
)
@then("the text search results should be empty")
def step_then_text_search_empty(context: Any) -> None:
assert context.search_results == []
@then("searching with empty query should raise ValueError")
def step_then_text_search_empty_query(context: Any) -> None:
try:
context.backend.search("", scope=frozenset({"RES01"}))
raise AssertionError("Expected ValueError")
except ValueError:
pass
@then("searching with max_results {n:d} should raise ValueError")
def step_then_text_search_bad_max(context: Any, n: int) -> None:
try:
context.backend.search("q", scope=frozenset({"RES01"}), max_results=n)
raise AssertionError("Expected ValueError")
except ValueError:
pass
# ---------------------------------------------------------------------------
# InMemoryVectorBackend steps
# ---------------------------------------------------------------------------
@given("an InMemoryVectorBackend instance")
def step_given_vector_backend(context: Any) -> None:
context.backend = InMemoryVectorBackend()
@then("it should be a VectorBackend")
def step_then_is_vector_backend(context: Any) -> None:
assert isinstance(context.backend, VectorBackend)
@when('I similarity search with embedding and scope "{scope}"')
def step_when_vector_search(context: Any, scope: str) -> None:
context.search_results = context.backend.similarity_search(
[0.1, 0.2, 0.3], scope=frozenset({scope})
)
@when('I similarity search with embedding and scope "{scope}" and top_k {k:d}')
def step_when_vector_search_topk(context: Any, scope: str, k: int) -> None:
context.search_results = context.backend.similarity_search(
[0.1, 0.2, 0.3], scope=frozenset({scope}), top_k=k
)
@then("the vector search results should be empty")
def step_then_vector_search_empty(context: Any) -> None:
assert context.search_results == []
@then("similarity searching with empty embedding should raise ValueError")
def step_then_vector_search_empty_embed(context: Any) -> None:
try:
context.backend.similarity_search([], scope=frozenset({"RES01"}))
raise AssertionError("Expected ValueError")
except ValueError:
pass
@then("similarity searching with top_k {k:d} should raise ValueError")
def step_then_vector_search_bad_topk(context: Any, k: int) -> None:
try:
context.backend.similarity_search([0.1], scope=frozenset({"RES01"}), top_k=k)
raise AssertionError("Expected ValueError")
except ValueError:
pass
# ---------------------------------------------------------------------------
# InMemoryGraphBackend steps
# ---------------------------------------------------------------------------
@given("an InMemoryGraphBackend instance")
def step_given_graph_backend(context: Any) -> None:
context.backend = InMemoryGraphBackend()
@then("it should be a GraphBackend")
def step_then_is_graph_backend(context: Any) -> None:
assert isinstance(context.backend, GraphBackend)
@when('I run sparql query "{query}" with scope "{scope}"')
def step_when_sparql_query(context: Any, query: str, scope: str) -> None:
context.graph_query_result = context.backend.sparql_query(
query, scope=frozenset({scope})
)
@then("the graph query result should have no triples")
def step_then_graph_query_empty(context: Any) -> None:
assert context.graph_query_result.triples == []
@then("sparql querying with empty query should raise ValueError")
def step_then_sparql_empty_query(context: Any) -> None:
try:
context.backend.sparql_query("", scope=frozenset({"RES01"}))
raise AssertionError("Expected ValueError")
except ValueError:
pass
@when('I get triples for subject "{subject}"')
def step_when_get_triples(context: Any, subject: str) -> None:
context.graph_query_result = context.backend.get_triples(subject)
@then("getting triples with empty subject should raise ValueError")
def step_then_get_triples_empty_subject(context: Any) -> None:
try:
context.backend.get_triples("")
raise AssertionError("Expected ValueError")
except ValueError:
pass
@when('I traverse from "{start}" with depth {depth:d}')
def step_when_traverse(context: Any, start: str, depth: int) -> None:
context.graph_query_result = context.backend.traverse(start, depth=depth)
@then("traversing with empty start should raise ValueError")
def step_then_traverse_empty_start(context: Any) -> None:
try:
context.backend.traverse("")
raise AssertionError("Expected ValueError")
except ValueError:
pass
@then("traversing with depth {depth:d} should raise ValueError")
def step_then_traverse_bad_depth(context: Any, depth: int) -> None:
try:
context.backend.traverse("uko:x", depth=depth)
raise AssertionError("Expected ValueError")
except ValueError:
pass
# ---------------------------------------------------------------------------
# DI Container steps
# ---------------------------------------------------------------------------
@given("the DI container")
def step_given_container(context: Any) -> None:
reset_container()
context.container = get_container()
@then("the container should provide a text_backend")
def step_then_container_text(context: Any) -> None:
context.resolved_text = context.container.text_backend()
assert context.resolved_text is not None
@then("the text_backend should be a TextBackend")
def step_then_text_is_protocol(context: Any) -> None:
assert isinstance(context.resolved_text, TextBackend)
@then("the container should provide a vector_backend")
def step_then_container_vector(context: Any) -> None:
context.resolved_vector = context.container.vector_backend()
assert context.resolved_vector is not None
@then("the vector_backend should be a VectorBackend")
def step_then_vector_is_protocol(context: Any) -> None:
assert isinstance(context.resolved_vector, VectorBackend)
@then("the container should provide a graph_backend")
def step_then_container_graph(context: Any) -> None:
context.resolved_graph = context.container.graph_backend()
assert context.resolved_graph is not None
@then("the graph_backend should be a GraphBackend")
def step_then_graph_is_protocol(context: Any) -> None:
assert isinstance(context.resolved_graph, GraphBackend)
@then("resolving text_backend twice should return the same instance")
def step_then_text_singleton(context: Any) -> None:
a = context.container.text_backend()
b = context.container.text_backend()
assert a is b
@then("resolving vector_backend twice should return the same instance")
def step_then_vector_singleton(context: Any) -> None:
a = context.container.vector_backend()
b = context.container.vector_backend()
assert a is b
@then("resolving graph_backend twice should return the same instance")
def step_then_graph_singleton(context: Any) -> None:
a = context.container.graph_backend()
b = context.container.graph_backend()
assert a is b
+57
View File
@@ -0,0 +1,57 @@
*** Settings ***
Documentation Smoke tests for ACMS Backend Abstraction Layer protocols
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_acms_backends.py
*** Test Cases ***
InMemoryTextBackend Protocol Compliance
[Documentation] Verify InMemoryTextBackend satisfies TextBackend protocol
${result}= Run Process ${PYTHON} ${HELPER} text-backend cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} acms-text-backend-ok
InMemoryVectorBackend Protocol Compliance
[Documentation] Verify InMemoryVectorBackend satisfies VectorBackend protocol
${result}= Run Process ${PYTHON} ${HELPER} vector-backend cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} acms-vector-backend-ok
InMemoryGraphBackend Protocol Compliance
[Documentation] Verify InMemoryGraphBackend satisfies GraphBackend protocol
${result}= Run Process ${PYTHON} ${HELPER} graph-backend cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} acms-graph-backend-ok
Result Dataclass Construction
[Documentation] Verify TextResult, VectorResult, and GraphResult creation
${result}= Run Process ${PYTHON} ${HELPER} result-types cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} acms-result-types-ok
DI Container Backend Resolution
[Documentation] Verify DI container resolves all three backends
${result}= Run Process ${PYTHON} ${HELPER} di-resolution cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} acms-di-resolution-ok
Backend Argument Validation
[Documentation] Verify backends reject invalid arguments
${result}= Run Process ${PYTHON} ${HELPER} validation cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} acms-validation-ok
+159
View File
@@ -0,0 +1,159 @@
"""Robot Framework helper for ACMS backend protocol smoke tests.
Provides a CLI-style interface for Robot to invoke backend creation,
protocol compliance, and DI resolution. Exit code 0 = success, 1 = failure.
Usage:
python robot/helper_acms_backends.py <command>
"""
from __future__ import annotations
import sys
from pathlib import Path
# Ensure the src directory is on the import path.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from cleveragents.application.container import ( # noqa: E402
get_container,
reset_container,
)
from cleveragents.domain.models.acms.backends import ( # noqa: E402
GraphBackend,
GraphResult,
TextBackend,
TextResult,
VectorBackend,
VectorResult,
)
from cleveragents.domain.models.acms.stubs import ( # noqa: E402
InMemoryGraphBackend,
InMemoryTextBackend,
InMemoryVectorBackend,
)
def main() -> int:
"""Entry point called by Robot Framework ``Run Process``."""
if len(sys.argv) < 2:
print("Usage: helper_acms_backends.py <command>")
return 1
command: str = sys.argv[1]
if command == "text-backend":
try:
backend = InMemoryTextBackend()
assert isinstance(backend, TextBackend)
results = backend.search("test", scope=frozenset({"RES01"}))
assert results == []
print("acms-text-backend-ok")
return 0
except Exception as exc:
print(f"acms-text-backend-fail: {exc}")
return 1
if command == "vector-backend":
try:
backend = InMemoryVectorBackend()
assert isinstance(backend, VectorBackend)
results = backend.similarity_search([0.1, 0.2], scope=frozenset({"RES01"}))
assert results == []
print("acms-vector-backend-ok")
return 0
except Exception as exc:
print(f"acms-vector-backend-fail: {exc}")
return 1
if command == "graph-backend":
try:
backend = InMemoryGraphBackend()
assert isinstance(backend, GraphBackend)
result = backend.sparql_query(
"SELECT ?s WHERE { ?s a uko:Container }",
scope=frozenset({"RES01"}),
)
assert result.triples == []
result2 = backend.get_triples("uko:subject")
assert result2.triples == []
result3 = backend.traverse("uko:start", depth=2)
assert result3.triples == []
print("acms-graph-backend-ok")
return 0
except Exception as exc:
print(f"acms-graph-backend-fail: {exc}")
return 1
if command == "result-types":
try:
tr = TextResult(uko_uri="uko:test", content="hello", score=0.5)
assert tr.uko_uri == "uko:test"
vr = VectorResult(uko_uri="uko:vec", content="embed", score=0.8)
assert vr.uko_uri == "uko:vec"
gr = GraphResult(triples=[("a", "b", "c")])
assert len(gr.triples) == 1
print("acms-result-types-ok")
return 0
except Exception as exc:
print(f"acms-result-types-fail: {exc}")
return 1
if command == "di-resolution":
try:
reset_container()
container = get_container()
tb = container.text_backend()
assert isinstance(tb, TextBackend)
vb = container.vector_backend()
assert isinstance(vb, VectorBackend)
gb = container.graph_backend()
assert isinstance(gb, GraphBackend)
print("acms-di-resolution-ok")
return 0
except Exception as exc:
print(f"acms-di-resolution-fail: {exc}")
return 1
if command == "validation":
try:
# Empty query should raise
backend = InMemoryTextBackend()
raised = False
try:
backend.search("", scope=frozenset({"RES01"}))
except ValueError:
raised = True
assert raised, "Expected ValueError for empty query"
# Empty embedding should raise
vb = InMemoryVectorBackend()
raised = False
try:
vb.similarity_search([], scope=frozenset({"RES01"}))
except ValueError:
raised = True
assert raised, "Expected ValueError for empty embedding"
# Invalid score should raise
raised = False
try:
TextResult(uko_uri="uko:x", content="x", score=2.0)
except ValueError:
raised = True
assert raised, "Expected ValueError for score > 1.0"
print("acms-validation-ok")
return 0
except Exception as exc:
print(f"acms-validation-fail: {exc}")
return 1
print(f"Unknown command: {command}")
return 1
if __name__ == "__main__":
sys.exit(main())
+12
View File
@@ -28,6 +28,11 @@ from cleveragents.application.services.resource_registry_service import (
from cleveragents.application.services.subplan_service import SubplanService
from cleveragents.application.services.vector_store_service import VectorStoreService
from cleveragents.config.settings import Settings, get_settings
from cleveragents.domain.models.acms.stubs import (
InMemoryGraphBackend,
InMemoryTextBackend,
InMemoryVectorBackend,
)
from cleveragents.domain.providers.ai_provider import AIProviderInterface
from cleveragents.infrastructure.database.repositories import (
CheckpointRepository,
@@ -296,6 +301,13 @@ class Container(containers.DeclarativeContainer):
AutonomyGuardrailService,
)
# ACMS Backend Abstraction Layer — configurable via provider selection.
# Default: in-memory stubs. Override with production backends
# (Tantivy, FAISS, Blazegraph, etc.) via ``override_providers()``.
text_backend = providers.Singleton(InMemoryTextBackend)
vector_backend = providers.Singleton(InMemoryVectorBackend)
graph_backend = providers.Singleton(InMemoryGraphBackend)
# Reactive routing
stream_router = providers.Singleton(ReactiveStreamRouter)
langgraph_bridge = providers.Singleton(
@@ -1,8 +1,10 @@
"""ACMS (Advanced Context Management System) domain models.
Provides the Context Request Protocol (CRP) data types used by actors,
skills, strategies, and the Context Assembly Pipeline:
Provides the Context Request Protocol (CRP) data types and the Backend
Abstraction Layer (BAL) protocols used by actors, skills, strategies,
and the Context Assembly Pipeline:
CRP types (from :mod:`~cleveragents.domain.models.acms.crp`):
- ``DetailLevelMap`` -- Named-level-to-integer resolution with inheritance
- ``ContextRequest`` -- Structured request for context via the CRP
- ``ContextFragment`` -- Atomic unit of retrieved context
@@ -10,11 +12,29 @@ skills, strategies, and the Context Assembly Pipeline:
- ``ContextBudget`` -- Token budget management with reservation support
- ``AssembledContext`` -- Final budget-respecting context payload
BAL types (from :mod:`~cleveragents.domain.models.acms.backends`):
- ``TextBackend`` / ``TextResult`` -- Full-text search protocol
- ``VectorBackend`` / ``VectorResult`` -- Embedding similarity search
- ``GraphBackend`` / ``GraphResult`` -- SPARQL queries and traversals
Stub backends (from :mod:`~cleveragents.domain.models.acms.stubs`):
- ``InMemoryTextBackend`` -- Zero-dependency text search stub
- ``InMemoryVectorBackend`` -- Zero-dependency vector search stub
- ``InMemoryGraphBackend`` -- Zero-dependency graph query stub
Based on ``docs/specification.md`` ACMS / CRP sections and ADR-014.
"""
from __future__ import annotations
from cleveragents.domain.models.acms.backends import (
GraphBackend,
GraphResult,
TextBackend,
TextResult,
VectorBackend,
VectorResult,
)
from cleveragents.domain.models.acms.crp import (
AssembledContext,
ContextBudget,
@@ -23,6 +43,11 @@ from cleveragents.domain.models.acms.crp import (
DetailLevelMap,
FragmentProvenance,
)
from cleveragents.domain.models.acms.stubs import (
InMemoryGraphBackend,
InMemoryTextBackend,
InMemoryVectorBackend,
)
__all__: list[str] = [
"AssembledContext",
@@ -31,4 +56,13 @@ __all__: list[str] = [
"ContextRequest",
"DetailLevelMap",
"FragmentProvenance",
"GraphBackend",
"GraphResult",
"InMemoryGraphBackend",
"InMemoryTextBackend",
"InMemoryVectorBackend",
"TextBackend",
"TextResult",
"VectorBackend",
"VectorResult",
]
@@ -0,0 +1,262 @@
"""Backend Abstraction Layer (BAL) protocol definitions for the ACMS.
The BAL provides a uniform interface to heterogeneous data stores used
by the Advanced Context Management System. Each backend type has a
protocol and a corresponding frozen result dataclass:
| Protocol | Result Type | Purpose |
|-------------------|----------------|---------------------------------|
| ``TextBackend`` | ``TextResult`` | Full-text search over resources |
| ``VectorBackend`` | ``VectorResult``| Embedding similarity search |
| ``GraphBackend`` | ``GraphResult``| SPARQL queries and traversals |
Strategies issue queries through ``ScopedView`` wrappers that inject
a ``resource_filter`` restricting results to in-scope resources. The
protocols below accept a ``scope`` parameter (``frozenset[str]`` of
resource ULIDs) to support this filtering.
Based on ``docs/specification.md`` > ACMS > Backend Abstraction Layer
(BAL) and ADR-014.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Protocol, runtime_checkable
# ---------------------------------------------------------------------------
# Result dataclasses (frozen / immutable)
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class TextResult:
"""Immutable result returned by a :class:`TextBackend` search.
Attributes:
uko_uri: UKO URI of the matching information unit.
content: Matched text content (snippet or full).
score: Relevance score (0.0 -- 1.0, higher is more relevant).
metadata: Arbitrary backend-specific metadata.
"""
uko_uri: str
content: str
score: float
metadata: dict[str, str] = field(default_factory=dict)
def __post_init__(self) -> None:
"""Validate field constraints."""
if not self.uko_uri:
raise ValueError("uko_uri must be a non-empty string")
if self.score < 0.0 or self.score > 1.0:
raise ValueError(f"score must be between 0.0 and 1.0, got {self.score}")
@dataclass(frozen=True)
class VectorResult:
"""Immutable result returned by a :class:`VectorBackend` search.
Attributes:
uko_uri: UKO URI of the matching information unit.
content: Text content of the matched chunk.
score: Cosine similarity score (0.0 -- 1.0).
metadata: Arbitrary backend-specific metadata.
"""
uko_uri: str
content: str
score: float
metadata: dict[str, str] = field(default_factory=dict)
def __post_init__(self) -> None:
"""Validate field constraints."""
if not self.uko_uri:
raise ValueError("uko_uri must be a non-empty string")
if self.score < 0.0 or self.score > 1.0:
raise ValueError(f"score must be between 0.0 and 1.0, got {self.score}")
@dataclass(frozen=True)
class GraphResult:
"""Immutable result returned by a :class:`GraphBackend` query.
Attributes:
triples: List of (subject, predicate, object) triples.
metadata: Arbitrary backend-specific metadata.
"""
triples: list[tuple[str, str, str]] = field(default_factory=list)
metadata: dict[str, str] = field(default_factory=dict)
def __post_init__(self) -> None:
"""Validate field constraints."""
for triple in self.triples:
if len(triple) != 3:
raise ValueError(
f"Each triple must have exactly 3 elements, got {len(triple)}"
)
if not all(isinstance(elem, str) for elem in triple):
raise ValueError("All triple elements must be strings")
# ---------------------------------------------------------------------------
# Backend protocols
# ---------------------------------------------------------------------------
@runtime_checkable
class TextBackend(Protocol):
"""Protocol for full-text search backends.
Implementations wrap stores such as Tantivy, SQLite FTS, or
Elasticsearch. The ``scope`` parameter restricts results to
resources whose ULIDs are in the given set.
Lifecycle::
backend = InMemoryTextBackend()
results = backend.search("auth flow", scope=frozenset({"RES01"}))
"""
def search(
self,
query: str,
*,
scope: frozenset[str],
max_results: int = 20,
) -> list[TextResult]:
"""Search for text matching *query* within *scope*.
Args:
query: Natural-language or keyword query string.
Must be non-empty.
scope: Frozenset of resource ULIDs to restrict results to.
max_results: Maximum number of results to return.
Must be positive.
Returns:
List of :class:`TextResult` ordered by descending score.
Raises:
ValueError: If *query* is empty or *max_results* is not
positive.
"""
...
@runtime_checkable
class VectorBackend(Protocol):
"""Protocol for vector similarity search backends.
Implementations wrap stores such as FAISS, Qdrant, or Weaviate.
Lifecycle::
backend = InMemoryVectorBackend()
results = backend.similarity_search(
embedding=[0.1, 0.2, 0.3],
scope=frozenset({"RES01"}),
)
"""
def similarity_search(
self,
embedding: list[float],
*,
scope: frozenset[str],
top_k: int = 20,
) -> list[VectorResult]:
"""Find vectors most similar to *embedding* within *scope*.
Args:
embedding: Query embedding vector. Must be non-empty.
scope: Frozenset of resource ULIDs to restrict results to.
top_k: Maximum number of results to return.
Must be positive.
Returns:
List of :class:`VectorResult` ordered by descending score.
Raises:
ValueError: If *embedding* is empty or *top_k* is not
positive.
"""
...
@runtime_checkable
class GraphBackend(Protocol):
"""Protocol for knowledge-graph backends.
Implementations wrap stores such as Blazegraph, Jena, Neo4j, or
Stardog that hold UKO triples.
Lifecycle::
backend = InMemoryGraphBackend()
result = backend.sparql_query(
"SELECT ?s WHERE { ?s a uko:Container }",
scope=frozenset({"RES01"}),
)
"""
def sparql_query(
self,
query: str,
*,
scope: frozenset[str],
) -> GraphResult:
"""Execute a SPARQL query within *scope*.
Args:
query: SPARQL query string. Must be non-empty.
scope: Frozenset of resource ULIDs to restrict results to.
Returns:
A :class:`GraphResult` containing matched triples.
Raises:
ValueError: If *query* is empty.
"""
...
def get_triples(
self,
subject: str,
) -> GraphResult:
"""Retrieve all triples for a given *subject*.
Args:
subject: The UKO URI of the subject node. Must be
non-empty.
Returns:
A :class:`GraphResult` containing the subject's triples.
Raises:
ValueError: If *subject* is empty.
"""
...
def traverse(
self,
start: str,
*,
depth: int = 2,
) -> GraphResult:
"""Traverse the graph from *start* up to *depth* hops.
Args:
start: UKO URI to begin traversal from. Must be
non-empty.
depth: Maximum number of hops from *start*. Must be
non-negative.
Returns:
A :class:`GraphResult` containing discovered triples.
Raises:
ValueError: If *start* is empty or *depth* is negative.
"""
...
@@ -0,0 +1,166 @@
"""In-memory stub backends for the ACMS Backend Abstraction Layer.
These stubs satisfy the :class:`TextBackend`, :class:`VectorBackend`,
and :class:`GraphBackend` protocols with minimal, zero-dependency
implementations that return empty results. They serve as:
1. **Development placeholders** while physical store integrations are
built.
2. **Test doubles** for strategy and pipeline tests that need a
functioning backend without external infrastructure.
3. **Reference implementations** documenting the expected behaviour of
each protocol method, including argument validation.
Production backends (Tantivy, FAISS, Blazegraph, etc.) will replace
these stubs via DI container provider selection at startup.
Based on ``docs/specification.md`` > ACMS > Backend Abstraction Layer.
"""
from __future__ import annotations
from cleveragents.domain.models.acms.backends import (
GraphResult,
TextResult,
VectorResult,
)
class InMemoryTextBackend:
"""Stub :class:`TextBackend` that validates inputs and returns no results.
All searches return an empty list. This implementation validates
arguments to ensure callers conform to the protocol contract.
"""
def search(
self,
query: str,
*,
scope: frozenset[str],
max_results: int = 20,
) -> list[TextResult]:
"""Search for text matching *query* within *scope*.
Args:
query: Natural-language or keyword query string.
scope: Frozenset of resource ULIDs to restrict results to.
max_results: Maximum number of results to return.
Returns:
An empty list (stub implementation).
Raises:
ValueError: If *query* is empty or *max_results* < 1.
"""
if not query:
raise ValueError("query must be a non-empty string")
if max_results < 1:
raise ValueError(f"max_results must be positive, got {max_results}")
return []
class InMemoryVectorBackend:
"""Stub :class:`VectorBackend` that validates inputs and returns no results.
All similarity searches return an empty list.
"""
def similarity_search(
self,
embedding: list[float],
*,
scope: frozenset[str],
top_k: int = 20,
) -> list[VectorResult]:
"""Find vectors most similar to *embedding* within *scope*.
Args:
embedding: Query embedding vector.
scope: Frozenset of resource ULIDs to restrict results to.
top_k: Maximum number of results to return.
Returns:
An empty list (stub implementation).
Raises:
ValueError: If *embedding* is empty or *top_k* < 1.
"""
if not embedding:
raise ValueError("embedding must be a non-empty list")
if top_k < 1:
raise ValueError(f"top_k must be positive, got {top_k}")
return []
class InMemoryGraphBackend:
"""Stub :class:`GraphBackend` that validates inputs and returns empty results.
All queries and traversals return a :class:`GraphResult` with an
empty triple list.
"""
def sparql_query(
self,
query: str,
*,
scope: frozenset[str],
) -> GraphResult:
"""Execute a SPARQL query within *scope*.
Args:
query: SPARQL query string.
scope: Frozenset of resource ULIDs to restrict results to.
Returns:
A :class:`GraphResult` with no triples (stub).
Raises:
ValueError: If *query* is empty.
"""
if not query:
raise ValueError("query must be a non-empty string")
return GraphResult()
def get_triples(
self,
subject: str,
) -> GraphResult:
"""Retrieve all triples for *subject*.
Args:
subject: The UKO URI of the subject node.
Returns:
A :class:`GraphResult` with no triples (stub).
Raises:
ValueError: If *subject* is empty.
"""
if not subject:
raise ValueError("subject must be a non-empty string")
return GraphResult()
def traverse(
self,
start: str,
*,
depth: int = 2,
) -> GraphResult:
"""Traverse the graph from *start* up to *depth* hops.
Args:
start: UKO URI to begin traversal from.
depth: Maximum number of hops.
Returns:
A :class:`GraphResult` with no triples (stub).
Raises:
ValueError: If *start* is empty or *depth* is negative.
"""
if not start:
raise ValueError("start must be a non-empty string")
if depth < 0:
raise ValueError(f"depth must be non-negative, got {depth}")
return GraphResult()
+18
View File
@@ -517,3 +517,21 @@ resolve_server_mode # noqa: B018, F821
server_connect # noqa: B018, F821
server_status # noqa: B018, F821
_STUB_WARNING # noqa: B018, F821
# ACMS Backend Abstraction Layer — public API (issue #498)
TextBackend # noqa: B018, F821
VectorBackend # noqa: B018, F821
GraphBackend # noqa: B018, F821
TextResult # noqa: B018, F821
VectorResult # noqa: B018, F821
GraphResult # noqa: B018, F821
InMemoryTextBackend # noqa: B018, F821
InMemoryVectorBackend # noqa: B018, F821
InMemoryGraphBackend # noqa: B018, F821
text_backend # noqa: B018, F821
vector_backend # noqa: B018, F821
graph_backend # noqa: B018, F821
similarity_search # noqa: B018, F821
sparql_query # noqa: B018, F821
get_triples # noqa: B018, F821
traverse # noqa: B018, F821