cryogenic22 commited on
Commit
48d4cca
ยท
verified ยท
1 Parent(s): dfd1848

Update utils_core/llm_utils.py

Browse files
Files changed (1) hide show
  1. utils_core/llm_utils.py +89 -263
utils_core/llm_utils.py CHANGED
@@ -1,285 +1,111 @@
1
- """
2
- LLM API Utility Functions
3
-
4
- This module provides utility functions for interacting with LLM APIs,
5
- particularly Claude API, and processing responses.
6
- """
7
-
8
  import os
9
- import json
10
- import re
11
- from typing import Dict, List, Any, Optional, Union, Tuple
12
  import requests
13
- import anthropic
14
- from langchain_anthropic import ChatAnthropic
15
 
16
- def call_claude_api(
17
- prompt: str,
18
- system_prompt: Optional[str] = None,
19
- temperature: float = 0.0,
20
- max_tokens: int = 4000,
21
- model: str = "claude-3-7-sonnet-20250219"
22
- ) -> str:
23
- """
24
- Call the Claude API directly using the Anthropic Python client
25
-
26
- Parameters:
27
- -----------
28
- prompt : str
29
- The human prompt to send to Claude
30
- system_prompt : str, optional
31
- System prompt for Claude
32
- temperature : float
33
- Temperature setting (0.0 = deterministic, 1.0 = creative)
34
- max_tokens : int
35
- Maximum tokens in the response
36
- model : str
37
- Claude model to use
38
-
39
- Returns:
40
- --------
41
- str
42
- Claude's response text
43
- """
44
- # Get API key from environment
45
- api_key = os.getenv("ANTHROPIC_API_KEY")
46
- if not api_key:
47
- raise ValueError("ANTHROPIC_API_KEY environment variable not set")
48
-
49
- # Initialize client
50
- client = anthropic.Anthropic(api_key=api_key)
51
-
52
- # Create messages
53
- messages = []
54
- if system_prompt:
55
- messages.append({"role": "system", "content": system_prompt})
56
-
57
- messages.append({"role": "user", "content": prompt})
58
-
59
- # Call API
60
- response = client.messages.create(
61
- model=model,
62
- messages=messages,
63
- max_tokens=max_tokens,
64
- temperature=temperature
65
- )
66
-
67
- # Return content
68
- return response.content[0].text
69
 
70
- def create_claude_client(
71
- model: str = "claude-3-haiku-20240307",
72
- temperature: float = 0.0,
73
- anthropic_api_key: Optional[str] = None
74
- ) -> ChatAnthropic:
75
  """
76
- Create a LangChain ChatAnthropic client
77
 
78
- Parameters:
79
- -----------
80
- model : str
81
- Claude model to use
82
- temperature : float
83
- Temperature setting (0.0 = deterministic, 1.0 = creative)
84
- anthropic_api_key : str, optional
85
- API key (will use environment variable if not provided)
86
-
87
  Returns:
88
- --------
89
- ChatAnthropic
90
- Configured LangChain ChatAnthropic client
91
- """
92
- # Get API key from environment if not provided
93
- api_key = anthropic_api_key or os.getenv("ANTHROPIC_API_KEY")
94
- if not api_key:
95
- raise ValueError("ANTHROPIC_API_KEY environment variable not set")
96
-
97
- # Create LangChain client
98
- client = ChatAnthropic(
99
- model=model,
100
- anthropic_api_key=api_key,
101
- temperature=temperature
102
- )
103
-
104
- return client
105
-
106
- def extract_json(text: str) -> Dict[str, Any]:
107
- """
108
- Extract JSON from text that might contain additional content
109
-
110
- Parameters:
111
- -----------
112
- text : str
113
- Text that may contain JSON
114
-
115
- Returns:
116
- --------
117
- dict
118
- Extracted JSON as a dictionary
119
-
120
- Raises:
121
- -------
122
- ValueError
123
- If no valid JSON can be extracted
124
  """
125
  try:
126
- # First, try to parse the entire text as JSON
127
- return json.loads(text)
128
- except json.JSONDecodeError:
129
- # If that fails, look for JSON block
130
- json_pattern = r'```json\s*([\s\S]*?)\s*```'
131
- match = re.search(json_pattern, text, re.DOTALL)
132
- if match:
133
- try:
134
- return json.loads(match.group(1))
135
- except json.JSONDecodeError:
136
- pass
137
-
138
- # Try a more aggressive approach to find JSON-like content
139
- json_pattern = r'({[\s\S]*})'
140
- match = re.search(json_pattern, text)
141
- if match:
142
- try:
143
- return json.loads(match.group(1))
144
- except json.JSONDecodeError:
145
- pass
146
 
147
- raise ValueError(f"Could not extract JSON from response: {text}")
148
-
149
- def extract_code(text: str, language: str = None) -> str:
150
- """
151
- Extract code block from text
152
-
153
- Parameters:
154
- -----------
155
- text : str
156
- Text that may contain code blocks
157
- language : str, optional
158
- Specific language to extract (e.g., 'python', 'sql')
159
 
160
- Returns:
161
- --------
162
- str
163
- Extracted code as a string
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
164
 
165
- Raises:
166
- -------
167
- ValueError
168
- If no valid code block can be extracted
169
- """
170
- # Try to extract code with specific language
171
- if language:
172
- pattern = rf'```{language}\s*(.*?)\s*```'
173
- match = re.search(pattern, text, re.DOTALL)
174
- if match:
175
- return match.group(1).strip()
176
-
177
- # Try to extract code from any language block
178
- match = re.search(r'```(\w*)\s*(.*?)\s*```', text, re.DOTALL)
179
- if match:
180
- return match.group(2).strip()
181
-
182
- # If no code blocks, look for indented text
183
- lines = text.split('\n')
184
- indented_lines = []
185
- in_indented_block = False
186
-
187
- for line in lines:
188
- if line.startswith(' ') or line.startswith('\t'):
189
- indented_lines.append(line.replace(' ', '', 1) if line.startswith(' ') else line.replace('\t', '', 1))
190
- in_indented_block = True
191
- elif line.strip() == '' and in_indented_block:
192
- indented_lines.append('')
193
- elif in_indented_block:
194
- break
195
-
196
- if indented_lines:
197
- return '\n'.join(indented_lines).strip()
198
-
199
- # If all else fails, return empty string
200
- return ""
201
 
202
- def summarize_code(code: str, language: str = "python") -> str:
203
  """
204
- Generate a summary of code using Claude
205
 
206
- Parameters:
207
- -----------
208
- code : str
209
- The code to summarize
210
- language : str
211
- The programming language
212
-
213
- Returns:
214
- --------
215
- str
216
- Summary of the code
217
  """
218
- system_prompt = """You are an expert code analyzer. Your task is to provide a concise summary of the provided code,
219
- explaining what it does, its key components, and any important patterns or techniques used."""
220
 
221
- prompt = f"""Please provide a brief summary of this {language} code:
222
-
223
- ```{language}
224
- {code}
225
- ```
226
-
227
- Focus on:
228
- 1. The main purpose of the code
229
- 2. Key functions/classes and what they do
230
- 3. Any notable algorithms or design patterns used
231
- 4. Potential use cases
232
-
233
- Keep your response under 150 words."""
234
-
235
- return call_claude_api(prompt, system_prompt=system_prompt, max_tokens=500)
236
-
237
- def generate_unit_tests(code: str, language: str = "python") -> str:
238
- """
239
- Generate unit tests for code using Claude
240
 
241
- Parameters:
242
- -----------
243
- code : str
244
- The code to generate tests for
245
- language : str
246
- The programming language
247
-
248
- Returns:
249
- --------
250
- str
251
- Generated unit tests
252
- """
253
- system_prompt = """You are an expert at writing unit tests. Your task is to create comprehensive unit tests
254
- for the provided code, ensuring good test coverage and following best practices."""
255
 
256
- prompt = f"""Please generate unit tests for this {language} code:
257
-
258
- ```{language}
259
- {code}
260
- ```
261
-
262
- Please include:
263
- 1. Test cases for normal operation
264
- 2. Test cases for edge cases and potential errors
265
- 3. Use appropriate mocking where needed
266
- 4. Follow best practices for {language} unit testing
267
-
268
- Return ONLY the test code without explanation."""
269
 
270
- return call_claude_api(prompt, system_prompt=system_prompt, max_tokens=2000)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
271
 
272
- # Example usage
273
  if __name__ == "__main__":
274
- # Test if API key is available
275
- api_key = os.getenv("ANTHROPIC_API_KEY")
276
- if not api_key:
277
- print("Warning: ANTHROPIC_API_KEY environment variable not set")
278
- else:
279
- # Test API call
280
- response = call_claude_api(
281
- "Write a one-sentence explanation of what an LLM is.",
282
- temperature=0.0,
283
- max_tokens=100
284
- )
285
- print(f"Claude API response: {response}")
 
 
 
 
 
 
 
 
1
  import os
2
+ import logging
3
+ import traceback
 
4
  import requests
5
+ from typing import Tuple, Optional
 
6
 
7
+ # Configure logging
8
+ logging.basicConfig(level=logging.DEBUG,
9
+ format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
10
+ logger = logging.getLogger("llm_connectivity")
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
11
 
12
+ def comprehensive_anthropic_connectivity_test() -> Tuple[bool, Optional[str]]:
 
 
 
 
13
  """
14
+ Perform a comprehensive connectivity test for Anthropic API
15
 
 
 
 
 
 
 
 
 
 
16
  Returns:
17
+ - Boolean indicating success
18
+ - Error message (if any)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
19
  """
20
  try:
21
+ # 1. Check API Key
22
+ api_key = os.getenv("ANTHROPIC_API_KEY")
23
+ if not api_key:
24
+ return False, "No Anthropic API key found in environment variables"
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
25
 
26
+ # 2. Test network connectivity to Anthropic
27
+ try:
28
+ # Check basic internet connectivity
29
+ requests.get("https://www.google.com", timeout=5)
30
+ except requests.RequestException:
31
+ return False, "No internet connection detected"
 
 
 
 
 
 
32
 
33
+ # 3. Test Anthropic API endpoint
34
+ try:
35
+ import anthropic
36
+
37
+ # Create client
38
+ client = anthropic.Anthropic(api_key=api_key)
39
+
40
+ # Attempt a simple API call
41
+ response = client.messages.create(
42
+ model="claude-3-haiku-20240307",
43
+ max_tokens=10,
44
+ messages=[
45
+ {"role": "user", "content": "Confirm API connectivity"}
46
+ ]
47
+ )
48
+
49
+ # Check response
50
+ if not response or not response.content:
51
+ return False, "Received empty response from Anthropic API"
52
+
53
+ logger.info("Anthropic API connectivity test successful")
54
+ return True, None
55
 
56
+ except Exception as api_error:
57
+ logger.error(f"Anthropic API connection failed: {api_error}")
58
+ return False, f"Anthropic API connection error: {str(api_error)}"
59
+
60
+ except Exception as e:
61
+ logger.error(f"Unexpected error in connectivity test: {e}")
62
+ logger.error(traceback.format_exc())
63
+ return False, f"Unexpected error: {str(e)}"
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
64
 
65
+ def debug_anthropic_connection():
66
  """
67
+ Comprehensive debug function for Anthropic API connection
68
 
69
+ Prints detailed diagnostic information
 
 
 
 
 
 
 
 
 
 
70
  """
71
+ print("๐Ÿ” Anthropic API Connectivity Diagnostics")
72
+ print("=" * 50)
73
 
74
+ # 1. Environment Check
75
+ print("\n๐Ÿ“‹ Environment Variables:")
76
+ api_key = os.getenv("ANTHROPIC_API_KEY")
77
+ print(f"ANTHROPIC_API_KEY: {'SET' if api_key else 'NOT SET'}")
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
78
 
79
+ # 2. Connectivity Test
80
+ print("\n๐ŸŒ Connectivity Test:")
81
+ success, error = comprehensive_anthropic_connectivity_test()
 
 
 
 
 
 
 
 
 
 
 
82
 
83
+ if success:
84
+ print("โœ… API Connection: Successful")
85
+ else:
86
+ print(f"โŒ API Connection: Failed")
87
+ print(f"Error Details: {error}")
 
 
 
 
 
 
 
 
88
 
89
+ # 3. Library Versions
90
+ print("\n๐Ÿ“ฆ Library Versions:")
91
+ try:
92
+ import anthropic
93
+ import langchain
94
+ print(f"Anthropic Library: {anthropic.__version__}")
95
+ print(f"LangChain Library: {langchain.__version__}")
96
+ except ImportError as e:
97
+ print(f"Error checking library versions: {e}")
98
+
99
+ # 4. Network Diagnostics
100
+ print("\n๐Ÿ”ฌ Network Diagnostics:")
101
+ try:
102
+ import socket
103
+ print("Hostname:", socket.gethostname())
104
+ print("IP Addresses:")
105
+ print(socket.gethostbyname(socket.gethostname()))
106
+ except Exception as e:
107
+ print(f"Network diagnostic error: {e}")
108
 
109
+ # Run diagnostics if script is executed directly
110
  if __name__ == "__main__":
111
+ debug_anthropic_connection()