"""Groq LLM integration for DS Task AI News""" import os from typing import List, Dict, Any, Optional from groq import Groq from config import settings class GroqLLMService: def __init__(self): self.client = None self.model = "llama3-8b-8192" # Default Groq model # Initialize Groq client if API key is available if settings.groq_api_key: try: self.client = Groq(api_key=settings.groq_api_key) print("✅ Groq LLM service initialized") except Exception as e: print(f"⚠️ Groq initialization failed: {e}") self.client = None else: print("⚠️ Groq API key not provided") def is_available(self) -> bool: """Check if Groq service is available""" return self.client is not None def summarize_article(self, article: Dict[str, Any]) -> Optional[str]: """Generate a summary for an article""" if not self.is_available(): return None try: title = article.get('title', '') content = article.get('content', '') prompt = f""" Please provide a concise summary of this news article in 2-3 sentences: Title: {title} Content: {content} Summary: """ response = self.client.chat.completions.create( messages=[ {"role": "user", "content": prompt} ], model=self.model, max_tokens=150, temperature=0.3 ) summary = response.choices[0].message.content.strip() return summary except Exception as e: print(f"Error generating summary: {e}") return None def analyze_sentiment(self, article: Dict[str, Any]) -> Optional[str]: """Analyze sentiment of an article""" if not self.is_available(): return None try: title = article.get('title', '') content = article.get('content', '') prompt = f""" Analyze the sentiment of this news article. Respond with only one word: "positive", "negative", or "neutral". Title: {title} Content: {content} Sentiment: """ response = self.client.chat.completions.create( messages=[ {"role": "user", "content": prompt} ], model=self.model, max_tokens=10, temperature=0.1 ) sentiment = response.choices[0].message.content.strip().lower() # Validate response if sentiment in ['positive', 'negative', 'neutral']: return sentiment else: return 'neutral' # Default fallback except Exception as e: print(f"Error analyzing sentiment: {e}") return None def extract_keywords(self, article: Dict[str, Any]) -> Optional[List[str]]: """Extract key topics/keywords from an article""" if not self.is_available(): return None try: title = article.get('title', '') content = article.get('content', '') prompt = f""" Extract 3-5 key topics or keywords from this news article. Return them as a comma-separated list. Title: {title} Content: {content} Keywords: """ response = self.client.chat.completions.create( messages=[ {"role": "user", "content": prompt} ], model=self.model, max_tokens=50, temperature=0.3 ) keywords_text = response.choices[0].message.content.strip() keywords = [kw.strip() for kw in keywords_text.split(',') if kw.strip()] return keywords[:5] # Limit to 5 keywords except Exception as e: print(f"Error extracting keywords: {e}") return None def generate_insights(self, articles: List[Dict[str, Any]]) -> Optional[str]: """Generate insights from multiple articles""" if not self.is_available() or not articles: return None try: # Create a summary of article titles titles = [article.get('title', '') for article in articles[:10]] # Limit to 10 articles titles_text = '\n'.join([f"- {title}" for title in titles]) prompt = f""" Based on these recent news headlines, provide 2-3 key insights about current trends or themes: Headlines: {titles_text} Key Insights: """ response = self.client.chat.completions.create( messages=[ {"role": "user", "content": prompt} ], model=self.model, max_tokens=200, temperature=0.4 ) insights = response.choices[0].message.content.strip() return insights except Exception as e: print(f"Error generating insights: {e}") return None def enhance_article(self, article: Dict[str, Any]) -> Dict[str, Any]: """Enhance article with AI-generated metadata""" enhanced_article = article.copy() if self.is_available(): # Add summary summary = self.summarize_article(article) if summary: enhanced_article['ai_summary'] = summary # Add sentiment sentiment = self.analyze_sentiment(article) if sentiment: enhanced_article['sentiment'] = sentiment # Add keywords keywords = self.extract_keywords(article) if keywords: enhanced_article['ai_keywords'] = keywords return enhanced_article def batch_enhance_articles(self, articles: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Enhance multiple articles with AI features""" enhanced_articles = [] for article in articles: enhanced = self.enhance_article(article) enhanced_articles.append(enhanced) return enhanced_articles # Test function if __name__ == "__main__": # Test Groq integration groq_service = GroqLLMService() if groq_service.is_available(): print("✅ Groq service is available") # Test with sample article sample_article = { "title": "AI Technology Advances in Healthcare", "content": "Recent developments in artificial intelligence are transforming the healthcare industry with new diagnostic tools and treatment methods." } enhanced = groq_service.enhance_article(sample_article) print(f"Enhanced article: {enhanced}") else: print("⚠️ Groq service not available (API key needed)")