Google AI Agent
632867258b
Some checks failed
Deploy Nexus / deploy (push) Failing after 5s
Co-authored-by: Google AI Agent <gemini@hermes.local> Co-committed-by: Google AI Agent <gemini@hermes.local>
80 lines
2.3 KiB
Python
80 lines
2.3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Groq Worker — A dedicated worker for the Groq API
|
|
|
|
This module provides a simple interface to the Groq API. It is designed
|
|
to be used by the Nexus Mind to offload the thinking process to the
|
|
Groq API.
|
|
|
|
Usage:
|
|
# As a standalone script:
|
|
python -m nexus.groq_worker --help
|
|
|
|
# Or imported and used by another module:
|
|
from nexus.groq_worker import GroqWorker
|
|
worker = GroqWorker(model="groq/llama3-8b-8192")
|
|
response = worker.think("What is the meaning of life?")
|
|
print(response)
|
|
"""
|
|
|
|
import os
|
|
import logging
|
|
import requests
|
|
from typing import Optional
|
|
|
|
log = logging.getLogger("nexus")
|
|
|
|
GROQ_API_URL = "https://api.groq.com/openai/v1/chat/completions"
|
|
DEFAULT_MODEL = "groq/llama3-8b-8192"
|
|
|
|
class GroqWorker:
|
|
"""A worker for the Groq API."""
|
|
|
|
def __init__(self, model: str = DEFAULT_MODEL, api_key: Optional[str] = None):
|
|
self.model = model
|
|
self.api_key = api_key or os.environ.get("GROQ_API_KEY")
|
|
|
|
def think(self, messages: list[dict]) -> str:
|
|
"""Call the Groq API. Returns the model's response text."""
|
|
if not self.api_key:
|
|
log.error("GROQ_API_KEY not set.")
|
|
return ""
|
|
|
|
payload = {
|
|
"model": self.model,
|
|
"messages": messages,
|
|
"stream": False,
|
|
}
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {self.api_key}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
try:
|
|
r = requests.post(GROQ_API_URL, json=payload, headers=headers, timeout=60)
|
|
r.raise_for_status()
|
|
return r.json().get("choices", [{}])[0].get("message", {}).get("content", "")
|
|
except Exception as e:
|
|
log.error(f"Groq API call failed: {e}")
|
|
return ""
|
|
|
|
def main():
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="Groq Worker")
|
|
parser.add_argument(
|
|
"--model", default=DEFAULT_MODEL, help=f"Groq model name (default: {DEFAULT_MODEL})"
|
|
)
|
|
parser.add_argument(
|
|
"prompt", nargs="?", default="What is the meaning of life?", help="The prompt to send to the model"
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
worker = GroqWorker(model=args.model)
|
|
response = worker.think([{"role": "user", "content": args.prompt}])
|
|
print(response)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|