Coverage for integrations / coding_agent / aider_core / hart_model_adapter.py: 44.4%

1""" 

2HARTOS Model Adapter — Bridges HARTOS LLM infrastructure to Aider's Model interface. 

3 

4Aider's repomap.py calls self.main_model.token_count(text) for sizing the repo map. 

5This adapter provides that method using tiktoken (already a HARTOS dependency), 

6without requiring LiteLLM. 

7 

8For actual LLM completions (used by the HartCoder), this adapter routes through 

9HARTOS's existing OpenAI client and budget gate. 

10""" 

11import logging 

12import os 

13from typing import Dict, List, Optional 

14 

15logger = logging.getLogger('hevolve.coding_agent.aider_core') 

16 

17# Lazy tiktoken import — available in HARTOS venv 

18_encoder = None 

19 

20 

21def _get_encoder(): 

22 """Get or create a tiktoken encoder (cached).""" 

23 global _encoder 

24 if _encoder is None: 

25 try: 

26 import tiktoken 

27 _encoder = tiktoken.encoding_for_model('gpt-4') 

28 except ImportError: 

29 logger.warning("tiktoken not available, using approximate token counting") 

30 _encoder = _ApproximateEncoder() 

31 except Exception: 

32 _encoder = _ApproximateEncoder() 

33 return _encoder 

34 

35 

36class _ApproximateEncoder: 

37 """Fallback token counter when tiktoken isn't available (~4 chars per token).""" 

38 

39 def encode(self, text): 

40 return [0] * (len(text) // 4) 

41 

42 

43class HartModelAdapter: 

44 """Minimal model adapter that satisfies Aider's Model interface for repomap. 

45 

46 Only implements what repomap.py actually calls: 

47 - token_count(text) -> int 

48 """ 

49 

50 def __init__(self, model_name: str = 'gpt-4', max_context_window: int = 128000): 

51 self.name = model_name 

52 self.max_context_window = max_context_window 

53 

54 def token_count(self, text: str) -> int: 

55 """Count tokens in text using tiktoken.""" 

56 if not text: 

57 return 0 

58 encoder = _get_encoder() 

59 return len(encoder.encode(text)) 

60 

61 @classmethod 

62 def from_hartos_config(cls) -> 'HartModelAdapter': 

63 """Create adapter from HARTOS configuration. 

64 

65 Uses model_registry if available, falls back to env/defaults. 

66 """ 

67 model_name = os.environ.get('HEVOLVE_CODING_MODEL', 'gpt-4') 

68 max_ctx = 128000 

69 

70 try: 

71 from integrations.agent_engine.model_registry import get_model_by_policy 

72 model_info = get_model_by_policy() 

73 if model_info: 

74 model_name = model_info.get('model_id', model_name) 

75 max_ctx = model_info.get('context_window', max_ctx) 

76 except ImportError: 

77 pass 

78 

79 return cls(model_name=model_name, max_context_window=max_ctx) 

80 

81 

82def send_completion( 

83 messages: List[Dict], 

84 model: str = '', 

85 temperature: float = 0.0, 

86 max_tokens: int = 4096, 

87 user_id: str = '', 

88 prompt_id: str = '', 

89) -> Optional[str]: 

90 """Send a completion request through HARTOS's LLM infrastructure. 

91 

92 Routes through budget gate for cost tracking. Used by HartCoder 

93 for code editing tasks. 

94 

95 Returns: 

96 Response text or None on failure. 

97 """ 

98 if not model: 

99 model = os.environ.get('HEVOLVE_CODING_MODEL', 'gpt-4') 

100 

101 try: 

102 import openai 

103 from helper import get_openai_config 

104 

105 config = get_openai_config() 

106 client_kwargs = {} 

107 if config.get('api_key'): 

108 client_kwargs['api_key'] = config['api_key'] 

109 if config.get('api_base'): 

110 client_kwargs['base_url'] = config['api_base'] 

111 

112 client = openai.OpenAI(**client_kwargs) 

113 response = client.chat.completions.create( 

114 model=model, 

115 messages=messages, 

116 temperature=temperature, 

117 max_tokens=max_tokens, 

118 ) 

119 

120 result_text = response.choices[0].message.content 

121 

122 # Record metered usage if budget gate available 

123 try: 

124 from integrations.agent_engine.budget_gate import record_metered_usage 

125 usage = response.usage 

126 record_metered_usage( 

127 user_id=user_id or 'coding_agent', 

128 model=model, 

129 prompt_tokens=usage.prompt_tokens if usage else 0, 

130 completion_tokens=usage.completion_tokens if usage else 0, 

131 source='aider_native', 

132 ) 

133 except ImportError: 

134 pass 

135 

136 return result_text 

137 

138 except Exception as e: 

139 logger.error(f"Completion failed: {e}") 

140 return None