diff --git a/benchmarks/context_strategies_bench.py b/benchmarks/context_strategies_bench.py new file mode 100644 index 000000000..2aa2dc32d --- /dev/null +++ b/benchmarks/context_strategies_bench.py @@ -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"]}) diff --git a/features/context_strategies.feature b/features/context_strategies.feature new file mode 100644 index 000000000..0a61872af --- /dev/null +++ b/features/context_strategies.feature @@ -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" diff --git a/features/steps/context_strategies_steps.py b/features/steps/context_strategies_steps.py new file mode 100644 index 000000000..9e4c263bb --- /dev/null +++ b/features/steps/context_strategies_steps.py @@ -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())}" + ) diff --git a/robot/context_strategies.robot b/robot/context_strategies.robot new file mode 100644 index 000000000..1051abde3 --- /dev/null +++ b/robot/context_strategies.robot @@ -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 diff --git a/robot/helper_context_strategies.py b/robot/helper_context_strategies.py new file mode 100644 index 000000000..79383bcb8 --- /dev/null +++ b/robot/helper_context_strategies.py @@ -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]} ") + 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()) diff --git a/src/cleveragents/application/services/__init__.py b/src/cleveragents/application/services/__init__.py index 00f293def..2947ab239 100644 --- a/src/cleveragents/application/services/__init__.py +++ b/src/cleveragents/application/services/__init__.py @@ -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", diff --git a/src/cleveragents/application/services/context_strategies.py b/src/cleveragents/application/services/context_strategies.py new file mode 100644 index 000000000..3ed71229d --- /dev/null +++ b/src/cleveragents/application/services/context_strategies.py @@ -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 diff --git a/vulture_whitelist.py b/vulture_whitelist.py index f7d47163a..486dc8e97 100644 --- a/vulture_whitelist.py +++ b/vulture_whitelist.py @@ -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