Buckets:
| import os | |
| import json | |
| import time | |
| from typing import List, Dict, Any, Optional, Union | |
| from dataclasses import dataclass | |
| from enum import Enum | |
| try: | |
| import google.generativeai as genai | |
| GEMINI_AVAILABLE = True | |
| except ImportError: | |
| GEMINI_AVAILABLE = False | |
| print("⚠️ Google Generative AI not available. Install with: pip install google-generativeai") | |
| class GeminiModelType(Enum): | |
| GEMINI_PRO = "gemini-pro" | |
| GEMINI_PRO_VISION = "gemini-pro-vision" | |
| GEMINI_FLASH = "gemini-1.5-flash" | |
| class GeminiConfig: | |
| api_key: str | |
| model: GeminiModelType = GeminiModelType.GEMINI_PRO | |
| temperature: float = 0.7 | |
| max_output_tokens: int = 2048 | |
| top_p: float = 0.8 | |
| top_k: int = 40 | |
| class GeminiTTTGenerator: | |
| def __init__(self, config: GeminiConfig): | |
| self.config = config | |
| self.model = None | |
| self._initialize_gemini() | |
| def _initialize_gemini(self): | |
| if not GEMINI_AVAILABLE: | |
| print("❌ Gemini API not available") | |
| return | |
| try: | |
| genai.configure(api_key=self.config.api_key) | |
| self.model = genai.GenerativeModel( | |
| model_name=self.config.model.value, | |
| generation_config=genai.types.GenerationConfig( | |
| temperature=self.config.temperature, | |
| max_output_tokens=self.config.max_output_tokens, | |
| top_p=self.config.top_p, | |
| top_k=self.config.top_k | |
| ) | |
| ) | |
| print("✅ Gemini API initialized successfully") | |
| except Exception as e: | |
| print(f"❌ Failed to initialize Gemini: {e}") | |
| def generate_creative_examples(self, task_description: str, num_examples: int = 5, | |
| creativity_level: str = "medium") -> List[Dict[str, str]]: | |
| if not self.model: | |
| return [] | |
| creativity_prompts = { | |
| "low": "simple and straightforward", | |
| "medium": "creative and interesting", | |
| "high": "highly creative and unexpected" | |
| } | |
| creativity_desc = creativity_prompts.get(creativity_level, "creative and interesting") | |
| prompt = f | |
| try: | |
| response = self.model.generate_content(prompt) | |
| examples_text = response.text.strip() | |
| if examples_text.startswith("```json"): | |
| examples_text = examples_text[7:] | |
| if examples_text.endswith("```"): | |
| examples_text = examples_text[:-3] | |
| examples = json.loads(examples_text) | |
| valid_examples = [] | |
| for ex in examples[:num_examples]: | |
| if isinstance(ex, dict) and "input" in ex and "output" in ex: | |
| valid_examples.append({ | |
| "input": str(ex["input"]), | |
| "output": str(ex["output"]) | |
| }) | |
| return valid_examples | |
| except json.JSONDecodeError as e: | |
| print(f"❌ Failed to parse JSON response: {e}") | |
| print(f"Response: {response.text if 'response' in locals() else 'No response'}") | |
| return [] | |
| except Exception as e: | |
| print(f"❌ Failed to generate examples: {e}") | |
| return [] | |
| def generate_task_variations(self, base_task: str, num_variations: int = 3) -> List[str]: | |
| if not self.model: | |
| return [] | |
| prompt = f | |
| try: | |
| response = self.model.generate_content(prompt) | |
| variations_text = response.text.strip() | |
| if variations_text.startswith("```json"): | |
| variations_text = variations_text[7:] | |
| if variations_text.endswith("```"): | |
| variations_text = variations_text[:-3] | |
| variations = json.loads(variations_text) | |
| return variations[:num_variations] | |
| except Exception as e: | |
| print(f"❌ Failed to generate task variations: {e}") | |
| return [] | |
| def generate_complex_reasoning_task(self, base_pattern: str, complexity_level: int = 3) -> Dict[str, Any]: | |
| if not self.model: | |
| return {} | |
| complexity_descriptions = { | |
| 1: "simple and straightforward", | |
| 2: "moderately complex", | |
| 3: "complex with multiple steps", | |
| 4: "highly complex with nested patterns", | |
| 5: "extremely complex with advanced reasoning" | |
| } | |
| complexity_desc = complexity_descriptions.get(complexity_level, "complex") | |
| prompt = f | |
| try: | |
| response = self.model.generate_content(prompt) | |
| task_text = response.text.strip() | |
| if task_text.startswith("```json"): | |
| task_text = task_text[7:] | |
| if task_text.endswith("```"): | |
| task_text = task_text[:-3] | |
| task_config = json.loads(task_text) | |
| return task_config | |
| except Exception as e: | |
| print(f"❌ Failed to generate complex task: {e}") | |
| return {} | |
| class GeminiTTTEvaluator: | |
| def __init__(self, config: GeminiConfig): | |
| self.config = config | |
| self.model = None | |
| self._initialize_gemini() | |
| def _initialize_gemini(self): | |
| if not GEMINI_AVAILABLE: | |
| return | |
| try: | |
| genai.configure(api_key=self.config.api_key) | |
| self.model = genai.GenerativeModel( | |
| model_name=self.config.model.value, | |
| generation_config=genai.types.GenerationConfig( | |
| temperature=0.3, | |
| max_output_tokens=self.config.max_output_tokens | |
| ) | |
| ) | |
| except Exception as e: | |
| print(f"❌ Failed to initialize Gemini evaluator: {e}") | |
| def evaluate_ttt_performance(self, task_config: Dict[str, Any], | |
| predictions: List[str], | |
| ground_truths: List[str]) -> Dict[str, Any]: | |
| if not self.model: | |
| return {"error": "Gemini not available"} | |
| correct_predictions = sum(1 for pred, gt in zip(predictions, ground_truths) if pred == gt) | |
| accuracy = correct_predictions / len(predictions) if predictions else 0 | |
| prompt = f | |
| try: | |
| response = self.model.generate_content(prompt) | |
| evaluation_text = response.text.strip() | |
| if evaluation_text.startswith("```json"): | |
| evaluation_text = evaluation_text[7:] | |
| if evaluation_text.endswith("```"): | |
| evaluation_text = evaluation_text[:-3] | |
| evaluation = json.loads(evaluation_text) | |
| evaluation["basic_metrics"] = { | |
| "accuracy": accuracy, | |
| "correct_predictions": correct_predictions, | |
| "total_predictions": len(predictions) | |
| } | |
| return evaluation | |
| except Exception as e: | |
| return { | |
| "error": f"Evaluation failed: {e}", | |
| "basic_metrics": { | |
| "accuracy": accuracy, | |
| "correct_predictions": correct_predictions, | |
| "total_predictions": len(predictions) | |
| } | |
| } | |
| def analyze_pattern_learning(self, examples: List[Dict[str, str]], | |
| predictions: List[str]) -> Dict[str, Any]: | |
| if not self.model: | |
| return {"error": "Gemini not available"} | |
| prompt = f | |
| try: | |
| response = self.model.generate_content(prompt) | |
| analysis_text = response.text.strip() | |
| if analysis_text.startswith("```json"): | |
| analysis_text = analysis_text[7:] | |
| if analysis_text.endswith("```"): | |
| analysis_text = analysis_text[:-3] | |
| analysis = json.loads(analysis_text) | |
| return analysis | |
| except Exception as e: | |
| return {"error": f"Pattern analysis failed: {e}"} | |
| class GeminiTTTInsights: | |
| def __init__(self, config: GeminiConfig): | |
| self.config = config | |
| self.model = None | |
| self._initialize_gemini() | |
| def _initialize_gemini(self): | |
| if not GEMINI_AVAILABLE: | |
| return | |
| try: | |
| genai.configure(api_key=self.config.api_key) | |
| self.model = genai.GenerativeModel( | |
| model_name=self.config.model.value, | |
| generation_config=genai.types.GenerationConfig( | |
| temperature=0.8, | |
| max_output_tokens=self.config.max_output_tokens | |
| ) | |
| ) | |
| except Exception as e: | |
| print(f"❌ Failed to initialize Gemini insights: {e}") | |
| def generate_improvement_recommendations(self, evaluation_results: Dict[str, Any], | |
| task_config: Dict[str, Any]) -> List[str]: | |
| if not self.model: | |
| return [] | |
| prompt = f | |
| try: | |
| response = self.model.generate_content(prompt) | |
| recommendations_text = response.text.strip() | |
| if recommendations_text.startswith("```json"): | |
| recommendations_text = recommendations_text[7:] | |
| if recommendations_text.endswith("```"): | |
| recommendations_text = recommendations_text[:-3] | |
| recommendations = json.loads(recommendations_text) | |
| return recommendations | |
| except Exception as e: | |
| print(f"❌ Failed to generate recommendations: {e}") | |
| return [] | |
| def generate_research_insights(self, experiment_results: Dict[str, Any]) -> Dict[str, Any]: | |
| if not self.model: | |
| return {"error": "Gemini not available"} | |
| prompt = f | |
| try: | |
| response = self.model.generate_content(prompt) | |
| insights_text = response.text.strip() | |
| if insights_text.startswith("```json"): | |
| insights_text = insights_text[7:] | |
| if insights_text.endswith("```"): | |
| insights_text = insights_text[:-3] | |
| insights = json.loads(insights_text) | |
| return insights | |
| except Exception as e: | |
| return {"error": f"Insights generation failed: {e}"} | |
| class AdvancedGeminiTTTManager: | |
| def __init__(self, api_key: str): | |
| self.config = GeminiConfig(api_key=api_key) | |
| self.generator = GeminiTTTGenerator(self.config) | |
| self.evaluator = GeminiTTTEvaluator(self.config) | |
| self.insights = GeminiTTTInsights(self.config) | |
| def run_comprehensive_ttt_experiment(self, base_task: str, test_queries: List[str]) -> Dict[str, Any]: | |
| results = { | |
| "base_task": base_task, | |
| "test_queries": test_queries, | |
| "generated_examples": [], | |
| "task_variations": [], | |
| "complex_task": {}, | |
| "evaluation": {}, | |
| "insights": {}, | |
| "recommendations": [] | |
| } | |
| print("🌟 Starting comprehensive Gemini TTT experiment...") | |
| print("📝 Generating creative examples...") | |
| creative_examples = self.generator.generate_creative_examples( | |
| base_task, num_examples=5, creativity_level="high" | |
| ) | |
| results["generated_examples"] = creative_examples | |
| print("🔄 Generating task variations...") | |
| variations = self.generator.generate_task_variations(base_task, num_variations=3) | |
| results["task_variations"] = variations | |
| print("🧠 Generating complex reasoning task...") | |
| complex_task = self.generator.generate_complex_reasoning_task(base_task, complexity_level=4) | |
| results["complex_task"] = complex_task | |
| print("🔮 Simulating TTT predictions...") | |
| simulated_predictions = [] | |
| ground_truths = [] | |
| for query in test_queries: | |
| prediction = query[::-1] + query[-1] if query else "" | |
| ground_truth = query[::-1] + query[-1] if query else "" | |
| simulated_predictions.append(prediction) | |
| ground_truths.append(ground_truth) | |
| print("📊 Evaluating performance...") | |
| task_config = { | |
| "description": base_task, | |
| "examples": creative_examples | |
| } | |
| evaluation = self.evaluator.evaluate_ttt_performance( | |
| task_config, simulated_predictions, ground_truths | |
| ) | |
| results["evaluation"] = evaluation | |
| print("🔍 Analyzing pattern learning...") | |
| pattern_analysis = self.evaluator.analyze_pattern_learning( | |
| creative_examples, simulated_predictions | |
| ) | |
| results["pattern_analysis"] = pattern_analysis | |
| print("💡 Generating recommendations...") | |
| recommendations = self.insights.generate_improvement_recommendations( | |
| evaluation, task_config | |
| ) | |
| results["recommendations"] = recommendations | |
| print("🔬 Generating research insights...") | |
| research_insights = self.insights.generate_research_insights(results) | |
| results["insights"] = research_insights | |
| print("✅ Comprehensive experiment completed!") | |
| return results | |
| def create_sample_gemini_config() -> GeminiConfig: | |
| api_key = os.getenv("GOOGLE_API_KEY", "") | |
| if not api_key: | |
| print("⚠️ GOOGLE_API_KEY not set. Using placeholder.") | |
| api_key = "your-gemini-api-key-here" | |
| return GeminiConfig( | |
| api_key=api_key, | |
| model=GeminiModelType.GEMINI_PRO, | |
| temperature=0.7, | |
| max_output_tokens=2048 | |
| ) | |
| if __name__ == "__main__": | |
| print("🌟 Testing Gemini API Integration for TTT") | |
| print("=" * 50) | |
| config = create_sample_gemini_config() | |
| if config.api_key == "your-gemini-api-key-here": | |
| print("⚠️ Please set GOOGLE_API_KEY environment variable to test Gemini features") | |
| print(" Example: export GOOGLE_API_KEY='your-actual-api-key'") | |
| else: | |
| manager = AdvancedGeminiTTTManager(config.api_key) | |
| base_task = "Reverse the word and append the last character" | |
| test_queries = ["gemini", "tensor", "neural", "adaptive"] | |
| print(f"📋 Base Task: {base_task}") | |
| print(f"🔍 Test Queries: {test_queries}") | |
| results = manager.run_comprehensive_ttt_experiment(base_task, test_queries) | |
| print("\n📊 Results Summary:") | |
| print(f" Generated Examples: {len(results['generated_examples'])}") | |
| print(f" Task Variations: {len(results['task_variations'])}") | |
| print(f" Complex Task Created: {bool(results['complex_task'])}") | |
| print(f" Evaluation Completed: {bool(results['evaluation'])}") | |
| print(f" Recommendations: {len(results['recommendations'])}") | |
| print(f" Research Insights: {bool(results['insights'])}") | |
| if results['generated_examples']: | |
| print("\n🎨 Sample Generated Examples:") | |
| for i, ex in enumerate(results['generated_examples'][:3]): | |
| print(f" {i+1}. {ex['input']} -> {ex['output']}") | |
| if results['recommendations']: | |
| print("\n💡 Sample Recommendations:") | |
| for i, rec in enumerate(results['recommendations'][:3]): | |
| print(f" {i+1}. {rec}") | |
| print("✅ Gemini integration test completed!") | |
Xet Storage Details
- Size:
- 15.9 kB
- Xet hash:
- f9f743965ca6dee10457163451bf953b879df018df8a630b3c5aab6ce2fe3611
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.