如果你是一名天文学家或数据科学家每天面对詹姆斯·韦伯太空望远镜JWST传回的海量数据是否曾想过有没有一种方法能让AI智能体自动识别其中的候选星系而不是手动在成千上万个模糊的光点中逐个筛选这正是Gemini 3.5 Flash结合Antigravity harness正在解决的问题。传统天文数据分析需要研究人员花费数周时间进行图像预处理、特征提取和分类验证而如今通过智能体协作框架这一过程可以压缩到几小时甚至几分钟内完成。但这里有个关键点容易被忽略很多人以为这只是另一个AI辅助工具实际上Gemini 3.5 Flash Antigravity harness的组合代表的是完全不同的工作范式。它不是简单地在现有流程中加入AI模块而是重新设计了整个数据分析流水线——从数据获取、多模态理解到决策验证全部由协同工作的子智能体自主完成。本文将深入解析这一技术组合在天文数据分析中的实际应用重点演示如何构建一个能够从JWST数据中自主识别候选星系的智能体系统。无论你是天文学研究者、数据科学家还是对AI智能体开发感兴趣的工程师都能从中获得可直接复用的技术方案和实践洞察。1. 为什么传统天文数据分析需要智能体革命天文数据处理长期以来面临三个核心挑战数据量巨大、分析流程复杂、人工验证耗时。以JWST为例单次观测就能产生数TB的数据包含数百万个潜在天体。传统方法下研究人员需要使用专业软件如Astropy、SExtractor进行图像预处理和源提取手动设置筛选阈值和分类规则对候选目标进行交叉验证和人工检查这个过程不仅效率低下还容易因主观判断引入偏差。更关键的是当发现新的天体类型或异常现象时整个分析流程可能需要重新设计。Gemini 3.5 Flash的突破在于其前沿智能与行动结合的设计理念。根据官方数据它在多模态理解基准CharXiv Reasoning上达到84.2%的准确率这意味着它能同时处理图像、文本和代码信息理解天文数据中的复杂模式。而Antigravity harness的作用是让多个 specialized agents协同工作。想象一下一个agent负责图像预处理另一个负责特征提取第三个进行交叉验证它们之间可以实时交换信息和调整策略。这种架构特别适合JWST数据分析这种需要多步骤推理的长周期任务。2. Gemini 3.5 Flash与Antigravity harness技术架构解析2.1 Gemini 3.5 Flash的核心能力Gemini 3.5 Flash不是通用大模型的简单升级而是专门为智能体任务优化的架构。其技术特点包括多模态理解深度整合能够同时处理JWST的FITS格式图像、观测元数据和科学论文上下文长上下文窗口支持百万级token的连续推理适合分析完整的观测数据集工具调用能力可以无缝集成天文数据分析工具链如AstroPy、photutils代码生成与执行直接生成数据预处理、特征提取的可执行Python代码与之前版本相比3.5 Flash在终端基准测试Terminal-Bench 2.1上达到76.2%的准确率这意味着它在执行复杂命令行任务和数据处理流程时更加可靠。2.2 Antigravity harness的协同智能体框架Antigravity harness的本质是一个智能体编排平台其核心创新在于动态子智能体分配根据任务复杂度自动决定需要多少专门agent协作上下文持久化各个子智能体共享工作记忆避免重复计算错误恢复机制当某个步骤失败时系统能够自动回滚并尝试替代方案人类监督接口关键决策点等待专家确认平衡自动化与质量控制在实际的JWST数据分析场景中这意味着系统可以同时部署图像处理agent、光谱分析agent和分类验证agent它们共同完成从原始数据到候选星系列表的完整流程。3. 环境准备与天文数据分析基础设置3.1 硬件与软件要求对于JWST级别的大规模数据分析建议配置# 最低硬件要求 CPU: 8核心以上 内存: 32GB以上 GPU: NVIDIA RTX 4080以上用于加速图像处理 存储: 1TB SSD用于缓存JWST数据集 # 软件环境 Python: 3.9 CUDA: 12.0如使用GPU加速3.2 核心依赖库安装创建专用的天文数据分析环境# 创建conda环境 conda create -n jwst-analysis python3.9 conda activate jwst-analysis # 安装天文数据分析核心库 pip install astropy photutils scikit-image pip install matplotlib seaborn pandas numpy # 安装机器学习相关库 pip install torch torchvision transformers # 安装Gemini API客户端 pip install google-generativeai # 安装JWST数据访问工具 pip install jwst mastquery3.3 Gemini API密钥配置获取Gemini 3.5 Flash访问权限后配置API密钥# config.py - API配置管理 import os from google import genai # 从环境变量读取API密钥 GEMINI_API_KEY os.getenv(GEMINI_API_KEY, your_api_key_here) # 初始化Gemini客户端 client genai.Client(api_keyGEMINI_API_KEY) # 配置模型参数 GEMINI_CONFIG { model: gemini-3.5-flash, temperature: 0.1, # 低随机性保证科学准确性 max_output_tokens: 8192, # 长输出支持复杂推理 top_p: 0.8 }4. JWST数据获取与预处理流水线设计4.1 自动化数据下载与验证JWST数据通过MASTMikulski Archive for Space Telescopes平台公开获取。以下是自动化下载流程# jwst_data_loader.py from mastquery import Observations import astropy.io.fits as fits import os class JWSTDataLoader: def __init__(self, data_dir./jwst_data): self.data_dir data_dir os.makedirs(data_dir, exist_okTrue) def query_observations(self, ra, dec, radius0.1, instrumentNIRCam): 查询指定天区的JWST观测数据 observations Observations.query_region( f{ra} {dec}, radiusradius, instrument_nameinstrument ) return observations def download_data(self, observation_ids, file_types[SCIENCE]): 下载指定的观测数据 downloaded_files [] for obs_id in observation_ids: try: manifest Observations.download_products( obs_id, productTypefile_types, download_dirself.data_dir ) downloaded_files.extend(manifest[Local Path].tolist()) except Exception as e: print(f下载观测{obs_id}失败: {e}) return downloaded_files def validate_fits_files(self, file_paths): 验证FITS文件完整性 valid_files [] for file_path in file_paths: try: with fits.open(file_path) as hdul: if len(hdul) 0 and hdul[0].data is not None: valid_files.append(file_path) else: print(f文件{file_path}数据为空或损坏) except Exception as e: print(f验证文件{file_path}时出错: {e}) return valid_files # 使用示例 if __name__ __main__: loader JWSTDataLoader() # 查询仙女座星系区域的观测 observations loader.query_observations(00h42m44.3s, 41d16m09s, radius0.2) files loader.download_data(observations[obsid][:5]) # 下载前5个观测 valid_files loader.validate_fits_files(files)4.2 图像预处理与质量过滤原始JWST数据需要经过一系列预处理才能用于科学分析# image_preprocessor.py import numpy as np from astropy.stats import sigma_clipped_stats from photutils import detect_sources, deblend_sources from scipy import ndimage class JWSTImagePreprocessor: def __init__(self, detection_threshold3.0): self.detection_threshold detection_threshold def calibrate_image(self, image_data, dark_frameNone, flat_fieldNone): 基本图像校准 calibrated image_data.copy() if dark_frame is not None: calibrated - dark_frame if flat_field is not None: calibrated / flat_field calibrated * np.mean(flat_field) return calibrated def remove_background(self, image_data): 背景扣除 mean, median, std sigma_clipped_stats(image_data, sigma3.0) background_removed image_data - median return background_removed, std def detect_sources(self, image_data, fwhm3.0): 源检测与去混叠 threshold self.detection_threshold mean, median, std sigma_clipped_stats(image_data, sigma3.0) # 使用photutils进行源检测 segmentation detect_sources(image_data, threshold, npixels10) if segmentation is not None and segmentation.nlabels 1: # 去混叠紧密源 segmented deblend_sources(image_data, segmentation, npixels10) else: segmented segmentation return segmented def extract_source_properties(self, image_data, segmentation): 提取检测源的基本属性 sources_properties [] labels np.unique(segmentation.data) labels labels[labels 0] # 排除背景 for label in labels: mask segmentation.data label y, x np.where(mask) properties { label: label, centroid: (np.mean(x), np.mean(y)), area: np.sum(mask), flux: np.sum(image_data[mask]), max_value: np.max(image_data[mask]), snr: np.max(image_data[mask]) / np.std(image_data[~mask]) } sources_properties.append(properties) return sources_properties # 预处理流水线示例 def run_preprocessing_pipeline(fits_file_path): 完整的预处理流水线 with fits.open(fits_file_path) as hdul: image_data hdul[1].data # 假设科学数据在第二个HDU preprocessor JWSTImagePreprocessor() # 步骤1: 图像校准 calibrated preprocessor.calibrate_image(image_data) # 步骤2: 背景扣除 background_removed, noise_std preprocessor.remove_background(calibrated) # 步骤3: 源检测 segmentation preprocessor.detect_sources(background_removed) # 步骤4: 特征提取 if segmentation is not None: sources preprocessor.extract_source_properties(background_removed, segmentation) return sources, segmentation else: return [], None5. 基于Antigravity harness的多智能体协作架构5.1 智能体角色定义与任务分配在JWST数据分析场景中我们设计四个核心智能体# agent_orchestrator.py from typing import List, Dict, Any import asyncio class JWSTAnalysisOrchestrator: def __init__(self, gemini_client): self.client gemini_client self.agents { data_manager: None, feature_analyzer: None, galaxy_classifier: None, quality_validator: None } async def initialize_agents(self): 初始化各专业智能体 # 数据管理智能体 - 负责数据获取和预处理 data_manager_prompt 你是JWST数据分析专家专门负责数据获取、校准和质量管理。 你的任务包括 1. 从MAST archive下载指定天区的JWST观测数据 2. 执行基本图像处理平场校正、背景扣除 3. 验证数据质量并过滤不合格图像 4. 为后续分析准备标准化数据格式 请严格按照天文数据处理标准操作确保数据可靠性。 self.agents[data_manager] await self.create_agent( data_manager, data_manager_prompt ) # 特征分析智能体 - 提取天体物理特征 feature_analyzer_prompt 你是天体物理特征提取专家擅长从天文图像中量化天体属性。 你的任务包括 1. 检测图像中的点源和延展源 2. 测量流量、大小、形状等形态学特征 3. 计算信噪比和检测显著性 4. 生成标准化的特征向量用于分类 请确保测量结果符合天体测量学标准。 self.agents[feature_analyzer] await self.create_agent( feature_analyzer, feature_analyzer_prompt ) # 星系分类智能体 - 识别候选星系 galaxy_classifier_prompt 你是星系分类专家精通各种星系类型的形态特征。 你的任务包括 1. 基于形态特征区分类星点、星系、 artifacts 2. 识别不同哈勃类型的星系椭圆、旋涡、不规则 3. 检测特殊天体引力透镜、并合星系、活动星系核 4. 为每个候选目标分配置信度分数 请参考标准星系分类框架如哈勃序列。 self.agents[galaxy_classifier] await self.create_agent( galaxy_classifier, galaxy_classifier_prompt ) # 质量验证智能体 - 交叉验证结果可靠性 quality_validator_prompt 你是科学数据质量验证专家负责确保分析结果的可靠性。 你的任务包括 1. 检查特征测量的一致性 2. 验证分类结果的统计显著性 3. 识别可能的误检和漏检 4. 生成质量评估报告 请采用保守的验证策略优先保证结果的科学性。 self.agents[quality_validator] await self.create_agent( quality_validator, quality_validator_prompt ) async def create_agent(self, role: str, system_prompt: str): 创建专用智能体 response self.client.models.generate_content( modelgemini-3.5-flash, contents[{ role: user, parts: [{text: f创建{role}智能体系统提示词{system_prompt}}] }] ) return response.text async def coordinate_analysis(self, target_region: Dict[str, Any]): 协调多智能体完成分析任务 results {} # 阶段1: 数据获取与预处理 data_task f 获取{target_region[ra]} {target_region[dec]}天区的JWST观测数据。 观测仪器{target_region.get(instrument, NIRCam)} 天区半径{target_region.get(radius, 0.1)}度 返回校准后的科学就绪数据。 results[raw_data] await self.query_agent(data_manager, data_task) # 阶段2: 特征提取 feature_task f 分析提供的JWST图像数据检测所有源并提取形态学特征。 重点测量位置、流量、半光半径、椭圆率、信噪比。 返回结构化特征表格。 results[features] await self.query_agent(feature_analyzer, feature_task) # 阶段3: 星系分类 classification_task f 基于提取的特征识别候选星系并分类。 区分类星点、前景恒星、背景星系。 对星系进一步分类椭圆、旋涡、不规则、特殊类型。 返回分类结果和置信度。 results[classifications] await self.query_agent(galaxy_classifier, classification_task) # 阶段4: 质量验证 validation_task f 验证分类结果的可靠性检查可能的系统误差。 特别关注低信噪比源、拥挤场区域、仪器伪影。 生成质量评估报告和改进建议。 results[validation] await self.query_agent(quality_validator, validation_task) return results async def query_agent(self, agent_name: str, task: str): 向特定智能体发送查询 if agent_name not in self.agents: raise ValueError(f未知智能体: {agent_name}) response self.client.models.generate_content( modelgemini-3.5-flash, contents[{ role: user, parts: [{text: f{task}}] }] ) return response.text5.2 智能体间通信与上下文管理Antigravity harness的核心优势在于智能体间的协同工作机制# context_manager.py import json from datetime import datetime from typing import Dict, Any, List class AnalysisContextManager: def __init__(self): self.shared_context {} self.agent_messages [] self.decision_points [] def update_context(self, agent: str, data_type: str, data: Any, confidence: float 1.0): 更新共享上下文 timestamp datetime.now().isoformat() context_entry { agent: agent, data_type: data_type, data: data, confidence: confidence, timestamp: timestamp, version: len([x for x in self.shared_context.get(data_type, [])]) 1 } if data_type not in self.shared_context: self.shared_context[data_type] [] self.shared_context[data_type].append(context_entry) # 记录消息用于审计 self.agent_messages.append({ from: agent, type: context_update, data_type: data_type, timestamp: timestamp }) def get_context(self, data_type: str, version: int -1): 获取指定类型的上下文 if data_type not in self.shared_context: return None entries self.shared_context[data_type] if version -1: # 获取最新版本 return entries[-1] if entries else None else: return next((e for e in entries if e[version] version), None) def create_decision_point(self, question: str, options: List[str], critical: bool False): 创建决策点可能需要人工干预 decision_id fdecision_{len(self.decision_points) 1:04d} decision { id: decision_id, question: question, options: options, critical: critical, created_at: datetime.now().isoformat(), resolved: False, choice: None, resolved_at: None } self.decision_points.append(decision) return decision_id def resolve_decision(self, decision_id: str, choice: str, notes: str ): 解决决策点 decision next((d for d in self.decision_points if d[id] decision_id), None) if decision: decision[resolved] True decision[choice] choice decision[notes] notes decision[resolved_at] datetime.now().isoformat() def generate_audit_report(self): 生成分析过程审计报告 report { summary: { total_agents: len(set(msg[from] for msg in self.agent_messages)), total_messages: len(self.agent_messages), context_updates: len(self.shared_context), decisions_made: len([d for d in self.decision_points if d[resolved]]), pending_decisions: len([d for d in self.decision_points if not d[resolved]]) }, timeline: self.agent_messages, context_snapshot: {k: v[-1] for k, v in self.shared_context.items() if v}, decisions: self.decision_points } return json.dumps(report, indent2, defaultstr)6. 完整示例从JWST数据到候选星系识别6.1 端到端分析流水线实现下面展示完整的智能体驱动分析流程# jwst_galaxy_pipeline.py import asyncio import json from config import client, GEMINI_CONFIG from agent_orchestrator import JWSTAnalysisOrchestrator from context_manager import AnalysisContextManager class JWSTGalaxyPipeline: def __init__(self): self.client client self.orchestrator JWSTAnalysisOrchestrator(client) self.context AnalysisContextManager() async def initialize(self): 初始化流水线 await self.orchestrator.initialize_agents() print(智能体系统初始化完成) async def run_analysis(self, target_config: Dict): 运行完整分析流程 print(f开始分析目标天区: {target_config[name]}) # 记录开始时间 self.context.update_context(pipeline, analysis_start, {target: target_config, timestamp: datetime.now()}) try: # 执行多智能体协同分析 results await self.orchestrator.coordinate_analysis(target_config) # 记录结果到上下文 self.context.update_context(pipeline, analysis_results, results, 0.9) # 生成最终报告 report self.generate_final_report(results, target_config) # 质量检查 quality_score self.assess_quality(report) self.context.update_context(quality_validator, quality_assessment, {score: quality_score, report: report}) print(f分析完成质量评分: {quality_score}/100) return report except Exception as e: error_info {error: str(e), target: target_config} self.context.update_context(pipeline, analysis_error, error_info, 0.1) raise def generate_final_report(self, results: Dict, target_config: Dict) - Dict: 生成最终科学报告 # 提取关键结果 raw_data_info results.get(raw_data, {}) features results.get(features, {}) classifications results.get(classifications, {}) validation results.get(validation, {}) # 统计候选星系 candidate_galaxies [ obj for obj in classifications.get(objects, []) if obj.get(type) galaxy and obj.get(confidence, 0) 0.7 ] report { target_info: target_config, execution_summary: { total_sources_detected: len(features.get(sources, [])), candidate_galaxies: len(candidate_galaxies), analysis_timestamp: datetime.now().isoformat(), data_quality: raw_data_info.get(quality_rating, unknown) }, candidate_galaxies: candidate_galaxies, quality_assessment: validation, context_snapshot: self.context.shared_context } return report def assess_quality(self, report: Dict) - float: 评估分析质量 score 0.0 # 基于检测数量评分 total_sources report[execution_summary][total_sources_detected] if total_sources 100: score 30 elif total_sources 50: score 20 elif total_sources 10: score 10 # 基于候选星系数量评分 candidate_galaxies report[execution_summary][candidate_galaxies] if candidate_galaxies 20: score 40 elif candidate_galaxies 10: score 30 elif candidate_galaxies 5: score 20 elif candidate_galaxies 0: score 10 # 基于数据质量评分 data_quality report[execution_summary][data_quality] if data_quality excellent: score 30 elif data_quality good: score 20 elif data_quality fair: score 10 return min(score, 100) # 使用示例 async def main(): # 初始化流水线 pipeline JWSTGalaxyPipeline() await pipeline.initialize() # 定义分析目标 - 以著名的哈勃深场区域为例 target_config { name: Hubble Deep Field North, ra: 12h36m49.4s, dec: 62d12m58s, radius: 0.05, # 度 instrument: NIRCam, filters: [F115W, F150W, F200W] # JWST近红外波段 } try: # 运行分析 report await pipeline.run_analysis(target_config) # 保存结果 with open(fgalaxy_report_{datetime.now().strftime(%Y%m%d_%H%M%S)}.json, w) as f: json.dump(report, f, indent2) # 打印摘要 summary report[execution_summary] print(f\n分析完成摘要:) print(f- 检测到总源数: {summary[total_sources_detected]}) print(f- 候选星系数: {summary[candidate_galaxies]}) print(f- 数据质量: {summary[data_quality]}) # 显示前5个候选星系 print(f\n前5个候选星系:) for i, galaxy in enumerate(report[candidate_galaxies][:5]): print(f{i1}. RA: {galaxy.get(ra, N/A)}, Dec: {galaxy.get(dec, N/A)}, f类型: {galaxy.get(subtype, 未知)}, 置信度: {galaxy.get(confidence, 0):.2f}) except Exception as e: print(f分析过程中出错: {e}) if __name__ __main__: asyncio.run(main())6.2 结果可视化与交互探索生成分析结果后提供交互式可视化界面# results_visualizer.py import matplotlib.pyplot as plt import numpy as np from astropy.wcs import WCS import astropy.io.fits as fits class GalaxyResultsVisualizer: def __init__(self, results_file: str): with open(results_file, r) as f: self.results json.load(f) def create_detection_map(self, fits_image_path: str): 创建检测结果叠加图 with fits.open(fits_image_path) as hdul: image_data hdul[1].data wcs WCS(hdul[1].header) fig plt.figure(figsize(12, 10)) ax fig.add_subplot(111, projectionwcs) # 显示原始图像 ax.imshow(image_data, cmapgray, originlower, vminnp.percentile(image_data, 5), vmaxnp.percentile(image_data, 95)) # 标记检测到的源 galaxies self.results[candidate_galaxies] ras [galaxy[ra] for galaxy in galaxies if ra in galaxy] decs [galaxy[dec] for galaxy in galaxies if dec in galaxy] if ras and decs: # 将坐标转换为像素位置简化处理实际需要精确转换 ax.scatter(ras, decs, s50, facecolorsnone, edgecolorsr, linewidths2, transformax.get_transform(world), labelf候选星系 ({len(galaxies)}个)) ax.set_xlabel(RA) ax.set_ylabel(Dec) ax.set_title(fJWST候选星系检测 - {self.results[target_info][name]}) ax.legend() plt.tight_layout() return fig def create_classification_pie_chart(self): 创建星系类型分布饼图 galaxies self.results[candidate_galaxies] types [galaxy.get(subtype, 未知) for galaxy in galaxies] type_counts {} for t in types: type_counts[t] type_counts.get(t, 0) 1 fig, ax plt.subplots(figsize(8, 8)) wedges, texts, autotexts ax.pie(type_counts.values(), labelstype_counts.keys(), autopct%1.1f%%, startangle90) ax.set_title(候选星系类型分布) return fig def generate_interactive_report(self): 生成交互式HTML报告 import plotly.express as px import plotly.graph_objects as go from plotly.subplots import make_subplots galaxies self.results[candidate_galaxies] if not galaxies: return p未检测到候选星系/p # 创建散点图亮度 vs 大小 fluxes [g.get(flux, 0) for g in galaxies] sizes [g.get(size, 0) for g in galaxies] types [g.get(subtype, 未知) for g in galaxies] confidences [g.get(confidence, 0) for g in galaxies] fig px.scatter(xfluxes, ysizes, colortypes, sizeconfidences, title候选星系分布亮度 vs 大小, labels{x: 流量, y: 角大小(弧秒)}) return fig.to_html(full_htmlFalse) # 使用示例 def visualize_results(results_file, fits_image_path): visualizer GalaxyResultsVisualizer(results_file) # 创建检测图 detection_fig visualizer.create_detection_map(fits_image_path) detection_fig.savefig(detection_map.png, dpi300, bbox_inchestight) # 创建分类饼图 pie_fig visualizer.create_classification_pie_chart() pie_fig.savefig(classification_pie.png, dpi300, bbox_inchestight) # 生成交互式报告 html_report visualizer.generate_interactive_report() with open(interactive_report.html, w) as f: f.write(html_report) print(可视化结果已保存)7. 性能优化与大规模处理策略7.1 并行处理与资源管理当处理大面积天区或多个观测时需要优化性能# parallel_processor.py import concurrent.futures from typing import List, Dict, Any import asyncio class JWSTParallelProcessor: def __init__(self, max_workers: int 4): self.max_workers max_workers self.executor concurrent.futures.ThreadPoolExecutor(max_workersmax_workers) async def process_multiple_regions(self, regions: List[Dict]) - Dict[str, Any]: 并行处理多个天区 tasks [] for region in regions: pipeline JWSTGalaxyPipeline() await pipeline.initialize() # 创建处理任务 task asyncio.create_task(self.process_single_region(pipeline, region)) tasks.append(task) # 等待所有任务完成 results await asyncio.gather(*tasks, return_exceptionsTrue) # 汇总结果 combined_results {} for i, result in enumerate(results): region_name regions[i][name] if isinstance(result, Exception): combined_results[region_name] {error: str(result)} else: combined_results[region_name] result return combined_results async def process_single_region(self, pipeline: JWSTGalaxyPipeline, region: Dict): 处理单个天区 try: result await pipeline.run_analysis(region) return result except Exception as e: return {error: str(e)} def optimize_memory_usage(self, image_data: np.ndarray, strategy: str adaptive): 优化内存使用策略 if strategy adaptive: # 自适应分块处理 chunk_size self.calculate_optimal_chunk_size(image_data.shape) processed_chunks [] for i in range(0, image_data.shape[0], chunk_size): for j in range(0, image_data.shape[1], chunk_size): chunk image_data[i:ichunk_size, j:jchunk_size] processed_chunk self.process_image_chunk(chunk) processed_chunks.append((i, j, processed_chunk)) return self.reconstruct_from_chunks(processed_chunks, image_data.shape) elif strategy compression: # 使用有损压缩减少内存占用 import zlib compressed zlib.compress(image_data.tobytes(), level1) return compressed def calculate_optimal_chunk_size(self, shape: tuple) - int: 计算最优分块大小 max_memory 2 * 1024 * 1024 * 1024 # 2GB pixel_size 8 # 假设float64 optimal_chunk int(np.sqrt(max_memory / pixel_size)) return min(optimal_chunk, shape[0] // 4, shape[1] // 4)7.2 缓存策略与增量处理对于大规模巡天数据实现智能缓存# cache_manager.py import pickle import hashlib import os from datetime import datetime, timedelta class AnalysisCacheManager: def __init__(self, cache_dir: str ./analysis_cache, ttl_hours: int 24): self.cache_dir cache_dir self.ttl timedelta(hoursttl_hours) os.makedirs(cache_dir, exist_okTrue) def get_cache_key(self, region: Dict, analysis_type: str) - str: 生成缓存键 key_data { region: region, analysis_type: analysis_type, timestamp: datetime.now().strftime(%Y%m%d) } key_str json.dumps(key_data, sort_keysTrue) return hashlib.md5(key_str.encode()).hexdigest() def get_cached_result(self, cache_key: str): 获取缓存结果 cache_file