138 lines
3.8 KiB
Python
138 lines
3.8 KiB
Python
from memabra.candidate_types import CandidateObject
|
|
from memabra.router import FeatureScoringRouter, TaskContext
|
|
|
|
|
|
def test_feature_scoring_router_computes_score_breakdown_and_selects_best():
|
|
router = FeatureScoringRouter()
|
|
memory = CandidateObject(
|
|
id="mem-1",
|
|
type="memory",
|
|
title="m1",
|
|
summary="s1",
|
|
confidence=0.9,
|
|
success_rate=0.9,
|
|
freshness=0.9,
|
|
cost=0.1,
|
|
risk=0.1,
|
|
)
|
|
tool = CandidateObject(
|
|
id="tool-1",
|
|
type="tool",
|
|
title="t1",
|
|
summary="s1",
|
|
confidence=0.8,
|
|
success_rate=0.8,
|
|
freshness=0.8,
|
|
cost=0.1,
|
|
risk=0.1,
|
|
)
|
|
decision = router.choose(
|
|
TaskContext(user_input="do something"),
|
|
memory_candidates=[memory],
|
|
skill_candidates=[],
|
|
tool_candidates=[tool],
|
|
)
|
|
assert decision.decision_type == "inject_memory"
|
|
assert "mem-1" in decision.score_breakdown
|
|
assert "tool-1" in decision.score_breakdown
|
|
assert decision.score_breakdown["mem-1"] > decision.score_breakdown["tool-1"]
|
|
|
|
|
|
def test_feature_scoring_router_applies_failure_penalty():
|
|
router = FeatureScoringRouter()
|
|
tool_a = CandidateObject(
|
|
id="tool-a",
|
|
type="tool",
|
|
title="ta",
|
|
summary="sa",
|
|
confidence=0.9,
|
|
success_rate=0.9,
|
|
freshness=0.9,
|
|
cost=0.0,
|
|
risk=0.0,
|
|
)
|
|
tool_b = CandidateObject(
|
|
id="tool-b",
|
|
type="tool",
|
|
title="tb",
|
|
summary="sb",
|
|
confidence=0.9,
|
|
success_rate=0.9,
|
|
freshness=0.9,
|
|
cost=0.0,
|
|
risk=0.0,
|
|
)
|
|
context = TaskContext(user_input="run tool", recent_failures=["tool-b"])
|
|
decision = router.choose(
|
|
context,
|
|
memory_candidates=[],
|
|
skill_candidates=[],
|
|
tool_candidates=[tool_a, tool_b],
|
|
)
|
|
assert decision.decision_type == "call_tool"
|
|
assert decision.selected_ids == ["tool-a"]
|
|
assert decision.score_breakdown["tool-b"] < decision.score_breakdown["tool-a"]
|
|
|
|
|
|
def test_feature_scoring_router_emits_composite_action_for_preconditions():
|
|
router = FeatureScoringRouter()
|
|
memory = CandidateObject(
|
|
id="mem-1",
|
|
type="memory",
|
|
title="m1",
|
|
summary="s1",
|
|
confidence=0.7,
|
|
success_rate=0.5,
|
|
freshness=0.3,
|
|
cost=0.0,
|
|
risk=0.0,
|
|
)
|
|
tool = CandidateObject(
|
|
id="tool-1",
|
|
type="tool",
|
|
title="t1",
|
|
summary="s1",
|
|
confidence=0.9,
|
|
success_rate=0.9,
|
|
freshness=0.9,
|
|
cost=0.0,
|
|
risk=0.0,
|
|
preconditions=["memory"],
|
|
)
|
|
decision = router.choose(
|
|
TaskContext(user_input="run tool"),
|
|
memory_candidates=[memory],
|
|
skill_candidates=[],
|
|
tool_candidates=[tool],
|
|
)
|
|
assert decision.decision_type == "composite_action"
|
|
assert len(decision.composite_steps) == 2
|
|
assert decision.composite_steps[0].decision_type == "inject_memory"
|
|
assert decision.composite_steps[0].selected_ids == ["mem-1"]
|
|
assert decision.composite_steps[1].decision_type == "call_tool"
|
|
assert decision.composite_steps[1].selected_ids == ["tool-1"]
|
|
|
|
|
|
def test_feature_scoring_router_fallback_when_precondition_missing():
|
|
router = FeatureScoringRouter()
|
|
tool = CandidateObject(
|
|
id="tool-1",
|
|
type="tool",
|
|
title="t1",
|
|
summary="s1",
|
|
confidence=0.9,
|
|
success_rate=0.9,
|
|
freshness=0.9,
|
|
cost=0.0,
|
|
risk=0.0,
|
|
preconditions=["memory"],
|
|
)
|
|
decision = router.choose(
|
|
TaskContext(user_input="run tool"),
|
|
memory_candidates=[],
|
|
skill_candidates=[],
|
|
tool_candidates=[tool],
|
|
)
|
|
assert decision.decision_type == "call_tool"
|
|
assert decision.selected_ids == ["tool-1"]
|