Spaces:
Running
Running
File size: 7,299 Bytes
2c72e40 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 |
import os
import json
import logging
from typing import Dict, Any, List, Optional
from openai import OpenAI
logger = logging.getLogger(__name__)
import os
from dotenv import load_dotenv
load_dotenv()
print("✅ API KEY LOADED:", os.getenv("OPENAI_API_KEY"))
class LLMService:
"""Service for interacting with OpenAI LLM to process and consolidate scraped data"""
def __init__(self, model_name: str = "gpt-4o"):
"""
Initialize LLM service
Args:
model_name: Name of the OpenAI model to use (default: gpt-4o)
"""
# the newest OpenAI model is "gpt-4o" which was released May 13, 2024.
# do not change this unless explicitly requested by the user
self.model_name = model_name
self.api_key = os.environ.get("OPENAI_API_KEY")
if not self.api_key:
logger.warning("OpenAI API key not found in environment variables")
self.client = OpenAI(api_key=self.api_key)
# This method will be implemented in api/horoscope_routes.py
def consolidate_horoscopes(self, horoscope_data):
"""Placeholder method for consolidating horoscopes"""
return {"error": "Method not implemented"}
def consolidate_data(self, scraped_data: List[Dict[str, Any]]) -> Dict[str, Any]:
"""
Consolidate data from multiple sources using LLM
Args:
scraped_data: List of scraped data from different sources
Returns:
Consolidated information as a dictionary
"""
if not scraped_data:
return {"error": "No data provided for consolidation"}
try:
# Prepare data for LLM
sources_text = ""
for i, data in enumerate(scraped_data, 1):
source_type = data.get("type", "unknown")
title = data.get("title", "Unknown Title")
source = data.get("source", "Unknown Source")
text = data.get("text_content", "No content available")
sources_text += f"SOURCE {i} ({source_type} from {source}):\n"
sources_text += f"Title: {title}\n"
sources_text += f"Content: {text[:2000]}...\n\n"
# Create prompt for consolidation
prompt = f"""
Please analyze and consolidate the following information from multiple sources.
{sources_text}
Provide a comprehensive consolidation of this information in JSON format with the following structure:
{{
"main_topics": [list of main topics covered],
"key_points": [list of key factual points from all sources],
"summary": "A 2-3 paragraph summary that synthesizes the information",
"analysis": "Brief analysis of the information and any discrepancies between sources",
"sources": [list of sources used]
}}
Only include factual information present in the sources. Do not add any speculative or additional information.
"""
# Call OpenAI API
response = self.client.chat.completions.create(
model=self.model_name,
messages=[
{"role": "system", "content": "You are a data analysis expert specializing in consolidating information from multiple sources."},
{"role": "user", "content": prompt}
],
response_format={"type": "json_object"},
temperature=0.2
)
# Parse the response
content = response.choices[0].message.content
if content:
result = json.loads(content)
return result
return {"error": "Empty response from LLM"}
except Exception as e:
logger.error(f"Error consolidating data with LLM: {str(e)}")
return {"error": f"Failed to consolidate data: {str(e)}"}
def summarize_content(self, text: str, max_length: int = 500) -> str:
"""
Summarize a single piece of content
Args:
text: Text to summarize
max_length: Maximum length of summary in characters
Returns:
Summarized text
"""
if not text:
return "No content to summarize"
try:
prompt = f"""
Please summarize the following text concisely in no more than {max_length} characters,
while maintaining all key information:
{text[:10000]}
"""
response = self.client.chat.completions.create(
model=self.model_name,
messages=[
{"role": "system", "content": "You are a summarization expert."},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=max_length // 2 # Approximate token count
)
return response.choices[0].message.content
except Exception as e:
logger.error(f"Error summarizing content with LLM: {str(e)}")
return f"Failed to summarize content: {str(e)}"
def extract_key_information(self, text: str, info_type: Optional[str] = None) -> Dict[str, Any]:
"""
Extract specific type of information from content
Args:
text: Text to extract information from
info_type: Type of information to extract (e.g., "news", "product", "research")
Returns:
Extracted information as dictionary
"""
if not text:
return {"error": "No content provided"}
try:
type_instruction = ""
if info_type:
type_instruction = f"This is {info_type} content. "
prompt = f"""
{type_instruction}Please extract key structured information from the following text.
Return the result as a JSON object with appropriate fields based on the content type.
{text[:8000]}
"""
response = self.client.chat.completions.create(
model=self.model_name,
messages=[
{"role": "system", "content": "You are a data extraction expert."},
{"role": "user", "content": prompt}
],
response_format={"type": "json_object"},
temperature=0.1
)
# Parse the response
content = response.choices[0].message.content
if content:
result = json.loads(content)
return result
return {"error": "Empty response from LLM"}
except Exception as e:
logger.error(f"Error extracting information with LLM: {str(e)}")
return {"error": f"Failed to extract information: {str(e)}"}
# Create a singleton instance
llm_service = LLMService()
|