fix(test): move advanced context strategy test doubles to features/mocks
- Extract FakeEmbeddings, RelevanceScoringStrategy, AdaptiveContextSelector, ContextFusionStrategy, and _pack_budget from features/steps/ into new features/mocks/advanced_context_strategies_mocks.py per mock-placement rules - Remove sys.path manipulation from robot/helper_advanced_context_strategies.py; import directly from features.mocks instead of features/steps - Add None guard before selected.assemble() in step_assemble_context_query - Add explicit ValueError for unknown strategy types in step_load_yaml_strategy and load_strategy_from_yaml_impl ISSUES CLOSED: #7574
This commit is contained in:
Forgejo
@@ -6,6 +6,7 @@ Changed `wf10_batch.robot` to be less likely to create files, and
|
||||
`plan_generation_graph.robot` to give more test answers.
|
||||
|
||||
## [Unreleased]
|
||||
- **fix(test): move advanced context strategy test doubles to features/mocks** (#7574): Extracted `FakeEmbeddings`, `RelevanceScoringStrategy`, `AdaptiveContextSelector`, `ContextFusionStrategy`, and `_pack_budget` from `features/steps/advanced_context_strategies_steps.py` into a new `features/mocks/advanced_context_strategies_mocks.py` file per CONTRIBUTING.md mock-placement rules. Updated the Robot Framework helper `robot/helper_advanced_context_strategies.py` to import directly from `features.mocks` rather than manipulating `sys.path` to reach the Behave steps file. Added `None` guard in `step_assemble_context_query` before calling `selected.assemble()`, and added explicit `ValueError` for unknown strategy types in both `step_load_yaml_strategy` and `load_strategy_from_yaml_impl`.
|
||||
- **fix(a2a): regression tests for stale cleveragents.acp removal** (#5566): Added two Behave BDD scenarios verifying that `cleveragents.acp` is not importable (raises `ImportError`) and that `src/cleveragents/acp/` does not exist in the source tree. These guard against regression of the `__pycache__`-based import that allowed the removed ACP module to still be loaded from bytecode after the v3.6.0 rename to `a2a`.
|
||||
- **Virtual Resource Type Base Class** (#8610): Implemented `VirtualResource` base class with two example concrete implementations (`MetricResource`, `APIEndpointResource`) for abstract/computed resources that are derived rather than mapped to physical files. Virtual resources are computed on demand via a `compute_fn` callable. Includes Behave BDD scenarios in `features/resource_virtual_types.feature` exercising construction, computation, name validation, kwargs passthrough, exception handling, string representation, and subclassing. Resource names are validated against `^[a-zA-Z][a-zA-Z0-9_-]*$` (must start with a letter; alphanumeric, hyphens, and underscores otherwise).
|
||||
- **test(e2e): restore complete M2 acceptance test** (#11191): Restored the truncated M2 full actor compiler and LLM integration e2e acceptance test to its complete 10-step form. Added dynamic LLM provider selection via `Resolve LLM Actor` (falls back to Anthropic when OpenAI is unavailable or quota-exhausted), replacing hardcoded `gpt-4` / `openai/gpt-4` references in the actor config and action YAML. Added explicit return-code validation (`Should Be Equal As Integers ${r_actor.rc} 0`) for the actor registration step.
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
# Contributors
|
||||
|
||||
* HAL9000 <HAL9000@cleverthis.com> has contributed fix for #7574 — move advanced context strategy test doubles to features/mocks and resolve lint violation in Robot Framework helper.
|
||||
|
||||
* HAL9000 <HAL9000@cleverthis.com>
|
||||
* Aditya Chhabra <aditya.chhabra@cleverthis.com>
|
||||
* Brent E. Edwards <brent.edwards@cleverthis.com>
|
||||
|
||||
@@ -0,0 +1,159 @@
|
||||
"""Mock implementations for advanced context strategies tests.
|
||||
|
||||
FakeEmbeddings provides deterministic word-overlap embeddings so tests never
|
||||
hit a real embedding API. The three strategy classes are test-only
|
||||
implementations that satisfy the strategy duck-type contract used by the
|
||||
Behave and Robot Framework test layers.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
BreadthDepthNavigatorStrategy,
|
||||
SemanticEmbeddingStrategy,
|
||||
)
|
||||
from cleveragents.domain.models.core.context_fragment import (
|
||||
ContextBudget,
|
||||
ContextFragment,
|
||||
)
|
||||
|
||||
|
||||
class FakeEmbeddings:
|
||||
"""Deterministic fake embeddings for testing without real API calls."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._cache: dict[str, list[float]] = {}
|
||||
|
||||
def embed_query(self, text: str) -> list[float]:
|
||||
if text not in self._cache:
|
||||
hash_val = hash(text) % 1000
|
||||
self._cache[text] = [float((hash_val + i) % 100) / 100.0 for i in range(10)]
|
||||
return self._cache[text]
|
||||
|
||||
def embed_documents(self, texts: list[str]) -> list[list[float]]:
|
||||
return [self.embed_query(text) for text in texts]
|
||||
|
||||
|
||||
def _pack_budget(
|
||||
fragments: list[ContextFragment], budget: ContextBudget
|
||||
) -> list[ContextFragment]:
|
||||
result: list[ContextFragment] = []
|
||||
used_tokens = budget.reserved_tokens
|
||||
|
||||
for frag in fragments:
|
||||
if used_tokens + frag.token_count <= budget.max_tokens:
|
||||
result.append(frag)
|
||||
used_tokens += frag.token_count
|
||||
else:
|
||||
break
|
||||
|
||||
return result
|
||||
|
||||
|
||||
class RelevanceScoringStrategy:
|
||||
"""Strategy that ranks fragments purely by relevance score."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
pass
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return "relevance-scoring"
|
||||
|
||||
def can_handle(self, request: dict[str, Any]) -> float:
|
||||
return 0.5
|
||||
|
||||
def assemble(
|
||||
self,
|
||||
fragments: list[ContextFragment],
|
||||
budget: ContextBudget,
|
||||
) -> list[ContextFragment]:
|
||||
if not fragments:
|
||||
return []
|
||||
sorted_frags = sorted(fragments, key=lambda f: f.relevance_score, reverse=True)
|
||||
return _pack_budget(sorted_frags, budget)
|
||||
|
||||
def explain(self) -> str:
|
||||
return "Ranks fragments purely by relevance score."
|
||||
|
||||
|
||||
class AdaptiveContextSelector:
|
||||
"""Selects the best strategy based on request characteristics."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._strategies: dict[str, Any] = {
|
||||
"semantic-embedding": SemanticEmbeddingStrategy(),
|
||||
"relevance-scoring": RelevanceScoringStrategy(),
|
||||
"breadth-depth-navigator": BreadthDepthNavigatorStrategy(),
|
||||
}
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return "adaptive-selector"
|
||||
|
||||
def select_strategy(self, request: dict[str, Any]) -> tuple[str, Any]:
|
||||
best_name = "relevance-scoring"
|
||||
best_confidence = 0.0
|
||||
|
||||
for name, strategy in self._strategies.items():
|
||||
confidence = strategy.can_handle(request)
|
||||
if confidence > best_confidence:
|
||||
best_confidence = confidence
|
||||
best_name = name
|
||||
|
||||
return best_name, self._strategies[best_name]
|
||||
|
||||
|
||||
class ContextFusionStrategy:
|
||||
"""Fuses results from multiple strategies."""
|
||||
|
||||
def __init__(self, strategy_names: list[str]) -> None:
|
||||
self._strategy_names = strategy_names
|
||||
self._strategies: dict[str, Any] = {
|
||||
"semantic-embedding": SemanticEmbeddingStrategy(),
|
||||
"relevance-scoring": RelevanceScoringStrategy(),
|
||||
"breadth-depth-navigator": BreadthDepthNavigatorStrategy(),
|
||||
}
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return "context-fusion"
|
||||
|
||||
def assemble(
|
||||
self,
|
||||
fragments: list[ContextFragment],
|
||||
budget: ContextBudget,
|
||||
query: str = "",
|
||||
) -> list[ContextFragment]:
|
||||
if not fragments:
|
||||
return []
|
||||
|
||||
all_results: dict[str, ContextFragment] = {}
|
||||
remaining_budget = budget.max_tokens - budget.reserved_tokens
|
||||
|
||||
for strategy_name in self._strategy_names:
|
||||
if remaining_budget <= 0:
|
||||
break
|
||||
|
||||
strategy = self._strategies.get(strategy_name)
|
||||
if not strategy:
|
||||
continue
|
||||
|
||||
if hasattr(strategy, "set_query"):
|
||||
strategy.set_query(query)
|
||||
|
||||
strategy_budget = ContextBudget(
|
||||
max_tokens=remaining_budget,
|
||||
reserved_tokens=0,
|
||||
)
|
||||
|
||||
results = strategy.assemble(fragments, strategy_budget)
|
||||
|
||||
for frag in results:
|
||||
if frag.uko_node not in all_results:
|
||||
all_results[frag.uko_node] = frag
|
||||
remaining_budget -= frag.token_count
|
||||
|
||||
return list(all_results.values())
|
||||
@@ -7,7 +7,6 @@ context fusion strategies using FakeEmbeddings for deterministic behavior.
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from behave import given, then, when
|
||||
from behave.runner import Context
|
||||
@@ -21,6 +20,12 @@ from cleveragents.domain.models.core.context_fragment import (
|
||||
ContextFragment,
|
||||
FragmentProvenance,
|
||||
)
|
||||
from features.mocks.advanced_context_strategies_mocks import (
|
||||
AdaptiveContextSelector,
|
||||
ContextFusionStrategy,
|
||||
FakeEmbeddings,
|
||||
RelevanceScoringStrategy,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -28,167 +33,6 @@ logger = logging.getLogger(__name__)
|
||||
_TEST_PROVENANCE = FragmentProvenance(resource_uri="test://fixture")
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Fixtures and Helpers
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class FakeEmbeddings:
|
||||
"""Deterministic fake embeddings for testing without real API calls."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._cache: dict[str, list[float]] = {}
|
||||
|
||||
def embed_query(self, text: str) -> list[float]:
|
||||
"""Generate deterministic embedding for query."""
|
||||
if text not in self._cache:
|
||||
# Simple deterministic hash-based embedding
|
||||
hash_val = hash(text) % 1000
|
||||
self._cache[text] = [
|
||||
float((hash_val + i) % 100) / 100.0 for i in range(10)
|
||||
]
|
||||
return self._cache[text]
|
||||
|
||||
def embed_documents(self, texts: list[str]) -> list[list[float]]:
|
||||
"""Generate deterministic embeddings for documents."""
|
||||
return [self.embed_query(text) for text in texts]
|
||||
|
||||
|
||||
class RelevanceScoringStrategy:
|
||||
"""Strategy that ranks fragments purely by relevance score."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
pass
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return "relevance-scoring"
|
||||
|
||||
def can_handle(self, request: dict[str, Any]) -> float:
|
||||
"""Always handle with moderate confidence."""
|
||||
return 0.5
|
||||
|
||||
def assemble(
|
||||
self,
|
||||
fragments: list[ContextFragment],
|
||||
budget: ContextBudget,
|
||||
) -> list[ContextFragment]:
|
||||
"""Rank fragments by relevance score."""
|
||||
if not fragments:
|
||||
return []
|
||||
|
||||
sorted_frags = sorted(
|
||||
fragments, key=lambda f: f.relevance_score, reverse=True
|
||||
)
|
||||
return _pack_budget(sorted_frags, budget)
|
||||
|
||||
def explain(self) -> str:
|
||||
return "Ranks fragments purely by relevance score."
|
||||
|
||||
|
||||
class AdaptiveContextSelector:
|
||||
"""Selects the best strategy based on request characteristics."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._strategies: dict[str, Any] = {
|
||||
"semantic-embedding": SemanticEmbeddingStrategy(),
|
||||
"relevance-scoring": RelevanceScoringStrategy(),
|
||||
"breadth-depth-navigator": BreadthDepthNavigatorStrategy(),
|
||||
}
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return "adaptive-selector"
|
||||
|
||||
def select_strategy(self, request: dict[str, Any]) -> tuple[str, Any]:
|
||||
"""Select best strategy based on request."""
|
||||
best_name = "relevance-scoring"
|
||||
best_confidence = 0.0
|
||||
|
||||
for name, strategy in self._strategies.items():
|
||||
confidence = strategy.can_handle(request)
|
||||
if confidence > best_confidence:
|
||||
best_confidence = confidence
|
||||
best_name = name
|
||||
|
||||
return best_name, self._strategies[best_name]
|
||||
|
||||
|
||||
class ContextFusionStrategy:
|
||||
"""Fuses results from multiple strategies."""
|
||||
|
||||
def __init__(self, strategy_names: list[str]) -> None:
|
||||
self._strategy_names = strategy_names
|
||||
self._strategies: dict[str, Any] = {
|
||||
"semantic-embedding": SemanticEmbeddingStrategy(),
|
||||
"relevance-scoring": RelevanceScoringStrategy(),
|
||||
"breadth-depth-navigator": BreadthDepthNavigatorStrategy(),
|
||||
}
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return "context-fusion"
|
||||
|
||||
def assemble(
|
||||
self,
|
||||
fragments: list[ContextFragment],
|
||||
budget: ContextBudget,
|
||||
query: str = "",
|
||||
) -> list[ContextFragment]:
|
||||
"""Fuse results from multiple strategies."""
|
||||
if not fragments:
|
||||
return []
|
||||
|
||||
all_results: dict[str, ContextFragment] = {}
|
||||
remaining_budget = budget.max_tokens - budget.reserved_tokens
|
||||
|
||||
for strategy_name in self._strategy_names:
|
||||
if remaining_budget <= 0:
|
||||
break
|
||||
|
||||
strategy = self._strategies.get(strategy_name)
|
||||
if not strategy:
|
||||
continue
|
||||
|
||||
# Set query if strategy supports it
|
||||
if hasattr(strategy, "set_query"):
|
||||
strategy.set_query(query)
|
||||
|
||||
# Create budget for this strategy
|
||||
strategy_budget = ContextBudget(
|
||||
max_tokens=remaining_budget,
|
||||
reserved_tokens=0,
|
||||
)
|
||||
|
||||
# Get results from strategy
|
||||
results = strategy.assemble(fragments, strategy_budget)
|
||||
|
||||
# Add to all results (deduplication by uko_node)
|
||||
for frag in results:
|
||||
if frag.uko_node not in all_results:
|
||||
all_results[frag.uko_node] = frag
|
||||
remaining_budget -= frag.token_count
|
||||
|
||||
return list(all_results.values())
|
||||
|
||||
|
||||
def _pack_budget(
|
||||
fragments: list[ContextFragment], budget: ContextBudget
|
||||
) -> list[ContextFragment]:
|
||||
"""Pack fragments within token budget."""
|
||||
result: list[ContextFragment] = []
|
||||
used_tokens = budget.reserved_tokens
|
||||
|
||||
for frag in fragments:
|
||||
if used_tokens + frag.token_count <= budget.max_tokens:
|
||||
result.append(frag)
|
||||
used_tokens += frag.token_count
|
||||
else:
|
||||
break
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def _make_fragment(
|
||||
uko_node: str,
|
||||
content: str,
|
||||
@@ -438,6 +282,8 @@ def step_load_yaml_strategy(context: Context) -> None:
|
||||
elif strategy_type == "context-fusion":
|
||||
strategies = context.yaml_config.get("strategies", [])
|
||||
context.loaded_strategy = ContextFusionStrategy(strategies)
|
||||
else:
|
||||
raise ValueError(f"Unknown strategy type: {strategy_type!r}")
|
||||
|
||||
|
||||
@when("I assemble context with query {query}")
|
||||
@@ -466,6 +312,8 @@ def step_assemble_context_query(context: Context, query: str) -> None:
|
||||
if selected and hasattr(selected, "set_query"):
|
||||
selected.set_query(query_str)
|
||||
|
||||
if selected is None:
|
||||
raise AssertionError(f"No strategy found with name {best_name!r}")
|
||||
context.selected_strategy_name = best_name
|
||||
context.results = selected.assemble(context.fragments, context.budget)
|
||||
|
||||
|
||||
@@ -2,10 +2,14 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from features.mocks.advanced_context_strategies_mocks import (
|
||||
AdaptiveContextSelector,
|
||||
ContextFusionStrategy,
|
||||
RelevanceScoringStrategy,
|
||||
)
|
||||
|
||||
from cleveragents.application.services.context_strategies import (
|
||||
BreadthDepthNavigatorStrategy,
|
||||
SemanticEmbeddingStrategy,
|
||||
@@ -19,16 +23,6 @@ from cleveragents.domain.models.core.context_fragment import (
|
||||
# Default provenance used for test fragments (no real resource needed).
|
||||
_TEST_PROVENANCE = FragmentProvenance(resource_uri="test://fixture")
|
||||
|
||||
# Import from step definitions
|
||||
features_path = Path(__file__).parent.parent / "features" / "steps"
|
||||
sys.path.insert(0, str(features_path))
|
||||
|
||||
from advanced_context_strategies_steps import ( # noqa: E402
|
||||
AdaptiveContextSelector,
|
||||
ContextFusionStrategy,
|
||||
RelevanceScoringStrategy,
|
||||
)
|
||||
|
||||
|
||||
def create_semantic_search_strategy_impl() -> SemanticEmbeddingStrategy:
|
||||
"""Create a semantic search strategy with FakeEmbeddings."""
|
||||
@@ -80,9 +74,7 @@ def create_test_fragments_impl(args: list[str]) -> list[ContextFragment]:
|
||||
return fragments
|
||||
|
||||
|
||||
def create_context_budget_impl(
|
||||
max_tokens: int, reserved_tokens: int
|
||||
) -> ContextBudget:
|
||||
def create_context_budget_impl(max_tokens: int, reserved_tokens: int) -> ContextBudget:
|
||||
"""Create a context budget."""
|
||||
return ContextBudget(
|
||||
max_tokens=int(max_tokens),
|
||||
@@ -159,8 +151,8 @@ def load_strategy_from_yaml_impl(config: dict[str, Any]) -> Any:
|
||||
elif strategy_type == "context-fusion":
|
||||
strategies = config.get("strategies", [])
|
||||
return ContextFusionStrategy(strategies)
|
||||
|
||||
return None
|
||||
else:
|
||||
raise ValueError(f"Unknown strategy type: {strategy_type!r}")
|
||||
|
||||
|
||||
def create_context_assembler_with_advanced_strategies_impl() -> dict[str, Any]:
|
||||
@@ -193,7 +185,6 @@ def assemble_context_with_query_impl(
|
||||
"""Assemble context with a query."""
|
||||
request = {"query": query}
|
||||
|
||||
# Select best strategy
|
||||
best_strategy = None
|
||||
best_confidence = 0.0
|
||||
|
||||
@@ -206,7 +197,6 @@ def assemble_context_with_query_impl(
|
||||
if best_strategy is None:
|
||||
best_strategy = assembler["strategies"][0]
|
||||
|
||||
# Set query if needed
|
||||
if hasattr(best_strategy, "set_query"):
|
||||
best_strategy.set_query(query)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user