Spaces:
Running
Running
File size: 4,086 Bytes
ddaf3fc c0d8669 ddaf3fc c0d8669 ddaf3fc 1a96a66 ddaf3fc 5187aa2 ddaf3fc c0d8669 ddaf3fc 5187aa2 c0d8669 ddaf3fc c0d8669 ddaf3fc c0d8669 5187aa2 ddaf3fc 5187aa2 ddaf3fc c0d8669 ddaf3fc c0d8669 ddaf3fc 5187aa2 ddaf3fc c0d8669 ddaf3fc c0d8669 ddaf3fc 5187aa2 ddaf3fc 5187aa2 |
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 |
"""
A robust, resilient analyzer using the Google Gemini Pro API.
This module prompts Gemini for a simple key-value format and then constructs
the final JSON object in Python, making it resilient to LLM syntax errors.
"""
import os
import logging
import httpx
from typing import Optional, TypedDict, List, Union
logger = logging.getLogger(__name__)
class AnalysisResult(TypedDict):
sentiment: str
sentiment_score: float
reason: str
entities: List[str]
topic: str
impact: str
summary: str
error: Optional[str]
class GeminiAnalyzer:
API_URL = "https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-pro-latest:generateContent"
def __init__(self, client: httpx.AsyncClient, api_key: Optional[str] = None):
self.client = client
self.api_key = api_key or os.getenv("GEMINI_API_KEY")
if not self.api_key:
raise ValueError("GEMINI_API_KEY is not set.")
self.params = {"key": self.api_key}
self.headers = {"Content-Type": "application/json"}
def _build_prompt(self, text: str) -> dict:
"""Creates a prompt asking for a simple, parsable text format."""
return {
"contents": [{
"parts": [{
"text": f"""
Analyze the following financial text from the cryptocurrency world.
Respond using a simple key::value format, with each key-value pair on a new line. Do NOT use JSON.
KEYS:
sentiment:: [POSITIVE, NEGATIVE, or NEUTRAL]
sentiment_score:: [A float between -1.0 and 1.0]
reason:: [A brief, one-sentence explanation for the sentiment.]
entities:: [A comma-separated list of cryptocurrencies mentioned, e.g., Bitcoin, ETH]
topic:: [One of: Regulation, Partnership, Technical Update, Market Hype, Security, General News]
impact:: [One of: LOW, MEDIUM, HIGH]
summary:: [A concise, one-sentence summary of the text.]
TEXT TO ANALYZE: "{text}"
"""
}]
}]
}
def _parse_structured_text(self, text: str) -> AnalysisResult:
"""Parses the key::value text response from Gemini into a structured dict."""
data = {}
for line in text.splitlines():
if '::' in line:
key, value = line.split('::', 1)
data[key.strip()] = value.strip()
# Build the final, validated object
return {
"sentiment": data.get("sentiment", "NEUTRAL").upper(),
"sentiment_score": float(data.get("sentiment_score", 0.0)),
"reason": data.get("reason", "N/A"),
"entities": [e.strip() for e in data.get("entities", "").split(',') if e.strip()],
"topic": data.get("topic", "General News"),
"impact": data.get("impact", "LOW").upper(),
"summary": data.get("summary", "Summary not available."),
"error": None
}
async def analyze_text(self, text: str) -> AnalysisResult:
"""Sends text to Gemini and returns a structured analysis."""
prompt = self._build_prompt(text)
try:
response = await self.client.post(self.API_URL, headers=self.headers, params=self.params, json=prompt, timeout=60.0)
response.raise_for_status()
full_response = response.json()
response_text = full_response["candidates"][0]["content"]["parts"][0]["text"]
# Use our new, robust parser
return self._parse_structured_text(response_text)
except Exception as e:
logger.error(f"❌ Gemini API or Parsing Error: {e}")
return {
"sentiment": "ERROR", "sentiment_score": 0.0, "reason": str(e),
"entities": [], "topic": "Unknown", "impact": "Unknown",
"summary": "Failed to perform analysis.", "error": str(e)
} |