Coverage for integrations / agent_lightning / config.py: 39.4%

1""" 

2Agent Lightning Configuration 

3 

4Central configuration for Agent Lightning integration with our agent system. 

5""" 

6 

7import os 

8from typing import Dict, Any, Optional 

9 

10# Agent Lightning feature flag 

11AGENT_LIGHTNING_ENABLED = os.getenv('AGENT_LIGHTNING_ENABLED', 'false').lower() == 'true' 

12 

13# Configuration dictionary 

14AGENT_LIGHTNING_CONFIG: Dict[str, Any] = { 

15 # Global settings 

16 'enabled': AGENT_LIGHTNING_ENABLED, 

17 'auto_trace': True, # Enable automatic tracing 

18 'store_backend': os.getenv('AGENT_LIGHTNING_STORE', 'json'), # redis, json, or memory 

19 

20 # Storage paths 

21 'store_path': os.getenv('AGENT_LIGHTNING_STORE_PATH', './agent_data/lightning_store'), 

22 'traces_path': os.getenv('AGENT_LIGHTNING_TRACES_PATH', './agent_data/lightning_traces'), 

23 

24 # Training configuration 

25 'training': { 

26 'enabled': True, 

27 'algorithm': 'prompt_opt', # ppo, prompt_opt, sft 

28 'batch_size': 32, 

29 'learning_rate': 1e-4, 

30 'update_frequency': '1 hour', # How often to retrain 

31 'min_samples': 100, # Minimum samples before training 

32 }, 

33 

34 # Reward configuration 

35 'rewards': { 

36 'task_completion': 1.0, 

37 'task_failure': -0.5, 

38 'tool_use_efficiency': 0.1, 

39 'response_quality': 0.3, 

40 'execution_time': -0.1, # Negative reward for slow execution 

41 'user_feedback': 0.5, 

42 }, 

43 

44 # Agent-specific configuration 

45 'agents': { 

46 'create_recipe_assistant': { 

47 'optimize_prompts': True, 

48 'optimize_tools': False, 

49 'track_tool_usage': True, 

50 'collect_feedback': True, 

51 }, 

52 'reuse_recipe_assistant': { 

53 'optimize_prompts': True, 

54 'optimize_tools': True, 

55 'track_tool_usage': True, 

56 'collect_feedback': True, 

57 }, 

58 'default': { 

59 'optimize_prompts': True, 

60 'optimize_tools': False, 

61 'track_tool_usage': True, 

62 'collect_feedback': False, 

63 } 

64 }, 

65 

66 # Monitoring and logging 

67 'monitoring': { 

68 'enabled': True, 

69 'log_level': 'INFO', 

70 'metrics_interval': 60, # seconds 

71 'save_traces': True, 

72 }, 

73 

74 # Performance settings 

75 'performance': { 

76 'async_emit': True, # Emit events asynchronously 

77 'batch_emit': True, # Batch events before emitting 

78 'batch_size': 10, 

79 'emit_timeout': 5, # seconds 

80 }, 

81 

82 # Integration settings 

83 'integration': { 

84 'autogen_compatible': True, 

85 'task_ledger_integration': True, 

86 'a2a_integration': True, 

87 'ap2_integration': True, 

88 } 

89} 

90 

91 

92def is_enabled() -> bool: 

93 """Check if Agent Lightning is enabled""" 

94 return AGENT_LIGHTNING_CONFIG['enabled'] 

95 

96 

97def get_agent_config(agent_id: str) -> Dict[str, Any]: 

98 """ 

99 Get configuration for a specific agent 

100 

101 Args: 

102 agent_id: Agent identifier 

103 

104 Returns: 

105 Agent-specific configuration 

106 """ 

107 agents_config = AGENT_LIGHTNING_CONFIG.get('agents', {}) 

108 

109 # Try exact match first 

110 if agent_id in agents_config: 

111 return agents_config[agent_id] 

112 

113 # Check for partial match (e.g., "reuse_8888_assistant" matches "reuse_recipe_assistant") 

114 for agent_pattern, config in agents_config.items(): 

115 if agent_pattern in agent_id or agent_id.startswith(agent_pattern.split('_')[0]): 

116 return config 

117 

118 # Return default configuration 

119 return agents_config.get('default', { 

120 'optimize_prompts': True, 

121 'optimize_tools': False, 

122 'track_tool_usage': True, 

123 'collect_feedback': False, 

124 }) 

125 

126 

127def get_reward_value(reward_type: str) -> float: 

128 """ 

129 Get reward value for a specific reward type 

130 

131 Args: 

132 reward_type: Type of reward 

133 

134 Returns: 

135 Reward value 

136 """ 

137 rewards = AGENT_LIGHTNING_CONFIG.get('rewards', {}) 

138 return rewards.get(reward_type, 0.0) 

139 

140 

141def get_store_backend() -> str: 

142 """Get the configured store backend""" 

143 return AGENT_LIGHTNING_CONFIG.get('store_backend', 'json') 

144 

145 

146def get_training_config() -> Dict[str, Any]: 

147 """Get training configuration""" 

148 return AGENT_LIGHTNING_CONFIG.get('training', {}) 

149 

150 

151def update_config(updates: Dict[str, Any]) -> None: 

152 """ 

153 Update configuration dynamically 

154 

155 Args: 

156 updates: Dictionary of configuration updates 

157 """ 

158 def deep_update(d: Dict, u: Dict) -> Dict: 

159 """Recursively update nested dictionaries""" 

160 for k, v in u.items(): 

161 if isinstance(v, dict) and k in d and isinstance(d[k], dict): 

162 d[k] = deep_update(d[k], v) 

163 else: 

164 d[k] = v 

165 return d 

166 

167 deep_update(AGENT_LIGHTNING_CONFIG, updates) 

168 

169 

170# Environment-based overrides 

171if os.getenv('AGENT_LIGHTNING_DEBUG', 'false').lower() == 'true': 

172 AGENT_LIGHTNING_CONFIG['monitoring']['log_level'] = 'DEBUG' 

173 AGENT_LIGHTNING_CONFIG['monitoring']['save_traces'] = True 

174 

175 

176__all__ = [ 

177 'AGENT_LIGHTNING_CONFIG', 

178 'AGENT_LIGHTNING_ENABLED', 

179 'is_enabled', 

180 'get_agent_config', 

181 'get_reward_value', 

182 'get_store_backend', 

183 'get_training_config', 

184 'update_config', 

185]