HAL9000
e1a7c7a3e2
Fixes multiple bugs in the Behave step definitions for RelevanceScoringStrategy:
- Use context.strategy_fragments and context.strategy_budget (not context.fragments/context.budget)
- Store assemble result in context.strategy_result (not context.result_fragments)
- Store can_handle result in context.confidence (not context.strategy_confidence)
- Fix step pattern for can_handle with query to use quoted string "{query}"
- Fix step pattern for explain to use quoted string "{text}"
- Fix register_strategy call to pass name and strategy (not just strategy)
- Add RelevanceScoringStrategy to top-level imports
- Add proper type annotations to all new step functions
406 lines
14 KiB
Python
406 lines
14 KiB
Python
"""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
|
|
* **RelevanceScoringStrategy** — relevance scoring via embeddings and metadata
|
|
|
|
Also covers pipeline registration of all four 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,
|
|
RelevanceScoringStrategy,
|
|
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())}"
|
|
)
|
|
|
|
|
|
# ===========================================================================
|
|
# 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)
|