Data_sheets / src /agents /genai_integration.py
Navya-Sree's picture
Create src/agents/genai_integration.py
2de8f47 verified
import openai
import pandas as pd
import numpy as np
from typing import Dict, List, Optional
import json
import logging
logger = logging.getLogger(__name__)
class ForecastingAIAssistant:
"""AI assistant for forecasting tasks using GenAI."""
def __init__(self, api_key: str, model_name: str = "gpt-3.5-turbo"):
self.api_key = api_key
self.model_name = model_name
openai.api_key = api_key
def generate_forecast_interpretation(self, data_summary: Dict,
forecast_results: Dict,
metrics: Dict) -> str:
"""Generate comprehensive interpretation of forecasting results."""
try:
prompt = f"""
As a senior data scientist with 35 years of experience, provide a comprehensive analysis of these forecasting results:
Data Summary: {json.dumps(data_summary, indent=2)}
Forecast Results: {json.dumps(forecast_results, indent=2)}
Performance Metrics: {json.dumps(metrics, indent=2)}
Please provide:
1. Key insights about the forecast quality
2. Potential business implications
3. Limitations of the current approach
4. Recommendations for improvement
5. Any anomalies or patterns worth noting
Write in a professional yet accessible tone suitable for both technical and business audiences.
"""
response = openai.ChatCompletion.create(
model=self.model_name,
messages=[
{"role": "system", "content": "You are a senior data scientist with expertise in time series forecasting."},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=1000
)
return response.choices[0].message.content
except Exception as e:
logger.error(f"Error generating interpretation: {str(e)}")
return f"Interpretation generation failed: {str(e)}"
def generate_business_recommendations(self, business_context: str,
forecast_results: Dict,
historical_data: pd.Series) -> str:
"""Generate business recommendations based on forecasts."""
try:
# Create historical data summary
hist_summary = {
"period": f"{historical_data.index.min()} to {historical_data.index.max()}" if hasattr(historical_data, 'index') else "N/A",
"data_points": len(historical_data),
"mean_value": historical_data.mean(),
"trend": "upward" if historical_data.iloc[-1] > historical_data.iloc[0] else "downward" if len(historical_data) > 1 else "stable"
}
prompt = f"""
Based on the forecasting results and business context, provide actionable recommendations:
Business Context: {business_context}
Forecast Results: {json.dumps(forecast_results, indent=2)}
Historical Trends: {json.dumps(hist_summary, indent=2)}
Provide specific, actionable recommendations including:
1. Operational adjustments
2. Risk mitigation strategies
3. Opportunities to capitalize on
4. Timeline considerations
5. Key metrics to monitor
Tailor recommendations to the specific business context.
"""
response = openai.ChatCompletion.create(
model=self.model_name,
messages=[
{"role": "system", "content": "You are a business strategy consultant with expertise in data-driven decision making."},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=1000
)
return response.choices[0].message.content
except Exception as e:
logger.error(f"Error generating recommendations: {str(e)}")
return f"Recommendation generation failed: {str(e)}"