"""Experience Library AI Enhancer (P3-M5). Uses AI to extract design knowledge, rules, and insights from historical simulation results, enhancing the experience library. """ import json from pathlib import Path from typing import Dict, List, Optional, Any from datetime import datetime from ..config import PROMPTS_DIR from ..services.ai_client import get_kimi_client class ExperienceEnhancer: """AI-powered experience library enhancer. Extracts: - Design rules and heuristics from successful cases - Failure patterns and constraint boundaries - Parameter sensitivity rankings - Optimal design regions - Comparative insights across topologies """ def __init__(self): self.ai_client = get_kimi_client() self._prompt_template = None def _load_prompt(self) -> str: if self._prompt_template is None: prompt_path = PROMPTS_DIR / "experience" / "extract.txt" if prompt_path.exists(): with open(prompt_path, "r", encoding="utf-8") as f: self._prompt_template = f.read() else: self._prompt_template = self._default_prompt() return self._prompt_template def _default_prompt(self) -> str: return """你是轴向磁通电机设计知识提取专家。从仿真结果数据中提取设计规则、失败模式、参数敏感性和最优设计区域。输出JSON格式。""" def extract_insights( self, results: List[Dict[str, Any]], project_context: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """Extract design insights from a batch of simulation results. Args: results: List of simulation result dicts project_context: Optional project context Returns: Structured insights including rules, patterns, and recommendations """ if not results: return {"error": "No results to analyze"} # Prepare condensed data condensed = self._condense_results(results) if project_context: condensed["project_context"] = project_context user_message = f"从以下轴向磁通电机仿真结果中提取设计知识：\n{json.dumps(condensed, ensure_ascii=False, indent=2)}" if not self.ai_client.is_configured: return self._fallback_insights(results) try: result = self.ai_client.chat_json( messages=[{"role": "user", "content": user_message}], system_prompt=self._load_prompt(), max_tokens=3000, ) insights = result.get("parsed_json", {}) if not insights: raw = result.get("raw_content", result.get("content", "")) try: start = raw.find("{") end = raw.rfind("}") + 1 if start >= 0 and end > start: insights = json.loads(raw[start:end]) except (json.JSONDecodeError, Exception): insights = {"summary": raw[:500]} insights["extracted_at"] = datetime.now().isoformat() insights["n_results_analyzed"] = len(results) insights["usage"] = result.get("usage", {}) return insights except Exception as e: return {"error": f"AI insight extraction failed: {str(e)}", **self._fallback_insights(results)} def _condense_results(self, results: List[Dict[str, Any]]) -> Dict[str, Any]: """Condense results for AI processing (avoid token overflow).""" # Extract key metrics metrics = [] for r in results: m = {} for key in ["tavg_nm", "efficiency_pct", "total_losses_w", "winding_temp_c", "airgap_mm", "magnet_thickness_mm", "current_a", "speed_rpm", "outer_diameter_mm", "inner_diameter_mm", "feasible"]: if key in r: m[key] = r[key] if m: metrics.append(m) # Sort by torque and take top/bottom 10 + random 10 sorted_by_torque = sorted(metrics, key=lambda x: x.get("tavg_nm", 0), reverse=True) top_10 = sorted_by_torque[:10] bottom_10 = sorted_by_torque[-10:] if len(sorted_by_torque) > 20 else [] # Calculate basic statistics all_torque = [m.get("tavg_nm", 0) for m in metrics if "tavg_nm" in m] all_efficiency = [m.get("efficiency_pct", 0) for m in metrics if "efficiency_pct" in m] stats = {} if all_torque: stats["torque"] = { "min": min(all_torque), "max": max(all_torque), "mean": sum(all_torque) / len(all_torque), } if all_efficiency: stats["efficiency"] = { "min": min(all_efficiency), "max": max(all_efficiency), "mean": sum(all_efficiency) / len(all_efficiency), } return { "total_points": len(results), "statistics": stats, "top_performers": top_10, "bottom_performers": bottom_10, } def _fallback_insights(self, results: List[Dict[str, Any]]) -> Dict[str, Any]: """Generate basic insights without AI.""" feasible = [r for r in results if r.get("feasible", True)] infeasible = [r for r in results if not r.get("feasible", True)] insights = { "summary": f"Analyzed {len(results)} points ({len(feasible)} feasible, {len(infeasible)} infeasible).", "feasibility_rate": len(feasible) / len(results) if results else 0, "design_rules": [], "failure_patterns": [], "extracted_at": datetime.now().isoformat(), "n_results_analyzed": len(results), "ai_used": False, } # Simple correlation analysis if len(feasible) >= 5: # Find parameters that correlate with high torque for param in ["airgap_mm", "magnet_thickness_mm", "current_a"]: values = [(r.get(param, 0), r.get("tavg_nm", 0)) for r in feasible if param in r] if len(values) >= 5: values.sort(key=lambda x: x[1], reverse=True) top_params = [v[0] for v in values[:len(values)//3]] avg_top = sum(top_params) / len(top_params) if top_params else 0 all_avg = sum(v[0] for v in values) / len(values) if abs(avg_top - all_avg) > 0.1 * abs(all_avg) if all_avg else False: direction = "higher" if avg_top > all_avg else "lower" insights["design_rules"].append( f"High torque designs tend to have {direction} {param} (avg top={avg_top:.2f} vs all={all_avg:.2f})" ) return insights def generate_experience_entry( self, insights: Dict[str, Any], project_name: str, topology: str, ) -> Dict[str, Any]: """Generate a structured experience library entry from insights. Args: insights: Extracted insights project_name: Project name topology: Motor topology Returns: Structured experience entry ready for database storage """ return { "title": f"{project_name} - {topology} Design Insights", "topology": topology, "project": project_name, "summary": insights.get("summary", ""), "design_rules": insights.get("design_rules", []), "failure_patterns": insights.get("failure_patterns", []), "parameter_sensitivity": insights.get("parameter_sensitivity", {}), "optimal_regions": insights.get("optimal_regions", []), "confidence": insights.get("confidence_grade", "C"), "source": "ai_extracted", "created_at": datetime.now().isoformat(), "tags": [topology, "ai-insights", project_name], } # Global singleton _enhancer: Optional[ExperienceEnhancer] = None def get_experience_enhancer() -> ExperienceEnhancer: """Get or create global ExperienceEnhancer singleton.""" global _enhancer if _enhancer is None: _enhancer = ExperienceEnhancer() return _enhancer