Create groq_api.py

groq_api.py

@@ -0,0 +1,218 @@
+import os
+import requests
+from typing import List, Dict, Any
+
+def grok_get_llm_response(
+    system_prompt: str,
+    user_input: str,
+    tools: List[Dict[str, Any]] = None,
+    tool_choice: str = "auto",
+    reasoning_effort: str = "default", #"default",json, text, verbose_json
+    response_format: Dict[str, Any] = None,
+    temperature: float = 0.3,
+    max_completion_tokens: int = 2000,
+    include_reasoning: str = False # True
+) -> str:
+    """
+    Make a request to the Grok API and return the response content, supporting tool usage and agentic features.
+    
+    Args:
+        system_prompt (str): The system prompt to set the context.
+        user_input (str): The user input to process.
+        tools (List[Dict[str, Any]], optional): List of tool definitions for tool-calling.
+        tool_choice (str, optional): Controls tool usage ("none", "auto", "required"). Defaults to "auto".
+        reasoning_effort (str, optional): Reasoning mode for Qwen3 models ("none", "default"). Defaults to "default".
+        response_format (Dict[str, Any], optional): Format for structured outputs (e.g., JSON schema).
+        temperature (float, optional): Sampling temperature (0 to 2). Defaults to 0.7 for determinism.
+        max_completion_tokens (int, optional): Max tokens in response. Defaults to 1000.
+    
+    Returns:
+        str: The content of the assistant's response or tool call results, or empty string on error.
+    """
+    # Retrieve API key from environment
+    api_key = os.getenv("GROQ_API_KEY")
+    if not api_key:
+        print("Grok API error: GROQ_API_KEY environment variable not set")
+        return ""
+
+    # API endpoint
+    api_url = "https://api.groq.com/openai/v1/chat/completions"
+    
+    # Prepare messages in Grok API format
+    messages = [
+        {"role": "system", "content": system_prompt},
+        {"role": "user", "content": user_input}
+    ]
+    
+    # Prepare payload
+    payload = {
+        "model": "qwen/qwen3-32b",
+        "messages": messages,
+        "temperature": max(0, min(temperature, 2)),  # Clamp to valid range [0, 2]
+        "include_reasoning": include_reasoning,
+        "max_completion_tokens": max_completion_tokens
+    }
+    
+    # Add tools and tool_choice if provided
+    if tools:
+        payload["tools"] = tools
+        if tool_choice in ["none", "auto", "required"]:
+            payload["tool_choice"] = tool_choice
+        else:
+            print(f"Grok API warning: Invalid tool_choice '{tool_choice}', defaulting to 'auto'")
+            payload["tool_choice"] = "auto"
+    
+    # Add reasoning_effort for Qwen3 models
+    if reasoning_effort in ["none", "default"]:
+        payload["reasoning_effort"] = reasoning_effort
+    else:
+        print(f"Grok API warning: Invalid reasoning_effort '{reasoning_effort}', defaulting to 'default'")
+        payload["reasoning_effort"] = "default"
+    
+    # Add response_format if provided
+    if response_format:
+        payload["response_format"] = response_format
+    
+    # Set headers
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {api_key}"
+    }
+    
+    try:
+        # Make API request
+        response = requests.post(api_url, headers=headers, json=payload, timeout=60)
+        response.raise_for_status()
+        
+        # Parse response
+        result = response.json()
+        choice = result.get("choices", [{}])[0]
+        message = choice.get("message", {})
+        
+        # Handle tool calls if present
+        if "tool_calls" in message:
+            tool_calls = message["tool_calls"]
+            tool_results = []
+            for tool_call in tool_calls:
+                tool_name = tool_call.get("function", {}).get("name", "")
+                tool_args = tool_call.get("function", {}).get("arguments", "{}")
+                tool_results.append(f"Tool Call: {tool_name} with args {tool_args}")
+            return "; ".join(tool_results)  # Combine tool call results into a single string
+        
+        # Return assistant content if no tool calls
+        content = message.get("content", "")
+        return content.strip()
+        
+    except requests.exceptions.HTTPError as e:
+        print(f"Grok API error: HTTP {e.response.status_code} - {e.response.text}")
+        return ""
+    except requests.exceptions.RequestException as e:
+        print(f"Grok API error: Network error - {e}")
+        return ""
+    except (KeyError, ValueError) as e:
+        print(f"Grok API error: Unexpected response format - {e}")
+        return ""
+    except Exception as e:
+        print(f"Grok API error: Unexpected error - {e}")
+        return ""
+
+
+
+def open_oss_get_llm_response(
+    system_prompt: str,
+    user_input: str,
+    tools: List[Dict[str, Any]] = None,
+    tool_choice: str = "auto",
+    temperature: float = 0.1,
+    max_completion_tokens: int = 3000
+) -> str:
+    """
+    Make a request to the Grok API and return the response content, supporting tool usage and agentic features.
+    
+    Args:
+        system_prompt (str): The system prompt to set the context.
+        user_input (str): The user input to process.
+        tools (List[Dict[str, Any]], optional): List of tool definitions for tool-calling.
+        tool_choice (str, optional): Controls tool usage ("none", "auto", "required"). Defaults to "auto".
+        temperature (float, optional): Sampling temperature (0 to 2). Defaults to 0.7 for determinism.
+        max_completion_tokens (int, optional): Max tokens in response. Defaults to 1000.
+    
+    Returns:
+        str: The content of the assistant's response or tool call results, or empty string on error.
+    """
+    # Retrieve API key from environment
+    api_key = os.getenv("GROQ_API_KEY")
+    if not api_key:
+        print("Grok API error: GROQ_API_KEY environment variable not set")
+        return ""
+
+    # API endpoint
+    api_url = "https://api.groq.com/openai/v1/chat/completions"
+    
+    # Prepare messages in Grok API format
+    messages = [
+        {"role": "system", "content": system_prompt},
+        {"role": "user", "content": user_input}
+    ]
+    
+    # Prepare payload
+    payload = {
+        "model": "openai/gpt-oss-20b",
+        "messages": messages,
+        "temperature": max(0, min(temperature, 2)),  # Clamp to valid range [0, 2]
+        "max_completion_tokens": max_completion_tokens,
+        "reasoning_effort": "medium"
+    }
+    
+    # Add tools and tool_choice if provided
+    if tools:
+        payload["tools"] = tools
+        if tool_choice in ["none", "auto", "required"]:
+            payload["tool_choice"] = tool_choice
+        else:
+            print(f"Grok API warning: Invalid tool_choice '{tool_choice}', defaulting to 'auto'")
+            payload["tool_choice"] = "auto"
+    
+   
+    # Set headers
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {api_key}"
+    }
+    
+    try:
+        # Make API request
+        response = requests.post(api_url, headers=headers, json=payload, timeout=60)
+        response.raise_for_status()
+        
+        # Parse response
+        result = response.json()
+        choice = result.get("choices", [{}])[0]
+        message = choice.get("message", {})
+        
+        # Handle tool calls if present
+        if "tool_calls" in message:
+            tool_calls = message["tool_calls"]
+            tool_results = []
+            for tool_call in tool_calls:
+                tool_name = tool_call.get("function", {}).get("name", "")
+                tool_args = tool_call.get("function", {}).get("arguments", "{}")
+                tool_results.append(f"Tool Call: {tool_name} with args {tool_args}")
+            return "; ".join(tool_results)  # Combine tool call results into a single string
+        
+        # Return assistant content if no tool calls
+        content = message.get("content", "")
+        return content.strip()
+        
+    except requests.exceptions.HTTPError as e:
+        print(f"Grok API error: HTTP {e.response.status_code} - {e.response.text}")
+        return ""
+    except requests.exceptions.RequestException as e:
+        print(f"Grok API error: Network error - {e}")
+        return ""
+    except (KeyError, ValueError) as e:
+        print(f"Grok API error: Unexpected response format - {e}")
+        return ""
+    except Exception as e:
+        print(f"Grok API error: Unexpected error - {e}")
+        return ""