feat(acms): implement context strategies batch 1 #605
@@ -0,0 +1,136 @@
|
||||
"""ASV benchmarks for built-in context strategies batch 1.
|
||||
|
||||
Measures the performance of:
|
||||
- SimpleKeywordStrategy.assemble at varying fragment counts
|
||||
- SemanticEmbeddingStrategy.assemble at varying fragment counts
|
||||
- BreadthDepthNavigatorStrategy.assemble at varying fragment counts
|
||||
- can_handle() call overhead for each strategy
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_SRC = str(Path(__file__).resolve().parents[1] / "src")
|
||||
if _SRC not in sys.path:
|
||||
sys.path.insert(0, _SRC)
|
||||
|
||||
import cleveragents # noqa: E402
|
||||
|
||||
importlib.reload(cleveragents)
|
||||
|
||||
from cleveragents.application.services.context_strategies import ( # noqa: E402
|
||||
BreadthDepthNavigatorStrategy,
|
||||
SemanticEmbeddingStrategy,
|
||||
SimpleKeywordStrategy,
|
||||
)
|
||||
from cleveragents.domain.models.core.context_fragment import ( # noqa: E402
|
||||
ContextBudget,
|
||||
ContextFragment,
|
||||
FragmentProvenance,
|
||||
)
|
||||
|
||||
_DEFAULT_PROV = FragmentProvenance(resource_uri="bench://default")
|
||||
|
||||
|
||||
def _make_fragments(count: int) -> list[ContextFragment]:
|
||||
"""Build *count* test fragments with distinct URIs and content."""
|
||||
frags: list[ContextFragment] = []
|
||||
for i in range(count):
|
||||
frags.append(
|
||||
ContextFragment(
|
||||
uko_node=f"project://app/mod{i}.py",
|
||||
content=f"Module {i} with async database connection handler pool",
|
||||
token_count=20,
|
||||
detail_depth=min(i % 10, 9),
|
||||
relevance_score=round(0.1 + (i % 9) * 0.1, 2),
|
||||
provenance=_DEFAULT_PROV,
|
||||
)
|
||||
)
|
||||
return frags
|
||||
|
||||
|
||||
class SimpleKeywordSuite:
|
||||
"""Benchmark SimpleKeywordStrategy assemble throughput."""
|
||||
|
||||
def setup(self) -> None:
|
||||
self.strategy = SimpleKeywordStrategy()
|
||||
self.strategy.set_query("async database connection")
|
||||
self.budget = ContextBudget(max_tokens=50_000, reserved_tokens=0)
|
||||
self.frags_10 = _make_fragments(10)
|
||||
self.frags_100 = _make_fragments(100)
|
||||
self.frags_1000 = _make_fragments(1000)
|
||||
|
||||
def time_assemble_10_fragments(self) -> None:
|
||||
"""Assemble 10 fragments with SimpleKeywordStrategy."""
|
||||
self.strategy.assemble(self.frags_10, self.budget)
|
||||
|
||||
def time_assemble_100_fragments(self) -> None:
|
||||
"""Assemble 100 fragments with SimpleKeywordStrategy."""
|
||||
self.strategy.assemble(self.frags_100, self.budget)
|
||||
|
||||
def time_assemble_1000_fragments(self) -> None:
|
||||
"""Assemble 1000 fragments with SimpleKeywordStrategy."""
|
||||
self.strategy.assemble(self.frags_1000, self.budget)
|
||||
|
||||
def time_can_handle(self) -> None:
|
||||
"""Benchmark can_handle call overhead."""
|
||||
self.strategy.can_handle({"query": "test search"})
|
||||
|
||||
|
||||
class SemanticEmbeddingSuite:
|
||||
"""Benchmark SemanticEmbeddingStrategy assemble throughput."""
|
||||
|
||||
def setup(self) -> None:
|
||||
self.strategy = SemanticEmbeddingStrategy()
|
||||
self.strategy.set_query("async database connection")
|
||||
self.budget = ContextBudget(max_tokens=50_000, reserved_tokens=0)
|
||||
self.frags_10 = _make_fragments(10)
|
||||
self.frags_100 = _make_fragments(100)
|
||||
self.frags_1000 = _make_fragments(1000)
|
||||
|
||||
def time_assemble_10_fragments(self) -> None:
|
||||
"""Assemble 10 fragments with SemanticEmbeddingStrategy."""
|
||||
self.strategy.assemble(self.frags_10, self.budget)
|
||||
|
||||
def time_assemble_100_fragments(self) -> None:
|
||||
"""Assemble 100 fragments with SemanticEmbeddingStrategy."""
|
||||
self.strategy.assemble(self.frags_100, self.budget)
|
||||
|
||||
def time_assemble_1000_fragments(self) -> None:
|
||||
"""Assemble 1000 fragments with SemanticEmbeddingStrategy."""
|
||||
self.strategy.assemble(self.frags_1000, self.budget)
|
||||
|
||||
def time_can_handle(self) -> None:
|
||||
"""Benchmark can_handle call overhead."""
|
||||
self.strategy.can_handle({"query": "test search"})
|
||||
|
||||
|
||||
class BreadthDepthNavigatorSuite:
|
||||
"""Benchmark BreadthDepthNavigatorStrategy assemble throughput."""
|
||||
|
||||
def setup(self) -> None:
|
||||
self.strategy = BreadthDepthNavigatorStrategy()
|
||||
self.strategy.set_focus(["project://app/mod0.py"])
|
||||
self.budget = ContextBudget(max_tokens=50_000, reserved_tokens=0)
|
||||
self.frags_10 = _make_fragments(10)
|
||||
self.frags_100 = _make_fragments(100)
|
||||
self.frags_1000 = _make_fragments(1000)
|
||||
|
||||
def time_assemble_10_fragments(self) -> None:
|
||||
"""Assemble 10 fragments with BreadthDepthNavigatorStrategy."""
|
||||
self.strategy.assemble(self.frags_10, self.budget)
|
||||
|
||||
def time_assemble_100_fragments(self) -> None:
|
||||
"""Assemble 100 fragments with BreadthDepthNavigatorStrategy."""
|
||||
self.strategy.assemble(self.frags_100, self.budget)
|
||||
|
||||
def time_assemble_1000_fragments(self) -> None:
|
||||
"""Assemble 1000 fragments with BreadthDepthNavigatorStrategy."""
|
||||
self.strategy.assemble(self.frags_1000, self.budget)
|
||||
|
||||
def time_can_handle(self) -> None:
|
||||
"""Benchmark can_handle call overhead."""
|
||||
self.strategy.can_handle({"focus": ["project://app/mod0.py"]})
|
||||
@@ -0,0 +1,257 @@
|
||||
@phase2 @acms @context_strategies
|
||||
Feature: Built-in Context Strategies Batch 1
|
||||
As a CleverAgents developer
|
||||
I want built-in context strategies
|
||||
So that the ACMS pipeline can rank fragments using different approaches
|
||||
|
||||
# ===========================================================================
|
||||
# SimpleKeywordStrategy
|
||||
# ===========================================================================
|
||||
|
||||
@simple_keyword
|
||||
Scenario: SimpleKeyword ranks by keyword match count
|
||||
Given a SimpleKeywordStrategy with query "async IO"
|
||||
And the following strategy fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/io.py | Use async IO pattern | 0.5 | 20 | 3 |
|
||||
| project://app/main.py | Main entry point | 0.8 | 15 | 3 |
|
||||
| project://app/net.py | Async network handler | 0.6 | 25 | 3 |
|
||||
And a strategy budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I assemble with the SimpleKeywordStrategy
|
||||
Then the first result fragment should have uko_node "project://app/io.py"
|
||||
|
||||
@simple_keyword
|
||||
Scenario: SimpleKeyword without query falls back to word density
|
||||
Given a SimpleKeywordStrategy without query
|
||||
And the following strategy fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/io.py | io | 0.5 | 10 | 3 |
|
||||
| project://app/main.py | Main entry point with many words here | 0.5 | 10 | 3 |
|
||||
And a strategy budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I assemble with the SimpleKeywordStrategy
|
||||
Then 2 fragments should be returned by strategy
|
||||
|
||||
@simple_keyword
|
||||
Scenario: SimpleKeyword returns empty for empty input
|
||||
Given a SimpleKeywordStrategy with query "test"
|
||||
And an empty strategy fragment list
|
||||
And a strategy budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I assemble with the SimpleKeywordStrategy
|
||||
Then 0 fragments should be returned by strategy
|
||||
|
||||
@simple_keyword
|
||||
Scenario: SimpleKeyword respects budget
|
||||
Given a SimpleKeywordStrategy with query "hello"
|
||||
And the following strategy fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/a.py | hello world | 0.9 | 100 | 3 |
|
||||
| project://app/b.py | hello there | 0.7 | 100 | 3 |
|
||||
| project://app/c.py | hello again | 0.5 | 100 | 3 |
|
||||
And a strategy budget with max_tokens 250 and reserved_tokens 0
|
||||
When I assemble with the SimpleKeywordStrategy
|
||||
Then 2 fragments should be returned by strategy
|
||||
|
||||
@simple_keyword
|
||||
Scenario: SimpleKeyword can_handle returns 0.3
|
||||
Given a SimpleKeywordStrategy without query
|
||||
When I check can_handle on SimpleKeywordStrategy with query "test"
|
||||
Then the strategy confidence should be 0.3
|
||||
|
||||
@simple_keyword
|
||||
Scenario: SimpleKeyword reports capabilities
|
||||
Given a SimpleKeywordStrategy without query
|
||||
Then the SimpleKeywordStrategy should not support semantic search
|
||||
And the SimpleKeywordStrategy name should be "simple-keyword"
|
||||
|
||||
# ===========================================================================
|
||||
# SemanticEmbeddingStrategy
|
||||
# ===========================================================================
|
||||
|
||||
@semantic_embedding
|
||||
Scenario: SemanticEmbedding ranks by word similarity
|
||||
Given a SemanticEmbeddingStrategy with query "database connection pool"
|
||||
And the following strategy fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/db.py | Database connection pool manager | 0.5 | 20 | 3 |
|
||||
| project://app/io.py | File input output handler | 0.8 | 15 | 3 |
|
||||
| project://app/sql.py | SQL database query executor | 0.6 | 25 | 3 |
|
||||
And a strategy budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I assemble with the SemanticEmbeddingStrategy
|
||||
Then the first result fragment should have uko_node "project://app/db.py"
|
||||
|
||||
@semantic_embedding
|
||||
Scenario: SemanticEmbedding filters below similarity threshold
|
||||
Given a SemanticEmbeddingStrategy with query "quantum computing"
|
||||
And the following strategy fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/db.py | database handler | 0.9 | 20 | 3 |
|
||||
| project://app/io.py | file io module | 0.8 | 15 | 3 |
|
||||
And a strategy budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I assemble with the SemanticEmbeddingStrategy
|
||||
Then 0 fragments should be returned by strategy
|
||||
|
||||
@semantic_embedding
|
||||
Scenario: SemanticEmbedding without query falls back to relevance
|
||||
Given a SemanticEmbeddingStrategy without query
|
||||
And the following strategy fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/a.py | alpha | 0.3 | 10 | 3 |
|
||||
| project://app/b.py | beta | 0.9 | 10 | 3 |
|
||||
And a strategy budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I assemble with the SemanticEmbeddingStrategy
|
||||
Then the first result fragment should have uko_node "project://app/b.py"
|
||||
|
||||
@semantic_embedding
|
||||
Scenario: SemanticEmbedding can_handle returns 0.6 with query
|
||||
Given a SemanticEmbeddingStrategy without query
|
||||
When I check can_handle on SemanticEmbeddingStrategy with query "test"
|
||||
Then the strategy confidence should be 0.6
|
||||
|
||||
@semantic_embedding
|
||||
Scenario: SemanticEmbedding can_handle returns 0.1 without query
|
||||
Given a SemanticEmbeddingStrategy without query
|
||||
When I check can_handle on SemanticEmbeddingStrategy without query
|
||||
Then the strategy confidence should be 0.1
|
||||
|
||||
@semantic_embedding
|
||||
Scenario: SemanticEmbedding reports capabilities
|
||||
Given a SemanticEmbeddingStrategy without query
|
||||
Then the SemanticEmbeddingStrategy should support semantic search
|
||||
And the SemanticEmbeddingStrategy name should be "semantic-embedding"
|
||||
|
||||
# ===========================================================================
|
||||
# BreadthDepthNavigatorStrategy
|
||||
# ===========================================================================
|
||||
|
||||
@breadth_depth
|
||||
Scenario: BreadthDepthNavigator prioritises near focus nodes
|
||||
Given a BreadthDepthNavigatorStrategy with focus "project://app/io.py"
|
||||
And the following strategy fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/io.py | io module | 0.5 | 20 | 5 |
|
||||
| project://app/main.py | main entry | 0.9 | 15 | 3 |
|
||||
| project://other/lib.py | library | 0.7 | 25 | 9 |
|
||||
And a strategy budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I assemble with the BreadthDepthNavigatorStrategy
|
||||
Then the first result fragment should have uko_node "project://app/io.py"
|
||||
|
||||
@breadth_depth
|
||||
Scenario: BreadthDepthNavigator prefers higher depth near focus
|
||||
Given a BreadthDepthNavigatorStrategy with focus "project://app"
|
||||
And the following strategy fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/a.py | alpha | 0.5 | 10 | 9 |
|
||||
| project://app/b.py | beta | 0.5 | 10 | 1 |
|
||||
And a strategy budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I assemble with the BreadthDepthNavigatorStrategy
|
||||
Then the first result fragment should have uko_node "project://app/a.py"
|
||||
|
||||
@breadth_depth
|
||||
Scenario: BreadthDepthNavigator without focus falls back to depth+relevance
|
||||
Given a BreadthDepthNavigatorStrategy without focus
|
||||
And the following strategy fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/a.py | alpha | 0.5 | 10 | 9 |
|
||||
| project://app/b.py | beta | 0.9 | 10 | 2 |
|
||||
And a strategy budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I assemble with the BreadthDepthNavigatorStrategy
|
||||
Then the first result fragment should have uko_node "project://app/a.py"
|
||||
|
||||
@breadth_depth
|
||||
Scenario: BreadthDepthNavigator can_handle returns 0.85 with focus
|
||||
Given a BreadthDepthNavigatorStrategy without focus
|
||||
When I check can_handle on BreadthDepthNavigator with focus
|
||||
Then the strategy confidence should be 0.85
|
||||
|
||||
@breadth_depth
|
||||
Scenario: BreadthDepthNavigator can_handle returns 0.2 without focus
|
||||
Given a BreadthDepthNavigatorStrategy without focus
|
||||
When I check can_handle on BreadthDepthNavigator without focus
|
||||
Then the strategy confidence should be 0.2
|
||||
|
||||
@breadth_depth
|
||||
Scenario: BreadthDepthNavigator reports capabilities
|
||||
Given a BreadthDepthNavigatorStrategy without focus
|
||||
Then the BreadthDepthNavigatorStrategy should support graph navigation
|
||||
And the BreadthDepthNavigatorStrategy name should be "breadth-depth-navigator"
|
||||
|
||||
@breadth_depth
|
||||
Scenario: BreadthDepthNavigator respects budget
|
||||
Given a BreadthDepthNavigatorStrategy with focus "project://app"
|
||||
And the following strategy fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/a.py | alpha | 0.9 | 100 | 5 |
|
||||
| project://app/b.py | beta | 0.7 | 100 | 5 |
|
||||
| project://app/c.py | gamma | 0.5 | 100 | 5 |
|
||||
And a strategy budget with max_tokens 250 and reserved_tokens 0
|
||||
When I assemble with the BreadthDepthNavigatorStrategy
|
||||
Then 2 fragments should be returned by strategy
|
||||
|
||||
@breadth_depth
|
||||
Scenario: BreadthDepthNavigator returns empty for empty input
|
||||
Given a BreadthDepthNavigatorStrategy with focus "project://app"
|
||||
And an empty strategy fragment list
|
||||
And a strategy budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I assemble with the BreadthDepthNavigatorStrategy
|
||||
Then 0 fragments should be returned by strategy
|
||||
|
||||
@breadth_depth
|
||||
Scenario: BreadthDepthNavigator can_handle accepts focus as string
|
||||
Given a BreadthDepthNavigatorStrategy without focus
|
||||
When I check can_handle on BreadthDepthNavigator with string focus
|
||||
Then the strategy confidence should be 0.85
|
||||
|
||||
@semantic_embedding
|
||||
Scenario: SemanticEmbedding returns empty for empty input
|
||||
Given a SemanticEmbeddingStrategy with query "test"
|
||||
And an empty strategy fragment list
|
||||
And a strategy budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I assemble with the SemanticEmbeddingStrategy
|
||||
Then 0 fragments should be returned by strategy
|
||||
|
||||
@semantic_embedding
|
||||
Scenario: SemanticEmbedding handles fragment with empty content
|
||||
Given a SemanticEmbeddingStrategy with query "test"
|
||||
And the following strategy fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/a.py | | 0.5 | 10 | 3 |
|
||||
And a strategy budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I assemble with the SemanticEmbeddingStrategy
|
||||
Then 0 fragments should be returned by strategy
|
||||
|
||||
# ===========================================================================
|
||||
# explain() coverage
|
||||
# ===========================================================================
|
||||
|
||||
@simple_keyword
|
||||
Scenario: SimpleKeyword explain returns description
|
||||
Given a SimpleKeywordStrategy without query
|
||||
Then the SimpleKeywordStrategy explain should contain "keyword"
|
||||
|
||||
@semantic_embedding
|
||||
Scenario: SemanticEmbedding explain returns description
|
||||
Given a SemanticEmbeddingStrategy without query
|
||||
Then the SemanticEmbeddingStrategy explain should contain "semantic"
|
||||
|
||||
@breadth_depth
|
||||
Scenario: BreadthDepthNavigator explain returns description
|
||||
Given a BreadthDepthNavigatorStrategy without focus
|
||||
Then the BreadthDepthNavigatorStrategy explain should contain "hierarchy"
|
||||
|
||||
# ===========================================================================
|
||||
# Pipeline Registration
|
||||
# ===========================================================================
|
||||
|
||||
@registration
|
||||
Scenario: Register SimpleKeywordStrategy with pipeline
|
||||
Given an ACMS pipeline for strategy tests
|
||||
When I register SimpleKeywordStrategy with the pipeline
|
||||
Then the pipeline should have strategy "simple-keyword"
|
||||
|
||||
@registration
|
||||
Scenario: Register all three strategies with pipeline
|
||||
Given an ACMS pipeline for strategy tests
|
||||
When I register all batch 1 strategies with the pipeline
|
||||
Then the pipeline should have strategy "simple-keyword"
|
||||
And the pipeline should have strategy "semantic-embedding"
|
||||
And the pipeline should have strategy "breadth-depth-navigator"
|
||||
@@ -0,0 +1,314 @@
|
||||
"""Step definitions for ``features/context_strategies.feature``.
|
||||
|
||||
Covers the first batch of built-in context strategies:
|
||||
|
||||
* **SimpleKeywordStrategy** — keyword matching / word-density fallback
|
||||
* **SemanticEmbeddingStrategy** — Jaccard word-overlap similarity
|
||||
* **BreadthDepthNavigatorStrategy** — UKO hierarchy navigation
|
||||
|
||||
Also covers pipeline registration of all three strategies.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from behave import given, then, when
|
||||
from behave.runner import Context
|
||||
|
||||
from cleveragents.application.services.acms_service import ACMSPipeline
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
BreadthDepthNavigatorStrategy,
|
||||
SemanticEmbeddingStrategy,
|
||||
SimpleKeywordStrategy,
|
||||
)
|
||||
from cleveragents.domain.models.core.context_fragment import (
|
||||
ContextBudget,
|
||||
ContextFragment,
|
||||
FragmentProvenance,
|
||||
)
|
||||
|
||||
__all__: list[str] = []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_DEFAULT_PROVENANCE = FragmentProvenance(resource_uri="test://default")
|
||||
|
||||
|
||||
def _make_fragment(
|
||||
uko_node: str,
|
||||
content: str,
|
||||
score: float,
|
||||
tokens: int,
|
||||
depth: int,
|
||||
) -> ContextFragment:
|
||||
"""Build a ``ContextFragment`` for test purposes."""
|
||||
return ContextFragment(
|
||||
uko_node=uko_node,
|
||||
content=content,
|
||||
relevance_score=score,
|
||||
token_count=tokens,
|
||||
detail_depth=depth,
|
||||
provenance=FragmentProvenance(resource_uri=uko_node),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Given steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given('a SimpleKeywordStrategy with query "{query}"')
|
||||
def step_simple_keyword_with_query(context: Context, query: str) -> None:
|
||||
strategy = SimpleKeywordStrategy()
|
||||
strategy.set_query(query)
|
||||
context.strategy = strategy
|
||||
|
||||
|
||||
@given("a SimpleKeywordStrategy without query")
|
||||
def step_simple_keyword_no_query(context: Context) -> None:
|
||||
context.strategy = SimpleKeywordStrategy()
|
||||
|
||||
|
||||
@given('a SemanticEmbeddingStrategy with query "{query}"')
|
||||
def step_semantic_embedding_with_query(context: Context, query: str) -> None:
|
||||
strategy = SemanticEmbeddingStrategy()
|
||||
strategy.set_query(query)
|
||||
context.strategy = strategy
|
||||
|
||||
|
||||
@given("a SemanticEmbeddingStrategy without query")
|
||||
def step_semantic_embedding_no_query(context: Context) -> None:
|
||||
context.strategy = SemanticEmbeddingStrategy()
|
||||
|
||||
|
||||
@given('a BreadthDepthNavigatorStrategy with focus "{focus}"')
|
||||
def step_breadth_depth_with_focus(context: Context, focus: str) -> None:
|
||||
strategy = BreadthDepthNavigatorStrategy()
|
||||
strategy.set_focus([focus])
|
||||
context.strategy = strategy
|
||||
|
||||
|
||||
@given("a BreadthDepthNavigatorStrategy without focus")
|
||||
def step_breadth_depth_no_focus(context: Context) -> None:
|
||||
context.strategy = BreadthDepthNavigatorStrategy()
|
||||
|
||||
|
||||
@given("the following strategy fragments:")
|
||||
def step_strategy_fragments_table(context: Context) -> None:
|
||||
context.strategy_fragments = []
|
||||
for row in context.table:
|
||||
frag = _make_fragment(
|
||||
uko_node=row["uko_node"],
|
||||
content=row["content"],
|
||||
score=float(row["score"]),
|
||||
tokens=int(row["tokens"]),
|
||||
depth=int(row["depth"]),
|
||||
)
|
||||
context.strategy_fragments.append(frag)
|
||||
|
||||
|
||||
@given("an empty strategy fragment list")
|
||||
def step_empty_strategy_fragments(context: Context) -> None:
|
||||
context.strategy_fragments = []
|
||||
|
||||
|
||||
@given(
|
||||
"a strategy budget with max_tokens {max_tokens:d} and reserved_tokens {reserved:d}"
|
||||
)
|
||||
def step_strategy_budget(context: Context, max_tokens: int, reserved: int) -> None:
|
||||
context.strategy_budget = ContextBudget(
|
||||
max_tokens=max_tokens, reserved_tokens=reserved
|
||||
)
|
||||
|
||||
|
||||
@given("an ACMS pipeline for strategy tests")
|
||||
def step_pipeline_for_strategies(context: Context) -> None:
|
||||
context.pipeline = ACMSPipeline()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# When steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when("I assemble with the SimpleKeywordStrategy")
|
||||
def step_assemble_simple_keyword(context: Context) -> None:
|
||||
context.strategy_result = list(
|
||||
context.strategy.assemble(context.strategy_fragments, context.strategy_budget)
|
||||
)
|
||||
|
||||
|
||||
@when("I assemble with the SemanticEmbeddingStrategy")
|
||||
def step_assemble_semantic_embedding(context: Context) -> None:
|
||||
context.strategy_result = list(
|
||||
context.strategy.assemble(context.strategy_fragments, context.strategy_budget)
|
||||
)
|
||||
|
||||
|
||||
@when("I assemble with the BreadthDepthNavigatorStrategy")
|
||||
def step_assemble_breadth_depth(context: Context) -> None:
|
||||
context.strategy_result = list(
|
||||
context.strategy.assemble(context.strategy_fragments, context.strategy_budget)
|
||||
)
|
||||
|
||||
|
||||
@when('I check can_handle on SimpleKeywordStrategy with query "{query}"')
|
||||
def step_can_handle_simple_keyword(context: Context, query: str) -> None:
|
||||
request: dict[str, Any] = {"query": query}
|
||||
context.confidence = context.strategy.can_handle(request)
|
||||
|
||||
|
||||
@when('I check can_handle on SemanticEmbeddingStrategy with query "{query}"')
|
||||
def step_can_handle_semantic_with_query(context: Context, query: str) -> None:
|
||||
request: dict[str, Any] = {"query": query}
|
||||
context.confidence = context.strategy.can_handle(request)
|
||||
|
||||
|
||||
@when("I check can_handle on SemanticEmbeddingStrategy without query")
|
||||
def step_can_handle_semantic_no_query(context: Context) -> None:
|
||||
request: dict[str, Any] = {}
|
||||
context.confidence = context.strategy.can_handle(request)
|
||||
|
||||
|
||||
@when("I check can_handle on BreadthDepthNavigator with focus")
|
||||
def step_can_handle_breadth_depth_with_focus(context: Context) -> None:
|
||||
request: dict[str, Any] = {"focus": ["project://app/io.py"]}
|
||||
context.confidence = context.strategy.can_handle(request)
|
||||
|
||||
|
||||
@when("I check can_handle on BreadthDepthNavigator with string focus")
|
||||
def step_can_handle_breadth_depth_string_focus(context: Context) -> None:
|
||||
request: dict[str, Any] = {"focus": "project://app/io.py"}
|
||||
context.confidence = context.strategy.can_handle(request)
|
||||
|
||||
|
||||
@when("I check can_handle on BreadthDepthNavigator without focus")
|
||||
def step_can_handle_breadth_depth_no_focus(context: Context) -> None:
|
||||
request: dict[str, Any] = {}
|
||||
context.confidence = context.strategy.can_handle(request)
|
||||
|
||||
|
||||
@when("I register SimpleKeywordStrategy with the pipeline")
|
||||
def step_register_simple_keyword(context: Context) -> None:
|
||||
context.pipeline.register_strategy("simple-keyword", SimpleKeywordStrategy())
|
||||
|
||||
|
||||
@when("I register all batch 1 strategies with the pipeline")
|
||||
def step_register_all_batch1(context: Context) -> None:
|
||||
context.pipeline.register_strategy("simple-keyword", SimpleKeywordStrategy())
|
||||
context.pipeline.register_strategy(
|
||||
"semantic-embedding", SemanticEmbeddingStrategy()
|
||||
)
|
||||
context.pipeline.register_strategy(
|
||||
"breadth-depth-navigator", BreadthDepthNavigatorStrategy()
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Then steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then('the first result fragment should have uko_node "{uko_node}"')
|
||||
def step_first_result_uko_node(context: Context, uko_node: str) -> None:
|
||||
assert len(context.strategy_result) > 0, "Expected at least one result fragment"
|
||||
actual = context.strategy_result[0].uko_node
|
||||
assert actual == uko_node, (
|
||||
f"Expected first fragment uko_node '{uko_node}', got '{actual}'"
|
||||
)
|
||||
|
||||
|
||||
@then("{count:d} fragments should be returned by strategy")
|
||||
def step_fragment_count(context: Context, count: int) -> None:
|
||||
actual = len(context.strategy_result)
|
||||
assert actual == count, f"Expected {count} fragments, got {actual}"
|
||||
|
||||
|
||||
@then("the strategy confidence should be {expected:g}")
|
||||
def step_strategy_confidence(context: Context, expected: float) -> None:
|
||||
actual = context.confidence
|
||||
assert abs(actual - expected) < 1e-6, (
|
||||
f"Expected strategy confidence {expected}, got {actual}"
|
||||
)
|
||||
|
||||
|
||||
@then("the SimpleKeywordStrategy should not support semantic search")
|
||||
def step_simple_keyword_no_semantic(context: Context) -> None:
|
||||
caps = context.strategy.capabilities
|
||||
assert not caps.supports_semantic_search, (
|
||||
"SimpleKeywordStrategy should NOT support semantic search"
|
||||
)
|
||||
|
||||
|
||||
@then('the SimpleKeywordStrategy name should be "{name}"')
|
||||
def step_simple_keyword_name(context: Context, name: str) -> None:
|
||||
actual = context.strategy.name
|
||||
assert actual == name, f"Expected name '{name}', got '{actual}'"
|
||||
|
||||
|
||||
@then("the SemanticEmbeddingStrategy should support semantic search")
|
||||
def step_semantic_supports_search(context: Context) -> None:
|
||||
caps = context.strategy.capabilities
|
||||
assert caps.supports_semantic_search, (
|
||||
"SemanticEmbeddingStrategy should support semantic search"
|
||||
)
|
||||
|
||||
|
||||
@then('the SemanticEmbeddingStrategy name should be "{name}"')
|
||||
def step_semantic_name(context: Context, name: str) -> None:
|
||||
actual = context.strategy.name
|
||||
assert actual == name, f"Expected name '{name}', got '{actual}'"
|
||||
|
||||
|
||||
@then("the BreadthDepthNavigatorStrategy should support graph navigation")
|
||||
def step_breadth_depth_supports_graph(context: Context) -> None:
|
||||
caps = context.strategy.capabilities
|
||||
assert caps.supports_graph_navigation, (
|
||||
"BreadthDepthNavigatorStrategy should support graph navigation"
|
||||
)
|
||||
|
||||
|
||||
@then('the BreadthDepthNavigatorStrategy name should be "{name}"')
|
||||
def step_breadth_depth_name(context: Context, name: str) -> None:
|
||||
actual = context.strategy.name
|
||||
assert actual == name, f"Expected name '{name}', got '{actual}'"
|
||||
|
||||
|
||||
@then('the SimpleKeywordStrategy explain should contain "{text}"')
|
||||
def step_simple_keyword_explain(context: Context, text: str) -> None:
|
||||
explanation = context.strategy.explain()
|
||||
assert text.lower() in explanation.lower(), (
|
||||
f"Expected explain to contain '{text}', got: {explanation}"
|
||||
)
|
||||
|
||||
|
||||
@then('the SemanticEmbeddingStrategy explain should contain "{text}"')
|
||||
def step_semantic_explain(context: Context, text: str) -> None:
|
||||
explanation = context.strategy.explain()
|
||||
assert text.lower() in explanation.lower(), (
|
||||
f"Expected explain to contain '{text}', got: {explanation}"
|
||||
)
|
||||
|
||||
|
||||
@then('the BreadthDepthNavigatorStrategy explain should contain "{text}"')
|
||||
def step_breadth_depth_explain(context: Context, text: str) -> None:
|
||||
explanation = context.strategy.explain()
|
||||
assert text.lower() in explanation.lower(), (
|
||||
f"Expected explain to contain '{text}', got: {explanation}"
|
||||
)
|
||||
|
||||
|
||||
@then('the pipeline should have strategy "{name}"')
|
||||
def step_pipeline_has_strategy(context: Context, name: str) -> None:
|
||||
strategies = context.pipeline.BUILTIN_STRATEGIES
|
||||
registered = context.pipeline._strategies
|
||||
has_builtin = name in strategies
|
||||
has_registered = name in registered
|
||||
assert has_builtin or has_registered, (
|
||||
f"Pipeline does not have strategy '{name}'. "
|
||||
f"Builtins: {list(strategies.keys())}, Registered: {list(registered.keys())}"
|
||||
)
|
||||
@@ -0,0 +1,81 @@
|
||||
*** Settings ***
|
||||
Documentation Integration tests for built-in context strategies batch 1
|
||||
Resource ${CURDIR}/common.resource
|
||||
Suite Setup Setup Test Environment
|
||||
Suite Teardown Cleanup Test Environment
|
||||
|
||||
*** Variables ***
|
||||
${HELPER} ${CURDIR}/helper_context_strategies.py
|
||||
|
||||
*** Test Cases ***
|
||||
SimpleKeyword Ranks By Keyword Match
|
||||
[Documentation] SimpleKeywordStrategy ranks keyword-matching fragments first
|
||||
${result}= Run Process ${PYTHON} ${HELPER} simple-keyword-rank cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} context-strategies-ok: simple-keyword-rank
|
||||
|
||||
SimpleKeyword Respects Budget
|
||||
[Documentation] SimpleKeywordStrategy respects token budget
|
||||
${result}= Run Process ${PYTHON} ${HELPER} simple-keyword-budget cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} context-strategies-ok: simple-keyword-budget
|
||||
|
||||
SemanticEmbedding Ranks By Similarity
|
||||
[Documentation] SemanticEmbeddingStrategy ranks by word similarity
|
||||
${result}= Run Process ${PYTHON} ${HELPER} semantic-embedding-rank cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} context-strategies-ok: semantic-embedding-rank
|
||||
|
||||
SemanticEmbedding Filters Unrelated
|
||||
[Documentation] SemanticEmbeddingStrategy filters below similarity threshold
|
||||
${result}= Run Process ${PYTHON} ${HELPER} semantic-embedding-filter cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} context-strategies-ok: semantic-embedding-filter
|
||||
|
||||
BreadthDepth Prioritises Near Focus
|
||||
[Documentation] BreadthDepthNavigatorStrategy prioritises fragments near focus
|
||||
${result}= Run Process ${PYTHON} ${HELPER} breadth-depth-rank cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} context-strategies-ok: breadth-depth-rank
|
||||
|
||||
BreadthDepth Respects Budget
|
||||
[Documentation] BreadthDepthNavigatorStrategy respects token budget
|
||||
${result}= Run Process ${PYTHON} ${HELPER} breadth-depth-budget cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} context-strategies-ok: breadth-depth-budget
|
||||
|
||||
Pipeline Registration
|
||||
[Documentation] Register all batch 1 strategies with pipeline
|
||||
${result}= Run Process ${PYTHON} ${HELPER} pipeline-register cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} context-strategies-ok: pipeline-register
|
||||
|
||||
Can Handle Confidence Values
|
||||
[Documentation] Verify can_handle confidence values for all strategies
|
||||
${result}= Run Process ${PYTHON} ${HELPER} can-handle cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} context-strategies-ok: can-handle
|
||||
|
||||
Strategy Capabilities
|
||||
[Documentation] Verify capability flags and names for all strategies
|
||||
${result}= Run Process ${PYTHON} ${HELPER} capabilities cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} context-strategies-ok: capabilities
|
||||
@@ -0,0 +1,288 @@
|
||||
"""Robot Framework integration helper for context strategies batch 1.
|
||||
|
||||
Each ``_cmd_*`` function exercises one strategy end-to-end and prints a
|
||||
sentinel string on success. The Robot ``.robot`` file invokes this script
|
||||
as a subprocess via ``Run Process`` and asserts exit-code + sentinel.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Bootstrap src/ so domain imports resolve when executed standalone.
|
||||
_src = str(Path(__file__).resolve().parent.parent / "src")
|
||||
if _src not in sys.path:
|
||||
sys.path.insert(0, _src)
|
||||
|
||||
from cleveragents.application.services.acms_service import ACMSPipeline # noqa: E402
|
||||
from cleveragents.application.services.context_strategies import ( # noqa: E402
|
||||
BreadthDepthNavigatorStrategy,
|
||||
SemanticEmbeddingStrategy,
|
||||
SimpleKeywordStrategy,
|
||||
)
|
||||
from cleveragents.domain.models.core.context_fragment import ( # noqa: E402
|
||||
ContextBudget,
|
||||
ContextFragment,
|
||||
FragmentProvenance,
|
||||
)
|
||||
|
||||
|
||||
def _make_frag(
|
||||
uko_node: str,
|
||||
content: str,
|
||||
*,
|
||||
score: float = 0.5,
|
||||
tokens: int = 20,
|
||||
depth: int = 3,
|
||||
) -> ContextFragment:
|
||||
return ContextFragment(
|
||||
uko_node=uko_node,
|
||||
content=content,
|
||||
relevance_score=score,
|
||||
token_count=tokens,
|
||||
detail_depth=depth,
|
||||
provenance=FragmentProvenance(resource_uri=uko_node),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Command functions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _cmd_simple_keyword_rank() -> int:
|
||||
"""SimpleKeywordStrategy ranks keyword-matching fragments first."""
|
||||
strategy = SimpleKeywordStrategy()
|
||||
strategy.set_query("async IO")
|
||||
|
||||
frags = [
|
||||
_make_frag("project://app/io.py", "Use async IO pattern"),
|
||||
_make_frag("project://app/main.py", "Main entry point"),
|
||||
_make_frag("project://app/net.py", "Async network handler"),
|
||||
]
|
||||
budget = ContextBudget(max_tokens=1000, reserved_tokens=0)
|
||||
|
||||
result = list(strategy.assemble(frags, budget))
|
||||
if result[0].uko_node != "project://app/io.py":
|
||||
print(f"FAIL: expected io.py first, got {result[0].uko_node}")
|
||||
return 1
|
||||
print("context-strategies-ok: simple-keyword-rank")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_simple_keyword_budget() -> int:
|
||||
"""SimpleKeywordStrategy respects token budget."""
|
||||
strategy = SimpleKeywordStrategy()
|
||||
strategy.set_query("hello")
|
||||
|
||||
frags = [
|
||||
_make_frag("project://app/a.py", "hello world", tokens=100),
|
||||
_make_frag("project://app/b.py", "hello there", tokens=100),
|
||||
_make_frag("project://app/c.py", "hello again", tokens=100),
|
||||
]
|
||||
budget = ContextBudget(max_tokens=250, reserved_tokens=0)
|
||||
|
||||
result = list(strategy.assemble(frags, budget))
|
||||
if len(result) != 2:
|
||||
print(f"FAIL: expected 2 fragments, got {len(result)}")
|
||||
return 1
|
||||
print("context-strategies-ok: simple-keyword-budget")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_semantic_embedding_rank() -> int:
|
||||
"""SemanticEmbeddingStrategy ranks by word similarity."""
|
||||
strategy = SemanticEmbeddingStrategy()
|
||||
strategy.set_query("database connection pool")
|
||||
|
||||
frags = [
|
||||
_make_frag("project://app/db.py", "Database connection pool manager"),
|
||||
_make_frag("project://app/io.py", "File input output handler"),
|
||||
_make_frag("project://app/sql.py", "SQL database query executor"),
|
||||
]
|
||||
budget = ContextBudget(max_tokens=1000, reserved_tokens=0)
|
||||
|
||||
result = list(strategy.assemble(frags, budget))
|
||||
if result[0].uko_node != "project://app/db.py":
|
||||
print(f"FAIL: expected db.py first, got {result[0].uko_node}")
|
||||
return 1
|
||||
print("context-strategies-ok: semantic-embedding-rank")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_semantic_embedding_filter() -> int:
|
||||
"""SemanticEmbeddingStrategy filters unrelated fragments."""
|
||||
strategy = SemanticEmbeddingStrategy()
|
||||
strategy.set_query("quantum computing")
|
||||
|
||||
frags = [
|
||||
_make_frag("project://app/db.py", "database handler"),
|
||||
_make_frag("project://app/io.py", "file io module"),
|
||||
]
|
||||
budget = ContextBudget(max_tokens=1000, reserved_tokens=0)
|
||||
|
||||
result = list(strategy.assemble(frags, budget))
|
||||
if len(result) != 0:
|
||||
print(f"FAIL: expected 0 fragments, got {len(result)}")
|
||||
return 1
|
||||
print("context-strategies-ok: semantic-embedding-filter")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_breadth_depth_rank() -> int:
|
||||
"""BreadthDepthNavigatorStrategy prioritises near focus."""
|
||||
strategy = BreadthDepthNavigatorStrategy()
|
||||
strategy.set_focus(["project://app/io.py"])
|
||||
|
||||
frags = [
|
||||
_make_frag("project://app/io.py", "io module", depth=5),
|
||||
_make_frag("project://app/main.py", "main entry", score=0.9, depth=3),
|
||||
_make_frag("project://other/lib.py", "library", score=0.7, depth=9),
|
||||
]
|
||||
budget = ContextBudget(max_tokens=1000, reserved_tokens=0)
|
||||
|
||||
result = list(strategy.assemble(frags, budget))
|
||||
if result[0].uko_node != "project://app/io.py":
|
||||
print(f"FAIL: expected io.py first, got {result[0].uko_node}")
|
||||
return 1
|
||||
print("context-strategies-ok: breadth-depth-rank")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_breadth_depth_budget() -> int:
|
||||
"""BreadthDepthNavigatorStrategy respects budget."""
|
||||
strategy = BreadthDepthNavigatorStrategy()
|
||||
strategy.set_focus(["project://app"])
|
||||
|
||||
frags = [
|
||||
_make_frag("project://app/a.py", "alpha", score=0.9, tokens=100, depth=5),
|
||||
_make_frag("project://app/b.py", "beta", score=0.7, tokens=100, depth=5),
|
||||
_make_frag("project://app/c.py", "gamma", score=0.5, tokens=100, depth=5),
|
||||
]
|
||||
budget = ContextBudget(max_tokens=250, reserved_tokens=0)
|
||||
|
||||
result = list(strategy.assemble(frags, budget))
|
||||
if len(result) != 2:
|
||||
print(f"FAIL: expected 2 fragments, got {len(result)}")
|
||||
return 1
|
||||
print("context-strategies-ok: breadth-depth-budget")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_pipeline_register() -> int:
|
||||
"""Register all batch 1 strategies with pipeline."""
|
||||
pipeline = ACMSPipeline()
|
||||
pipeline.register_strategy("simple-keyword", SimpleKeywordStrategy())
|
||||
pipeline.register_strategy("semantic-embedding", SemanticEmbeddingStrategy())
|
||||
pipeline.register_strategy(
|
||||
"breadth-depth-navigator", BreadthDepthNavigatorStrategy()
|
||||
)
|
||||
|
||||
registered = pipeline._strategies
|
||||
for name in ("simple-keyword", "semantic-embedding", "breadth-depth-navigator"):
|
||||
if name not in registered:
|
||||
print(f"FAIL: strategy '{name}' not registered")
|
||||
return 1
|
||||
print("context-strategies-ok: pipeline-register")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_can_handle() -> int:
|
||||
"""Verify can_handle confidence values for all strategies."""
|
||||
sk = SimpleKeywordStrategy()
|
||||
se = SemanticEmbeddingStrategy()
|
||||
bd = BreadthDepthNavigatorStrategy()
|
||||
|
||||
# SimpleKeyword always returns 0.3
|
||||
if abs(sk.can_handle({"query": "test"}) - 0.3) > 1e-6:
|
||||
print("FAIL: SimpleKeyword can_handle != 0.3")
|
||||
return 1
|
||||
|
||||
# SemanticEmbedding returns 0.6 with query
|
||||
if abs(se.can_handle({"query": "test"}) - 0.6) > 1e-6:
|
||||
print("FAIL: SemanticEmbedding can_handle with query != 0.6")
|
||||
return 1
|
||||
|
||||
# SemanticEmbedding returns 0.1 without query
|
||||
if abs(se.can_handle({}) - 0.1) > 1e-6:
|
||||
print("FAIL: SemanticEmbedding can_handle without query != 0.1")
|
||||
return 1
|
||||
|
||||
# BreadthDepth returns 0.85 with focus
|
||||
if abs(bd.can_handle({"focus": ["project://app"]}) - 0.85) > 1e-6:
|
||||
print("FAIL: BreadthDepth can_handle with focus != 0.85")
|
||||
return 1
|
||||
|
||||
# BreadthDepth returns 0.2 without focus
|
||||
if abs(bd.can_handle({}) - 0.2) > 1e-6:
|
||||
print("FAIL: BreadthDepth can_handle without focus != 0.2")
|
||||
return 1
|
||||
|
||||
print("context-strategies-ok: can-handle")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_capabilities() -> int:
|
||||
"""Verify capability flags for all strategies."""
|
||||
sk = SimpleKeywordStrategy()
|
||||
se = SemanticEmbeddingStrategy()
|
||||
bd = BreadthDepthNavigatorStrategy()
|
||||
|
||||
if sk.capabilities.supports_semantic_search:
|
||||
print("FAIL: SimpleKeyword should not support semantic search")
|
||||
return 1
|
||||
if not se.capabilities.supports_semantic_search:
|
||||
print("FAIL: SemanticEmbedding should support semantic search")
|
||||
return 1
|
||||
if not bd.capabilities.supports_graph_navigation:
|
||||
print("FAIL: BreadthDepth should support graph navigation")
|
||||
return 1
|
||||
if sk.name != "simple-keyword":
|
||||
print(f"FAIL: SimpleKeyword name = {sk.name}")
|
||||
return 1
|
||||
if se.name != "semantic-embedding":
|
||||
print(f"FAIL: SemanticEmbedding name = {se.name}")
|
||||
return 1
|
||||
if bd.name != "breadth-depth-navigator":
|
||||
print(f"FAIL: BreadthDepth name = {bd.name}")
|
||||
return 1
|
||||
|
||||
print("context-strategies-ok: capabilities")
|
||||
return 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Dispatch
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_COMMANDS: dict[str, object] = {
|
||||
"simple-keyword-rank": _cmd_simple_keyword_rank,
|
||||
"simple-keyword-budget": _cmd_simple_keyword_budget,
|
||||
"semantic-embedding-rank": _cmd_semantic_embedding_rank,
|
||||
"semantic-embedding-filter": _cmd_semantic_embedding_filter,
|
||||
"breadth-depth-rank": _cmd_breadth_depth_rank,
|
||||
"breadth-depth-budget": _cmd_breadth_depth_budget,
|
||||
"pipeline-register": _cmd_pipeline_register,
|
||||
"can-handle": _cmd_can_handle,
|
||||
"capabilities": _cmd_capabilities,
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
if len(sys.argv) < 2:
|
||||
print(f"Usage: {sys.argv[0]} <command>")
|
||||
print(f"Commands: {', '.join(sorted(_COMMANDS))}")
|
||||
return 1
|
||||
|
||||
cmd = sys.argv[1]
|
||||
handler = _COMMANDS.get(cmd)
|
||||
if handler is None:
|
||||
print(f"Unknown command: {cmd}")
|
||||
return 1
|
||||
|
||||
return handler() # type: ignore[operator]
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -15,6 +15,11 @@ from cleveragents.application.services.config_service import (
|
||||
ConfigService,
|
||||
ResolvedValue,
|
||||
)
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
BreadthDepthNavigatorStrategy,
|
||||
SemanticEmbeddingStrategy,
|
||||
SimpleKeywordStrategy,
|
||||
)
|
||||
from cleveragents.application.services.correction_service import (
|
||||
CorrectionService,
|
||||
)
|
||||
@@ -153,6 +158,7 @@ __all__ = [
|
||||
"AttachmentScope",
|
||||
"AutonomyController",
|
||||
"AutonomyGuardrailService",
|
||||
"BreadthDepthNavigatorStrategy",
|
||||
"BrokenReferenceRule",
|
||||
"ClusterStrategy",
|
||||
"ClusteringStrategy",
|
||||
@@ -194,12 +200,14 @@ __all__ = [
|
||||
"RuntimeExecuteActor",
|
||||
"RuntimeExecuteResult",
|
||||
"SemanticCheckResult",
|
||||
"SemanticEmbeddingStrategy",
|
||||
"SemanticRuleRegistry",
|
||||
"SemanticValidationCache",
|
||||
"SemanticValidationRule",
|
||||
"SemanticValidationService",
|
||||
"SemanticValidationSeverity",
|
||||
"SequenceConflictError",
|
||||
"SimpleKeywordStrategy",
|
||||
"SkeletonCompressorService",
|
||||
"SkillRegistryService",
|
||||
"SnapshotStore",
|
||||
|
||||
@@ -0,0 +1,405 @@
|
||||
"""Built-in context strategies batch 1.
|
||||
|
||||
Implements the first three built-in context strategies from the spec
|
||||
(§25207-25216):
|
||||
|
||||
1. **SimpleKeywordStrategy** (quality 0.3) — Keyword matching on fragment
|
||||
content and UKO node URIs. Universal fallback that works without
|
||||
specialised backends.
|
||||
2. **SemanticEmbeddingStrategy** (quality 0.6) — Approximate semantic
|
||||
similarity using word-overlap scoring between fragments. In v1,
|
||||
operates on pre-fetched fragments without actual embedding backends.
|
||||
3. **BreadthDepthNavigatorStrategy** (quality 0.85) — Navigates the UKO
|
||||
node hierarchy, prioritising fragments near designated focus nodes
|
||||
with higher detail depths. Primary strategy for code projects.
|
||||
|
||||
All strategies implement the v1 ``ContextStrategy`` Protocol defined in
|
||||
``acms_service.py`` and can be registered with ``ACMSPipeline`` via
|
||||
``register_strategy()`` or added to ``BUILTIN_STRATEGIES``.
|
||||
|
||||
Based on ``docs/specification.md`` §25207-25216.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
from collections.abc import Sequence
|
||||
from typing import Any
|
||||
|
||||
from cleveragents.application.services.acms_service import (
|
||||
StrategyCapabilities,
|
||||
_pack_budget,
|
||||
)
|
||||
from cleveragents.domain.models.core.context_fragment import (
|
||||
ContextBudget,
|
||||
ContextFragment,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_WORD_RE = re.compile(r"\w+", re.UNICODE)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 1. SimpleKeywordStrategy (quality 0.3)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class SimpleKeywordStrategy:
|
||||
"""Keyword matching on fragment content and UKO node URIs.
|
||||
|
||||
Scores each fragment by the density of distinct words in its content,
|
||||
treating fragments with richer keyword coverage as more informative.
|
||||
When a query is provided via ``set_query()``, matching is narrowed to
|
||||
query keywords. Otherwise, a word-density heuristic is used.
|
||||
|
||||
This is the universal fallback strategy — it always produces results
|
||||
regardless of backend availability.
|
||||
|
||||
Implements ``ContextStrategy`` protocol.
|
||||
|
||||
Based on ``docs/specification.md`` §25207 — ``simple-keyword``.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._query: str = ""
|
||||
|
||||
def set_query(self, query: str) -> None:
|
||||
"""Set the query string for keyword matching (optional)."""
|
||||
self._query = query
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return "simple-keyword"
|
||||
|
||||
@property
|
||||
def capabilities(self) -> StrategyCapabilities:
|
||||
return StrategyCapabilities(supports_semantic_search=False)
|
||||
|
||||
def can_handle(self, request: dict[str, Any]) -> float:
|
||||
"""Return 0.3 confidence — universal fallback."""
|
||||
query = str(request.get("query", "") or "")
|
||||
self._query = query
|
||||
return 0.3
|
||||
|
||||
def assemble(
|
||||
self,
|
||||
fragments: Sequence[ContextFragment],
|
||||
budget: ContextBudget,
|
||||
) -> Sequence[ContextFragment]:
|
||||
"""Rank fragments by keyword match count or word density."""
|
||||
if not fragments:
|
||||
return list(fragments)
|
||||
|
||||
keywords = _extract_keywords(self._query) if self._query else []
|
||||
|
||||
if keywords:
|
||||
# Score by keyword match count
|
||||
scored = [
|
||||
(frag, _keyword_match_score(frag, keywords)) for frag in fragments
|
||||
]
|
||||
scored.sort(
|
||||
key=lambda pair: (pair[1], pair[0].relevance_score),
|
||||
reverse=True,
|
||||
)
|
||||
sorted_frags = [frag for frag, _ in scored]
|
||||
else:
|
||||
# No query — score by word density (unique words / token_count)
|
||||
scored_density = [(frag, _word_density(frag)) for frag in fragments]
|
||||
scored_density.sort(
|
||||
key=lambda pair: (pair[1], pair[0].relevance_score),
|
||||
reverse=True,
|
||||
)
|
||||
sorted_frags = [frag for frag, _ in scored_density]
|
||||
|
||||
logger.info(
|
||||
"SimpleKeyword ranked fragments",
|
||||
extra={
|
||||
"keyword_count": len(keywords),
|
||||
"fragment_count": len(fragments),
|
||||
"has_query": bool(self._query),
|
||||
},
|
||||
)
|
||||
return _pack_budget(sorted_frags, budget)
|
||||
|
||||
def explain(self) -> str:
|
||||
return (
|
||||
"Ranks fragments by keyword match count in content and UKO node "
|
||||
"URI. Falls back to word-density ordering when no query is "
|
||||
"provided. Universal fallback strategy (quality 0.3)."
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 2. SemanticEmbeddingStrategy (quality 0.6)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class SemanticEmbeddingStrategy:
|
||||
"""Approximate semantic similarity via word-overlap scoring.
|
||||
|
||||
In the v1 pipeline, strategies receive pre-fetched fragments rather
|
||||
than querying vector backends directly. This strategy approximates
|
||||
embedding-based similarity by computing word-level Jaccard similarity
|
||||
between a query (set via ``set_query()`` or ``can_handle()``) and
|
||||
each fragment's content.
|
||||
|
||||
Fragments with similarity below ``min_similarity`` are excluded.
|
||||
Remaining fragments are sorted by similarity (descending) and packed
|
||||
within the token budget. When no query is available, falls back to
|
||||
relevance-based ordering.
|
||||
|
||||
Implements ``ContextStrategy`` protocol.
|
||||
|
||||
Based on ``docs/specification.md`` §25207 — ``semantic-embedding``.
|
||||
"""
|
||||
|
||||
def __init__(self, *, min_similarity: float = 0.05) -> None:
|
||||
self._min_similarity = min_similarity
|
||||
self._query: str = ""
|
||||
|
||||
def set_query(self, query: str) -> None:
|
||||
"""Set the query string for similarity scoring (optional)."""
|
||||
self._query = query
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return "semantic-embedding"
|
||||
|
||||
@property
|
||||
def capabilities(self) -> StrategyCapabilities:
|
||||
return StrategyCapabilities(supports_semantic_search=True)
|
||||
|
||||
def can_handle(self, request: dict[str, Any]) -> float:
|
||||
"""Return 0.6 confidence when a query is present, else 0.1."""
|
||||
query = str(request.get("query", "") or "")
|
||||
self._query = query
|
||||
return 0.6 if query else 0.1
|
||||
|
||||
def assemble(
|
||||
self,
|
||||
fragments: Sequence[ContextFragment],
|
||||
budget: ContextBudget,
|
||||
) -> Sequence[ContextFragment]:
|
||||
"""Rank fragments by word-overlap similarity to query."""
|
||||
if not fragments:
|
||||
return list(fragments)
|
||||
|
||||
query_words = _tokenize(self._query) if self._query else set()
|
||||
|
||||
if not query_words:
|
||||
# No query — fall back to relevance ordering
|
||||
sorted_frags = sorted(
|
||||
fragments, key=lambda f: f.relevance_score, reverse=True
|
||||
)
|
||||
return _pack_budget(sorted_frags, budget)
|
||||
|
||||
# Compute similarity for each fragment
|
||||
scored: list[tuple[ContextFragment, float]] = []
|
||||
for frag in fragments:
|
||||
sim = _jaccard_similarity(query_words, _tokenize(frag.content))
|
||||
if sim >= self._min_similarity:
|
||||
scored.append((frag, sim))
|
||||
|
||||
scored.sort(key=lambda pair: (pair[1], pair[0].relevance_score), reverse=True)
|
||||
sorted_frags = [frag for frag, _ in scored]
|
||||
|
||||
logger.info(
|
||||
"SemanticEmbedding ranked fragments",
|
||||
extra={
|
||||
"query_word_count": len(query_words),
|
||||
"eligible_count": len(sorted_frags),
|
||||
"filtered_count": len(fragments) - len(sorted_frags),
|
||||
},
|
||||
)
|
||||
return _pack_budget(sorted_frags, budget)
|
||||
|
||||
def explain(self) -> str:
|
||||
return (
|
||||
"Approximates semantic similarity using word-level Jaccard "
|
||||
"similarity between query and fragment content. Filters by "
|
||||
f"minimum similarity threshold ({self._min_similarity}). "
|
||||
"Quality 0.6."
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 3. BreadthDepthNavigatorStrategy (quality 0.85)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class BreadthDepthNavigatorStrategy:
|
||||
"""Navigate UKO node hierarchy with breadth-first + depth-first phases.
|
||||
|
||||
This strategy treats fragment UKO node URIs as a hierarchical
|
||||
namespace. Given focus nodes (set via ``set_focus()`` or
|
||||
``can_handle()``), it:
|
||||
|
||||
1. **Breadth phase**: Discovers fragments at related nodes (shared
|
||||
URI prefix within ``max_hops`` path segments).
|
||||
2. **Depth phase**: Prefers higher ``detail_depth`` for fragments
|
||||
closer to focus nodes.
|
||||
|
||||
The combined score prioritises proximity (breadth, weight 0.6) and
|
||||
detail (depth, weight 0.3), with a small relevance contribution
|
||||
(weight 0.1). This makes it the primary strategy for code projects
|
||||
where UKO nodes represent a file/module hierarchy.
|
||||
|
||||
Implements ``ContextStrategy`` protocol.
|
||||
|
||||
Based on ``docs/specification.md`` §25207 — ``breadth-depth-navigator``.
|
||||
"""
|
||||
|
||||
def __init__(self, *, max_hops: int = 4) -> None:
|
||||
self._max_hops = max_hops
|
||||
self._focus: list[str] = []
|
||||
|
||||
def set_focus(self, focus: list[str]) -> None:
|
||||
"""Set the focus nodes for graph navigation (optional)."""
|
||||
self._focus = list(focus)
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return "breadth-depth-navigator"
|
||||
|
||||
@property
|
||||
def capabilities(self) -> StrategyCapabilities:
|
||||
return StrategyCapabilities(supports_graph_navigation=True)
|
||||
|
||||
def can_handle(self, request: dict[str, Any]) -> float:
|
||||
"""Return 0.85 when focus nodes are present, else 0.2."""
|
||||
focus = request.get("focus", [])
|
||||
if isinstance(focus, str):
|
||||
self._focus = [focus] if focus else []
|
||||
else:
|
||||
self._focus = list(focus) if focus else []
|
||||
return 0.85 if self._focus else 0.2
|
||||
|
||||
def assemble(
|
||||
self,
|
||||
fragments: Sequence[ContextFragment],
|
||||
budget: ContextBudget,
|
||||
) -> Sequence[ContextFragment]:
|
||||
"""Rank fragments by breadth proximity and depth detail."""
|
||||
if not fragments:
|
||||
return list(fragments)
|
||||
|
||||
if not self._focus:
|
||||
# No focus — fall back to depth-weighted relevance
|
||||
sorted_frags = sorted(
|
||||
fragments,
|
||||
key=lambda f: (f.detail_depth, f.relevance_score),
|
||||
reverse=True,
|
||||
)
|
||||
return _pack_budget(sorted_frags, budget)
|
||||
|
||||
# Score each fragment by proximity to focus nodes and detail depth
|
||||
scored: list[tuple[ContextFragment, float]] = []
|
||||
for frag in fragments:
|
||||
proximity = _max_proximity(frag.uko_node, self._focus, self._max_hops)
|
||||
# Normalise depth contribution: depth/9 gives 0.0-1.0
|
||||
depth_score = frag.detail_depth / 9.0
|
||||
# Combined: proximity dominates (0.6), depth adds (0.3),
|
||||
# relevance contributes (0.1)
|
||||
combined = proximity * 0.6 + depth_score * 0.3 + frag.relevance_score * 0.1
|
||||
scored.append((frag, combined))
|
||||
|
||||
scored.sort(key=lambda pair: pair[1], reverse=True)
|
||||
sorted_frags = [frag for frag, _ in scored]
|
||||
|
||||
logger.info(
|
||||
"BreadthDepthNavigator ranked fragments",
|
||||
extra={
|
||||
"focus_count": len(self._focus),
|
||||
"max_hops": self._max_hops,
|
||||
"fragment_count": len(fragments),
|
||||
},
|
||||
)
|
||||
return _pack_budget(sorted_frags, budget)
|
||||
|
||||
def explain(self) -> str:
|
||||
return (
|
||||
"Navigates the UKO node hierarchy with breadth-first discovery "
|
||||
"and depth-first detail retrieval. Prioritises fragments near "
|
||||
f"focus nodes (max {self._max_hops} hops) with higher detail "
|
||||
"depth. Primary strategy for code projects (quality 0.85)."
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Internal helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _extract_keywords(query: str) -> list[str]:
|
||||
"""Extract lowercase keywords from a query string."""
|
||||
return _WORD_RE.findall(query.lower())
|
||||
|
||||
|
||||
def _tokenize(text: str) -> set[str]:
|
||||
"""Tokenise text into a set of lowercase words."""
|
||||
return set(_WORD_RE.findall(text.lower()))
|
||||
|
||||
|
||||
def _keyword_match_score(frag: ContextFragment, keywords: list[str]) -> int:
|
||||
"""Count how many keywords match in fragment content + uko_node."""
|
||||
text = (frag.content + " " + frag.uko_node).lower()
|
||||
return sum(1 for kw in keywords if kw in text)
|
||||
|
||||
|
||||
def _word_density(frag: ContextFragment) -> float:
|
||||
"""Compute word density: unique words / max(token_count, 1)."""
|
||||
words = _tokenize(frag.content)
|
||||
return len(words) / max(frag.token_count, 1)
|
||||
|
||||
|
||||
def _jaccard_similarity(set_a: set[str], set_b: set[str]) -> float:
|
||||
"""Compute Jaccard similarity between two word sets."""
|
||||
if not set_a or not set_b:
|
||||
return 0.0
|
||||
intersection = len(set_a & set_b)
|
||||
union = len(set_a | set_b)
|
||||
return intersection / union if union > 0 else 0.0
|
||||
|
||||
|
||||
def _uri_segments(uri: str) -> list[str]:
|
||||
"""Split a UKO URI into path segments for hierarchy comparison."""
|
||||
# Strip scheme (e.g. "project://")
|
||||
if "://" in uri:
|
||||
uri = uri.split("://", 1)[1]
|
||||
return [seg for seg in uri.replace("\\", "/").split("/") if seg]
|
||||
|
||||
|
||||
def _max_proximity(node_uri: str, focus_nodes: list[str], max_hops: int) -> float:
|
||||
"""Compute the maximum proximity of a node to any focus node.
|
||||
|
||||
Proximity is based on the number of shared URI path segments.
|
||||
Returns 1.0 for exact match, decreasing toward 0.0 as distance
|
||||
increases up to ``max_hops``.
|
||||
"""
|
||||
node_segs = _uri_segments(node_uri)
|
||||
best = 0.0
|
||||
|
||||
for focus in focus_nodes:
|
||||
focus_segs = _uri_segments(focus)
|
||||
# Count shared prefix segments
|
||||
shared = 0
|
||||
for a, b in zip(node_segs, focus_segs, strict=False):
|
||||
if a == b:
|
||||
shared += 1
|
||||
else:
|
||||
break
|
||||
|
||||
# Distance = max of the two tail lengths
|
||||
distance = max(
|
||||
len(node_segs) - shared,
|
||||
len(focus_segs) - shared,
|
||||
)
|
||||
|
||||
if distance <= max_hops:
|
||||
# Proximity: 1.0 for exact match, decaying with distance
|
||||
proximity = 1.0 - (distance / (max_hops + 1))
|
||||
best = max(best, proximity)
|
||||
|
||||
return best
|
||||
@@ -787,3 +787,17 @@ async_job_ttl # noqa: B018, F821
|
||||
validate_phase # noqa: B018, F821
|
||||
validate_payload_json # noqa: B018, F821
|
||||
AsyncJobModel # noqa: B018, F821
|
||||
|
||||
# Context strategies batch 1 -- public API (issue #541)
|
||||
SimpleKeywordStrategy # noqa: B018, F821
|
||||
SemanticEmbeddingStrategy # noqa: B018, F821
|
||||
BreadthDepthNavigatorStrategy # noqa: B018, F821
|
||||
set_query # noqa: B018, F821
|
||||
set_focus # noqa: B018, F821
|
||||
_extract_keywords # noqa: B018, F821
|
||||
_tokenize # noqa: B018, F821
|
||||
_keyword_match_score # noqa: B018, F821
|
||||
_word_density # noqa: B018, F821
|
||||
_jaccard_similarity # noqa: B018, F821
|
||||
_uri_segments # noqa: B018, F821
|
||||
_max_proximity # noqa: B018, F821
|
||||
|
||||
Reference in New Issue
Block a user