gemini_service.py•8.56 kB
import os
import logging
from typing import Dict, List, Optional, Any
from dotenv import load_dotenv
import google.generativeai as genai
from google.generativeai.types import HarmCategory, HarmBlockThreshold
load_dotenv()
logger = logging.getLogger(__name__)
class GeminiService:
def __init__(self):
self.api_key = os.getenv("GEMINI_API_KEY")
if not self.api_key:
raise ValueError("GEMINI_API_KEY not found in environment variables")
genai.configure(api_key=self.api_key)
# Initialize model with grounding capabilities
self.model = genai.GenerativeModel(
'gemini-2.0-flash-exp',
generation_config={
"temperature": 0.7,
"top_p": 0.95,
"top_k": 40,
"max_output_tokens": 8192,
},
safety_settings={
HarmCategory.HARM_CATEGORY_HATE_SPEECH: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: HarmBlockThreshold.BLOCK_NONE,
HarmCategory.HARM_CATEGORY_HARASSMENT: HarmBlockThreshold.BLOCK_NONE,
}
)
logger.info("Gemini service initialized successfully")
async def search_with_grounding(self, query: str, guides_data: List[Dict]) -> Dict[str, Any]:
"""
Use Gemini's grounding to search through AI guides with semantic understanding
"""
try:
# Create context from guides data
guides_context = "\n\n".join([
f"Title: {guide['title']}\n"
f"Publisher: {guide['publisher']}\n"
f"Description: {guide['description']}\n"
f"Topics: {', '.join(guide['topics'])}"
for guide in guides_data
])
prompt = f"""You are an AI assistant helping to search through AI guides.
Based on the following query and the available guides, provide the most relevant guides.
Query: {query}
Available Guides:
{guides_context}
Instructions:
1. Analyze the query for intent and keywords
2. Find guides that best match the query semantically
3. Return a JSON response with:
- matched_guides: Array of matching guide titles
- relevance_scores: Object mapping titles to relevance scores (0-1)
- search_reasoning: Brief explanation of why these guides match
Respond only with valid JSON."""
response = self.model.generate_content(prompt)
# Parse the response
import json
result = json.loads(response.text)
return {
"success": True,
"grounded_search": True,
"results": result
}
except Exception as e:
logger.error(f"Grounding search error: {str(e)}")
return {
"success": False,
"error": str(e),
"grounded_search": True
}
async def analyze_guide_url(self, url: str) -> Dict[str, Any]:
"""
Extract and analyze content from a guide URL using Gemini
"""
try:
# First, fetch the URL content
import requests
response = requests.get(url, timeout=10)
if response.status_code != 200:
return {
"success": False,
"error": f"Failed to fetch URL: {response.status_code}"
}
# For demonstration, we'll analyze the page content
# In a real implementation, you'd parse PDFs, extract text, etc.
prompt = f"""Analyze this AI guide URL and provide a comprehensive summary.
URL: {url}
Content Preview: {response.text[:2000]}
Provide:
1. Main topics covered
2. Key takeaways
3. Target audience
4. Practical applications
5. Prerequisites or recommended knowledge
Format as JSON with these keys: topics, takeaways, audience, applications, prerequisites"""
response = self.model.generate_content(prompt)
import json
analysis = json.loads(response.text)
return {
"success": True,
"url": url,
"analysis": analysis,
"analyzed_with": "gemini-2.0-flash-exp"
}
except Exception as e:
logger.error(f"URL analysis error: {str(e)}")
return {
"success": False,
"error": str(e),
"url": url
}
async def generate_guide_summary(self, guide: Dict) -> Dict[str, Any]:
"""
Generate an enhanced summary of a guide using Gemini
"""
try:
prompt = f"""Create a comprehensive summary for this AI guide:
Title: {guide['title']}
Publisher: {guide['publisher']}
Description: {guide['description']}
Topics: {', '.join(guide['topics'])}
Generate:
1. Extended summary (2-3 paragraphs)
2. Learning objectives
3. Who should read this guide
4. Estimated reading time
5. Related guides or resources
Format as JSON."""
response = self.model.generate_content(prompt)
import json
summary = json.loads(response.text)
return {
"success": True,
"guide_title": guide['title'],
"enhanced_summary": summary
}
except Exception as e:
logger.error(f"Summary generation error: {str(e)}")
return {
"success": False,
"error": str(e),
"guide_title": guide.get('title', 'Unknown')
}
async def compare_guides(self, guide_titles: List[str], guides_data: List[Dict]) -> Dict[str, Any]:
"""
Compare multiple guides and provide recommendations
"""
try:
# Filter guides by titles
selected_guides = [g for g in guides_data if g['title'] in guide_titles]
if not selected_guides:
return {
"success": False,
"error": "No matching guides found"
}
guides_info = "\n\n".join([
f"Guide {i+1}: {guide['title']}\n"
f"Publisher: {guide['publisher']}\n"
f"Description: {guide['description']}\n"
f"Topics: {', '.join(guide['topics'])}"
for i, guide in enumerate(selected_guides)
])
prompt = f"""Compare these AI guides and provide insights:
{guides_info}
Provide:
1. Key differences between the guides
2. Overlapping content areas
3. Recommended reading order
4. Which guide is best for beginners vs advanced users
5. Complementary aspects
Format as JSON with keys: differences, overlaps, reading_order, audience_fit, complementary_aspects"""
response = self.model.generate_content(prompt)
import json
comparison = json.loads(response.text)
return {
"success": True,
"compared_guides": guide_titles,
"comparison": comparison
}
except Exception as e:
logger.error(f"Guide comparison error: {str(e)}")
return {
"success": False,
"error": str(e)
}
# Singleton instance
_gemini_service = None
def get_gemini_service() -> GeminiService:
global _gemini_service
if _gemini_service is None:
_gemini_service = GeminiService()
return _gemini_service