greenintellect / app /services /llm_generator.py
Tanxshh's picture
Deploy GreenIntellect Backend API with ML models and scraping
02cc7f6
Raw
History Blame Contribute Delete
9 kB
import google.generativeai as genai
import os
from dotenv import load_dotenv
load_dotenv()
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
# if GEMINI_API_KEY:
# genai.configure(api_key=GEMINI_API_KEY)
# model = genai.GenerativeModel('gemini-2.0-flash')
# else:
model = None
print("Gemini LLM Disabled by user request.")
# print("Warning: GEMINI_API_KEY not found in .env. LLM features will be disabled.")
def generate_company_description(company_name: str) -> str:
"""
Generates a brief 2-3 sentence description of the company using Gemini.
"""
if not model:
return "AI description unavailable (API Key missing)."
try:
prompt = f"Provide a factual, neutral 2-3 sentence description of the company '{company_name}', focusing on its industry and main products. Do not mention sentiment or controversies."
response = model.generate_content(prompt)
return response.text.strip()
except Exception as e:
print(f"Error generating description for {company_name}: {e}")
return "AI description unavailable due to an error."
def generate_ai_recommendations(company_name: str, analysis_data: dict) -> dict:
"""
Generates tailored recommendations for Customers, Investors, and Leadership based on the analysis.
"""
if not model:
return {
"for_customers": ["Review provided evidence links."],
"for_investors": ["Analyze financial risks mentioned in report."],
"for_company_leadership": ["Address flagged contradictions."]
}
try:
# Construct a summary context for the LLM
context = f"""
Company: {company_name}
Greenwashing Risk: {'High' if analysis_data.get('greenwashingLabel') == 1 else 'Low'}
Reason: {analysis_data.get('internal_documents_analysis', {}).get('major_findings', ['N/A'])[0]}
Contradictions: {len(analysis_data.get('contradictions_detected', []))} found.
Sentiment: {analysis_data.get('external_summary', {}).get('public_sentiment', 'N/A')}
"""
prompt = f"""
Based on the following analysis of '{company_name}', provide 3 specific, actionable recommendations for each group (Customers, Investors, Leadership).
Focus on greenwashing, transparency, and sustainability accountability.
Analysis Context:
{context}
Output purely as JSON format with keys: "for_customers", "for_investors", "for_company_leadership". Each key should have a list of strings.
Do not allow Markdown code blocks. Just raw JSON.
"""
response = model.generate_content(prompt)
text = response.text.strip()
# Clean potential markdown wrapping
if text.startswith("```json"):
text = text[7:]
if text.endswith("```"):
text = text[:-3]
import json
return json.loads(text)
except Exception as e:
print(f"Error generating recommendations for {company_name}: {e}")
# Fallback
return {
"for_customers": ["Review provided evidence links.", "Cross-check claims."],
"for_investors": ["Monitor reputational risks.", "Demand clearer impact reports."],
"for_company_leadership": ["Address detected contradictions.", "Improve transparency."]
}
def generate_combined_insights(company_name: str, analysis_data: dict) -> dict:
"""
Combines description and recommendations into a single API call to reduce rate limit usage.
Returns: { "description": str, "recommendations": dict }
"""
if not model:
return {
"description": "AI description unavailable (API Key missing).",
"recommendations": generate_ai_recommendations(company_name, analysis_data) # Fallback to default
}
try:
context = f"""
Company: {company_name}
Greenwashing Risk: {'High' if analysis_data.get('greenwashingLabel') == 1 else 'Low'}
Reason: {analysis_data.get('internal_documents_analysis', {}).get('major_findings', ['N/A'])[0]}
"""
prompt = f"""
Analyze '{company_name}' based on this context:
{context}
Provide 2 outputs in a single JSON object:
1. "description": A factual 2-sentence description of the company.
2. "recommendations": A dictionary with keys "for_customers", "for_investors", "for_company_leadership", containing 3 actionable tips for each.
Output purely JSON. No markdown.
"""
response = model.generate_content(prompt)
text = response.text.strip()
if text.startswith("```json"): text = text[7:]
if text.endswith("```"): text = text[:-3]
return json.loads(text)
except Exception as e:
print(f"Error generating combined insights for {company_name}: {e}")
return {
"description": "AI description unavailable due to high traffic.",
"recommendations": {
"for_customers": ["Review evidence links."],
"for_investors": ["Analyze risks."],
"for_company_leadership": ["Address contradictions."]
}
}
def generate_batch_insights(companies_data: list) -> dict:
"""
Generates insights for a batch of companies (up to 10-15 recommended) in a SINGLE prompt.
Input: list of {name, context: str}
Output: dict { company_name: { "description": ..., "recommendations": ... } }
"""
import json
from .hugchat_client import generate_hugchat_response
# Try HuggingChat if Gemini is disabled
if not model:
# Construct Prompt for HuggingChat
batch_context = ""
for i, c in enumerate(companies_data):
batch_context += f"\n--- Company {i+1}: {c['name']} ---\n{c['context']}\n"
prompt = f"""
You are a sustainability analyst. Analyze these {len(companies_data)} companies.
{batch_context}
Return a valid JSON OBJECT where keys are company names.
For each company, provide:
1. "description": A factual 2-sentence summary.
2. "recommendations": Object with keys "for_customers", "for_investors", "for_company_leadership" (list of 3 tips each).
Example JSON Structure:
{{
"Company Name": {{
"description": "...",
"recommendations": {{ "for_customers": [...], ... }}
}}
}}
IMPORTANT: Output ONLY valid JSON. No Markdown. No Intro.
"""
print("Using HuggingChat for Batch Analysis...")
response_text = generate_hugchat_response(prompt)
try:
# clean json
text = response_text.strip()
if text.startswith("```json"): text = text[7:]
if text.endswith("```"): text = text[:-3]
if "{" not in text: raise Exception("Invalid JSON format")
return json.loads(text)
except Exception as e:
print(f"HuggingChat Parsing Error: {e}")
# Fallthrogh to fallback
pass
if not model and not 'response_text' in locals():
# Return fallback for all
return {c['name']: {
"description": "AI unavailable (Key missing)",
"recommendations": {
"for_customers": ["Review evidence."],
"for_investors": ["Check risks."],
"for_company_leadership": ["Monitor compliance."]
}
} for c in companies_data}
try:
# ... (Gemini Logic remains as backup if re-enabled) ...
# Construct simplified context list
batch_context = ""
for i, c in enumerate(companies_data):
batch_context += f"\n--- Company {i+1}: {c['name']} ---\n{c['context']}\n"
prompt = f"""
Analyze the following {len(companies_data)} companies based on the provided contexts.
{batch_context}
For EACH company, provide:
1. "description": A factual 2-sentence summary.
2. "recommendations": 3 specific actionable tips per group (Customers, Investors, Leadership).
Output purely as a JSON OBJECT where keys are the exact company names and values are the insight objects.
Example:
{{
"Company A": {{ "description": "...", "recommendations": {{ ... }} }},
"Company B": ...
}}
No markdown formatting. Just JSON.
"""
response = model.generate_content(prompt)
text = response.text.strip()
if text.startswith("```json"): text = text[7:]
if text.endswith("```"): text = text[:-3]
results = json.loads(text)
return results
except Exception as e:
print(f"Batch generation error: {e}")
return {}