| |
| |
| |
|
|
| |
| |
| |
|
|
| |
| import subprocess |
| import sys |
|
|
| def install_packages(): |
| """Install required packages""" |
| packages = ['openai', 'gradio', 'python-dotenv', 'requests', 'pandas'] |
| for package in packages: |
| try: |
| __import__(package) |
| except ImportError: |
| print(f"Installing {package}...") |
| subprocess.check_call([sys.executable, "-m", "pip", "install", package]) |
|
|
| |
| install_packages() |
|
|
| |
| import gradio as gr |
| import json |
| import random |
| import re |
| import time |
| import os |
| import io |
| import zipfile |
| from datetime import datetime |
| from typing import Dict, List, Any, Optional, Tuple |
| from openai import OpenAI |
| import pandas as pd |
|
|
| |
| |
| |
|
|
| class MedicalLiteratureSimulator: |
| """Simulates medical literature research for health dataset generation""" |
| |
| def __init__(self): |
| self.research_domains = { |
| "longevity": { |
| "interventions": ["NAD+ supplementation", "resveratrol", "metformin", "caloric restriction"], |
| "biomarkers": ["telomere length", "cellular senescence", "inflammatory markers", "mitochondrial function"], |
| "outcomes": ["biological age reduction", "improved healthspan", "enhanced cellular repair"] |
| }, |
| "metabolic_health": { |
| "interventions": ["berberine", "intermittent fasting", "alpha-lipoic acid", "chromium"], |
| "biomarkers": ["glucose levels", "insulin sensitivity", "HbA1c", "HOMA-IR"], |
| "outcomes": ["improved glucose control", "enhanced insulin sensitivity", "reduced inflammation"] |
| }, |
| "cardiovascular": { |
| "interventions": ["omega-3 fatty acids", "coenzyme Q10", "magnesium", "nattokinase"], |
| "biomarkers": ["blood pressure", "cholesterol levels", "CRP", "endothelial function"], |
| "outcomes": ["reduced blood pressure", "improved lipid profile", "decreased inflammation"] |
| }, |
| "cognitive": { |
| "interventions": ["lion's mane mushroom", "phosphatidylserine", "bacopa monnieri", "acetyl-L-carnitine"], |
| "biomarkers": ["cognitive performance", "BDNF levels", "neuroinflammation", "memory function"], |
| "outcomes": ["enhanced memory", "improved cognitive function", "neuroprotection"] |
| }, |
| "hormonal": { |
| "interventions": ["ashwagandha", "vitamin D", "DHEA", "maca root"], |
| "biomarkers": ["cortisol levels", "thyroid hormones", "sex hormones", "stress markers"], |
| "outcomes": ["hormone balance", "improved energy", "better sleep quality"] |
| }, |
| "inflammation": { |
| "interventions": ["curcumin", "omega-3", "quercetin", "boswellia"], |
| "biomarkers": ["CRP", "IL-6", "TNF-alpha", "oxidative stress"], |
| "outcomes": ["reduced inflammation", "improved immune function", "enhanced recovery"] |
| } |
| } |
| |
| def generate_study_data(self, domain: str) -> Dict[str, Any]: |
| """Generate realistic medical study data""" |
| if domain not in self.research_domains: |
| domain = "longevity" |
| |
| domain_data = self.research_domains[domain] |
| |
| study = { |
| "pmid": f"PMID{random.randint(35000000, 40000000)}", |
| "title": self._generate_study_title(domain, domain_data), |
| "abstract": self._generate_study_abstract(domain, domain_data), |
| "journal": random.choice([ |
| "Nature Medicine", "Cell Metabolism", "Journal of Clinical Medicine", |
| "Circulation", "Aging Cell", "Nutrients", "Clinical Nutrition" |
| ]), |
| "year": random.choice([2023, 2024]), |
| "domain": domain, |
| "interventions": random.sample(domain_data["interventions"], min(2, len(domain_data["interventions"]))), |
| "biomarkers": random.sample(domain_data["biomarkers"], min(3, len(domain_data["biomarkers"]))), |
| "outcomes": random.sample(domain_data["outcomes"], min(2, len(domain_data["outcomes"]))), |
| "participant_count": random.randint(50, 300), |
| "duration_weeks": random.choice([8, 12, 16, 24]), |
| "dosages": self._generate_dosages(domain_data["interventions"][0]) |
| } |
| |
| return study |
| |
| def _generate_study_title(self, domain: str, domain_data: Dict) -> str: |
| intervention = random.choice(domain_data["interventions"]) |
| outcome = random.choice(domain_data["outcomes"]) |
| |
| titles = [ |
| f"Effects of {intervention} on {outcome}: A randomized controlled trial", |
| f"{intervention} supplementation improves {outcome} in healthy adults", |
| f"Clinical evaluation of {intervention} for {outcome} optimization", |
| f"Randomized trial of {intervention} in {outcome} enhancement" |
| ] |
| |
| return random.choice(titles) |
| |
| def _generate_study_abstract(self, domain: str, domain_data: Dict) -> str: |
| intervention = domain_data["interventions"][0] |
| biomarker = random.choice(domain_data["biomarkers"]) |
| outcome = random.choice(domain_data["outcomes"]) |
| |
| abstract = f""" |
| Background: {intervention} has shown promise in preliminary studies for health optimization. |
| |
| Objective: To evaluate the effects of {intervention} supplementation on {biomarker} and related health outcomes. |
| |
| Methods: Randomized, double-blind, placebo-controlled trial with {random.randint(120, 250)} participants aged 40-65 years. |
| Subjects received {intervention} or placebo for {random.randint(12, 24)} weeks. |
| |
| Results: {intervention} supplementation significantly improved {outcome} compared to placebo (p<0.05). |
| {biomarker.capitalize()} showed {random.randint(15, 35)}% improvement from baseline. |
| Secondary outcomes included improved quality of life and no serious adverse events. |
| |
| Conclusions: {intervention} supplementation provides significant benefits for {outcome} with excellent safety profile. |
| """.strip() |
| |
| return abstract |
| |
| def _generate_dosages(self, intervention: str) -> List[str]: |
| dosage_ranges = { |
| "NAD+": ["250mg", "500mg", "1000mg"], |
| "resveratrol": ["100mg", "250mg", "500mg"], |
| "berberine": ["500mg", "1000mg", "1500mg"], |
| "omega-3": ["1000mg", "2000mg", "3000mg"], |
| "magnesium": ["200mg", "400mg", "600mg"], |
| "curcumin": ["500mg", "1000mg", "1500mg"] |
| } |
| |
| for key in dosage_ranges: |
| if key.lower() in intervention.lower(): |
| return random.sample(dosage_ranges[key], min(2, len(dosage_ranges[key]))) |
| |
| return ["500mg", "1000mg"] |
|
|
| class HealthProfileGenerator: |
| """Generates realistic health profiles based on medical studies""" |
| |
| def __init__(self): |
| self.severity_levels = { |
| "optimal": {"multiplier": 1.0, "description": "excellent baseline health with optimization focus"}, |
| "mild": {"multiplier": 1.2, "description": "minor health concerns with good overall function"}, |
| "moderate": {"multiplier": 1.5, "description": "noticeable health issues requiring intervention"}, |
| "severe": {"multiplier": 2.0, "description": "significant health challenges needing intensive protocols"} |
| } |
| |
| def generate_profile_from_study(self, study: Dict[str, Any], severity: str = "moderate") -> Dict[str, Any]: |
| """Generate complete health profile based on study data and severity level""" |
| domain = study.get("domain", "longevity") |
| severity_data = self.severity_levels.get(severity, self.severity_levels["moderate"]) |
| multiplier = severity_data["multiplier"] |
| |
| age = random.randint(35, 65) |
| gender = random.choice(["male", "female"]) |
| |
| labs = self._generate_lab_values(domain, multiplier) |
| |
| health_profile = { |
| "user_tests_result_data": { |
| "Labs": labs, |
| "gut_microbiome": self._generate_gut_microbiome(severity), |
| "epigenetics": self._generate_epigenetics(severity), |
| "wearables": self._generate_wearables(severity), |
| "cgm": self._generate_cgm(severity) |
| }, |
| "user_query": self._generate_user_query(study, age, gender, severity), |
| "source_study": { |
| "pmid": study.get("pmid"), |
| "domain": domain, |
| "severity": severity, |
| "title": study.get("title") |
| } |
| } |
| |
| return health_profile |
| |
| def _generate_lab_values(self, domain: str, multiplier: float) -> Dict[str, Any]: |
| """Generate realistic lab values based on domain and severity""" |
| base_labs = { |
| "blood_tests": { |
| "systolic_bp": int(random.randint(120, 140) * multiplier), |
| "diastolic_bp": int(random.randint(70, 90) * multiplier), |
| "total_cholesterol": int(random.randint(180, 220) * multiplier), |
| "ldl": int(random.randint(100, 140) * multiplier), |
| "hdl": int(random.randint(40, 60) / multiplier), |
| "triglycerides": int(random.randint(80, 150) * multiplier), |
| "apoB": int(random.randint(70, 110) * multiplier), |
| "lp_a": random.randint(10, 50) |
| }, |
| "inflammatory": { |
| "hscrp": round(random.uniform(1.0, 4.0) * multiplier, 1), |
| "esr": int(random.randint(5, 25) * multiplier), |
| "il6": round(random.uniform(1.0, 5.0) * multiplier, 1), |
| "tnf_alpha": round(random.uniform(1.0, 3.0) * multiplier, 1), |
| "oxidative_stress_markers": "elevated" if multiplier > 1.3 else "normal", |
| "homocysteine": round(random.uniform(8, 15) * multiplier, 1) |
| }, |
| "nutritional": { |
| "vitamin_d": int(random.randint(25, 50) / multiplier), |
| "b12": random.randint(250, 400), |
| "folate": round(random.uniform(6, 14), 1), |
| "iron": random.randint(60, 120), |
| "ferritin": random.randint(30, 100), |
| "selenium": random.randint(80, 120), |
| "zinc": random.randint(70, 110), |
| "magnesium": round(random.uniform(1.5, 2.2), 1), |
| "omega3_index": round(random.uniform(4, 8) / multiplier, 1) |
| } |
| } |
| |
| if domain == "metabolic_health": |
| base_labs["metabolic"] = { |
| "fasting_glucose": int(random.randint(85, 110) * multiplier), |
| "hba1c": round(random.uniform(5.2, 6.0) * min(multiplier, 1.4), 1), |
| "insulin_fasting": round(random.uniform(5, 15) * multiplier, 1), |
| "homa_ir": round(random.uniform(1.5, 4.0) * multiplier, 1) |
| } |
| |
| return base_labs |
| |
| def _generate_gut_microbiome(self, severity: str) -> str: |
| scores = { |
| "optimal": random.uniform(8.5, 9.5), |
| "mild": random.uniform(7.0, 8.5), |
| "moderate": random.uniform(5.5, 7.0), |
| "severe": random.uniform(3.5, 5.5) |
| } |
| |
| score = scores.get(severity, 6.5) |
| |
| descriptions = { |
| "optimal": "excellent diversity with optimal bacterial balance", |
| "mild": "good diversity with minor imbalances", |
| "moderate": "moderate dysbiosis with reduced beneficial bacteria", |
| "severe": "significant dysbiosis with pathogenic overgrowth" |
| } |
| |
| desc = descriptions.get(severity, "moderate dysbiosis") |
| return f"Diversity score {score:.1f}/10, {desc}, beneficial bacteria {random.randint(60, 90)}%" |
| |
| def _generate_epigenetics(self, severity: str) -> str: |
| age_acceleration = { |
| "optimal": random.randint(-2, 1), |
| "mild": random.randint(1, 3), |
| "moderate": random.randint(3, 6), |
| "severe": random.randint(6, 12) |
| } |
| |
| acceleration = age_acceleration.get(severity, 4) |
| telomere_percentile = max(10, random.randint(30, 80) - acceleration * 5) |
| |
| return f"Biological age acceleration: {acceleration} years, telomere length: {telomere_percentile}th percentile, DunedinPACE: {round(random.uniform(0.9, 1.4), 2)}" |
| |
| def _generate_wearables(self, severity: str) -> Dict[str, int]: |
| base_ranges = { |
| "optimal": {"hrv": (55, 75), "rhr": (45, 60), "sleep": (85, 95)}, |
| "mild": {"hrv": (45, 65), "rhr": (55, 70), "sleep": (75, 85)}, |
| "moderate": {"hrv": (30, 50), "rhr": (65, 80), "sleep": (60, 75)}, |
| "severe": {"hrv": (20, 35), "rhr": (75, 95), "sleep": (45, 65)} |
| } |
| |
| ranges = base_ranges.get(severity, base_ranges["moderate"]) |
| |
| return { |
| "hrv_avg": random.randint(*ranges["hrv"]), |
| "rhr": random.randint(*ranges["rhr"]), |
| "sleep_score": random.randint(*ranges["sleep"]), |
| "recovery_score": random.randint(ranges["sleep"][0]-10, ranges["sleep"][1]-5), |
| "stress_score": random.randint(100-ranges["sleep"][1], 100-ranges["sleep"][0]+20), |
| "vo2_max": random.randint(25, 50), |
| "fitness_age": random.randint(30, 65) |
| } |
| |
| def _generate_cgm(self, severity: str) -> str: |
| glucose_ranges = { |
| "optimal": (80, 95, 92, 98), |
| "mild": (85, 105, 85, 95), |
| "moderate": (95, 120, 70, 85), |
| "severe": (110, 140, 55, 75) |
| } |
| |
| avg_min, avg_max, tir_min, tir_max = glucose_ranges.get(severity, glucose_ranges["moderate"]) |
| return f"Average glucose {random.randint(avg_min, avg_max)} mg/dL, time in range {random.randint(tir_min, tir_max)}%" |
| |
| def _generate_user_query(self, study: Dict[str, Any], age: int, gender: str, severity: str) -> str: |
| domain = study.get("domain", "longevity") |
| |
| base_queries = { |
| "longevity": f"I'm a {age}-year-old {gender} interested in longevity optimization and anti-aging protocols", |
| "metabolic_health": f"I'm a {age}-year-old {gender} with metabolic dysfunction seeking evidence-based glucose control", |
| "cardiovascular": f"I'm a {age}-year-old {gender} with cardiovascular risk factors wanting heart health optimization", |
| "cognitive": f"I'm a {age}-year-old {gender} seeking cognitive enhancement and brain health optimization", |
| "hormonal": f"I'm a {age}-year-old {gender} with hormonal imbalances needing optimization protocols", |
| "inflammation": f"I'm a {age}-year-old {gender} with chronic inflammation seeking anti-inflammatory interventions" |
| } |
| |
| base_query = base_queries.get(domain, base_queries["longevity"]) |
| |
| severity_context = { |
| "optimal": "I have excellent baseline health but want to push the boundaries of optimization", |
| "mild": "I have minor health concerns and want targeted interventions", |
| "moderate": "I have noticeable health issues and need comprehensive protocols", |
| "severe": "I have significant health challenges and require intensive interventions" |
| } |
| |
| context = severity_context.get(severity, "") |
| return f"{base_query}. {context}." |
|
|
| class AIProtocolGenerator: |
| """Uses OpenAI to generate health optimization protocols""" |
| |
| def __init__(self, api_key: str, model: str = "gpt-4"): |
| self.client = OpenAI(api_key=api_key) |
| self.model = model |
| self.total_cost = 0.0 |
| |
| def generate_protocol(self, health_profile: Dict[str, Any], study_context: Dict[str, Any], progress_callback=None) -> Optional[str]: |
| """Generate comprehensive health optimization protocol""" |
| |
| system_prompt = self._create_system_prompt(study_context) |
| user_prompt = self._create_user_prompt(health_profile, study_context) |
| |
| try: |
| if progress_callback: |
| progress_callback(f"🔄 Generating protocol using {self.model}...") |
| |
| response = self.client.chat.completions.create( |
| model=self.model, |
| messages=[ |
| {"role": "system", "content": system_prompt}, |
| {"role": "user", "content": user_prompt} |
| ], |
| max_tokens=4000, |
| temperature=0.7, |
| top_p=0.9 |
| ) |
| |
| self._update_cost(response.usage) |
| |
| if progress_callback: |
| progress_callback(f"✅ Protocol generated ({response.usage.total_tokens} tokens)") |
| |
| return response.choices[0].message.content |
| |
| except Exception as e: |
| if progress_callback: |
| progress_callback(f"❌ Error generating protocol: {e}") |
| return None |
| |
| def _create_system_prompt(self, study_context: Dict[str, Any]) -> str: |
| domain = study_context.get("domain", "health") |
| interventions = ", ".join(study_context.get("interventions", [])) |
| |
| return f"""You are an advanced AI health optimization system specializing in evidence-based medicine and personalized protocols. |
| |
| RESEARCH CONTEXT: |
| - Domain: {domain} optimization |
| - Key Interventions: {interventions} |
| - Evidence Level: Peer-reviewed clinical research |
| |
| PROTOCOL REQUIREMENTS: |
| 1. Executive Summary with current health assessment |
| 2. Multi-Phase Protocol: |
| - Phase 1: Foundation (0-3 months) |
| - Phase 2: Optimization (3-6 months) |
| - Phase 3: Advanced Enhancement (6-12 months) |
| 3. Specific supplement protocols with dosages and timing |
| 4. Lifestyle interventions (exercise, nutrition, sleep) |
| 5. Monitoring and assessment plans |
| 6. Expected outcomes with realistic timelines |
| |
| STYLE: Professional, authoritative, using Medicine 3.0 terminology. Reference biological age, biomarkers, and cellular health. |
| |
| SAFETY: Keep dosages within evidence-based safe ranges. Include monitoring recommendations. |
| |
| Generate comprehensive protocols (3000+ words) with actionable precision medicine recommendations.""" |
| |
| def _create_user_prompt(self, health_profile: Dict[str, Any], study_context: Dict[str, Any]) -> str: |
| return f""" |
| COMPREHENSIVE HEALTH OPTIMIZATION REQUEST: |
| |
| Health Profile Analysis: |
| {json.dumps(health_profile, indent=2)} |
| |
| Research Context: |
| - Study: {study_context.get('title', 'Health Optimization Study')} |
| - Domain: {study_context.get('domain', 'general health')} |
| - Key Findings: Based on clinical research showing significant improvements in health biomarkers |
| |
| Please analyze this health profile and generate a detailed, personalized optimization protocol. Address the specific biomarker patterns, deficiencies, and health challenges identified in the data. Provide targeted interventions with precise dosing, timing, and monitoring protocols. |
| """ |
| |
| def _update_cost(self, usage): |
| pricing = { |
| "gpt-3.5-turbo": {"input": 0.0015, "output": 0.002}, |
| "gpt-4": {"input": 0.03, "output": 0.06}, |
| "gpt-4-turbo": {"input": 0.01, "output": 0.03} |
| } |
| |
| model_pricing = pricing.get(self.model, pricing["gpt-4"]) |
| input_cost = usage.prompt_tokens * model_pricing["input"] / 1000 |
| output_cost = usage.completion_tokens * model_pricing["output"] / 1000 |
| |
| self.total_cost += input_cost + output_cost |
|
|
| class HealthDatasetGenerator: |
| """Complete system that orchestrates the entire dataset generation process""" |
| |
| def __init__(self, api_key: str, model: str = "gpt-4"): |
| self.literature_sim = MedicalLiteratureSimulator() |
| self.profile_gen = HealthProfileGenerator() |
| self.protocol_gen = AIProtocolGenerator(api_key, model) |
| self.generated_examples = [] |
| |
| def generate_dataset(self, |
| domains: List[str] = None, |
| examples_per_domain: int = 2, |
| rate_limit_delay: float = 2.0, |
| progress_callback=None) -> Tuple[List[Dict[str, Any]], str]: |
| """Generate complete health optimization dataset with progress updates""" |
| |
| if domains is None: |
| domains = ["longevity", "metabolic_health", "cardiovascular", "cognitive"] |
| |
| if progress_callback: |
| progress_callback(f"🚀 Starting Health Dataset Generation") |
| progress_callback(f"Domains: {domains}") |
| progress_callback(f"Examples per domain: {examples_per_domain}") |
| progress_callback(f"Total examples to generate: {len(domains) * examples_per_domain}") |
| |
| examples = [] |
| total_examples = len(domains) * examples_per_domain |
| current_example = 0 |
| |
| for domain in domains: |
| if progress_callback: |
| progress_callback(f"\n📂 Processing domain: {domain}") |
| |
| for i in range(examples_per_domain): |
| current_example += 1 |
| try: |
| if progress_callback: |
| progress_callback(f" Creating example {i+1}/{examples_per_domain} (Overall: {current_example}/{total_examples})") |
| |
| |
| study = self.literature_sim.generate_study_data(domain) |
| if progress_callback: |
| progress_callback(f" 📚 Generated study: {study['title'][:50]}...") |
| |
| |
| severity = random.choice(["mild", "moderate", "severe"]) |
| health_profile = self.profile_gen.generate_profile_from_study(study, severity) |
| if progress_callback: |
| progress_callback(f" 👤 Created {severity} health profile") |
| |
| |
| protocol = self.protocol_gen.generate_protocol(health_profile, study, progress_callback) |
| |
| if protocol: |
| training_example = { |
| "user_context": health_profile, |
| "response": protocol, |
| "citations": self._generate_citations(study), |
| "metadata": { |
| "domain": domain, |
| "severity": severity, |
| "study_pmid": study["pmid"], |
| "generated_at": datetime.now().isoformat() |
| } |
| } |
| |
| examples.append(training_example) |
| if progress_callback: |
| progress_callback(f" ✅ Complete example generated") |
| |
| |
| if i < examples_per_domain - 1: |
| if progress_callback: |
| progress_callback(f" ⏳ Rate limit delay: {rate_limit_delay}s") |
| time.sleep(rate_limit_delay) |
| |
| except Exception as e: |
| if progress_callback: |
| progress_callback(f" ❌ Error generating example: {e}") |
| continue |
| |
| if progress_callback: |
| progress_callback(f"\n🎉 Dataset generation complete!") |
| progress_callback(f"Generated: {len(examples)} examples") |
| progress_callback(f"Total cost: ${self.protocol_gen.total_cost:.4f}") |
| |
| self.generated_examples = examples |
| return examples, f"Generated {len(examples)} examples. Total cost: ${self.protocol_gen.total_cost:.4f}" |
| |
| def _generate_citations(self, study: Dict[str, Any]) -> Dict[str, List[str]]: |
| return { |
| "tier_1_peer_reviewed": [study["pmid"], f"PMC{random.randint(1000000, 9999999)}"], |
| "tier_2_rct": [f"{study['domain'].upper()}.2024.{random.randint(100000, 999999)}"], |
| "tier_3_cohort": [f"HEALTH.2023.{random.randint(100000, 999999)}"], |
| "real_world_cases": ["Evidence-based health optimization protocols"] |
| } |
| |
| def export_dataset(self, filename: str = None) -> Tuple[str, List[str]]: |
| """Export dataset and return zip file path and file list""" |
| |
| if not filename: |
| timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") |
| filename = f"health_dataset_{timestamp}" |
| |
| |
| files_created = [] |
| |
| |
| raw_data = json.dumps(self.generated_examples, indent=2, ensure_ascii=False) |
| files_created.append((f"{filename}.json", raw_data)) |
| |
| |
| fine_tune_lines = [] |
| for example in self.generated_examples: |
| fine_tune_example = { |
| "messages": [ |
| { |
| "role": "system", |
| "content": "You are an advanced AI health optimization system that creates evidence-based protocols." |
| }, |
| { |
| "role": "user", |
| "content": f"Create a health optimization protocol for this profile:\n\n{json.dumps(example['user_context'], indent=2)}" |
| }, |
| { |
| "role": "assistant", |
| "content": example["response"] |
| } |
| ] |
| } |
| fine_tune_lines.append(json.dumps(fine_tune_example, ensure_ascii=False)) |
| |
| fine_tune_data = '\n'.join(fine_tune_lines) |
| files_created.append((f"{filename}_fine_tuning.jsonl", fine_tune_data)) |
| |
| |
| sample_size = min(3, len(self.generated_examples)) |
| sample_data = json.dumps(self.generated_examples[:sample_size], indent=2, ensure_ascii=False) |
| files_created.append((f"{filename}_samples.json", sample_data)) |
| |
| |
| metadata = { |
| "generation_info": { |
| "generated_at": datetime.now().isoformat(), |
| "total_examples": len(self.generated_examples), |
| "total_cost": self.protocol_gen.total_cost, |
| "model_used": self.protocol_gen.model |
| }, |
| "domains_covered": list(set(ex["metadata"]["domain"] for ex in self.generated_examples)), |
| "severity_distribution": { |
| severity: sum(1 for ex in self.generated_examples if ex["metadata"]["severity"] == severity) |
| for severity in ["mild", "moderate", "severe"] |
| } |
| } |
| |
| metadata_data = json.dumps(metadata, indent=2, ensure_ascii=False) |
| files_created.append((f"{filename}_metadata.json", metadata_data)) |
| |
| |
| zip_buffer = io.BytesIO() |
| with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file: |
| for file_name, file_content in files_created: |
| zip_file.writestr(file_name, file_content) |
| |
| |
| zip_filename = f"{filename}.zip" |
| with open(zip_filename, 'wb') as f: |
| f.write(zip_buffer.getvalue()) |
| |
| file_list = [f[0] for f in files_created] |
| return zip_filename, file_list |
|
|
| |
| |
| |
|
|
| class HealthDatasetGradioInterface: |
| """Gradio web interface for the health dataset generator""" |
| |
| def __init__(self): |
| self.generator = None |
| self.available_domains = list(MedicalLiteratureSimulator().research_domains.keys()) |
| |
| def estimate_cost(self, domains, examples_per_domain, model): |
| """Estimate generation cost""" |
| if not domains: |
| return "Please select at least one domain" |
| |
| total_examples = len(domains) * examples_per_domain |
| |
| cost_per_example = { |
| "gpt-3.5-turbo": 0.05, |
| "gpt-4": 0.25, |
| "gpt-4-turbo": 0.15 |
| } |
| |
| estimated_cost = total_examples * cost_per_example.get(model, 0.25) |
| |
| return f"💰 Estimated cost: ${estimated_cost:.2f} for {total_examples} examples" |
| |
| def validate_inputs(self, api_key, domains, examples_per_domain): |
| """Validate user inputs""" |
| if not api_key or not api_key.strip(): |
| return False, "❌ Please provide your OpenAI API key" |
| |
| if not domains: |
| return False, "❌ Please select at least one domain" |
| |
| if examples_per_domain < 1 or examples_per_domain > 10: |
| return False, "❌ Examples per domain must be between 1 and 10" |
| |
| return True, "✅ Inputs are valid" |
| |
| def generate_dataset_interface(self, api_key, domains, examples_per_domain, model, rate_limit): |
| """Main dataset generation function for Gradio interface""" |
| |
| |
| is_valid, message = self.validate_inputs(api_key, domains, examples_per_domain) |
| if not is_valid: |
| yield message, "", "", None, None |
| return |
| |
| |
| try: |
| self.generator = HealthDatasetGenerator(api_key.strip(), model) |
| except Exception as e: |
| yield f"❌ Error initializing generator: {e}", "", "", None, None |
| return |
| |
| |
| progress_messages = [] |
| |
| def progress_callback(message): |
| progress_messages.append(message) |
| progress_text = "\n".join(progress_messages[-20:]) |
| return progress_text |
| |
| try: |
| |
| yield "🚀 Starting dataset generation...", "", "", None, None |
| |
| dataset, summary = self.generator.generate_dataset( |
| domains=domains, |
| examples_per_domain=examples_per_domain, |
| rate_limit_delay=rate_limit, |
| progress_callback=progress_callback |
| ) |
| |
| if not dataset: |
| yield "❌ No examples generated", "", "", None, None |
| return |
| |
| |
| progress_callback("💾 Exporting dataset...") |
| zip_filename, file_list = self.generator.export_dataset() |
| |
| |
| preview = self.create_dataset_preview(dataset) |
| |
| |
| final_progress = progress_callback(f"🎉 Generation complete! Files: {', '.join(file_list)}") |
| |
| yield final_progress, summary, preview, zip_filename, file_list |
| |
| except Exception as e: |
| yield f"❌ Error during generation: {e}", "", "", None, None |
| |
| def create_dataset_preview(self, dataset): |
| """Create a preview of the generated dataset""" |
| if not dataset: |
| return "No data to preview" |
| |
| preview = "📄 **Dataset Preview**\n\n" |
| |
| |
| preview += f"**Total Examples:** {len(dataset)}\n" |
| |
| |
| domains = [ex['metadata']['domain'] for ex in dataset] |
| domain_counts = {d: domains.count(d) for d in set(domains)} |
| preview += f"**Domain Distribution:** {domain_counts}\n" |
| |
| |
| severities = [ex['metadata']['severity'] for ex in dataset] |
| severity_counts = {s: severities.count(s) for s in set(severities)} |
| preview += f"**Severity Distribution:** {severity_counts}\n\n" |
| |
| |
| if dataset: |
| example = dataset[0] |
| preview += "**Sample Example:**\n" |
| preview += f"- **Domain:** {example['metadata']['domain']}\n" |
| preview += f"- **Severity:** {example['metadata']['severity']}\n" |
| preview += f"- **User Query:** {example['user_context']['user_query'][:150]}...\n" |
| preview += f"- **Response Length:** {len(example['response'])} characters\n" |
| preview += f"- **PMID:** {example['metadata']['study_pmid']}\n" |
| |
| return preview |
| |
| def analyze_dataset_file(self, zip_file): |
| """Analyze uploaded dataset file""" |
| if zip_file is None: |
| return "No file uploaded" |
| |
| try: |
| |
| with zipfile.ZipFile(zip_file.name, 'r') as zip_ref: |
| |
| json_files = [f for f in zip_ref.namelist() if f.endswith('.json') and not f.endswith('_samples.json') and not f.endswith('_metadata.json')] |
| |
| if json_files: |
| dataset_file = json_files[0] |
| with zip_ref.open(dataset_file) as f: |
| dataset = json.load(f) |
| |
| analysis = "📊 **Dataset Analysis**\n\n" |
| analysis += f"**Total Examples:** {len(dataset)}\n" |
| analysis += f"**Average Response Length:** {sum(len(ex['response']) for ex in dataset) / len(dataset):.0f} characters\n" |
| |
| |
| long_responses = sum(1 for ex in dataset if len(ex['response']) > 2000) |
| has_phases = sum(1 for ex in dataset if "Phase" in ex['response']) |
| has_dosages = sum(1 for ex in dataset if re.search(r'\d+\s*mg', ex['response'])) |
| |
| analysis += f"**Quality Metrics:**\n" |
| analysis += f"- Responses >2000 chars: {long_responses}/{len(dataset)} ({long_responses/len(dataset)*100:.1f}%)\n" |
| analysis += f"- Responses with phases: {has_phases}/{len(dataset)} ({has_phases/len(dataset)*100:.1f}%)\n" |
| analysis += f"- Responses with dosages: {has_dosages}/{len(dataset)} ({has_dosages/len(dataset)*100:.1f}%)\n" |
| |
| return analysis |
| else: |
| return "No dataset JSON file found in zip" |
| |
| except Exception as e: |
| return f"Error analyzing file: {e}" |
| |
| def create_interface(self): |
| """Create the Gradio interface""" |
| |
| with gr.Blocks(title="Medical Literature Health Dataset Generator", theme=gr.themes.Soft()) as interface: |
| |
| gr.Markdown(""" |
| # 🏥 Medical Literature Health Dataset Generator |
| |
| This tool generates synthetic health optimization datasets based on medical literature patterns. |
| Perfect for training AI models on evidence-based health protocols. |
| |
| ⚠️ **Important:** Generated content is for research/educational purposes only. Not medical advice. |
| """) |
| |
| with gr.Tab("📊 Generate Dataset"): |
| |
| with gr.Row(): |
| with gr.Column(scale=1): |
| gr.Markdown("### ⚙️ Configuration") |
| |
| api_key = gr.Textbox( |
| label="OpenAI API Key", |
| placeholder="sk-...", |
| type="password", |
| info="Your OpenAI API key for generating protocols" |
| ) |
| |
| domains = gr.CheckboxGroup( |
| label="Research Domains", |
| choices=self.available_domains, |
| value=["longevity", "metabolic_health"], |
| info="Select medical research domains to include" |
| ) |
| |
| examples_per_domain = gr.Slider( |
| label="Examples per Domain", |
| minimum=1, |
| maximum=10, |
| value=2, |
| step=1, |
| info="Number of examples to generate for each domain" |
| ) |
| |
| model = gr.Dropdown( |
| label="OpenAI Model", |
| choices=["gpt-3.5-turbo", "gpt-4", "gpt-4-turbo"], |
| value="gpt-4", |
| info="Model for generating protocols (GPT-4 recommended for quality)" |
| ) |
| |
| rate_limit = gr.Slider( |
| label="Rate Limit Delay (seconds)", |
| minimum=0.5, |
| maximum=5.0, |
| value=2.0, |
| step=0.5, |
| info="Delay between API calls to avoid rate limits" |
| ) |
| |
| cost_estimate = gr.Textbox( |
| label="Cost Estimate", |
| value="Select domains and examples to see estimate", |
| interactive=False |
| ) |
| |
| generate_btn = gr.Button( |
| "🚀 Generate Dataset", |
| variant="primary", |
| size="lg" |
| ) |
| |
| with gr.Column(scale=2): |
| gr.Markdown("### 📈 Progress & Results") |
| |
| progress_output = gr.Textbox( |
| label="Generation Progress", |
| lines=15, |
| max_lines=20, |
| value="Ready to generate dataset...", |
| interactive=False |
| ) |
| |
| summary_output = gr.Textbox( |
| label="Generation Summary", |
| lines=3, |
| interactive=False |
| ) |
| |
| preview_output = gr.Markdown( |
| label="Dataset Preview", |
| value="Dataset preview will appear here..." |
| ) |
| |
| with gr.Row(): |
| download_file = gr.File( |
| label="📥 Download Generated Dataset", |
| interactive=False |
| ) |
| |
| file_list = gr.Textbox( |
| label="Generated Files", |
| placeholder="Files included in download will be listed here", |
| interactive=False |
| ) |
| |
| with gr.Tab("📊 Analyze Dataset"): |
| gr.Markdown("### 📋 Dataset Analysis") |
| gr.Markdown("Upload a generated dataset zip file to analyze its quality and structure.") |
| |
| with gr.Row(): |
| with gr.Column(): |
| upload_file = gr.File( |
| label="Upload Dataset Zip File", |
| file_types=[".zip"] |
| ) |
| |
| analyze_btn = gr.Button( |
| "🔍 Analyze Dataset", |
| variant="secondary" |
| ) |
| |
| with gr.Column(): |
| analysis_output = gr.Markdown( |
| label="Analysis Results", |
| value="Upload a dataset file to see analysis..." |
| ) |
| |
| with gr.Tab("ℹ️ Information"): |
| gr.Markdown(""" |
| ### 📚 How It Works |
| |
| 1. **Literature Simulation**: Creates realistic medical studies with proper abstracts, interventions, and outcomes |
| 2. **Health Profile Generation**: Generates comprehensive health profiles based on study domains and severity levels |
| 3. **AI Protocol Generation**: Uses OpenAI to create detailed health optimization protocols |
| 4. **Dataset Export**: Outputs data in multiple formats including OpenAI fine-tuning format |
| |
| ### 🎯 Output Files |
| |
| - **`dataset.json`**: Complete raw dataset |
| - **`dataset_fine_tuning.jsonl`**: OpenAI fine-tuning format |
| - **`dataset_samples.json`**: Sample examples for review |
| - **`dataset_metadata.json`**: Generation statistics and info |
| |
| ### 💰 Cost Information |
| |
| - **GPT-3.5-turbo**: ~$0.05 per example |
| - **GPT-4**: ~$0.25 per example |
| - **GPT-4-turbo**: ~$0.15 per example |
| |
| ### ⚠️ Important Notes |
| |
| - Generated content is for **research/educational purposes only** |
| - **Not medical advice** - always consult healthcare professionals |
| - Include appropriate medical disclaimers when using generated content |
| - Review sample outputs before using in production |
| |
| ### 🔧 Recommended Settings |
| |
| - **Start small**: Generate 2-4 examples first to test quality |
| - **Use GPT-4**: Better quality than GPT-3.5-turbo |
| - **Rate limiting**: Use 2+ second delays to avoid API limits |
| - **Multiple domains**: Include diverse domains for comprehensive dataset |
| """) |
| |
| |
| |
| |
| def update_cost_estimate(domains, examples_per_domain, model): |
| return self.estimate_cost(domains, examples_per_domain, model) |
| |
| for input_component in [domains, examples_per_domain, model]: |
| input_component.change( |
| fn=update_cost_estimate, |
| inputs=[domains, examples_per_domain, model], |
| outputs=[cost_estimate] |
| ) |
| |
| |
| generate_btn.click( |
| fn=self.generate_dataset_interface, |
| inputs=[api_key, domains, examples_per_domain, model, rate_limit], |
| outputs=[progress_output, summary_output, preview_output, download_file, file_list] |
| ) |
| |
| |
| analyze_btn.click( |
| fn=self.analyze_dataset_file, |
| inputs=[upload_file], |
| outputs=[analysis_output] |
| ) |
| |
| return interface |
|
|
| |
| |
| |
|
|
| def main(): |
| """Launch the Gradio interface""" |
| |
| print("🚀 Launching Medical Literature Health Dataset Generator") |
| print("This will start a web interface accessible through your browser") |
| |
| |
| interface_creator = HealthDatasetGradioInterface() |
| interface = interface_creator.create_interface() |
| |
| |
| interface.launch( |
| share=True, |
| server_name="0.0.0.0", |
| server_port=7860, |
| show_error=True, |
| quiet=False |
| ) |
|
|
| if __name__ == "__main__": |
| main() |
| |
| |
| |
