groq_api.py

import os
import requests
from typing import List, Dict, Any

def GrokApi(
    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 ""