| | |
| | |
| | |
| |
|
| | |
| | |
| | |
| |
|
| | |
| | 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() |
| | |
| | |
| | |
