Spaces:
Configuration error
Configuration error
File size: 12,374 Bytes
ceaf3bb 662a0bb ceaf3bb 662a0bb ceaf3bb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 | from io import BytesIO
from zipfile import ZipFile
from typing import Dict, Callable, Any, List
from hypotheses_generation import generate_hypotheses_from_system_charts
class HypothesesService:
"""
Service class for Scientific Hypotheses Generation.
Handles the orchestration of parsing system charts, generating hypotheses,
and packaging results using the new synergistic approach.
"""
def __init__(self):
pass
async def generate_hypotheses(
self,
system_charts_content_list: List[str],
llm,
material_goal_1: str = "Property #1 for xxx material",
material_goal_2: str = "Property #2 for xxx material",
iterations_count: int = 1,
progress_callback: Callable[[Dict[str, Any]], None] = None
) -> Dict[str, Any]:
"""
Main hypotheses generation pipeline using synergistic approach.
Args:
system_charts_content_list: List of content strings from multiple system chart files
llm: Language model for hypothesis generation
material_goal_1: First material design goal to guide hypothesis generation
material_goal_2: Second material design goal to guide hypothesis generation
iterations_count: Number of hypothesis iterations to generate
progress_callback: Function to report progress updates
Returns:
Dict containing generation results and metadata
"""
try:
# Initial progress update
if progress_callback:
progress_callback({
"status": "starting",
"progress": 0,
"message": "Preparing for synergistic hypothesis generation..."
})
# Handle both single content and list of contents for backward compatibility
if isinstance(system_charts_content_list, str):
system_charts_content_list = [system_charts_content_list]
# Create chart data structure for multiple files
chart_data = []
for i, content in enumerate(system_charts_content_list, 1):
if content and content.strip():
chart_data.append({
'content': content,
'citation': f'System chart file {i}',
'title': f'System Chart File {i}',
'raw_content': content # Pass the raw content
})
if not chart_data:
return {
"success": False,
"message": "No valid system chart content provided for hypothesis generation.",
"zip_data": None,
"filename": None
}
print(f"📊 Prepared {len(chart_data)} system chart files for synergistic hypothesis generation")
# Update progress after parsing
if progress_callback:
progress_callback({
"status": "parsing_complete",
"progress": 10,
"total_charts": len(chart_data),
"message": f"System charts prepared from {len(chart_data)} files. Starting synergistic hypothesis generation..."
})
# Create progress callback function for the generation process
def generation_progress_callback(progress_data):
if progress_callback:
# Adjust progress to fit within our overall range (10% to 95%)
adjusted_progress = 10 + int((progress_data.get("progress", 0) / 100) * 85)
progress_data["progress"] = adjusted_progress
progress_callback(progress_data)
# Generate hypotheses using the new synergistic approach
print(f"🧠 Starting synergistic hypothesis generation with goals: '{material_goal_1}' and '{material_goal_2}' ({iterations_count} iteration{'s' if iterations_count > 1 else ''})...")
hypotheses_content, hypotheses = await generate_hypotheses_from_system_charts(
chart_data,
llm,
material_goal_1,
material_goal_2,
iterations_count,
generation_progress_callback
)
# Note: Cross-chart analysis is now integrated into the main generation process
cross_analysis_content = "" # No longer separate
# Create ZIP file in memory
if progress_callback:
progress_callback({
"status": "packaging",
"progress": 95,
"message": "Creating results package..."
})
zip_data = self._create_zip_file(
hypotheses_content,
cross_analysis_content
)
# Prepare result
return {
"success": True,
"message": f"Generated {iterations_count} iteration{'s' if iterations_count > 1 else ''} of synergistic hypotheses from {len(chart_data)} system chart files using cross-paper analysis approach!",
"zip_data": zip_data,
"filename": "hypotheses.zip",
"hypotheses": hypotheses,
"charts_processed": len(chart_data),
"cross_analysis_included": True, # Now always included in main process
"method": "synergistic_analysis",
"iterations_count": iterations_count
}
except Exception as e:
return {
"success": False,
"message": f"Error during synergistic hypothesis generation: {str(e)}",
"zip_data": None,
"filename": None
}
def _create_zip_file(
self,
hypotheses_content: str,
cross_analysis_content: str = ""
) -> bytes:
"""Create ZIP file containing hypothesis generation results."""
zip_buffer = BytesIO()
with ZipFile(zip_buffer, "w") as zipf:
# Add main hypotheses file
zipf.writestr("hypotheses.txt", hypotheses_content)
print(f"📦 Created synergistic hypotheses ZIP package with {len(zipf.namelist())} files")
# Get ZIP data and close buffer
zip_data = zip_buffer.getvalue()
zip_buffer.close()
return zip_data
@staticmethod
def validate_multiple_system_charts_content(content_list: List[str]) -> Dict[str, Any]:
"""
Validate that the provided content list is suitable for synergistic hypothesis generation.
Args:
content_list: List of content strings to validate
Returns:
Dict with validation results
"""
try:
if not content_list:
return {
"valid": False,
"message": "No content provided for hypothesis generation.",
"chart_count": 0
}
valid_files = 0
total_data_rows = 0
file_details = []
for i, content in enumerate(content_list, 1):
if not content or not content.strip():
file_details.append(f"File {i}: Empty content")
continue
# Check for new format with 100 '=' delimiters
lines = content.strip().split('\n')
has_new_format = any(line.strip() == '=' * 100 for line in lines)
has_paper_keywords = any(line.strip().startswith('PAPER:') for line in lines)
has_chart_keywords = any(line.strip().startswith('SYSTEM_CHART:') for line in lines)
if has_new_format and has_paper_keywords and has_chart_keywords:
# Count entries in new format
paper_count = len([line for line in lines if line.strip().startswith('PAPER:')])
chart_count = len([line for line in lines if line.strip().startswith('SYSTEM_CHART:')])
if paper_count > 0 and chart_count > 0 and paper_count == chart_count:
valid_files += 1
total_data_rows += paper_count
file_details.append(f"File {i}: Valid format with {paper_count} entries")
else:
file_details.append(f"File {i}: Format issues - {paper_count} papers, {chart_count} charts")
else:
# Fallback to table-based validation
table_lines = [line for line in lines if '|' in line and not line.strip().startswith('|---')]
if table_lines:
data_rows = len([line for line in table_lines if line.strip() and (
('|' in line) and not any(
header_word in line.lower() for header_word in ['processing', 'structure', 'property', 'mechanism']
)
)])
if data_rows >= 2:
valid_files += 1
total_data_rows += data_rows
file_details.append(f"File {i}: Valid table format with {data_rows} rows")
else:
file_details.append(f"File {i}: Insufficient table data ({data_rows} rows)")
else:
file_details.append(f"File {i}: No valid format found")
if valid_files == 0:
return {
"valid": False,
"message": f"No valid system chart files found. Details: {'; '.join(file_details)}",
"chart_count": 0,
"valid_charts": 0,
"file_details": file_details
}
if total_data_rows < 2:
return {
"valid": False,
"message": f"Insufficient data across all files. Need multiple entries for synergistic analysis. Found {total_data_rows} total entries.",
"chart_count": len(content_list),
"valid_charts": valid_files,
"file_details": file_details
}
return {
"valid": True,
"message": f"Validated {valid_files}/{len(content_list)} system chart files for synergistic hypothesis generation. Found {total_data_rows} total entries.",
"chart_count": len(content_list),
"valid_charts": valid_files,
"data_entries": total_data_rows,
"file_details": file_details
}
except Exception as e:
return {
"valid": False,
"message": f"Error validating content: {str(e)}",
"chart_count": 0,
"file_details": []
}
@staticmethod
def validate_system_charts_content(content) -> Dict[str, Any]:
"""
Validate that the provided content is suitable for synergistic hypothesis generation.
Handles both single content string and list of content strings.
Args:
content: Content to validate (str or List[str])
Returns:
Dict with validation results
"""
# Handle both single content and list of contents
if isinstance(content, str):
return HypothesesService.validate_multiple_system_charts_content([content])
elif isinstance(content, list):
return HypothesesService.validate_multiple_system_charts_content(content)
else:
return {
"valid": False,
"message": "Invalid content type. Expected string or list of strings.",
"chart_count": 0
} |