feat(acms): implement core ACMS pipeline components (StrategySelector, BudgetAllocator, FragmentScorer, BudgetPacker, FragmentOrderer)
Implemented core ACMS pipeline components: ActorPhaseStrategySelector, SpecBudgetAllocator, RelevanceRecencyPriorityScorer, ConstrainedKnapsackPacker, and PriorityCoherenceOrderer. These components implement the Protocol interfaces from acms_service.py and coordinate strategy selection, budget allocation, fragment scoring, content packing, and fragment ordering to optimize LLM usage. The StrategySelector selects context strategies based on actor type and plan phase with confidence boosts; the BudgetAllocator computes per-strategy budgets using the spec formula (confidence * quality_score proportional allocation); the FragmentScorer scores fragments by a weighted composite of relevance, recency, and priority; the Packer performs greedy knapsack packing respecting max_file_size and max_total_size. The Orderer groups related content to maximize coherence and overall throughput. The work also includes a 44-scenario BDD feature file covering all components and edge cases. All quality gates pass: lint, typecheck, unit tests. ISSUES CLOSED: #10015
This commit is contained in:
Forgejo
@@ -0,0 +1,364 @@
|
||||
@acms @acms_core_pipeline
|
||||
Feature: ACMS Core Pipeline Components
|
||||
As a CleverAgents developer
|
||||
I want core ACMS pipeline components implemented
|
||||
So that the pipeline can select strategies, allocate budgets, score fragments,
|
||||
pack within constraints, and order for optimal LLM consumption
|
||||
|
||||
# ===========================================================================
|
||||
# ActorPhaseStrategySelector
|
||||
# ===========================================================================
|
||||
|
||||
@strategy_selector
|
||||
Scenario: Select strategies with no actor type or plan phase
|
||||
Given a set of core pipeline strategies
|
||||
When I select strategies with no actor_type or plan_phase
|
||||
Then all core strategies with positive confidence should be returned
|
||||
And the core strategies should be sorted by confidence descending
|
||||
|
||||
@strategy_selector
|
||||
Scenario: Select strategies boosts preferred strategies for planner actor
|
||||
Given a set of core pipeline strategies
|
||||
When I select strategies with actor_type "planner" only
|
||||
Then the "relevance" strategy should have boosted confidence
|
||||
|
||||
@strategy_selector
|
||||
Scenario: Select strategies boosts preferred strategies for executor actor
|
||||
Given a set of core pipeline strategies
|
||||
When I select strategies with actor_type "executor" only
|
||||
Then the "tiered" strategy should have boosted confidence
|
||||
|
||||
@strategy_selector
|
||||
Scenario: Select strategies boosts preferred strategies for strategize phase
|
||||
Given a set of core pipeline strategies
|
||||
When I select strategies with plan_phase "strategize" only
|
||||
Then the "relevance" strategy should have boosted confidence
|
||||
|
||||
@strategy_selector
|
||||
Scenario: Select strategies applies both actor and phase boosts
|
||||
Given a set of core pipeline strategies
|
||||
When I select strategies with actor_type "planner" and plan_phase "strategize"
|
||||
Then the "relevance" strategy should have maximum boosted confidence
|
||||
|
||||
@strategy_selector
|
||||
Scenario: Select strategies excludes zero-confidence strategies
|
||||
Given a set of core pipeline strategies including a zero-confidence strategy
|
||||
When I select strategies with no actor_type or plan_phase
|
||||
Then the core zero-confidence strategy should not be in the results
|
||||
|
||||
@strategy_selector
|
||||
Scenario: Select strategies with empty strategy list returns empty
|
||||
Given an empty strategy list
|
||||
When I select strategies with no actor_type or plan_phase
|
||||
Then 0 strategies should be selected
|
||||
|
||||
@strategy_selector
|
||||
Scenario: Confidence is clamped to 1.0 after boost
|
||||
Given a strategy with confidence 0.9
|
||||
When I select strategies with actor_type "planner" and plan_phase "strategize"
|
||||
Then the strategy confidence should not exceed 1.0
|
||||
|
||||
# ===========================================================================
|
||||
# SpecBudgetAllocator
|
||||
# ===========================================================================
|
||||
|
||||
@budget_allocator
|
||||
Scenario: Allocate budget to empty candidates returns empty
|
||||
Given an empty candidate list
|
||||
When I allocate budget 1000 with SpecBudgetAllocator
|
||||
Then 0 allocations should be returned
|
||||
|
||||
@budget_allocator
|
||||
Scenario: Allocate budget to single candidate gives full budget
|
||||
Given a single candidate with confidence 0.8
|
||||
When I allocate budget 1000 with SpecBudgetAllocator
|
||||
Then 1 allocation should be returned
|
||||
And the single allocation should receive 1000 tokens
|
||||
|
||||
@budget_allocator
|
||||
Scenario: Allocate budget proportionally to two candidates
|
||||
Given two candidates with equal confidence 0.5
|
||||
When I allocate budget 1000 with SpecBudgetAllocator
|
||||
Then 2 allocations should be returned
|
||||
And the total allocated tokens should equal 1000
|
||||
|
||||
@budget_allocator
|
||||
Scenario: Allocate budget uses spec formula with quality scores
|
||||
Given two candidates with different quality scores
|
||||
When I allocate budget 1000 with SpecBudgetAllocator
|
||||
Then the higher quality candidate should receive more tokens
|
||||
|
||||
@budget_allocator
|
||||
Scenario: Allocate budget with zero total weight falls back to equal split
|
||||
Given two candidates with zero confidence
|
||||
When I allocate budget 1000 with SpecBudgetAllocator
|
||||
Then 2 allocations should be returned
|
||||
And the total allocated tokens should equal 1000
|
||||
|
||||
@budget_allocator
|
||||
Scenario: Allocate budget with min_useful_budget excludes small candidates
|
||||
Given three candidates where one would receive very few tokens
|
||||
When I allocate budget 100 with SpecBudgetAllocator and min_useful_budget 30
|
||||
Then the small candidate should be excluded from allocations
|
||||
|
||||
@budget_allocator
|
||||
Scenario: Allocate budget total never exceeds budget
|
||||
Given three candidates with varying confidence
|
||||
When I allocate budget 500 with SpecBudgetAllocator
|
||||
Then the total allocated tokens should equal 500
|
||||
|
||||
# ===========================================================================
|
||||
# RelevanceRecencyPriorityScorer
|
||||
# ===========================================================================
|
||||
|
||||
@scorer
|
||||
Scenario: Score empty fragment list returns empty
|
||||
Given an empty core pipeline fragment list
|
||||
When I score the fragments with RelevanceRecencyPriorityScorer
|
||||
Then 0 core scored fragments should be returned
|
||||
|
||||
@scorer
|
||||
Scenario: Score single fragment preserves metadata
|
||||
Given a core pipeline fragment with relevance 0.8
|
||||
When I score the fragments with RelevanceRecencyPriorityScorer
|
||||
Then the scored fragment should have _original_relevance metadata "0.8"
|
||||
|
||||
@scorer
|
||||
Scenario: Score produces composite from relevance recency and priority
|
||||
Given a core pipeline fragment with relevance 0.8 and priority 0.9
|
||||
When I score the fragments with RelevanceRecencyPriorityScorer
|
||||
Then the scored fragment composite score should be between 0.0 and 1.0
|
||||
|
||||
@scorer
|
||||
Scenario: Score is deterministic for identical inputs
|
||||
Given two identical core pipeline fragments
|
||||
When I score both fragment sets with RelevanceRecencyPriorityScorer
|
||||
Then both scored fragments should have identical composite scores
|
||||
|
||||
@scorer
|
||||
Scenario: Score recency normalises across fragment timestamps
|
||||
Given two core pipeline fragments with different timestamps
|
||||
When I score the fragments with RelevanceRecencyPriorityScorer
|
||||
Then the newer fragment should have higher recency score
|
||||
|
||||
@scorer
|
||||
Scenario: Score with same timestamps gives recency 1.0
|
||||
Given two core pipeline fragments with identical timestamps
|
||||
When I score the fragments with RelevanceRecencyPriorityScorer
|
||||
Then all scored fragments should have recency score 1.0
|
||||
|
||||
@scorer
|
||||
Scenario: Score composite is clamped to 1.0
|
||||
Given a core pipeline fragment with maximum relevance and priority
|
||||
When I score the fragments with RelevanceRecencyPriorityScorer
|
||||
Then the scored fragment composite score should be at most 1.0
|
||||
|
||||
@scorer
|
||||
Scenario: Score composite is clamped to 0.0
|
||||
Given a core pipeline fragment with zero relevance and zero priority
|
||||
When I score the fragments with RelevanceRecencyPriorityScorer
|
||||
Then the scored fragment composite score should be at least 0.0
|
||||
|
||||
@scorer
|
||||
Scenario: Score with custom weights uses configured weights
|
||||
Given a core pipeline fragment with relevance 1.0 and priority 0.0
|
||||
And scorer configured with relevance_weight 1.0 recency_weight 0.0 priority_weight 0.0
|
||||
When I score the fragments with custom RelevanceRecencyPriorityScorer
|
||||
Then the scored fragment composite score should be 1.0
|
||||
|
||||
@scorer
|
||||
Scenario: Score extracts priority from metadata
|
||||
Given a core pipeline fragment with metadata priority "0.9"
|
||||
When I score the fragments with RelevanceRecencyPriorityScorer
|
||||
Then the scored fragment should have _score_priority metadata near "0.9"
|
||||
|
||||
@scorer
|
||||
Scenario: Score defaults priority to 0.5 when not in metadata
|
||||
Given a core pipeline fragment with no priority metadata
|
||||
When I score the fragments with RelevanceRecencyPriorityScorer
|
||||
Then the scored fragment should have _score_priority metadata "0.5"
|
||||
|
||||
# ===========================================================================
|
||||
# ConstrainedKnapsackPacker
|
||||
# ===========================================================================
|
||||
|
||||
@packer
|
||||
Scenario: Pack empty fragment list returns empty
|
||||
Given an empty core pipeline fragment list
|
||||
And a core pipeline budget with max_tokens 1000 and reserved_tokens 0
|
||||
When I pack the fragments with ConstrainedKnapsackPacker
|
||||
Then 0 core packed fragments should be returned
|
||||
|
||||
@packer
|
||||
Scenario: Pack fragments within token budget
|
||||
Given the following core pipeline fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/main.py | alpha | 0.9 | 100 | 3 |
|
||||
| project://app/io.py | beta | 0.7 | 100 | 3 |
|
||||
| project://app/util.py | gamma | 0.5 | 100 | 3 |
|
||||
And a core pipeline budget with max_tokens 250 and reserved_tokens 0
|
||||
When I pack the fragments with ConstrainedKnapsackPacker
|
||||
Then 2 core packed fragments should be returned
|
||||
And the core packed total tokens should be at most 250
|
||||
|
||||
@packer
|
||||
Scenario: Pack respects max_file_size constraint
|
||||
Given the following core pipeline fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/main.py | short content | 0.9 | 10 | 3 |
|
||||
| project://app/io.py | this is a much longer text | 0.7 | 20 | 3 |
|
||||
And a core pipeline budget with max_tokens 1000 and reserved_tokens 0
|
||||
And a ConstrainedKnapsackPacker with max_file_size 15
|
||||
When I pack the fragments with ConstrainedKnapsackPacker
|
||||
Then 1 core packed fragment should be returned
|
||||
And the core packed fragment uko_node should be "project://app/main.py"
|
||||
|
||||
@packer
|
||||
Scenario: Pack respects max_total_size constraint
|
||||
Given the following core pipeline fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/main.py | hello | 0.9 | 10 | 3 |
|
||||
| project://app/io.py | world | 0.7 | 10 | 3 |
|
||||
| project://app/util.py | extra | 0.5 | 10 | 3 |
|
||||
And a core pipeline budget with max_tokens 1000 and reserved_tokens 0
|
||||
And a ConstrainedKnapsackPacker with max_total_size 4
|
||||
When I pack the fragments with ConstrainedKnapsackPacker
|
||||
Then 0 core packed fragments should be returned
|
||||
|
||||
@packer
|
||||
Scenario: Pack prefers higher scored fragments
|
||||
Given the following core pipeline fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/main.py | alpha | 0.3 | 100 | 3 |
|
||||
| project://app/io.py | beta | 0.9 | 100 | 3 |
|
||||
And a core pipeline budget with max_tokens 150 and reserved_tokens 0
|
||||
When I pack the fragments with ConstrainedKnapsackPacker
|
||||
Then 1 core packed fragment should be returned
|
||||
And the core packed fragment uko_node should be "project://app/io.py"
|
||||
|
||||
@packer
|
||||
Scenario: Pack with context_view uses view constraints
|
||||
Given the following core pipeline fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/main.py | hello | 0.9 | 10 | 3 |
|
||||
| project://app/io.py | world | 0.7 | 10 | 3 |
|
||||
And a core pipeline budget with max_tokens 1000 and reserved_tokens 0
|
||||
And a ConstrainedKnapsackPacker with context_view max_total_size 4
|
||||
When I pack the fragments with ConstrainedKnapsackPacker
|
||||
Then 0 core packed fragments should be returned
|
||||
|
||||
@packer
|
||||
Scenario: Pack with budget=0 returns empty
|
||||
Given the following core pipeline fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/main.py | hello | 0.9 | 10 | 3 |
|
||||
And a core pipeline budget with max_tokens 1 and reserved_tokens 0
|
||||
When I pack the fragments with ConstrainedKnapsackPacker
|
||||
Then 0 core packed fragments should be returned
|
||||
|
||||
# ===========================================================================
|
||||
# PriorityCoherenceOrderer
|
||||
# ===========================================================================
|
||||
|
||||
@orderer
|
||||
Scenario: Order empty fragment list returns empty
|
||||
Given an empty core pipeline fragment list
|
||||
When I order the fragments with PriorityCoherenceOrderer
|
||||
Then 0 core ordered fragments should be returned
|
||||
|
||||
@orderer
|
||||
Scenario: Order single fragment returns it unchanged
|
||||
Given a core pipeline fragment with relevance 0.8
|
||||
When I order the fragments with PriorityCoherenceOrderer
|
||||
Then 1 core ordered fragment should be returned
|
||||
|
||||
@orderer
|
||||
Scenario: Order preserves all fragments
|
||||
Given the following core pipeline fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/src/main.py | alpha | 0.9 | 10 | 3 |
|
||||
| project://app/src/io.py | beta | 0.7 | 10 | 3 |
|
||||
| project://lib/util/util.py | gamma | 0.5 | 10 | 3 |
|
||||
When I order the fragments with PriorityCoherenceOrderer
|
||||
Then 3 core ordered fragments should be returned
|
||||
|
||||
@orderer
|
||||
Scenario: Order places high-priority fragments first
|
||||
Given the following core pipeline fragments with priorities:
|
||||
| uko_node | content | score | tokens | depth | priority |
|
||||
| project://app/src/main.py | alpha | 0.5 | 10 | 3 | 0.9 |
|
||||
| project://app/src/io.py | beta | 0.9 | 10 | 3 | 0.1 |
|
||||
When I order the fragments with PriorityCoherenceOrderer
|
||||
Then the first core ordered fragment should have uko_node "project://app/src/main.py"
|
||||
|
||||
@orderer
|
||||
Scenario: Order groups related fragments together
|
||||
Given the following core pipeline fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/src/main.py | alpha | 0.9 | 10 | 3 |
|
||||
| project://lib/util/util.py | beta | 0.8 | 10 | 3 |
|
||||
| project://app/src/io.py | gamma | 0.7 | 10 | 3 |
|
||||
When I order the fragments with PriorityCoherenceOrderer
|
||||
Then fragments from "project://app" should be adjacent
|
||||
|
||||
@orderer
|
||||
Scenario: Order with default priority 0.5 falls back to relevance
|
||||
Given the following core pipeline fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/src/main.py | alpha | 0.3 | 10 | 3 |
|
||||
| project://lib/util/util.py | beta | 0.9 | 10 | 3 |
|
||||
When I order the fragments with PriorityCoherenceOrderer
|
||||
Then the first core ordered fragment should have uko_node "project://lib/util/util.py"
|
||||
|
||||
# ===========================================================================
|
||||
# Pipeline Integration
|
||||
# ===========================================================================
|
||||
|
||||
@pipeline_integration
|
||||
Scenario: Inject ActorPhaseStrategySelector into pipeline
|
||||
Given the ACMS pipeline with an ActorPhaseStrategySelector
|
||||
And the following core pipeline fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/main.py | hello | 0.8 | 10 | 3 |
|
||||
When I assemble context through the core pipeline
|
||||
Then the core pipeline output should contain 1 fragment
|
||||
|
||||
@pipeline_integration
|
||||
Scenario: Inject SpecBudgetAllocator into pipeline
|
||||
Given the ACMS pipeline with a SpecBudgetAllocator
|
||||
And the following core pipeline fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/main.py | hello | 0.8 | 10 | 3 |
|
||||
When I assemble context through the core pipeline
|
||||
Then the core pipeline output should contain 1 fragment
|
||||
|
||||
@pipeline_integration
|
||||
Scenario: Inject RelevanceRecencyPriorityScorer into pipeline
|
||||
Given the ACMS pipeline with a RelevanceRecencyPriorityScorer
|
||||
And the following core pipeline fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/main.py | hello | 0.8 | 10 | 3 |
|
||||
When I assemble context through the core pipeline
|
||||
Then the core pipeline output fragments should have updated relevance scores
|
||||
|
||||
@pipeline_integration
|
||||
Scenario: Inject ConstrainedKnapsackPacker into pipeline
|
||||
Given the ACMS pipeline with a ConstrainedKnapsackPacker
|
||||
And the following core pipeline fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/main.py | alpha | 0.9 | 100 | 3 |
|
||||
| project://app/io.py | beta | 0.7 | 100 | 3 |
|
||||
| project://app/util.py | gamma | 0.5 | 100 | 3 |
|
||||
And a core pipeline budget with max_tokens 250 and reserved_tokens 0
|
||||
When I assemble context through the core pipeline with budget
|
||||
Then the core pipeline output total tokens should be at most 250
|
||||
|
||||
@pipeline_integration
|
||||
Scenario: Inject PriorityCoherenceOrderer into pipeline
|
||||
Given the ACMS pipeline with a PriorityCoherenceOrderer
|
||||
And the following core pipeline fragments:
|
||||
| uko_node | content | score | tokens | depth |
|
||||
| project://app/main.py | hello | 0.8 | 10 | 3 |
|
||||
| project://lib/util.py | world | 0.6 | 15 | 5 |
|
||||
When I assemble context through the core pipeline
|
||||
Then the core pipeline output should contain 2 fragments
|
||||
@@ -0,0 +1,858 @@
|
||||
"""Step definitions for features/acms_core_pipeline_components.feature.
|
||||
|
||||
Tests the ACMS core pipeline components directly in-memory.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
|
||||
from behave import given, then, when
|
||||
from behave.runner import Context
|
||||
|
||||
from cleveragents.application.services.acms_core_pipeline import (
|
||||
ActorPhaseStrategySelector,
|
||||
ConstrainedKnapsackPacker,
|
||||
PriorityCoherenceOrderer,
|
||||
RelevanceRecencyPriorityScorer,
|
||||
SpecBudgetAllocator,
|
||||
)
|
||||
from cleveragents.application.services.acms_service import (
|
||||
ACMSPipeline,
|
||||
StrategyCapabilities,
|
||||
)
|
||||
from cleveragents.domain.models.core.context_fragment import (
|
||||
ContextBudget,
|
||||
ContextFragment,
|
||||
FragmentProvenance,
|
||||
)
|
||||
from cleveragents.domain.models.core.context_policy import ContextView
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_DEFAULT_PROVENANCE = FragmentProvenance(resource_uri="test://core-pipeline")
|
||||
_TEST_PLAN_ID = "01JQTESTCP00000000000000CC"
|
||||
|
||||
|
||||
def _make_core_fragment(**kwargs: Any) -> ContextFragment:
|
||||
"""Create a ContextFragment with sensible core pipeline test defaults."""
|
||||
kwargs.setdefault("uko_node", "test://default")
|
||||
kwargs.setdefault("content", "test content")
|
||||
kwargs.setdefault("token_count", 10)
|
||||
kwargs.setdefault("provenance", _DEFAULT_PROVENANCE)
|
||||
return ContextFragment(**kwargs)
|
||||
|
||||
|
||||
class _MockStrategy:
|
||||
"""A simple mock strategy for testing."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
name: str,
|
||||
confidence: float,
|
||||
quality_score: float = 1.0,
|
||||
) -> None:
|
||||
self._name = name
|
||||
self._confidence = confidence
|
||||
self._quality_score = quality_score
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def capabilities(self) -> StrategyCapabilities:
|
||||
return StrategyCapabilities(quality_score=self._quality_score)
|
||||
|
||||
def can_handle(self, request: dict[str, Any]) -> float:
|
||||
return self._confidence
|
||||
|
||||
def assemble(self, fragments: Any, budget: Any) -> list[Any]:
|
||||
return list(fragments)
|
||||
|
||||
def explain(self) -> str:
|
||||
return f"Mock strategy: {self._name}"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ActorPhaseStrategySelector — Given steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("a set of core pipeline strategies")
|
||||
def step_given_core_strategies(context: Context) -> None:
|
||||
context.core_strategies = [
|
||||
_MockStrategy("relevance", 0.8),
|
||||
_MockStrategy("tiered", 0.7),
|
||||
_MockStrategy("recency", 0.6),
|
||||
_MockStrategy("arce", 0.95),
|
||||
]
|
||||
|
||||
|
||||
@given("a set of core pipeline strategies including a zero-confidence strategy")
|
||||
def step_given_core_strategies_with_zero(context: Context) -> None:
|
||||
context.core_strategies = [
|
||||
_MockStrategy("relevance", 0.8),
|
||||
_MockStrategy("zero-strategy", 0.0),
|
||||
]
|
||||
|
||||
|
||||
@given("an empty strategy list")
|
||||
def step_given_empty_strategy_list(context: Context) -> None:
|
||||
context.core_strategies = []
|
||||
|
||||
|
||||
@given("a strategy with confidence {conf:g}")
|
||||
def step_given_strategy_with_confidence(context: Context, conf: float) -> None:
|
||||
context.core_strategies = [
|
||||
_MockStrategy("relevance", conf),
|
||||
]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ActorPhaseStrategySelector — When steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when("I select strategies with no actor_type or plan_phase")
|
||||
def step_select_no_context(context: Context) -> None:
|
||||
selector = ActorPhaseStrategySelector()
|
||||
context.selector_result = selector.select(context.core_strategies, {})
|
||||
|
||||
|
||||
@when('I select strategies with actor_type "{actor_type}" only')
|
||||
def step_select_with_actor_only(context: Context, actor_type: str) -> None:
|
||||
selector = ActorPhaseStrategySelector()
|
||||
context.selector_result = selector.select(
|
||||
context.core_strategies, {"actor_type": actor_type}
|
||||
)
|
||||
|
||||
|
||||
@when('I select strategies with plan_phase "{plan_phase}" only')
|
||||
def step_select_with_phase_only(context: Context, plan_phase: str) -> None:
|
||||
selector = ActorPhaseStrategySelector()
|
||||
context.selector_result = selector.select(
|
||||
context.core_strategies, {"plan_phase": plan_phase}
|
||||
)
|
||||
|
||||
|
||||
@when(
|
||||
'I select strategies with actor_type "{actor_type}" '
|
||||
'and plan_phase "{plan_phase_val}"'
|
||||
)
|
||||
def step_select_with_actor_and_phase(
|
||||
context: Context, actor_type: str, plan_phase_val: str
|
||||
) -> None:
|
||||
selector = ActorPhaseStrategySelector()
|
||||
context.selector_result = selector.select(
|
||||
context.core_strategies,
|
||||
{"actor_type": actor_type, "plan_phase": plan_phase_val},
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ActorPhaseStrategySelector — Then steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then("all core strategies with positive confidence should be returned")
|
||||
def step_all_positive_confidence(context: Context) -> None:
|
||||
result_names = {s.name for s, _ in context.selector_result}
|
||||
for strategy in context.core_strategies:
|
||||
if strategy.can_handle({}) > 0.0:
|
||||
assert strategy.name in result_names, (
|
||||
f"Expected {strategy.name} in results"
|
||||
)
|
||||
|
||||
|
||||
@then("the core strategies should be sorted by confidence descending")
|
||||
def step_sorted_by_confidence(context: Context) -> None:
|
||||
confidences = [c for _, c in context.selector_result]
|
||||
assert confidences == sorted(confidences, reverse=True), (
|
||||
f"Not sorted: {confidences}"
|
||||
)
|
||||
|
||||
|
||||
@then('the "{name}" strategy should have boosted confidence')
|
||||
def step_strategy_boosted(context: Context, name: str) -> None:
|
||||
for strategy, confidence in context.selector_result:
|
||||
if strategy.name == name:
|
||||
base = strategy.can_handle({})
|
||||
assert confidence > base, (
|
||||
f"Expected {name} confidence > {base}, got {confidence}"
|
||||
)
|
||||
return
|
||||
msg = f"Strategy {name} not found in results"
|
||||
raise AssertionError(msg)
|
||||
|
||||
|
||||
@then('the "{name}" strategy should have maximum boosted confidence')
|
||||
def step_strategy_max_boosted(context: Context, name: str) -> None:
|
||||
for strategy, confidence in context.selector_result:
|
||||
if strategy.name == name:
|
||||
base = strategy.can_handle({})
|
||||
assert confidence > base, (
|
||||
f"Expected {name} confidence > {base}, got {confidence}"
|
||||
)
|
||||
assert confidence <= 1.0, f"Confidence {confidence} exceeds 1.0"
|
||||
return
|
||||
msg = f"Strategy {name} not found in results"
|
||||
raise AssertionError(msg)
|
||||
|
||||
|
||||
@then("the core zero-confidence strategy should not be in the results")
|
||||
def step_zero_confidence_excluded(context: Context) -> None:
|
||||
result_names = {s.name for s, _ in context.selector_result}
|
||||
assert "zero-strategy" not in result_names, (
|
||||
"Zero-confidence strategy should be excluded"
|
||||
)
|
||||
|
||||
|
||||
@then("{count:d} strategies should be selected")
|
||||
def step_strategies_count(context: Context, count: int) -> None:
|
||||
assert len(context.selector_result) == count, (
|
||||
f"Expected {count}, got {len(context.selector_result)}"
|
||||
)
|
||||
|
||||
|
||||
@then("the strategy confidence should not exceed 1.0")
|
||||
def step_confidence_not_exceed_one(context: Context) -> None:
|
||||
for _, confidence in context.selector_result:
|
||||
assert confidence <= 1.0, f"Confidence {confidence} exceeds 1.0"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SpecBudgetAllocator — Given steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("an empty candidate list")
|
||||
def step_given_empty_candidates(context: Context) -> None:
|
||||
context.budget_candidates: list[tuple[Any, float]] = []
|
||||
|
||||
|
||||
@given("a single candidate with confidence {conf:g}")
|
||||
def step_given_single_candidate(context: Context, conf: float) -> None:
|
||||
context.budget_candidates = [(_MockStrategy("relevance", conf), conf)]
|
||||
|
||||
|
||||
@given("two candidates with equal confidence {conf:g}")
|
||||
def step_given_two_equal_candidates(context: Context, conf: float) -> None:
|
||||
context.budget_candidates = [
|
||||
(_MockStrategy("relevance", conf), conf),
|
||||
(_MockStrategy("tiered", conf), conf),
|
||||
]
|
||||
|
||||
|
||||
@given("two candidates with different quality scores")
|
||||
def step_given_two_different_quality(context: Context) -> None:
|
||||
context.budget_candidates = [
|
||||
(_MockStrategy("high-quality", 0.8, quality_score=0.9), 0.8),
|
||||
(_MockStrategy("low-quality", 0.8, quality_score=0.1), 0.8),
|
||||
]
|
||||
|
||||
|
||||
@given("two candidates with zero confidence")
|
||||
def step_given_two_zero_confidence(context: Context) -> None:
|
||||
context.budget_candidates = [
|
||||
(_MockStrategy("a", 0.0), 0.0),
|
||||
(_MockStrategy("b", 0.0), 0.0),
|
||||
]
|
||||
|
||||
|
||||
@given("three candidates where one would receive very few tokens")
|
||||
def step_given_three_candidates_one_small(context: Context) -> None:
|
||||
context.budget_candidates = [
|
||||
(_MockStrategy("big-a", 0.9), 0.9),
|
||||
(_MockStrategy("big-b", 0.8), 0.8),
|
||||
(_MockStrategy("tiny", 0.01), 0.01),
|
||||
]
|
||||
|
||||
|
||||
@given("three candidates with varying confidence")
|
||||
def step_given_three_varying_candidates(context: Context) -> None:
|
||||
context.budget_candidates = [
|
||||
(_MockStrategy("a", 0.9), 0.9),
|
||||
(_MockStrategy("b", 0.6), 0.6),
|
||||
(_MockStrategy("c", 0.3), 0.3),
|
||||
]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SpecBudgetAllocator — When steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when("I allocate budget {budget:d} with SpecBudgetAllocator")
|
||||
def step_allocate_budget(context: Context, budget: int) -> None:
|
||||
allocator = SpecBudgetAllocator()
|
||||
context.allocation_result = allocator.allocate(
|
||||
context.budget_candidates, budget
|
||||
)
|
||||
|
||||
|
||||
@when(
|
||||
"I allocate budget {budget:d} with SpecBudgetAllocator "
|
||||
"and min_useful_budget {min_budget:d}"
|
||||
)
|
||||
def step_allocate_budget_with_min(
|
||||
context: Context, budget: int, min_budget: int
|
||||
) -> None:
|
||||
allocator = SpecBudgetAllocator(min_useful_budget=min_budget)
|
||||
context.allocation_result = allocator.allocate(
|
||||
context.budget_candidates, budget
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SpecBudgetAllocator — Then steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then("{count:d} allocation should be returned")
|
||||
def step_allocations_count_singular(context: Context, count: int) -> None:
|
||||
assert len(context.allocation_result) == count, (
|
||||
f"Expected {count}, got {len(context.allocation_result)}"
|
||||
)
|
||||
|
||||
|
||||
@then("{count:d} allocations should be returned")
|
||||
def step_allocations_count_plural(context: Context, count: int) -> None:
|
||||
assert len(context.allocation_result) == count, (
|
||||
f"Expected {count}, got {len(context.allocation_result)}"
|
||||
)
|
||||
|
||||
|
||||
@then("the single allocation should receive {tokens:d} tokens")
|
||||
def step_single_allocation_tokens(context: Context, tokens: int) -> None:
|
||||
assert len(context.allocation_result) == 1
|
||||
_, _, allocated = context.allocation_result[0]
|
||||
assert allocated == tokens, f"Expected {tokens}, got {allocated}"
|
||||
|
||||
|
||||
@then("the total allocated tokens should equal {total:d}")
|
||||
def step_total_allocated_tokens(context: Context, total: int) -> None:
|
||||
actual = sum(t for _, _, t in context.allocation_result)
|
||||
assert actual == total, f"Expected {total}, got {actual}"
|
||||
|
||||
|
||||
@then("the higher quality candidate should receive more tokens")
|
||||
def step_higher_quality_more_tokens(context: Context) -> None:
|
||||
assert len(context.allocation_result) == 2
|
||||
_, _, high_tokens = context.allocation_result[0]
|
||||
_, _, low_tokens = context.allocation_result[1]
|
||||
assert high_tokens > low_tokens, (
|
||||
f"Expected high-quality ({high_tokens}) > low-quality ({low_tokens})"
|
||||
)
|
||||
|
||||
|
||||
@then("the small candidate should be excluded from allocations")
|
||||
def step_small_candidate_excluded(context: Context) -> None:
|
||||
names = {s.name for s, _, _ in context.allocation_result}
|
||||
assert "tiny" not in names, f"Expected 'tiny' to be excluded, got {names}"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# RelevanceRecencyPriorityScorer — Given steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("an empty core pipeline fragment list")
|
||||
def step_given_empty_core_fragments(context: Context) -> None:
|
||||
context.core_fragments: list[ContextFragment] = []
|
||||
|
||||
|
||||
@given("a core pipeline fragment with relevance {rel:g}")
|
||||
def step_given_core_fragment_relevance(context: Context, rel: float) -> None:
|
||||
context.core_fragments = [
|
||||
_make_core_fragment(relevance_score=rel)
|
||||
]
|
||||
|
||||
|
||||
@given("a core pipeline fragment with relevance {rel:g} and priority {pri:g}")
|
||||
def step_given_core_fragment_relevance_priority(
|
||||
context: Context, rel: float, pri: float
|
||||
) -> None:
|
||||
context.core_fragments = [
|
||||
_make_core_fragment(
|
||||
relevance_score=rel,
|
||||
metadata={"priority": str(pri)},
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
@given("two identical core pipeline fragments")
|
||||
def step_given_two_identical_fragments(context: Context) -> None:
|
||||
ts = datetime(2024, 1, 1, tzinfo=UTC)
|
||||
frag = _make_core_fragment(relevance_score=0.7, created_at=ts)
|
||||
context.core_fragments = [frag]
|
||||
context.core_fragments2 = [frag]
|
||||
|
||||
|
||||
@given("two core pipeline fragments with different timestamps")
|
||||
def step_given_two_different_timestamps(context: Context) -> None:
|
||||
ts_old = datetime(2024, 1, 1, tzinfo=UTC)
|
||||
ts_new = datetime(2024, 6, 1, tzinfo=UTC)
|
||||
context.core_fragments = [
|
||||
_make_core_fragment(
|
||||
uko_node="test://old",
|
||||
relevance_score=0.5,
|
||||
created_at=ts_old,
|
||||
),
|
||||
_make_core_fragment(
|
||||
uko_node="test://new",
|
||||
relevance_score=0.5,
|
||||
created_at=ts_new,
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
@given("two core pipeline fragments with identical timestamps")
|
||||
def step_given_two_identical_timestamps(context: Context) -> None:
|
||||
ts = datetime(2024, 1, 1, tzinfo=UTC)
|
||||
context.core_fragments = [
|
||||
_make_core_fragment(
|
||||
uko_node="test://a",
|
||||
relevance_score=0.5,
|
||||
created_at=ts,
|
||||
),
|
||||
_make_core_fragment(
|
||||
uko_node="test://b",
|
||||
relevance_score=0.7,
|
||||
created_at=ts,
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
@given("a core pipeline fragment with maximum relevance and priority")
|
||||
def step_given_max_relevance_priority(context: Context) -> None:
|
||||
context.core_fragments = [
|
||||
_make_core_fragment(
|
||||
relevance_score=1.0,
|
||||
metadata={"priority": "1.0"},
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
@given("a core pipeline fragment with zero relevance and zero priority")
|
||||
def step_given_zero_relevance_priority(context: Context) -> None:
|
||||
context.core_fragments = [
|
||||
_make_core_fragment(
|
||||
relevance_score=0.0,
|
||||
metadata={"priority": "0.0"},
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
@given(
|
||||
"scorer configured with relevance_weight {rel:g} recency_weight {rec:g} "
|
||||
"priority_weight {pri:g}"
|
||||
)
|
||||
def step_given_custom_scorer_weights(
|
||||
context: Context, rel: float, rec: float, pri: float
|
||||
) -> None:
|
||||
context.custom_scorer = RelevanceRecencyPriorityScorer(
|
||||
relevance_weight=rel,
|
||||
recency_weight=rec,
|
||||
priority_weight=pri,
|
||||
)
|
||||
|
||||
|
||||
@given('a core pipeline fragment with metadata priority "{priority}"')
|
||||
def step_given_fragment_with_priority_metadata(
|
||||
context: Context, priority: str
|
||||
) -> None:
|
||||
context.core_fragments = [
|
||||
_make_core_fragment(
|
||||
relevance_score=0.5,
|
||||
metadata={"priority": priority},
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
@given("a core pipeline fragment with no priority metadata")
|
||||
def step_given_fragment_no_priority(context: Context) -> None:
|
||||
context.core_fragments = [
|
||||
_make_core_fragment(relevance_score=0.5)
|
||||
]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# RelevanceRecencyPriorityScorer — When steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when("I score the fragments with RelevanceRecencyPriorityScorer")
|
||||
def step_score_with_rrp_scorer(context: Context) -> None:
|
||||
scorer = RelevanceRecencyPriorityScorer()
|
||||
context.core_scored = list(scorer.score(context.core_fragments))
|
||||
|
||||
|
||||
@when("I score both fragment sets with RelevanceRecencyPriorityScorer")
|
||||
def step_score_both_sets(context: Context) -> None:
|
||||
scorer = RelevanceRecencyPriorityScorer()
|
||||
context.core_scored = list(scorer.score(context.core_fragments))
|
||||
context.core_scored2 = list(scorer.score(context.core_fragments2))
|
||||
|
||||
|
||||
@when("I score the fragments with custom RelevanceRecencyPriorityScorer")
|
||||
def step_score_with_custom_scorer(context: Context) -> None:
|
||||
scorer = context.custom_scorer
|
||||
context.core_scored = list(scorer.score(context.core_fragments))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# RelevanceRecencyPriorityScorer — Then steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then("{count:d} core scored fragments should be returned")
|
||||
def step_scored_count_core(context: Context, count: int) -> None:
|
||||
assert len(context.core_scored) == count, (
|
||||
f"Expected {count}, got {len(context.core_scored)}"
|
||||
)
|
||||
|
||||
|
||||
@then('the scored fragment should have _original_relevance metadata "{val}"')
|
||||
def step_scored_original_relevance_core(context: Context, val: str) -> None:
|
||||
assert len(context.core_scored) >= 1
|
||||
meta = context.core_scored[0].metadata
|
||||
assert meta.get("_original_relevance") == val, (
|
||||
f"Expected {val}, got {meta.get('_original_relevance')}"
|
||||
)
|
||||
|
||||
|
||||
@then("the scored fragment composite score should be between 0.0 and 1.0")
|
||||
def step_scored_composite_range(context: Context) -> None:
|
||||
assert len(context.core_scored) >= 1
|
||||
score = context.core_scored[0].relevance_score
|
||||
assert 0.0 <= score <= 1.0, f"Score {score} out of range"
|
||||
|
||||
|
||||
@then("both scored fragments should have identical composite scores")
|
||||
def step_scored_identical(context: Context) -> None:
|
||||
assert len(context.core_scored) >= 1
|
||||
assert len(context.core_scored2) >= 1
|
||||
assert (
|
||||
context.core_scored[0].relevance_score
|
||||
== context.core_scored2[0].relevance_score
|
||||
)
|
||||
|
||||
|
||||
@then("the newer fragment should have higher recency score")
|
||||
def step_newer_higher_recency(context: Context) -> None:
|
||||
assert len(context.core_scored) == 2
|
||||
old_recency = float(context.core_scored[0].metadata.get("_score_recency", "0"))
|
||||
new_recency = float(context.core_scored[1].metadata.get("_score_recency", "0"))
|
||||
assert new_recency > old_recency, (
|
||||
f"Expected new ({new_recency}) > old ({old_recency})"
|
||||
)
|
||||
|
||||
|
||||
@then("all scored fragments should have recency score 1.0")
|
||||
def step_all_recency_one(context: Context) -> None:
|
||||
for frag in context.core_scored:
|
||||
recency = float(frag.metadata.get("_score_recency", "0"))
|
||||
assert abs(recency - 1.0) < 1e-6, f"Expected recency 1.0, got {recency}"
|
||||
|
||||
|
||||
@then("the scored fragment composite score should be at most 1.0")
|
||||
def step_scored_at_most_one(context: Context) -> None:
|
||||
assert len(context.core_scored) >= 1
|
||||
assert context.core_scored[0].relevance_score <= 1.0
|
||||
|
||||
|
||||
@then("the scored fragment composite score should be at least 0.0")
|
||||
def step_scored_at_least_zero(context: Context) -> None:
|
||||
assert len(context.core_scored) >= 1
|
||||
assert context.core_scored[0].relevance_score >= 0.0
|
||||
|
||||
|
||||
@then("the scored fragment composite score should be {expected:g}")
|
||||
def step_scored_exact_composite(context: Context, expected: float) -> None:
|
||||
assert len(context.core_scored) >= 1
|
||||
actual = context.core_scored[0].relevance_score
|
||||
assert abs(actual - expected) < 1e-4, f"Expected {expected}, got {actual}"
|
||||
|
||||
|
||||
@then('the scored fragment should have _score_priority metadata near "{val}"')
|
||||
def step_scored_priority_near(context: Context, val: str) -> None:
|
||||
assert len(context.core_scored) >= 1
|
||||
meta = context.core_scored[0].metadata
|
||||
actual = float(meta.get("_score_priority", "0"))
|
||||
expected = float(val)
|
||||
assert abs(actual - expected) < 0.01, f"Expected ~{expected}, got {actual}"
|
||||
|
||||
|
||||
@then('the scored fragment should have _score_priority metadata "{val}"')
|
||||
def step_scored_priority_exact(context: Context, val: str) -> None:
|
||||
assert len(context.core_scored) >= 1
|
||||
meta = context.core_scored[0].metadata
|
||||
actual = float(meta.get("_score_priority", "0"))
|
||||
expected = float(val)
|
||||
assert abs(actual - expected) < 0.01, f"Expected {expected}, got {actual}"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ConstrainedKnapsackPacker — Given steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("the following core pipeline fragments:")
|
||||
def step_given_core_pipeline_fragments(context: Context) -> None:
|
||||
frags: list[ContextFragment] = []
|
||||
for row in context.table:
|
||||
frags.append(
|
||||
_make_core_fragment(
|
||||
uko_node=row["uko_node"],
|
||||
content=row["content"],
|
||||
relevance_score=float(row["score"]),
|
||||
token_count=int(row["tokens"]),
|
||||
detail_depth=int(row["depth"]),
|
||||
)
|
||||
)
|
||||
context.core_fragments = frags
|
||||
|
||||
|
||||
@given("the following core pipeline fragments with priorities:")
|
||||
def step_given_core_fragments_with_priorities(context: Context) -> None:
|
||||
frags: list[ContextFragment] = []
|
||||
for row in context.table:
|
||||
frags.append(
|
||||
_make_core_fragment(
|
||||
uko_node=row["uko_node"],
|
||||
content=row["content"],
|
||||
relevance_score=float(row["score"]),
|
||||
token_count=int(row["tokens"]),
|
||||
detail_depth=int(row["depth"]),
|
||||
metadata={"priority": row["priority"]},
|
||||
)
|
||||
)
|
||||
context.core_fragments = frags
|
||||
|
||||
|
||||
@given(
|
||||
"a core pipeline budget with max_tokens {max_t:d} and reserved_tokens {res_t:d}"
|
||||
)
|
||||
def step_given_core_budget(context: Context, max_t: int, res_t: int) -> None:
|
||||
context.core_budget = ContextBudget(max_tokens=max_t, reserved_tokens=res_t)
|
||||
|
||||
|
||||
@given("a ConstrainedKnapsackPacker with max_file_size {max_size:d}")
|
||||
def step_given_packer_max_file_size(context: Context, max_size: int) -> None:
|
||||
context.core_packer = ConstrainedKnapsackPacker(max_file_size=max_size)
|
||||
|
||||
|
||||
@given("a ConstrainedKnapsackPacker with max_total_size {max_size:d}")
|
||||
def step_given_packer_max_total_size(context: Context, max_size: int) -> None:
|
||||
context.core_packer = ConstrainedKnapsackPacker(max_total_size=max_size)
|
||||
|
||||
|
||||
@given("a ConstrainedKnapsackPacker with context_view max_total_size {max_size:d}")
|
||||
def step_given_packer_context_view(context: Context, max_size: int) -> None:
|
||||
view = ContextView(max_total_size=max_size)
|
||||
context.core_packer = ConstrainedKnapsackPacker(context_view=view)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ConstrainedKnapsackPacker — When steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when("I pack the fragments with ConstrainedKnapsackPacker")
|
||||
def step_pack_with_constrained_packer(context: Context) -> None:
|
||||
packer = getattr(context, "core_packer", ConstrainedKnapsackPacker())
|
||||
budget = getattr(
|
||||
context, "core_budget", ContextBudget(max_tokens=100000, reserved_tokens=0)
|
||||
)
|
||||
context.core_packed = list(packer.pack(context.core_fragments, budget))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ConstrainedKnapsackPacker — Then steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then("{count:d} core packed fragment should be returned")
|
||||
def step_packed_count_singular_core(context: Context, count: int) -> None:
|
||||
assert len(context.core_packed) == count, (
|
||||
f"Expected {count}, got {len(context.core_packed)}"
|
||||
)
|
||||
|
||||
|
||||
@then("{count:d} core packed fragments should be returned")
|
||||
def step_packed_count_plural_core(context: Context, count: int) -> None:
|
||||
assert len(context.core_packed) == count, (
|
||||
f"Expected {count}, got {len(context.core_packed)}"
|
||||
)
|
||||
|
||||
|
||||
@then("the core packed total tokens should be at most {max_tokens:d}")
|
||||
def step_packed_total_at_most_core(context: Context, max_tokens: int) -> None:
|
||||
total = sum(f.token_count for f in context.core_packed)
|
||||
assert total <= max_tokens, f"Expected <= {max_tokens}, got {total}"
|
||||
|
||||
|
||||
@then('the core packed fragment uko_node should be "{expected}"')
|
||||
def step_packed_single_node_core(context: Context, expected: str) -> None:
|
||||
assert len(context.core_packed) >= 1
|
||||
assert context.core_packed[0].uko_node == expected, (
|
||||
f"Expected {expected}, got {context.core_packed[0].uko_node}"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# PriorityCoherenceOrderer — When steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when("I order the fragments with PriorityCoherenceOrderer")
|
||||
def step_order_with_priority_orderer(context: Context) -> None:
|
||||
orderer = PriorityCoherenceOrderer()
|
||||
context.core_ordered = list(orderer.order(context.core_fragments))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# PriorityCoherenceOrderer — Then steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then("{count:d} core ordered fragment should be returned")
|
||||
def step_ordered_count_singular(context: Context, count: int) -> None:
|
||||
assert len(context.core_ordered) == count, (
|
||||
f"Expected {count}, got {len(context.core_ordered)}"
|
||||
)
|
||||
|
||||
|
||||
@then("{count:d} core ordered fragments should be returned")
|
||||
def step_ordered_count_plural(context: Context, count: int) -> None:
|
||||
assert len(context.core_ordered) == count, (
|
||||
f"Expected {count}, got {len(context.core_ordered)}"
|
||||
)
|
||||
|
||||
|
||||
@then('the first core ordered fragment should have uko_node "{expected}"')
|
||||
def step_first_ordered_node(context: Context, expected: str) -> None:
|
||||
assert len(context.core_ordered) >= 1
|
||||
actual = context.core_ordered[0].uko_node
|
||||
assert actual == expected, f"Expected {expected}, got {actual}"
|
||||
|
||||
|
||||
@then('fragments from "{prefix}" should be adjacent')
|
||||
def step_fragments_adjacent(context: Context, prefix: str) -> None:
|
||||
matching_indices = [
|
||||
i
|
||||
for i, f in enumerate(context.core_ordered)
|
||||
if f.uko_node.startswith(prefix)
|
||||
]
|
||||
if len(matching_indices) <= 1:
|
||||
return
|
||||
for i in range(len(matching_indices) - 1):
|
||||
assert matching_indices[i + 1] == matching_indices[i] + 1, (
|
||||
f"Fragments from {prefix} are not adjacent: {matching_indices}"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pipeline Integration — Given steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given("the ACMS pipeline with an ActorPhaseStrategySelector")
|
||||
def step_pipeline_with_actor_selector(context: Context) -> None:
|
||||
context.core_pipeline = ACMSPipeline(
|
||||
strategy_selector=ActorPhaseStrategySelector(),
|
||||
)
|
||||
|
||||
|
||||
@given("the ACMS pipeline with a SpecBudgetAllocator")
|
||||
def step_pipeline_with_spec_allocator(context: Context) -> None:
|
||||
context.core_pipeline = ACMSPipeline(
|
||||
budget_allocator=SpecBudgetAllocator(),
|
||||
)
|
||||
|
||||
|
||||
@given("the ACMS pipeline with a RelevanceRecencyPriorityScorer")
|
||||
def step_pipeline_with_rrp_scorer(context: Context) -> None:
|
||||
context.core_pipeline = ACMSPipeline(
|
||||
scorer=RelevanceRecencyPriorityScorer(),
|
||||
)
|
||||
|
||||
|
||||
@given("the ACMS pipeline with a ConstrainedKnapsackPacker")
|
||||
def step_pipeline_with_constrained_packer(context: Context) -> None:
|
||||
context.core_pipeline = ACMSPipeline(
|
||||
packer=ConstrainedKnapsackPacker(),
|
||||
)
|
||||
|
||||
|
||||
@given("the ACMS pipeline with a PriorityCoherenceOrderer")
|
||||
def step_pipeline_with_priority_orderer(context: Context) -> None:
|
||||
context.core_pipeline = ACMSPipeline(
|
||||
orderer=PriorityCoherenceOrderer(),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pipeline Integration — When steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when("I assemble context through the core pipeline")
|
||||
def step_assemble_core_pipeline(context: Context) -> None:
|
||||
budget = getattr(
|
||||
context, "core_budget", ContextBudget(max_tokens=100000, reserved_tokens=0)
|
||||
)
|
||||
payload = context.core_pipeline.assemble(
|
||||
plan_id=_TEST_PLAN_ID,
|
||||
fragments=context.core_fragments,
|
||||
budget=budget,
|
||||
)
|
||||
context.core_payload = payload
|
||||
|
||||
|
||||
@when("I assemble context through the core pipeline with budget")
|
||||
def step_assemble_core_pipeline_with_budget(context: Context) -> None:
|
||||
payload = context.core_pipeline.assemble(
|
||||
plan_id=_TEST_PLAN_ID,
|
||||
fragments=context.core_fragments,
|
||||
budget=context.core_budget,
|
||||
)
|
||||
context.core_payload = payload
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pipeline Integration — Then steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then("the core pipeline output should contain {count:d} fragment")
|
||||
def step_core_pipeline_count_singular(context: Context, count: int) -> None:
|
||||
assert len(context.core_payload.fragments) == count, (
|
||||
f"Expected {count}, got {len(context.core_payload.fragments)}"
|
||||
)
|
||||
|
||||
|
||||
@then("the core pipeline output should contain {count:d} fragments")
|
||||
def step_core_pipeline_count_plural(context: Context, count: int) -> None:
|
||||
assert len(context.core_payload.fragments) == count, (
|
||||
f"Expected {count}, got {len(context.core_payload.fragments)}"
|
||||
)
|
||||
|
||||
|
||||
@then("the core pipeline output fragments should have updated relevance scores")
|
||||
def step_core_pipeline_scorer_applied(context: Context) -> None:
|
||||
for frag in context.core_payload.fragments:
|
||||
assert "_original_relevance" in frag.metadata, (
|
||||
f"Fragment {frag.uko_node} missing _original_relevance metadata"
|
||||
)
|
||||
|
||||
|
||||
@then("the core pipeline output total tokens should be at most {max_tokens:d}")
|
||||
def step_core_pipeline_total_at_most(context: Context, max_tokens: int) -> None:
|
||||
assert context.core_payload.total_tokens <= max_tokens, (
|
||||
f"Expected <= {max_tokens}, got {context.core_payload.total_tokens}"
|
||||
)
|
||||
@@ -0,0 +1,582 @@
|
||||
"""ACMS core pipeline component implementations.
|
||||
|
||||
Provides the five core ACMS pipeline components that form the central
|
||||
assembly logic of the context assembly pipeline:
|
||||
|
||||
1. **ActorPhaseStrategySelector** -- Selects context strategies based on
|
||||
actor type and plan phase from the request context. Applies actor-type
|
||||
and phase-specific confidence boosts to guide strategy selection.
|
||||
|
||||
2. **SpecBudgetAllocator** -- Computes per-strategy token budgets using the
|
||||
spec formula: ``budget_i = total_budget * (confidence_i * quality_i) /
|
||||
sum(confidence_j * quality_j)``. Handles edge cases including empty
|
||||
candidates, zero total weight, and single-candidate scenarios.
|
||||
|
||||
3. **RelevanceRecencyPriorityScorer** -- Scores context fragments by a
|
||||
weighted composite of relevance, recency (normalised from ``created_at``),
|
||||
and priority (from fragment metadata). Produces deterministic scores for
|
||||
identical inputs.
|
||||
|
||||
4. **ConstrainedKnapsackPacker** -- Greedy knapsack algorithm that packs
|
||||
fragments within the token budget while respecting ``max_file_size`` and
|
||||
``max_total_size`` byte-size constraints from a ``ContextView``.
|
||||
|
||||
5. **PriorityCoherenceOrderer** -- Orders packed fragments for optimal LLM
|
||||
consumption by grouping related fragments (same UKO node prefix) and
|
||||
ordering groups by priority then relevance.
|
||||
|
||||
All components satisfy the Protocol interfaces defined in ``acms_service.py``
|
||||
and can be injected into ``ACMSPipeline`` via constructor dependency injection.
|
||||
|
||||
Based on ``docs/specification.md`` SS42630-42653.
|
||||
|
||||
ISSUES CLOSED: #10015
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from collections import defaultdict
|
||||
from collections.abc import Sequence
|
||||
from datetime import datetime
|
||||
from typing import Any, Final
|
||||
|
||||
from cleveragents.application.services.acms_service import (
|
||||
ContextStrategy,
|
||||
)
|
||||
from cleveragents.domain.models.core.context_fragment import (
|
||||
ContextBudget,
|
||||
ContextFragment,
|
||||
)
|
||||
from cleveragents.domain.models.core.context_policy import ContextView
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Configuration constants
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Default scoring weights for RelevanceRecencyPriorityScorer
|
||||
DEFAULT_RELEVANCE_WEIGHT: Final[float] = 0.5
|
||||
DEFAULT_RECENCY_WEIGHT: Final[float] = 0.3
|
||||
DEFAULT_PRIORITY_WEIGHT: Final[float] = 0.2
|
||||
|
||||
# Minimum token count for a fragment to be eligible for packing
|
||||
DEFAULT_MIN_FRAGMENT_TOKENS: Final[int] = 1
|
||||
|
||||
# Actor-type strategy preference mappings
|
||||
_ACTOR_STRATEGY_PREFERENCES: Final[dict[str, list[str]]] = {
|
||||
"planner": ["relevance", "arce", "tiered"],
|
||||
"executor": ["tiered", "relevance", "recency"],
|
||||
"reviewer": ["relevance", "recency", "arce"],
|
||||
"analyst": ["arce", "temporal-archaeology", "relevance"],
|
||||
"corrector": ["plan-decision-context", "relevance", "recency"],
|
||||
}
|
||||
|
||||
# Plan-phase strategy preference mappings
|
||||
_PHASE_STRATEGY_PREFERENCES: Final[dict[str, list[str]]] = {
|
||||
"strategize": ["relevance", "arce"],
|
||||
"execute": ["tiered", "relevance"],
|
||||
"apply": ["tiered", "recency"],
|
||||
"review": ["relevance", "recency"],
|
||||
"correct": ["plan-decision-context", "relevance"],
|
||||
}
|
||||
|
||||
# Confidence boost applied to preferred strategies
|
||||
_PREFERENCE_BOOST: Final[float] = 0.2
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 1. ActorPhaseStrategySelector (StrategySelector protocol)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class ActorPhaseStrategySelector:
|
||||
"""Select context strategies based on actor type and plan phase.
|
||||
|
||||
Evaluates all strategies via ``can_handle`` and applies confidence
|
||||
boosts to strategies preferred by the current actor type and plan
|
||||
phase. The request dict may contain:
|
||||
|
||||
- ``actor_type``: The type of actor making the request.
|
||||
- ``plan_phase``: The current plan phase.
|
||||
|
||||
Implements ``StrategySelector`` protocol from ``acms_service.py``.
|
||||
|
||||
Based on ``docs/specification.md`` SS42630.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
preference_boost: float = _PREFERENCE_BOOST,
|
||||
) -> None:
|
||||
self._preference_boost = preference_boost
|
||||
|
||||
@property
|
||||
def preference_boost(self) -> float:
|
||||
"""Return the configured preference boost value."""
|
||||
return self._preference_boost
|
||||
|
||||
def select(
|
||||
self,
|
||||
strategies: Sequence[ContextStrategy],
|
||||
request: dict[str, Any],
|
||||
) -> list[tuple[ContextStrategy, float]]:
|
||||
"""Return (strategy, confidence) pairs sorted by priority."""
|
||||
actor_type: str = request.get("actor_type", "")
|
||||
plan_phase: str = request.get("plan_phase", "")
|
||||
|
||||
actor_preferred: list[str] = _ACTOR_STRATEGY_PREFERENCES.get(actor_type, [])
|
||||
phase_preferred: list[str] = _PHASE_STRATEGY_PREFERENCES.get(plan_phase, [])
|
||||
|
||||
scored: list[tuple[ContextStrategy, float]] = []
|
||||
|
||||
for strategy in strategies:
|
||||
base_confidence = strategy.can_handle(request)
|
||||
if base_confidence <= 0.0:
|
||||
continue
|
||||
|
||||
effective = base_confidence
|
||||
|
||||
if strategy.name in actor_preferred:
|
||||
effective = min(effective + self._preference_boost, 1.0)
|
||||
|
||||
if strategy.name in phase_preferred:
|
||||
effective = min(effective + self._preference_boost, 1.0)
|
||||
|
||||
scored.append((strategy, effective))
|
||||
|
||||
scored.sort(key=lambda x: x[1], reverse=True)
|
||||
|
||||
logger.debug(
|
||||
"ActorPhaseStrategySelector: selected %d strategies",
|
||||
len(scored),
|
||||
)
|
||||
return scored
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 2. SpecBudgetAllocator (BudgetAllocator protocol)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class SpecBudgetAllocator:
|
||||
"""Compute per-strategy token budgets using the spec formula.
|
||||
|
||||
Implements the spec budget allocation formula:
|
||||
|
||||
budget_i = total_budget * (confidence_i * quality_i) /
|
||||
sum(confidence_j * quality_j for all j)
|
||||
|
||||
Implements ``BudgetAllocator`` protocol from ``acms_service.py``.
|
||||
|
||||
Based on ``docs/specification.md`` SS42632.
|
||||
"""
|
||||
|
||||
def __init__(self, *, min_useful_budget: int = 0) -> None:
|
||||
self._min_useful_budget = min_useful_budget
|
||||
|
||||
@property
|
||||
def min_useful_budget(self) -> int:
|
||||
"""Return the minimum useful budget threshold."""
|
||||
return self._min_useful_budget
|
||||
|
||||
@staticmethod
|
||||
def _weight(strategy: ContextStrategy, confidence: float) -> float:
|
||||
"""Compute allocation weight: confidence * quality_score."""
|
||||
quality: float = getattr(strategy.capabilities, "quality_score", 1.0)
|
||||
return confidence * quality
|
||||
|
||||
def allocate(
|
||||
self,
|
||||
candidates: list[tuple[ContextStrategy, float]],
|
||||
total_budget: int,
|
||||
request: Any = None,
|
||||
) -> list[tuple[ContextStrategy, float, int]]:
|
||||
"""Return (strategy, confidence, allocated_tokens) triples."""
|
||||
if not candidates:
|
||||
return []
|
||||
|
||||
if len(candidates) == 1:
|
||||
strategy, confidence = candidates[0]
|
||||
return [(strategy, confidence, total_budget)]
|
||||
|
||||
working = self._filter_by_min_budget(candidates, total_budget)
|
||||
if not working:
|
||||
best = max(candidates, key=lambda x: x[1])
|
||||
return [(best[0], best[1], total_budget)]
|
||||
|
||||
total_weight = sum(self._weight(s, c) for s, c in working)
|
||||
if total_weight <= 0.0:
|
||||
return self._equal_split(working, total_budget)
|
||||
|
||||
return self._proportional_split(working, total_budget, total_weight)
|
||||
|
||||
def _filter_by_min_budget(
|
||||
self,
|
||||
candidates: list[tuple[ContextStrategy, float]],
|
||||
total_budget: int,
|
||||
) -> list[tuple[ContextStrategy, float]]:
|
||||
"""Remove candidates whose proportional share < min_useful_budget."""
|
||||
if self._min_useful_budget <= 0:
|
||||
return list(candidates)
|
||||
|
||||
total_weight = sum(self._weight(s, c) for s, c in candidates)
|
||||
if total_weight <= 0.0:
|
||||
return list(candidates)
|
||||
|
||||
kept: list[tuple[ContextStrategy, float]] = []
|
||||
for strategy, confidence in candidates:
|
||||
w = self._weight(strategy, confidence)
|
||||
share = int(total_budget * w / total_weight)
|
||||
if share >= self._min_useful_budget:
|
||||
kept.append((strategy, confidence))
|
||||
|
||||
return kept if kept else list(candidates)
|
||||
|
||||
def _equal_split(
|
||||
self,
|
||||
candidates: list[tuple[ContextStrategy, float]],
|
||||
total_budget: int,
|
||||
) -> list[tuple[ContextStrategy, float, int]]:
|
||||
"""Split budget equally with largest-remainder distribution."""
|
||||
n = len(candidates)
|
||||
share = total_budget // n
|
||||
remainder = total_budget - share * n
|
||||
return [
|
||||
(s, c, share + (1 if i < remainder else 0))
|
||||
for i, (s, c) in enumerate(candidates)
|
||||
]
|
||||
|
||||
def _proportional_split(
|
||||
self,
|
||||
candidates: list[tuple[ContextStrategy, float]],
|
||||
total_budget: int,
|
||||
total_weight: float,
|
||||
) -> list[tuple[ContextStrategy, float, int]]:
|
||||
"""Proportional allocation with largest-remainder rounding."""
|
||||
weights = [self._weight(s, c) for s, c in candidates]
|
||||
raw = [total_budget * w / total_weight for w in weights]
|
||||
floors = [int(r) for r in raw]
|
||||
remainder = total_budget - sum(floors)
|
||||
|
||||
fractions = sorted(
|
||||
((r - f, i) for i, (r, f) in enumerate(zip(raw, floors, strict=True))),
|
||||
reverse=True,
|
||||
)
|
||||
for _, i in fractions[:remainder]:
|
||||
floors[i] += 1
|
||||
|
||||
return [(s, c, floors[i]) for i, (s, c) in enumerate(candidates)]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 3. RelevanceRecencyPriorityScorer (FragmentScorer protocol)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class RelevanceRecencyPriorityScorer:
|
||||
"""Score context fragments by relevance, recency, and priority.
|
||||
|
||||
Computes a weighted composite score:
|
||||
|
||||
composite = (relevance_weight * relevance_score
|
||||
+ recency_weight * recency_norm
|
||||
+ priority_weight * priority)
|
||||
|
||||
Produces deterministic scores for identical inputs.
|
||||
|
||||
Implements ``FragmentScorer`` protocol from ``acms_service.py``.
|
||||
|
||||
Based on ``docs/specification.md`` SS42636.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
relevance_weight: float = DEFAULT_RELEVANCE_WEIGHT,
|
||||
recency_weight: float = DEFAULT_RECENCY_WEIGHT,
|
||||
priority_weight: float = DEFAULT_PRIORITY_WEIGHT,
|
||||
) -> None:
|
||||
self._relevance_weight = relevance_weight
|
||||
self._recency_weight = recency_weight
|
||||
self._priority_weight = priority_weight
|
||||
|
||||
@property
|
||||
def relevance_weight(self) -> float:
|
||||
"""Return the configured relevance weight."""
|
||||
return self._relevance_weight
|
||||
|
||||
@property
|
||||
def recency_weight(self) -> float:
|
||||
"""Return the configured recency weight."""
|
||||
return self._recency_weight
|
||||
|
||||
@property
|
||||
def priority_weight(self) -> float:
|
||||
"""Return the configured priority weight."""
|
||||
return self._priority_weight
|
||||
|
||||
def score(
|
||||
self,
|
||||
fragments: Sequence[ContextFragment],
|
||||
) -> Sequence[ContextFragment]:
|
||||
"""Return fragments re-scored with composite relevance scores."""
|
||||
if not fragments:
|
||||
return list(fragments)
|
||||
|
||||
timestamps = [f.created_at for f in fragments]
|
||||
min_ts = min(timestamps)
|
||||
max_ts = max(timestamps)
|
||||
ts_range = (max_ts - min_ts).total_seconds()
|
||||
|
||||
scored: list[ContextFragment] = []
|
||||
for frag in fragments:
|
||||
recency_norm = self._recency_norm(frag.created_at, min_ts, ts_range)
|
||||
priority = self._extract_priority(frag)
|
||||
composite = self._compute_composite(
|
||||
frag.relevance_score, recency_norm, priority
|
||||
)
|
||||
|
||||
new_meta = dict(frag.metadata)
|
||||
new_meta["_original_relevance"] = str(frag.relevance_score)
|
||||
new_meta["_score_relevance"] = str(round(frag.relevance_score, 6))
|
||||
new_meta["_score_recency"] = str(round(recency_norm, 6))
|
||||
new_meta["_score_priority"] = str(round(priority, 6))
|
||||
|
||||
scored.append(
|
||||
ContextFragment(
|
||||
fragment_id=frag.fragment_id,
|
||||
uko_node=frag.uko_node,
|
||||
content=frag.content,
|
||||
detail_depth=frag.detail_depth,
|
||||
token_count=frag.token_count,
|
||||
relevance_score=composite,
|
||||
provenance=frag.provenance,
|
||||
strategy_source=frag.strategy_source,
|
||||
tier=frag.tier,
|
||||
metadata=new_meta,
|
||||
created_at=frag.created_at,
|
||||
)
|
||||
)
|
||||
|
||||
return scored
|
||||
|
||||
def _recency_norm(
|
||||
self,
|
||||
created_at: datetime,
|
||||
min_ts: datetime,
|
||||
ts_range: float,
|
||||
) -> float:
|
||||
"""Compute normalised recency score in [0.0, 1.0]."""
|
||||
if ts_range <= 0.0:
|
||||
return 1.0
|
||||
elapsed = (created_at - min_ts).total_seconds()
|
||||
return elapsed / ts_range
|
||||
|
||||
def _extract_priority(self, frag: ContextFragment) -> float:
|
||||
"""Extract priority from fragment metadata, defaulting to 0.5."""
|
||||
raw = frag.metadata.get("priority", "0.5")
|
||||
try:
|
||||
value = float(raw)
|
||||
return max(0.0, min(1.0, value))
|
||||
except (ValueError, TypeError):
|
||||
return 0.5
|
||||
|
||||
def _compute_composite(
|
||||
self,
|
||||
relevance: float,
|
||||
recency: float,
|
||||
priority: float,
|
||||
) -> float:
|
||||
"""Compute weighted composite score clamped to [0.0, 1.0]."""
|
||||
raw = (
|
||||
self._relevance_weight * relevance
|
||||
+ self._recency_weight * recency
|
||||
+ self._priority_weight * priority
|
||||
)
|
||||
return max(0.0, min(1.0, round(raw, 6)))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 4. ConstrainedKnapsackPacker (BudgetPacker protocol)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class ConstrainedKnapsackPacker:
|
||||
"""Pack fragments within budget respecting max_file_size and max_total_size.
|
||||
|
||||
Extends the greedy knapsack algorithm with byte-size constraints
|
||||
from a ``ContextView``.
|
||||
|
||||
Implements ``BudgetPacker`` protocol from ``acms_service.py``.
|
||||
|
||||
Based on ``docs/specification.md`` SS42637.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
max_file_size: int | None = None,
|
||||
max_total_size: int | None = None,
|
||||
min_fragment_tokens: int = DEFAULT_MIN_FRAGMENT_TOKENS,
|
||||
context_view: ContextView | None = None,
|
||||
) -> None:
|
||||
self._max_file_size = max_file_size
|
||||
self._max_total_size = max_total_size
|
||||
self._min_fragment_tokens = min_fragment_tokens
|
||||
self._context_view = context_view
|
||||
|
||||
@property
|
||||
def max_file_size(self) -> int | None:
|
||||
"""Return the configured max_file_size constraint."""
|
||||
return self._max_file_size
|
||||
|
||||
@property
|
||||
def max_total_size(self) -> int | None:
|
||||
"""Return the configured max_total_size constraint."""
|
||||
return self._max_total_size
|
||||
|
||||
@property
|
||||
def min_fragment_tokens(self) -> int:
|
||||
"""Return the minimum fragment token threshold."""
|
||||
return self._min_fragment_tokens
|
||||
|
||||
def pack(
|
||||
self,
|
||||
fragments: Sequence[ContextFragment],
|
||||
budget: ContextBudget,
|
||||
) -> Sequence[ContextFragment]:
|
||||
"""Pack fragments into budget with byte-size constraints."""
|
||||
if not fragments:
|
||||
return list(fragments)
|
||||
|
||||
max_file = self._max_file_size
|
||||
max_total = self._max_total_size
|
||||
|
||||
if self._context_view is not None:
|
||||
if max_file is None:
|
||||
max_file = self._context_view.max_file_size
|
||||
if max_total is None:
|
||||
max_total = self._context_view.max_total_size
|
||||
|
||||
available_tokens = budget.available_tokens
|
||||
|
||||
eligible = [
|
||||
f for f in fragments if f.token_count >= self._min_fragment_tokens
|
||||
]
|
||||
|
||||
if max_file is not None:
|
||||
eligible = [
|
||||
f
|
||||
for f in eligible
|
||||
if len(f.content.encode("utf-8", errors="replace")) <= max_file
|
||||
]
|
||||
|
||||
sorted_frags = sorted(eligible, key=lambda f: f.relevance_score, reverse=True)
|
||||
|
||||
packed: list[ContextFragment] = []
|
||||
used_tokens = 0
|
||||
used_bytes = 0
|
||||
|
||||
for frag in sorted_frags:
|
||||
if used_tokens + frag.token_count > available_tokens:
|
||||
continue
|
||||
|
||||
content_bytes = len(frag.content.encode("utf-8", errors="replace"))
|
||||
if max_total is not None and used_bytes + content_bytes > max_total:
|
||||
continue
|
||||
|
||||
packed.append(frag)
|
||||
used_tokens += frag.token_count
|
||||
used_bytes += content_bytes
|
||||
|
||||
return packed
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 5. PriorityCoherenceOrderer (FragmentOrderer protocol)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class PriorityCoherenceOrderer:
|
||||
"""Order packed fragments for optimal LLM consumption.
|
||||
|
||||
Groups related fragments (same UKO node prefix) and sorts groups by
|
||||
priority then relevance.
|
||||
|
||||
Implements ``FragmentOrderer`` protocol from ``acms_service.py``.
|
||||
|
||||
Based on ``docs/specification.md`` SS42648.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
min_shared_segments: int = 2,
|
||||
) -> None:
|
||||
self._min_shared_segments = min_shared_segments
|
||||
|
||||
@property
|
||||
def min_shared_segments(self) -> int:
|
||||
"""Return the minimum shared URI segments for grouping."""
|
||||
return self._min_shared_segments
|
||||
|
||||
def order(
|
||||
self,
|
||||
fragments: Sequence[ContextFragment],
|
||||
) -> Sequence[ContextFragment]:
|
||||
"""Order fragments for optimal LLM consumption."""
|
||||
if not fragments or len(fragments) <= 1:
|
||||
return list(fragments)
|
||||
|
||||
groups: dict[str, list[ContextFragment]] = defaultdict(list)
|
||||
for frag in fragments:
|
||||
prefix = self._extract_prefix(frag.uko_node)
|
||||
groups[prefix].append(frag)
|
||||
|
||||
def group_sort_key(group: list[ContextFragment]) -> tuple[float, float]:
|
||||
max_priority = max(self._extract_priority(f) for f in group)
|
||||
max_relevance = max(f.relevance_score for f in group)
|
||||
return (-max_priority, -max_relevance)
|
||||
|
||||
sorted_groups = sorted(groups.values(), key=group_sort_key)
|
||||
|
||||
ordered: list[ContextFragment] = []
|
||||
for group in sorted_groups:
|
||||
sorted_group = sorted(
|
||||
group,
|
||||
key=lambda f: (-self._extract_priority(f), -f.relevance_score),
|
||||
)
|
||||
ordered.extend(sorted_group)
|
||||
|
||||
return ordered
|
||||
|
||||
def _extract_prefix(self, uko_node: str) -> str:
|
||||
"""Extract the grouping prefix from a UKO node URI."""
|
||||
segments = _uri_segments(uko_node)
|
||||
prefix_segments = segments[: self._min_shared_segments]
|
||||
return "/".join(prefix_segments) if prefix_segments else uko_node
|
||||
|
||||
@staticmethod
|
||||
def _extract_priority(frag: ContextFragment) -> float:
|
||||
"""Extract priority from fragment metadata, defaulting to 0.5."""
|
||||
raw = frag.metadata.get("priority", "0.5")
|
||||
try:
|
||||
value = float(raw)
|
||||
return max(0.0, min(1.0, value))
|
||||
except (ValueError, TypeError):
|
||||
return 0.5
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Internal helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _uri_segments(uri: str) -> list[str]:
|
||||
"""Split a UKO URI into path segments for hierarchy comparison."""
|
||||
if "://" in uri:
|
||||
uri = uri.split("://", 1)[1]
|
||||
return [seg for seg in uri.replace("\\", "/").split("/") if seg]
|
||||
Reference in New Issue
Block a user