Coverage for src / agent_contracts / routing.py: 100%

1"""Routing types for traceable supervisor decisions. 

2 

3These types provide structured, explainable routing decisions 

4for debugging and observability. 

5""" 

6from __future__ import annotations 

7 

8from pydantic import BaseModel, Field 

9 

10from typing import TYPE_CHECKING 

11 

12if TYPE_CHECKING: 

13 from agent_contracts.supervisor import SupervisorDecision 

14 

15 

16# ============================================================================= 

17# Traceable Routing Types 

18# ============================================================================= 

19 

20class MatchedRule(BaseModel): 

21 """A matched trigger condition. 

22  

23 Represents a single rule that matched during supervisor evaluation. 

24  

25 Attributes: 

26 node: Node name that matched 

27 condition: Human-readable condition description 

28 priority: Trigger priority (higher = evaluated first) 

29  

30 Example: 

31 rule = MatchedRule( 

32 node="search", 

33 condition="request.action=search", 

34 priority=100, 

35 ) 

36 """ 

37 node: str = Field(description="Node name") 

38 condition: str = Field(description="Human-readable condition description") 

39 priority: int = Field(description="Trigger priority") 

40 

41 

42class RoutingReason(BaseModel): 

43 """Detailed routing decision reason. 

44  

45 Provides structured explanation of why a particular node was selected. 

46  

47 Attributes: 

48 decision_type: Type of decision made 

49 matched_rules: List of rules that matched 

50 llm_used: Whether LLM was used for the decision 

51 llm_reasoning: LLM's reasoning if used 

52  

53 Decision Types: 

54 - terminal_state: Response type triggered exit 

55 - explicit_routing: Answer routed to question owner 

56 - rule_match: TriggerCondition matched 

57 - llm_decision: LLM made the choice 

58 - fallback: No match, using default 

59 """ 

60 decision_type: str = Field( 

61 description="Type of decision: terminal_state, explicit_routing, rule_match, llm_decision, fallback" 

62 ) 

63 matched_rules: list[MatchedRule] = Field( 

64 default_factory=list, 

65 description="List of matched trigger rules" 

66 ) 

67 llm_used: bool = Field(default=False, description="Whether LLM was used for decision") 

68 llm_reasoning: str | None = Field(default=None, description="LLM's reasoning if used") 

69 

70 

71class RoutingDecision(BaseModel): 

72 """Complete routing decision with traceability. 

73  

74 The main output of `GenericSupervisor.decide_with_trace()`. 

75 Provides full visibility into how a routing decision was made. 

76  

77 Attributes: 

78 selected_node: The node that was selected 

79 reason: Detailed reason for the decision 

80  

81 Example: 

82 decision = await supervisor.decide_with_trace(state) 

83  

84 print(f"Selected: {decision.selected_node}") 

85 print(f"Type: {decision.reason.decision_type}") 

86  

87 for rule in decision.reason.matched_rules: 

88 print(f" P{rule.priority}: {rule.node} - {rule.condition}") 

89 """ 

90 selected_node: str = Field(description="Selected node name") 

91 reason: RoutingReason = Field(description="Decision reason details") 

92 

93 def to_supervisor_decision(self) -> "SupervisorDecision": 

94 """Convert to SupervisorDecision for backward compatibility. 

95  

96 Returns: 

97 SupervisorDecision with condensed reasoning string 

98 """ 

99 # Import here to avoid circular dependency 

100 from agent_contracts.supervisor import SupervisorDecision 

101 

102 reasoning_parts = [self.reason.decision_type] 

103 if self.reason.matched_rules: 

104 rules_str = ", ".join(r.node for r in self.reason.matched_rules) 

105 reasoning_parts.append(f"candidates: [{rules_str}]") 

106 if self.reason.llm_used: 

107 reasoning_parts.append("LLM") 

108 return SupervisorDecision( 

109 next_node=self.selected_node, 

110 reasoning=" | ".join(reasoning_parts) 

111 )