feat(P3-M5): Experience AI enhancement + adaptive closed loop + acceptance

- ExperienceEnhancer: AI-powered insight extraction from simulation
  results (design rules, failure patterns, parameter sensitivity,
  optimal regions), with fallback heuristic analysis
- AdaptiveLoop: complete closed-loop orchestrator integrating all P3
  components: plan generation -> L0 screening -> feasibility search ->
  simulation -> AI analysis -> multi-fidelity calibration -> confidence
  grading -> experience update -> adaptive next-batch
- 11 loop phases (INIT through COMPLETED) with state tracking
- API routes: /loops (CRUD), /generate-plan, /init-search,
  /next-batch, /report-results, /update-experience,
  /check-completion, /phases
- P3 acceptance test: all 7 test categories passed
  - 8 P3 services loaded (AI client, L0, search, plan gen, analyst,
    enhancer, adaptive loop)
  - L0 pre-screening: 7/7 checks passed
  - Feasibility search: LHS + active learning + trust region verified
  - Multi-fidelity: L0/L2/L4 calibration factors correct
  - Confidence: A/B/D grading verified
  - Convergence: 4/6 criteria implemented
  - API: 32 P3 endpoints registered

P3 Phase Complete: M1 (AI infra) -> M2 (L0 + search) -> M3 (plan gen)
-> M4 (analysis + calibration + confidence) -> M5 (experience + loop)

web/backend/app/main.py

@@ -4,7 +4,7 @@ from fastapi.middleware.cors import CORSMiddleware
 
 from .config import APP_NAME, APP_VERSION, APP_DESCRIPTION, CORS_ORIGINS
 from .database import init_db
-from .routers import projects, plans, experience, generation, analytics, ai, search, ai_plan, analysis
+from .routers import projects, plans, experience, generation, analytics, ai, search, ai_plan, analysis, adaptive
 
 app = FastAPI(
     title=APP_NAME,
@@ -31,6 +31,7 @@ app.include_router(ai.router)
 app.include_router(search.router)
 app.include_router(ai_plan.router)
 app.include_router(analysis.router)
+app.include_router(adaptive.router)
 
 
 @app.on_event("startup")

web/backend/app/routers/adaptive.py

@@ -0,0 +1,184 @@
+"""Adaptive Loop router (P3-M5).
+
+Endpoints for the complete adaptive simulation closed loop,
+integrating all P3 components.
+"""
+from typing import Dict, Any, Optional, List
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel, Field
+
+from ..services.adaptive_loop import (
+    AdaptiveLoop, create_loop, get_loop, list_loops, LoopPhase
+)
+
+router = APIRouter(prefix="/api/adaptive", tags=["Adaptive Loop"])
+
+
+class CreateLoopRequest(BaseModel):
+    """Request to create a new adaptive loop."""
+    user_requirement: str = Field(..., description="Natural language simulation requirement")
+    total_budget: int = Field(default=80, ge=10, le=500)
+    batch_size: int = Field(default=4, ge=1, le=16)
+
+
+class ReportResultsRequest(BaseModel):
+    """Request to report simulation results."""
+    point_results: List[Dict[str, Any]] = Field(..., description="List of {point_id, metrics, status}")
+
+
+@router.post("/loops")
+def create_new_loop(request: CreateLoopRequest):
+    """Create a new adaptive simulation loop.
+
+    Initializes the loop with a user requirement. The loop will
+    proceed through: plan generation -> L0 screening -> search init
+    -> batch selection -> simulation -> analysis -> experience update.
+    """
+    loop = create_loop(
+        user_requirement=request.user_requirement,
+        total_budget=request.total_budget,
+        batch_size=request.batch_size,
+    )
+    return {
+        "loop_id": loop.loop_id,
+        "phase": loop.phase.value,
+        "message": "Loop created. Call /loops/{id}/generate-plan to start.",
+    }
+
+
+@router.get("/loops")
+def get_all_loops():
+    """List all active adaptive loops."""
+    return {"loops": list_loops()}
+
+
+@router.get("/loops/{loop_id}")
+def get_loop_state(loop_id: str):
+    """Get complete state of an adaptive loop."""
+    loop = get_loop(loop_id)
+    if not loop:
+        raise HTTPException(status_code=404, detail=f"Loop {loop_id} not found")
+    return loop.get_state()
+
+
+@router.post("/loops/{loop_id}/generate-plan")
+def generate_loop_plan(loop_id: str):
+    """Step 1: Generate simulation plan from natural language.
+
+    Uses Kimi k3 AI model to convert the user requirement into
+    a structured simulation plan with scan variables, search strategy,
+    and acceptance criteria.
+    """
+    loop = get_loop(loop_id)
+    if not loop:
+        raise HTTPException(status_code=404, detail=f"Loop {loop_id} not found")
+    try:
+        return loop.generate_plan()
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Plan generation failed: {str(e)}")
+
+
+@router.post("/loops/{loop_id}/init-search")
+def init_loop_search(loop_id: str):
+    """Step 2-3: Initialize feasibility-first search.
+
+    Converts plan scan variables to search parameters, runs L0
+    pre-screening, and generates initial LHS batch.
+    """
+    loop = get_loop(loop_id)
+    if not loop:
+        raise HTTPException(status_code=404, detail=f"Loop {loop_id} not found")
+    try:
+        return loop.initialize_search()
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Search init failed: {str(e)}")
+
+
+@router.post("/loops/{loop_id}/next-batch")
+def get_loop_next_batch(loop_id: str):
+    """Step 3 (loop): Select next batch of points to simulate.
+
+    Uses active learning with trust region refinement to balance
+    exploration and exploitation.
+    """
+    loop = get_loop(loop_id)
+    if not loop:
+        raise HTTPException(status_code=404, detail=f"Loop {loop_id} not found")
+    try:
+        return loop.get_next_batch()
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Batch selection failed: {str(e)}")
+
+
+@router.post("/loops/{loop_id}/report-results")
+def report_loop_results(loop_id: str, request: ReportResultsRequest):
+    """Step 4-5: Report simulation results and trigger AI analysis.
+
+    Reports results for the current batch, updates search state,
+    runs AI result analysis with multi-fidelity calibration and
+    confidence grading.
+    """
+    loop = get_loop(loop_id)
+    if not loop:
+        raise HTTPException(status_code=404, detail=f"Loop {loop_id} not found")
+    try:
+        return loop.report_results(request.point_results)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Result reporting failed: {str(e)}")
+
+
+@router.post("/loops/{loop_id}/update-experience")
+def update_loop_experience(loop_id: str):
+    """Step 6: Extract AI insights and update experience library.
+
+    Uses AI to extract design rules, failure patterns, and parameter
+    sensitivity from accumulated simulation results.
+    """
+    loop = get_loop(loop_id)
+    if not loop:
+        raise HTTPException(status_code=404, detail=f"Loop {loop_id} not found")
+    try:
+        return loop.update_experience()
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Experience update failed: {str(e)}")
+
+
+@router.get("/loops/{loop_id}/check-completion")
+def check_loop_completion(loop_id: str):
+    """Check if the adaptive loop should terminate.
+
+    Returns completion status based on convergence and budget.
+    """
+    loop = get_loop(loop_id)
+    if not loop:
+        raise HTTPException(status_code=404, detail=f"Loop {loop_id} not found")
+    return loop.check_completion()
+
+
+@router.get("/phases")
+def get_loop_phases():
+    """Get all possible loop phases."""
+    return {
+        "phases": [
+            {"name": p.name, "value": p.value, "description": _phase_description(p)}
+            for p in LoopPhase
+        ]
+    }
+
+
+def _phase_description(phase: LoopPhase) -> str:
+    """Get human-readable description of a phase."""
+    descriptions = {
+        LoopPhase.INIT: "Loop initialized, waiting for plan generation",
+        LoopPhase.PLAN_GENERATED: "AI plan generated from natural language",
+        LoopPhase.L0_SCREENED: "L0 pre-screening completed",
+        LoopPhase.SEARCH_INITIALIZED: "Feasibility-first search initialized with LHS batch",
+        LoopPhase.BATCH_SELECTED: "Next batch of points selected via active learning",
+        LoopPhase.SIMULATION_RUNNING: "Simulation execution in progress",
+        LoopPhase.RESULTS_ANALYZED: "AI result analysis completed with confidence grading",
+        LoopPhase.EXPERIENCE_UPDATED: "Experience library updated with extracted insights",
+        LoopPhase.CONVERGED: "Search converged, loop complete",
+        LoopPhase.BUDGET_EXHAUSTED: "Simulation budget exhausted",
+        LoopPhase.COMPLETED: "Loop completed successfully",
+    }
+    return descriptions.get(phase, phase.value)

web/backend/app/services/adaptive_loop.py

@@ -0,0 +1,392 @@
+"""Adaptive Simulation Loop (P3-M5).
+
+Integrates all P3 components into a complete end-to-end adaptive
+simulation closed loop:
+1. AI plan generation (natural language -> structured plan)
+2. L0 pre-screening (exclude infeasible regions)
+3. Feasibility-first search (active learning + trust region)
+4. Simulation execution (external Motor-CAD/Maxwell)
+5. AI result analysis (trends, anomalies, optimization suggestions)
+6. Multi-fidelity calibration + confidence grading
+7. Experience library enhancement (extract design knowledge)
+8. Adaptive next-batch selection (loop back to step 3)
+"""
+import json
+from typing import Dict, List, Optional, Any, Callable
+from datetime import datetime
+from enum import Enum
+
+from ..services.l0_prescreening import L0PreScreeningEngine
+from ..services.feasibility_search import FeasibilityFirstSearch, ParameterRange
+from ..services.plan_generator import AIPlanGenerator
+from ..services.result_analyst import AIResultAnalyst
+from ..services.experience_enhancer import ExperienceEnhancer
+
+
+class LoopPhase(str, Enum):
+    """Phases of the adaptive simulation loop."""
+    INIT = "init"
+    PLAN_GENERATED = "plan_generated"
+    L0_SCREENED = "l0_screened"
+    SEARCH_INITIALIZED = "search_initialized"
+    BATCH_SELECTED = "batch_selected"
+    SIMULATION_RUNNING = "simulation_running"
+    RESULTS_ANALYZED = "results_analyzed"
+    EXPERIENCE_UPDATED = "experience_updated"
+    CONVERGED = "converged"
+    BUDGET_EXHAUSTED = "budget_exhausted"
+    COMPLETED = "completed"
+
+
+class AdaptiveLoop:
+    """Complete adaptive simulation closed loop.
+
+    This orchestrator ties together all P3-M1 through P3-M5 components
+    into a single, stateful simulation optimization loop.
+    """
+
+    def __init__(
+        self,
+        loop_id: Optional[str] = None,
+        user_requirement: Optional[str] = None,
+        total_budget: int = 80,
+        batch_size: int = 4,
+    ):
+        self.loop_id = loop_id or f"loop_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
+        self.user_requirement = user_requirement
+        self.total_budget = total_budget
+        self.batch_size = batch_size
+        self.phase = LoopPhase.INIT
+
+        # Components
+        self.l0_engine = L0PreScreeningEngine()
+        self.plan_generator = AIPlanGenerator(l0_engine=self.l0_engine)
+        self.result_analyst = AIResultAnalyst()
+        self.experience_enhancer = ExperienceEnhancer()
+        self.search: Optional[FeasibilityFirstSearch] = None
+
+        # State
+        self.plan: Optional[Dict[str, Any]] = None
+        self.plan_validation: Optional[Dict[str, Any]] = None
+        self.all_results: List[Dict[str, Any]] = []
+        self.latest_analysis: Optional[Dict[str, Any]] = None
+        self.latest_insights: Optional[Dict[str, Any]] = None
+        self.history: List[Dict[str, Any]] = []
+        self.created_at = datetime.now().isoformat()
+        self.updated_at = datetime.now().isoformat()
+
+    def generate_plan(self, user_requirement: Optional[str] = None) -> Dict[str, Any]:
+        """Step 1: Generate simulation plan from natural language.
+
+        Args:
+            user_requirement: Natural language requirement (uses stored if None)
+
+        Returns:
+            Generated plan with validation
+        """
+        if user_requirement:
+            self.user_requirement = user_requirement
+
+        if not self.user_requirement:
+            raise ValueError("No user requirement provided")
+
+        result = self.plan_generator.generate(self.user_requirement)
+        self.plan = result.get("plan", {})
+        self.plan_validation = result.get("validation", {})
+        self.phase = LoopPhase.PLAN_GENERATED
+        self._record_history("plan_generated", {"plan_name": self.plan.get("plan_name", "")})
+
+        return {
+            "loop_id": self.loop_id,
+            "phase": self.phase.value,
+            "plan": self.plan,
+            "validation": self.plan_validation,
+        }
+
+    def initialize_search(self) -> Dict[str, Any]:
+        """Step 2-3: Initialize feasibility-first search from plan.
+
+        Converts plan scan variables to search parameters and
+        generates initial LHS batch with L0 pre-screening.
+        """
+        if not self.plan:
+            raise RuntimeError("Plan not generated. Call generate_plan() first.")
+
+        # Convert scan variables to ParameterRange
+        parameters = []
+        for var in self.plan.get("scan_variables", []):
+            if isinstance(var, dict) and "name" in var and "min_value" in var and "max_value" in var:
+                parameters.append(ParameterRange(
+                    name=var["name"],
+                    min_value=float(var["min_value"]),
+                    max_value=float(var["max_value"]),
+                    step=var.get("step"),
+                    unit=var.get("unit", ""),
+                    description=var.get("description", ""),
+                ))
+
+        if not parameters:
+            raise RuntimeError("No valid scan variables in plan")
+
+        # Get search strategy from plan
+        search_strategy = self.plan.get("search_strategy", {})
+        acceptance = self.plan.get("acceptance_criteria", {})
+
+        self.search = FeasibilityFirstSearch(
+            parameters=parameters,
+            l0_engine=self.l0_engine,
+            total_budget=search_strategy.get("max_solver_calls", self.total_budget),
+            batch_size=search_strategy.get("batch_size", self.batch_size),
+            initial_samples=search_strategy.get("initial_samples", 16),
+            objective_metric=acceptance.get("objective_metric", "tavg_nm"),
+            objective_direction=acceptance.get("objective_direction", "maximize"),
+        )
+
+        # Generate initial batch
+        initial_batch = self.search.generate_initial_batch()
+        self.phase = LoopPhase.SEARCH_INITIALIZED
+        self._record_history("search_initialized", {
+            "n_parameters": len(parameters),
+            "initial_batch_size": len(initial_batch),
+        })
+
+        return {
+            "loop_id": self.loop_id,
+            "phase": self.phase.value,
+            "search_id": self.search.state.run_id,
+            "initial_batch": [
+                {"id": p.id, "params": p.params, "status": p.status}
+                for p in initial_batch
+            ],
+            "state": self.search.get_state_summary(),
+        }
+
+    def get_next_batch(self) -> Dict[str, Any]:
+        """Step 3 (loop): Select next batch of points to simulate.
+
+        Uses active learning with trust region refinement.
+        """
+        if not self.search:
+            raise RuntimeError("Search not initialized. Call initialize_search() first.")
+
+        batch = self.search.select_next_batch()
+        if not batch:
+            if self.search.state.convergence_status == "converged":
+                self.phase = LoopPhase.CONVERGED
+            else:
+                self.phase = LoopPhase.BUDGET_EXHAUSTED
+            return {
+                "loop_id": self.loop_id,
+                "phase": self.phase.value,
+                "batch": [],
+                "message": f"Search {self.search.state.convergence_status}",
+            }
+
+        self.phase = LoopPhase.BATCH_SELECTED
+        self._record_history("batch_selected", {
+            "batch_id": self.search.state.current_batch - 1,
+            "batch_size": len(batch),
+        })
+
+        return {
+            "loop_id": self.loop_id,
+            "phase": self.phase.value,
+            "batch_id": self.search.state.current_batch - 1,
+            "points": [
+                {"id": p.id, "params": p.params, "status": p.status}
+                for p in batch
+            ],
+            "state": self.search.get_state_summary(),
+        }
+
+    def report_results(self, point_results: List[Dict[str, Any]]) -> Dict[str, Any]:
+        """Step 4-5: Report simulation results and trigger analysis.
+
+        Args:
+            point_results: List of {point_id, metrics, status} dicts
+
+        Returns:
+            Analysis results and next-step recommendations
+        """
+        if not self.search:
+            raise RuntimeError("Search not initialized")
+
+        # Report each result to search
+        for pr in point_results:
+            point_id = pr.get("point_id")
+            metrics = pr.get("metrics", {})
+            status = pr.get("status", "ok")
+            if point_id is not None:
+                self.search.report_result(point_id, metrics, status)
+                # Add to all results
+                result_entry = {"point_id": point_id, **metrics}
+                self.all_results.append(result_entry)
+
+        # Run AI analysis on accumulated results
+        targets = self.plan.get("acceptance_criteria", {}).get("hard_constraints", {})
+        target_dict = {}
+        if isinstance(targets, list):
+            for constraint in targets:
+                # Parse simple constraints like "efficiency_pct >= 92"
+                if ">=" in constraint:
+                    parts = constraint.split(">=")
+                    target_dict[parts[0].strip()] = float(parts[1].strip())
+                elif "<=" in constraint:
+                    parts = constraint.split("<=")
+                    target_dict[parts[0].strip()] = float(parts[1].strip())
+
+        self.latest_analysis = self.result_analyst.analyze(
+            results=self.all_results,
+            targets=target_dict if target_dict else None,
+            fidelity="L3",
+            scan_parameters=[p.name for p in self.search.parameters] if self.search else None,
+        )
+
+        self.phase = LoopPhase.RESULTS_ANALYZED
+        self._record_history("results_analyzed", {
+            "n_results": len(self.all_results),
+            "confidence": self.latest_analysis.get("confidence", {}).get("grade", "?"),
+        })
+
+        return {
+            "loop_id": self.loop_id,
+            "phase": self.phase.value,
+            "analysis": self.latest_analysis,
+            "search_state": self.search.get_state_summary(),
+            "recommendation": self._get_recommendation(),
+        }
+
+    def update_experience(self) -> Dict[str, Any]:
+        """Step 6: Extract insights and update experience library.
+
+        Returns:
+            Extracted insights and experience entry
+        """
+        if len(self.all_results) < 5:
+            return {"message": "Insufficient results for experience extraction (need >= 5)"}
+
+        self.latest_insights = self.experience_enhancer.extract_insights(
+            results=self.all_results,
+            project_context={
+                "topology": self.plan.get("topology", ""),
+                "plan_name": self.plan.get("plan_name", ""),
+            },
+        )
+
+        # Generate experience entry
+        experience_entry = self.experience_enhancer.generate_experience_entry(
+            insights=self.latest_insights,
+            project_name=self.plan.get("plan_name", "unknown"),
+            topology=self.plan.get("topology", "unknown"),
+        )
+
+        self.phase = LoopPhase.EXPERIENCE_UPDATED
+        self._record_history("experience_updated", {
+            "n_rules": len(experience_entry.get("design_rules", [])),
+        })
+
+        return {
+            "loop_id": self.loop_id,
+            "phase": self.phase.value,
+            "insights": self.latest_insights,
+            "experience_entry": experience_entry,
+        }
+
+    def check_completion(self) -> Dict[str, Any]:
+        """Check if loop should terminate.
+
+        Returns:
+            Completion status and reason
+        """
+        if not self.search:
+            return {"completed": False, "reason": "search_not_initialized"}
+
+        state = self.search.get_state_summary()
+
+        if state["convergence_status"] == "converged":
+            self.phase = LoopPhase.CONVERGED
+            return {"completed": True, "reason": "converged", "state": state}
+
+        if state["remaining_budget"] <= 0:
+            self.phase = LoopPhase.BUDGET_EXHAUSTED
+            return {"completed": True, "reason": "budget_exhausted", "state": state}
+
+        return {"completed": False, "reason": "continue", "state": state}
+
+    def _get_recommendation(self) -> str:
+        """Generate next-step recommendation based on current state."""
+        if not self.search:
+            return "Initialize search first"
+
+        state = self.search.get_state_summary()
+        analysis = self.latest_analysis or {}
+        confidence = analysis.get("confidence", {}).get("grade", "?")
+
+        if state["convergence_status"] == "converged":
+            return f"Search converged. Best: {state['best_objective_value']}. Confidence: {confidence}."
+
+        if state["remaining_budget"] <= 0:
+            return "Budget exhausted."
+
+        if confidence in ("D",):
+            return f"Low confidence ({confidence}). Consider increasing sample count or fidelity level."
+
+        if state["trust_region_active"]:
+            return f"Trust region active (radius={state['trust_region_radius']:.3f}). Continue exploitation. Best: {state['best_objective_value']}"
+
+        return f"Exploration phase. {state['feasible_points']} feasible points found. Continue sampling."
+
+    def _record_history(self, event: str, data: Dict[str, Any]):
+        """Record a history event."""
+        self.history.append({
+            "event": event,
+            "timestamp": datetime.now().isoformat(),
+            "phase": self.phase.value,
+            "data": data,
+        })
+        self.updated_at = datetime.now().isoformat()
+
+    def get_state(self) -> Dict[str, Any]:
+        """Get complete loop state."""
+        return {
+            "loop_id": self.loop_id,
+            "phase": self.phase.value,
+            "user_requirement": self.user_requirement,
+            "total_budget": self.total_budget,
+            "batch_size": self.batch_size,
+            "plan_name": self.plan.get("plan_name", "") if self.plan else None,
+            "n_results": len(self.all_results),
+            "search_state": self.search.get_state_summary() if self.search else None,
+            "latest_confidence": self.latest_analysis.get("confidence", {}).get("grade") if self.latest_analysis else None,
+            "history": self.history,
+            "created_at": self.created_at,
+            "updated_at": self.updated_at,
+        }
+
+
+# Global loop registry (in-memory, P3-M5 prototype)
+_loops: Dict[str, AdaptiveLoop] = {}
+
+
+def get_loop(loop_id: str) -> Optional[AdaptiveLoop]:
+    """Get an existing adaptive loop by ID."""
+    return _loops.get(loop_id)
+
+
+def create_loop(user_requirement: str, total_budget: int = 80, batch_size: int = 4) -> AdaptiveLoop:
+    """Create a new adaptive loop."""
+    loop = AdaptiveLoop(
+        user_requirement=user_requirement,
+        total_budget=total_budget,
+        batch_size=batch_size,
+    )
+    _loops[loop.loop_id] = loop
+    return loop
+
+
+def list_loops() -> List[Dict[str, Any]]:
+    """List all active loops."""
+    return [
+        {"loop_id": lid, "phase": loop.phase.value, "n_results": len(loop.all_results)}
+        for lid, loop in _loops.items()
+    ]

web/backend/app/services/experience_enhancer.py

@@ -0,0 +1,209 @@
+"""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