Coverage for src / agent_contracts / config / loader.py: 58%

1"""Configuration loader. 

2 

3YAML-based configuration loading with caching. 

4""" 

5from __future__ import annotations 

6 

7from pathlib import Path 

8 

9import yaml 

10 

11from agent_contracts.config.schema import FrameworkConfig, SupervisorConfig, InterviewConfig 

12from agent_contracts.config.questions import QuestionDefinition, QuestionsConfig 

13 

14 

15# ============================================================================= 

16# Framework Config 

17# ============================================================================= 

18 

19_config: FrameworkConfig | None = None 

20 

21 

22def load_config(path: Path | str) -> FrameworkConfig: 

23 """Load framework configuration from YAML. 

24  

25 Args: 

26 path: Path to YAML config file 

27  

28 Returns: 

29 FrameworkConfig instance 

30 """ 

31 with open(path, encoding="utf-8") as f: 

32 data = yaml.safe_load(f) or {} 

33 

34 # Parse supervisor config 

35 supervisor_data = data.get("supervisor", {}) if isinstance(data, dict) else {} 

36 response_types = data.get("response_types", {}) if isinstance(data, dict) else {} 

37 supervisor = SupervisorConfig( 

38 max_iterations=supervisor_data.get("max_iterations", 10), 

39 terminal_response_types=response_types.get("terminal_states", []), 

40 ) 

41 

42 # Parse interview configs 

43 interview_data = data.get("interview", {}) if isinstance(data, dict) else {} 

44 interview: dict[str, InterviewConfig] = {} 

45 for name, config in interview_data.items(): 

46 interview[name] = InterviewConfig( 

47 max_turns=config.get("max_turns", 10), 

48 max_questions=config.get("max_questions", 5), 

49 ) 

50 

51 return FrameworkConfig( 

52 supervisor=supervisor, 

53 interview=interview, 

54 ) 

55 

56 

57def set_config(config: FrameworkConfig | None) -> None: 

58 """Set global framework configuration. 

59  

60 Args: 

61 config: FrameworkConfig instance (or None to reset) 

62 """ 

63 global _config 

64 _config = config 

65 

66 

67def get_config() -> FrameworkConfig: 

68 """Get current framework configuration. 

69  

70 Returns: 

71 Current FrameworkConfig (or default if not set) 

72 """ 

73 return _config or FrameworkConfig() 

74 

75 

76# ============================================================================= 

77# Questions Config 

78# ============================================================================= 

79 

80_questions: QuestionsConfig | None = None 

81 

82 

83def load_questions(path: Path | str) -> QuestionsConfig: 

84 """Load questions configuration from YAML. 

85  

86 Args: 

87 path: Path to questions YAML file 

88  

89 Returns: 

90 QuestionsConfig (dict of question groups) 

91 """ 

92 with open(path, encoding="utf-8") as f: 

93 data = yaml.safe_load(f) or {} 

94 

95 # Validate with Pydantic 

96 result: QuestionsConfig = {} 

97 if not isinstance(data, dict): 

98 return {} 

99 

100 for group_name, questions in data.items(): 

101 if not isinstance(questions, dict): 

102 continue 

103 result[group_name] = {} 

104 for qid, qdef in questions.items(): 

105 if isinstance(qdef, dict): 

106 result[group_name][qid] = QuestionDefinition(**qdef) 

107 

108 return result 

109 

110 

111def set_questions(config: QuestionsConfig) -> None: 

112 """Set global questions configuration. 

113  

114 Args: 

115 config: QuestionsConfig instance 

116 """ 

117 global _questions 

118 _questions = config 

119 

120 

121def get_question_group(group_name: str) -> dict[str, QuestionDefinition]: 

122 """Get a question group by name. 

123  

124 Args: 

125 group_name: Name of the question group 

126  

127 Returns: 

128 Dict of question ID -> QuestionDefinition 

129 """ 

130 if _questions is None: 

131 return {} 

132 return _questions.get(group_name, {}) 

133 

134 

135def get_question(group_name: str, question_id: str) -> QuestionDefinition | None: 

136 """Get a specific question. 

137  

138 Args: 

139 group_name: Name of the question group 

140 question_id: Question ID 

141  

142 Returns: 

143 QuestionDefinition or None if not found 

144 """ 

145 group = get_question_group(group_name) 

146 return group.get(question_id)