"""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, KIMI_MAX_TOKENS 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 """\u4f60\u662f\u8f74\u5411\u78c1\u901a\u7535\u673a\u8bbe\u8ba1\u77e5\u8bc6\u63d0\u53d6\u4e13\u5bb6\u3002\u4ece\u4eff\u771f\u7ed3\u679c\u6570\u636e\u4e2d\u63d0\u53d6\u8bbe\u8ba1\u89c4\u5219\u3001\u5931\u8d25\u6a21\u5f0f\u3001\u53c2\u6570\u654f\u611f\u6027\u548c\u6700\u4f18\u8bbe\u8ba1\u533a\u57df\u3002\u8f93\u51faJSON\u683c\u5f0f\u3002""" 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"\u4ece\u4ee5\u4e0b\u8f74\u5411\u78c1\u901a\u7535\u673a\u4eff\u771f\u7ed3\u679c\u4e2d\u63d0\u53d6\u8bbe\u8ba1\u77e5\u8bc6\uff1a\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=KIMI_MAX_TOKENS, ) 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. Input params may arrive under Motor-CAD names # (Airgap, RMSCurrent); translate them to the BC-style keys below via # the shared single-source map so sensitivity analysis sees inputs. from src.afmcore.l0.prescreening import MOTORCAD_TO_L0 keys = ["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"] metrics = [] for r in results: r_norm = dict(r) for mc_name, l0_name in MOTORCAD_TO_L0.items(): if mc_name in r_norm and l0_name not in r_norm: r_norm[l0_name] = r_norm[mc_name] m = {} for key in keys: if key in r_norm: m[key] = r_norm[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