forked from cleveragents/cleveragents-core
freemo
a41fc02f11
Implement StrategyCoordinator and FusionEngine as named facades over the existing ACMS pipeline components, providing clean public APIs for parallel strategy execution with proportional budget allocation and fragment fusion with dedup/conflict resolution/knapsack packing. Key changes: - Add StrategyCoordinator with parallel execution and confidence-based budget allocation - Add FusionEngine with URI+hash dedup, max-depth conflict resolution, greedy knapsack packing - Add budget overage guard with lowest-relevance fragment dropping - Add per-strategy max caps enforcement - Wire into existing ContextAssemblyPipeline - Add Behave BDD tests, Robot integration tests, ASV benchmarks - Add docs/reference/acms_fusion.md ISSUES CLOSED: #192
353 lines
11 KiB
Python
353 lines
11 KiB
Python
"""Robot Framework helper for ACMS Fusion integration tests.
|
|
|
|
Provides a CLI-style interface for Robot to invoke StrategyCoordinator
|
|
and FusionEngine operations and verify the results.
|
|
|
|
Usage:
|
|
python robot/helper_acms_fusion.py coord-basic
|
|
python robot/helper_acms_fusion.py coord-budget
|
|
python robot/helper_acms_fusion.py coord-caps
|
|
python robot/helper_acms_fusion.py coord-circuit
|
|
python robot/helper_acms_fusion.py fuse-dedup
|
|
python robot/helper_acms_fusion.py fuse-depth
|
|
python robot/helper_acms_fusion.py fuse-pack
|
|
python robot/helper_acms_fusion.py fuse-overage
|
|
python robot/helper_acms_fusion.py integration
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from collections.abc import Callable, Sequence
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
_SRC = str(Path(__file__).resolve().parents[1] / "src")
|
|
if _SRC not in sys.path:
|
|
sys.path.insert(0, _SRC)
|
|
|
|
from cleveragents.application.services.acms_pipeline import ( # noqa: E402
|
|
CircuitBreaker,
|
|
ParallelStrategyExecutor,
|
|
)
|
|
from cleveragents.application.services.acms_service import ( # noqa: E402
|
|
StrategyCapabilities,
|
|
)
|
|
from cleveragents.application.services.fusion_engine import ( # noqa: E402
|
|
FusionConfig,
|
|
FusionEngine,
|
|
)
|
|
from cleveragents.application.services.strategy_coordinator import ( # noqa: E402
|
|
CoordinatorConfig,
|
|
StrategyCoordinator,
|
|
)
|
|
from cleveragents.domain.models.core.context_fragment import ( # noqa: E402
|
|
ContextBudget,
|
|
ContextFragment,
|
|
FragmentProvenance,
|
|
)
|
|
|
|
_DEFAULT_PROV = FragmentProvenance(resource_uri="test://robot-fusion")
|
|
|
|
|
|
def _make_frag(**kwargs: Any) -> ContextFragment:
|
|
kwargs.setdefault("uko_node", "test://robot-fusion")
|
|
kwargs.setdefault("token_count", 10)
|
|
kwargs.setdefault("provenance", _DEFAULT_PROV)
|
|
return ContextFragment(**kwargs)
|
|
|
|
|
|
class _TestStrategy:
|
|
"""Minimal strategy for robot tests."""
|
|
|
|
def __init__(self, name: str = "robot_strat", confidence: float = 0.7) -> None:
|
|
self._name = name
|
|
self._confidence = confidence
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return self._name
|
|
|
|
@property
|
|
def capabilities(self) -> StrategyCapabilities:
|
|
return StrategyCapabilities()
|
|
|
|
def can_handle(self, request: dict[str, Any]) -> float:
|
|
return self._confidence
|
|
|
|
def assemble(
|
|
self,
|
|
fragments: Sequence[ContextFragment],
|
|
budget: ContextBudget,
|
|
) -> Sequence[ContextFragment]:
|
|
result: list[ContextFragment] = []
|
|
total = 0
|
|
for frag in fragments:
|
|
if total + frag.token_count <= budget.available_tokens:
|
|
result.append(frag)
|
|
total += frag.token_count
|
|
return result
|
|
|
|
def explain(self) -> str:
|
|
return f"Robot test strategy '{self._name}'."
|
|
|
|
|
|
def _cmd_coord_basic() -> int:
|
|
"""StrategyCoordinator basic coordination."""
|
|
coordinator = StrategyCoordinator()
|
|
strategies = [_TestStrategy("a", 0.8), _TestStrategy("b", 0.6)]
|
|
frags = [
|
|
_make_frag(
|
|
uko_node="p://a.py", content="alpha", token_count=50, relevance_score=0.9
|
|
),
|
|
_make_frag(
|
|
uko_node="p://b.py", content="beta", token_count=50, relevance_score=0.5
|
|
),
|
|
]
|
|
b = ContextBudget(max_tokens=200, reserved_tokens=0)
|
|
result = coordinator.coordinate(
|
|
request={},
|
|
strategies=strategies,
|
|
budget=b,
|
|
fragments=frags,
|
|
)
|
|
assert len(result.fragments) > 0, "Expected fragments"
|
|
assert len(result.strategies_used) > 0, "Expected strategies used"
|
|
print("coord-basic-ok")
|
|
return 0
|
|
|
|
|
|
def _cmd_coord_budget() -> int:
|
|
"""StrategyCoordinator proportional budget allocation."""
|
|
coordinator = StrategyCoordinator()
|
|
strategies = [_TestStrategy("a", 0.8), _TestStrategy("b", 0.2)]
|
|
frags = [
|
|
_make_frag(
|
|
uko_node="p://a.py", content="alpha", token_count=50, relevance_score=0.9
|
|
),
|
|
]
|
|
b = ContextBudget(max_tokens=1000, reserved_tokens=0)
|
|
result = coordinator.coordinate(
|
|
request={},
|
|
strategies=strategies,
|
|
budget=b,
|
|
fragments=frags,
|
|
)
|
|
total_alloc = sum(a[2] for a in result.allocations)
|
|
assert total_alloc == 1000, f"Total {total_alloc} != 1000"
|
|
first_alloc = result.allocations[0][2]
|
|
assert abs(first_alloc - 800) <= 1, f"First {first_alloc} != ~800"
|
|
print("coord-budget-ok")
|
|
return 0
|
|
|
|
|
|
def _cmd_coord_caps() -> int:
|
|
"""StrategyCoordinator per-strategy max caps."""
|
|
config = CoordinatorConfig(per_strategy_max_cap=300)
|
|
coordinator = StrategyCoordinator(config=config)
|
|
strategies = [_TestStrategy("a", 0.9), _TestStrategy("b", 0.1)]
|
|
frags = [
|
|
_make_frag(
|
|
uko_node="p://a.py", content="alpha", token_count=50, relevance_score=0.9
|
|
),
|
|
]
|
|
b = ContextBudget(max_tokens=1000, reserved_tokens=0)
|
|
result = coordinator.coordinate(
|
|
request={},
|
|
strategies=strategies,
|
|
budget=b,
|
|
fragments=frags,
|
|
)
|
|
for name, _, tokens in result.allocations:
|
|
assert tokens <= 300, f"Strategy {name} got {tokens} > 300"
|
|
print("coord-caps-ok")
|
|
return 0
|
|
|
|
|
|
def _cmd_coord_circuit() -> int:
|
|
"""StrategyCoordinator circuit breaker reporting."""
|
|
cb = CircuitBreaker(failure_threshold=1)
|
|
cb.record_failure("broken")
|
|
executor = ParallelStrategyExecutor(circuit_breaker=cb)
|
|
coordinator = StrategyCoordinator(executor=executor)
|
|
strategies = [_TestStrategy("broken", 0.7)]
|
|
frags = [
|
|
_make_frag(
|
|
uko_node="p://a.py", content="alpha", token_count=50, relevance_score=0.9
|
|
),
|
|
]
|
|
b = ContextBudget(max_tokens=200, reserved_tokens=0)
|
|
result = coordinator.coordinate(
|
|
request={},
|
|
strategies=strategies,
|
|
budget=b,
|
|
fragments=frags,
|
|
)
|
|
assert "broken" in result.circuit_broken, "Expected 'broken' in circuit_broken"
|
|
print("coord-circuit-ok")
|
|
return 0
|
|
|
|
|
|
def _cmd_fuse_dedup() -> int:
|
|
"""FusionEngine deduplication."""
|
|
engine = FusionEngine(config=FusionConfig(min_fragment_tokens=1))
|
|
frags = [
|
|
_make_frag(
|
|
uko_node="p://dup.py", content="same", token_count=50, relevance_score=0.9
|
|
),
|
|
_make_frag(
|
|
uko_node="p://dup.py", content="same", token_count=50, relevance_score=0.7
|
|
),
|
|
_make_frag(
|
|
uko_node="p://other.py",
|
|
content="other",
|
|
token_count=50,
|
|
relevance_score=0.5,
|
|
),
|
|
]
|
|
b = ContextBudget(max_tokens=500, reserved_tokens=0)
|
|
result = engine.fuse(frags, b)
|
|
assert result.dedup_count > 0, f"dedup_count={result.dedup_count}"
|
|
assert len(result.fragments) < len(frags), "Expected fewer fragments"
|
|
print("fuse-dedup-ok")
|
|
return 0
|
|
|
|
|
|
def _cmd_fuse_depth() -> int:
|
|
"""FusionEngine depth resolution."""
|
|
engine = FusionEngine(config=FusionConfig(min_fragment_tokens=1))
|
|
frags = [
|
|
_make_frag(
|
|
uko_node="p://deep.py",
|
|
content="shallow",
|
|
token_count=50,
|
|
detail_depth=2,
|
|
relevance_score=0.8,
|
|
),
|
|
_make_frag(
|
|
uko_node="p://deep.py",
|
|
content="deep detail",
|
|
token_count=80,
|
|
detail_depth=5,
|
|
relevance_score=0.7,
|
|
),
|
|
_make_frag(
|
|
uko_node="p://other.py",
|
|
content="other",
|
|
token_count=50,
|
|
detail_depth=3,
|
|
relevance_score=0.6,
|
|
),
|
|
]
|
|
b = ContextBudget(max_tokens=500, reserved_tokens=0)
|
|
result = engine.fuse(frags, b)
|
|
assert result.depth_resolved_count > 0, (
|
|
f"depth_resolved={result.depth_resolved_count}"
|
|
)
|
|
deep_frags = [f for f in result.fragments if "deep" in f.uko_node]
|
|
if deep_frags:
|
|
assert max(f.detail_depth for f in deep_frags) == 5
|
|
print("fuse-depth-ok")
|
|
return 0
|
|
|
|
|
|
def _cmd_fuse_pack() -> int:
|
|
"""FusionEngine knapsack packing."""
|
|
engine = FusionEngine(config=FusionConfig(min_fragment_tokens=1))
|
|
frags = [
|
|
_make_frag(
|
|
uko_node=f"p://f{i}.py",
|
|
content=f"c{i}",
|
|
token_count=100,
|
|
relevance_score=round(0.9 - i * 0.1, 2),
|
|
)
|
|
for i in range(6)
|
|
]
|
|
b = ContextBudget(max_tokens=400, reserved_tokens=0)
|
|
result = engine.fuse(frags, b)
|
|
assert result.total_tokens <= 400, f"total_tokens={result.total_tokens}"
|
|
print("fuse-pack-ok")
|
|
return 0
|
|
|
|
|
|
def _cmd_fuse_overage() -> int:
|
|
"""FusionEngine budget overage guard."""
|
|
config = FusionConfig(overage_guard_enabled=True, min_fragment_tokens=1)
|
|
engine = FusionEngine(config=config)
|
|
frags = [
|
|
_make_frag(
|
|
uko_node=f"p://ov{i}.py",
|
|
content=f"ov{i}",
|
|
token_count=40,
|
|
relevance_score=round(0.9 - i * 0.2, 2),
|
|
)
|
|
for i in range(4)
|
|
]
|
|
b = ContextBudget(max_tokens=100, reserved_tokens=0)
|
|
result = engine.fuse(frags, b)
|
|
assert result.total_tokens <= 100, f"total_tokens={result.total_tokens}"
|
|
print("fuse-overage-ok")
|
|
return 0
|
|
|
|
|
|
def _cmd_integration() -> int:
|
|
"""Integration: coordinator -> fusion."""
|
|
coordinator = StrategyCoordinator()
|
|
strategies = [_TestStrategy("a", 0.8)]
|
|
frags = [
|
|
_make_frag(
|
|
uko_node="p://a.py", content="alpha", token_count=50, relevance_score=0.9
|
|
),
|
|
_make_frag(
|
|
uko_node="p://b.py", content="beta", token_count=50, relevance_score=0.5
|
|
),
|
|
]
|
|
b = ContextBudget(max_tokens=200, reserved_tokens=0)
|
|
coord_result = coordinator.coordinate(
|
|
request={},
|
|
strategies=strategies,
|
|
budget=b,
|
|
fragments=frags,
|
|
)
|
|
engine = FusionEngine(config=FusionConfig(min_fragment_tokens=1))
|
|
fuse_result = engine.fuse(coord_result.fragments, b)
|
|
assert fuse_result.total_tokens <= 200
|
|
assert len(fuse_result.fragments) > 0
|
|
print("integration-ok")
|
|
return 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI dispatcher
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_COMMANDS: dict[str, Callable[[], int]] = {
|
|
"coord-basic": _cmd_coord_basic,
|
|
"coord-budget": _cmd_coord_budget,
|
|
"coord-caps": _cmd_coord_caps,
|
|
"coord-circuit": _cmd_coord_circuit,
|
|
"fuse-dedup": _cmd_fuse_dedup,
|
|
"fuse-depth": _cmd_fuse_depth,
|
|
"fuse-pack": _cmd_fuse_pack,
|
|
"fuse-overage": _cmd_fuse_overage,
|
|
"integration": _cmd_integration,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
|
|
print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr)
|
|
return 1
|
|
try:
|
|
return _COMMANDS[sys.argv[1]]()
|
|
except Exception as exc:
|
|
print(f"FAIL: {exc}", file=sys.stderr)
|
|
import traceback
|
|
|
|
traceback.print_exc()
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|