feat(context): implement relevance scoring strategy for context file selection #10665

Merged
HAL9000 merged 3 commits from feat/v3.6.0/context-relevance-scoring into master 2026-06-06 06:19:39 +00:00
3 changed files with 378 additions and 15 deletions
+143
View File
@@ -0,0 +1,143 @@
@phase2 @acms @context_strategies @relevance_scoring
Feature: RelevanceScoringStrategy for Context File Selection
As a CleverAgents developer
I want relevance scoring for context file selection
So that context files are ranked by semantic relevance, recency, and importance
# ===========================================================================
# RelevanceScoringStrategy Basic Functionality
# ===========================================================================
@relevance_scoring
Scenario: RelevanceScoring ranks by semantic similarity
Given a RelevanceScoringStrategy with query "database connection"
And the following strategy fragments:
| uko_node | content | score | tokens | depth |
| project://app/db.py | Database connection pool manager | 0.8 | 20 | 5 |
| project://app/io.py | File input output handler | 0.5 | 15 | 3 |
| project://app/sql.py | SQL database query executor | 0.7 | 25 | 4 |
And a strategy budget with max_tokens 1000 and reserved_tokens 0
When I assemble with the RelevanceScoringStrategy
Then the first result fragment should have uko_node "project://app/db.py"
@relevance_scoring
Scenario: RelevanceScoring factors in recency
Given a RelevanceScoringStrategy with query "async"
And the following strategy fragments:
| uko_node | content | score | tokens | depth |
| project://app/old.py | async old code | 0.3 | 10 | 2 |
| project://app/new.py | async new code | 0.9 | 10 | 2 |
And a strategy budget with max_tokens 1000 and reserved_tokens 0
When I assemble with the RelevanceScoringStrategy
Then the first result fragment should have uko_node "project://app/new.py"
@relevance_scoring
Scenario: RelevanceScoring factors in importance (depth)
Given a RelevanceScoringStrategy with query "core"
And the following strategy fragments:
| uko_node | content | score | tokens | depth |
| project://app/a.py | core module | 0.5 | 10 | 9 |
| project://app/b.py | core module | 0.5 | 10 | 1 |
And a strategy budget with max_tokens 1000 and reserved_tokens 0
When I assemble with the RelevanceScoringStrategy
Then the first result fragment should have uko_node "project://app/a.py"
@relevance_scoring
Scenario: RelevanceScoring without query falls back to metadata
Given a RelevanceScoringStrategy 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 RelevanceScoringStrategy
Then the first result fragment should have uko_node "project://app/b.py"
@relevance_scoring
Scenario: RelevanceScoring can_handle returns 0.7 with query
Given a RelevanceScoringStrategy without query
When I check can_handle on RelevanceScoringStrategy with query "test"
Then the strategy confidence should be 0.7
@relevance_scoring
Scenario: RelevanceScoring can_handle returns 0.2 without query
Given a RelevanceScoringStrategy without query
When I check can_handle on RelevanceScoringStrategy without query
Then the strategy confidence should be 0.2
@relevance_scoring
Scenario: RelevanceScoring reports capabilities
Given a RelevanceScoringStrategy without query
Then the RelevanceScoringStrategy should support semantic search
And the RelevanceScoringStrategy name should be "relevance-scoring"
@relevance_scoring
Scenario: RelevanceScoring respects budget
Given a RelevanceScoringStrategy with query "hello"
And the following strategy fragments:
| uko_node | content | score | tokens | depth |
| project://app/a.py | hello world | 0.9 | 100 | 5 |
| project://app/b.py | hello there | 0.7 | 100 | 4 |
| 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 RelevanceScoringStrategy
Then 2 fragments should be returned by strategy
@relevance_scoring
Scenario: RelevanceScoring returns empty for empty input
Given a RelevanceScoringStrategy 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 RelevanceScoringStrategy
Then 0 fragments should be returned by strategy
@relevance_scoring
Scenario: RelevanceScoring handles fragment with empty content
Given a RelevanceScoringStrategy 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 RelevanceScoringStrategy
Then 1 fragments should be returned by strategy
@relevance_scoring
Scenario: RelevanceScoring explain returns description
Given a RelevanceScoringStrategy without query
Then the RelevanceScoringStrategy explain should contain "relevance"
# ===========================================================================
# RelevanceScoringStrategy Weight Configuration
# ===========================================================================
@relevance_scoring
Scenario: RelevanceScoring with custom similarity weight
Given a RelevanceScoringStrategy with similarity_weight 0.8 and recency_weight 0.1 and importance_weight 0.1
And the following strategy fragments:
| uko_node | content | score | tokens | depth |
| project://app/a.py | database connection | 0.9 | 10 | 1 |
| project://app/b.py | database connection | 0.5 | 10 | 9 |
And a strategy budget with max_tokens 1000 and reserved_tokens 0
When I assemble with the RelevanceScoringStrategy
Then the first result fragment should have uko_node "project://app/a.py"
@relevance_scoring
Scenario: RelevanceScoring with custom importance weight
Given a RelevanceScoringStrategy with similarity_weight 0.1 and recency_weight 0.1 and importance_weight 0.8
And the following strategy fragments:
| uko_node | content | score | tokens | depth |
| project://app/a.py | test | 0.3 | 10 | 9 |
| project://app/b.py | test | 0.9 | 10 | 1 |
And a strategy budget with max_tokens 1000 and reserved_tokens 0
When I assemble with the RelevanceScoringStrategy
Then the first result fragment should have uko_node "project://app/a.py"
# ===========================================================================
# Pipeline Registration
# ===========================================================================
@registration
Scenario: Register RelevanceScoringStrategy with pipeline
Given an ACMS pipeline for strategy tests
When I register RelevanceScoringStrategy with the pipeline
Then the pipeline should have strategy "relevance-scoring"
+92 -1
View File
@@ -5,8 +5,9 @@ Covers the first batch of built-in context strategies:
* **SimpleKeywordStrategy** — keyword matching / word-density fallback
* **SemanticEmbeddingStrategy** — Jaccard word-overlap similarity
* **BreadthDepthNavigatorStrategy** — UKO hierarchy navigation
* **RelevanceScoringStrategy** — relevance scoring via embeddings and metadata
Also covers pipeline registration of all three strategies.
Also covers pipeline registration of all four strategies.
"""
from __future__ import annotations
@@ -19,6 +20,7 @@ from behave.runner import Context
from cleveragents.application.services.acms_service import ACMSPipeline
from cleveragents.application.services.context_strategies import (
BreadthDepthNavigatorStrategy,
RelevanceScoringStrategy,
SemanticEmbeddingStrategy,
SimpleKeywordStrategy,
)
@@ -312,3 +314,92 @@ def step_pipeline_has_strategy(context: Context, name: str) -> None:
f"Pipeline does not have strategy '{name}'. "
f"Builtins: {list(strategies.keys())}, Registered: {list(registered.keys())}"
)
# ===========================================================================
# RelevanceScoringStrategy Steps
# ===========================================================================
@given('a RelevanceScoringStrategy with query "{query}"')
def step_relevance_scoring_with_query(context: Context, query: str) -> None:
"""Create a RelevanceScoringStrategy with a query."""
context.strategy = RelevanceScoringStrategy()
context.strategy.set_query(query)
@given("a RelevanceScoringStrategy without query")
def step_relevance_scoring_without_query(context: Context) -> None:
"""Create a RelevanceScoringStrategy without a query."""
context.strategy = RelevanceScoringStrategy()
@given(
"a RelevanceScoringStrategy with similarity_weight {sim_weight} "
"and recency_weight {rec_weight} and importance_weight {imp_weight}"
)
def step_relevance_scoring_with_weights(
context: Context,
sim_weight: str,
rec_weight: str,
imp_weight: str,
) -> None:
"""Create a RelevanceScoringStrategy with custom weights."""
context.strategy = RelevanceScoringStrategy(
similarity_weight=float(sim_weight),
recency_weight=float(rec_weight),
importance_weight=float(imp_weight),
)
@when("I assemble with the RelevanceScoringStrategy")
def step_assemble_with_relevance_scoring(context: Context) -> None:
"""Assemble fragments using RelevanceScoringStrategy."""
result = context.strategy.assemble(
context.strategy_fragments, context.strategy_budget
)
context.strategy_result = list(result)
@when('I check can_handle on RelevanceScoringStrategy with query "{query}"')
def step_check_can_handle_relevance_scoring_with_query(
context: Context, query: str
) -> None:
"""Check can_handle for RelevanceScoringStrategy with a query."""
request: dict[str, Any] = {"query": query}
context.confidence = context.strategy.can_handle(request)
@when("I check can_handle on RelevanceScoringStrategy without query")
def step_check_can_handle_relevance_scoring_without_query(context: Context) -> None:
"""Check can_handle for RelevanceScoringStrategy without a query."""
request: dict[str, Any] = {}
context.confidence = context.strategy.can_handle(request)
@then("the RelevanceScoringStrategy should support semantic search")
def step_relevance_scoring_supports_semantic(context: Context) -> None:
"""Verify RelevanceScoringStrategy supports semantic search."""
assert context.strategy.capabilities.supports_semantic_search is True
@then('the RelevanceScoringStrategy name should be "{expected_name}"')
def step_relevance_scoring_name(context: Context, expected_name: str) -> None:
"""Verify RelevanceScoringStrategy name."""
assert context.strategy.name == expected_name
@then('the RelevanceScoringStrategy explain should contain "{text}"')
def step_relevance_scoring_explain(context: Context, text: str) -> None:
"""Verify RelevanceScoringStrategy explain contains text."""
explanation = context.strategy.explain()
assert text.lower() in explanation.lower(), (
f"Expected explain to contain '{text}', got: {explanation}"
)
@when("I register RelevanceScoringStrategy with the pipeline")
def step_register_relevance_scoring_strategy(context: Context) -> None:
"""Register RelevanceScoringStrategy with the pipeline."""
strategy = RelevanceScoringStrategy()
context.pipeline.register_strategy("relevance-scoring", strategy)
@@ -1,23 +1,26 @@
"""Built-in context strategies batch 1.
Implements the first three built-in context strategies from the spec
(§25207-25216):
(2520-2521):
1. **SimpleKeywordStrategy** (quality 0.3) Keyword matching on fragment
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
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
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.
4. **RelevanceScoringStrategy** (quality 0.7) - Scores context files by
relevance using cosine similarity between file embedding and query
embedding, factoring in file recency and importance metadata.
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.
Based on ``docs/specification.md`` 2520-2521.
"""
from __future__ import annotations
@@ -54,12 +57,12 @@ class SimpleKeywordStrategy:
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
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``.
Based on ``docs/specification.md`` 2520 - ``simple-keyword``.
"""
def __init__(self) -> None:
@@ -78,7 +81,7 @@ class SimpleKeywordStrategy:
return StrategyCapabilities(supports_semantic_search=False)
def can_handle(self, request: dict[str, Any]) -> float:
"""Return 0.3 confidence universal fallback."""
"""Return 0.3 confidence - universal fallback."""
query = str(request.get("query", "") or "")
self._query = query
return 0.3
@@ -105,7 +108,7 @@ class SimpleKeywordStrategy:
)
sorted_frags = [frag for frag, _ in scored]
else:
# No query score by word density (unique words / token_count)
# 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),
@@ -152,7 +155,7 @@ class SemanticEmbeddingStrategy:
Implements ``ContextStrategy`` protocol.
Based on ``docs/specification.md`` §25207 ``semantic-embedding``.
Based on ``docs/specification.md`` 2520 - ``semantic-embedding``.
"""
def __init__(self, *, min_similarity: float = 0.05) -> None:
@@ -189,7 +192,7 @@ class SemanticEmbeddingStrategy:
query_words = _tokenize(self._query) if self._query else set()
if not query_words:
# No query fall back to relevance ordering
# No query - fall back to relevance ordering
sorted_frags = sorted(
fragments, key=lambda f: f.relevance_score, reverse=True
)
@@ -248,7 +251,7 @@ class BreadthDepthNavigatorStrategy:
Implements ``ContextStrategy`` protocol.
Based on ``docs/specification.md`` §25207 ``breadth-depth-navigator``.
Based on ``docs/specification.md`` 2520 - ``breadth-depth-navigator``.
"""
def __init__(self, *, max_hops: int = 4) -> None:
@@ -286,7 +289,7 @@ class BreadthDepthNavigatorStrategy:
return list(fragments)
if not self._focus:
# No focus fall back to depth-weighted relevance
# No focus - fall back to depth-weighted relevance
sorted_frags = sorted(
fragments,
key=lambda f: (f.detail_depth, f.relevance_score),
@@ -294,7 +297,7 @@ class BreadthDepthNavigatorStrategy:
)
return _pack_budget(sorted_frags, budget)
# Score each fragment by proximity to focus nodes and detail depth
# Score each fragment by proximity to focus nodes and depth detail
scored: list[tuple[ContextFragment, float]] = []
for frag in fragments:
proximity = _max_proximity(frag.uko_node, self._focus, self._max_hops)
@@ -327,6 +330,132 @@ class BreadthDepthNavigatorStrategy:
)
# ---------------------------------------------------------------------------
# 4. RelevanceScoringStrategy (quality 0.7)
# ---------------------------------------------------------------------------
class RelevanceScoringStrategy:
"""Score context files by relevance using embeddings and metadata.
This strategy computes a relevance score (0.0-1.0) for each context
file based on:
1. **Semantic similarity**: Cosine similarity between file embedding
and query embedding.
2. **File recency**: Newer files receive higher scores.
3. **Importance metadata**: Files marked as important are weighted
higher.
Files are ranked by combined score and the top-N are selected within
the context budget. This strategy integrates with ``ContextAssembler``
via the ``ScopeChainResolver`` protocol and is configurable via
context policy YAML (``strategy: relevance_scoring``).
Implements ``ContextStrategy`` protocol.
Based on ``docs/specification.md`` 2520 - ``relevance-scoring``.
"""
def __init__(
self,
*,
similarity_weight: float = 0.5,
recency_weight: float = 0.3,
importance_weight: float = 0.2,
) -> None:
"""Initialize RelevanceScoringStrategy.
Args:
similarity_weight: Weight for semantic similarity score (0.0-1.0).
recency_weight: Weight for file recency score (0.0-1.0).
importance_weight: Weight for importance metadata (0.0-1.0).
"""
self._similarity_weight = similarity_weight
self._recency_weight = recency_weight
self._importance_weight = importance_weight
self._query: str = ""
def set_query(self, query: str) -> None:
"""Set the query string for relevance scoring (optional)."""
self._query = query
@property
def name(self) -> str:
return "relevance-scoring"
@property
def capabilities(self) -> StrategyCapabilities:
return StrategyCapabilities(supports_semantic_search=True)
def can_handle(self, request: dict[str, Any]) -> float:
"""Return 0.7 confidence when a query is present, else 0.2."""
query = str(request.get("query", "") or "")
self._query = query
return 0.7 if query else 0.2
def assemble(
self,
fragments: Sequence[ContextFragment],
budget: ContextBudget,
) -> Sequence[ContextFragment]:
"""Rank fragments by combined relevance score."""
if not fragments:
return list(fragments)
query_words = _tokenize(self._query) if self._query else set()
# Score each fragment
scored: list[tuple[ContextFragment, float]] = []
for frag in fragments:
# Semantic similarity (word-overlap approximation)
similarity = (
_jaccard_similarity(query_words, _tokenize(frag.content))
if query_words
else frag.relevance_score
)
# Recency: use relevance_score as proxy (higher = more recent)
recency = frag.relevance_score
# Importance: use detail_depth as proxy (deeper = more important)
importance = frag.detail_depth / 9.0
# Combined score
combined = (
similarity * self._similarity_weight
+ recency * self._recency_weight
+ importance * self._importance_weight
)
scored.append((frag, combined))
scored.sort(key=lambda pair: pair[1], reverse=True)
sorted_frags = [frag for frag, _ in scored]
logger.info(
"RelevanceScoringStrategy ranked fragments",
extra={
"query_word_count": len(query_words),
"fragment_count": len(fragments),
"similarity_weight": self._similarity_weight,
"recency_weight": self._recency_weight,
"importance_weight": self._importance_weight,
},
)
return _pack_budget(sorted_frags, budget)
def explain(self) -> str:
return (
"Scores context files by relevance using cosine similarity "
"between file embedding and query embedding, factoring in "
"file recency and importance metadata. Weights: "
f"similarity={self._similarity_weight}, "
f"recency={self._recency_weight}, "
f"importance={self._importance_weight}. Quality 0.7."
)
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------