"""AI Plan Generator (P3-M3). Converts natural language requirements into structured simulation plans using Kimi k3 model, with L0 pre-screening validation. Output is converted to unified plan schema v2.0 compatible with src/plan_schema.py (fixed_params + variables with value lists). """ import json import math from pathlib import Path from typing import Dict, List, Optional, Any, Tuple from ..config import PROMPTS_DIR, KIMI_MAX_TOKENS from ..services.ai_client import get_kimi_client from ..services.l0_prescreening import L0PreScreeningEngine from ..services.rule_engine import SCAN_PARAMETERS, get_parameter, recommend_range, BoundaryConditions from ..services.fixed_params_template import build_default_fixed_params, get_fixed_param_template, CATEGORY_CN # Single-source strategy registry (mirrors src/plan_schema.py validation) so the # AI-generated method string is normalized against the platform registry instead # of being trusted verbatim. Avoids a second definition of the strategy whitelist. from src.afmcore.strategies import ( is_registered as _strategy_registered, normalize_method as _strategy_normalize, ) # Mapping from common AI-output variable names to exact Motor-CAD variable names _VARIABLE_NAME_MAP = { "airgap": "Airgap", "airgap_mm": "Airgap", "air_gap": "Airgap", "air_gap_mm": "Airgap", "airgap_length": "Airgap", "airgap_length_mm": "Airgap", "air_gap_length": "Airgap", "air_gap_length_mm": "Airgap", "gap_length": "Airgap", "gap_length_mm": "Airgap", "magnet_length": "Magnet_Length", "magnet_length_mm": "Magnet_Length", "magnet_axial_thickness": "Magnet_Length", "magnet_axial_thickness_mm": "Magnet_Length", "magnet_axial_length": "Magnet_Length", "magnet_axial_length_mm": "Magnet_Length", "magnet_thickness": "Magnet_Thickness", "magnet_thickness_mm": "Magnet_Thickness", "magnet_radial_depth": "Magnet_Thickness", "magnet_radial_depth_mm": "Magnet_Thickness", "magnet_radial_thickness": "Magnet_Thickness", "magnet_radial_thickness_mm": "Magnet_Thickness", "magnet_arc": "Magnet_Arc_[ED]", "magnet_arc_deg": "Magnet_Arc_[ED]", "magnet_arc_[ed]": "Magnet_Arc_[ED]", "magnet_pole_arc": "Magnet_Arc_[ED]", "magnet_pole_arc_ratio": "Magnet_Arc_[ED]", "pole_arc": "Magnet_Arc_[ED]", "pole_arc_deg": "Magnet_Arc_[ED]", "pole_arc_ratio": "Magnet_Arc_[ED]", "pole_arc_coefficient": "Magnet_Arc_[ED]", "rms_current": "RMSCurrent", "rms_current_a": "RMSCurrent", "current_a": "RMSCurrent", "current": "RMSCurrent", "phase_current": "RMSCurrent", "phase_current_a": "RMSCurrent", "rated_current": "RMSCurrent", "rated_current_a": "RMSCurrent", "shaft_speed": "Shaft_Speed", "shaft_speed_rpm": "Shaft_Speed", "speed_rpm": "Shaft_Speed", "speed": "Shaft_Speed", "rotational_speed": "Shaft_Speed", "rotational_speed_rpm": "Shaft_Speed", "rated_speed": "Shaft_Speed", "rated_speed_rpm": "Shaft_Speed", "magnet_temperature": "Magnet_Temperature", "magnet_temperature_c": "Magnet_Temperature", "magnet_temp_c": "Magnet_Temperature", "magnet_temp": "Magnet_Temperature", "torque_points_per_cycle": "TorquePointsPerCycle", "torque_points": "TorquePointsPerCycle", "sampling_points": "TorquePointsPerCycle", "torque_sampling_points": "TorquePointsPerCycle", # Geometry / winding variants the AI commonly emits (normalized to the # canonical names registered in rule_engine.SCAN_PARAMETERS). "stator_outer_diameter": "Stator_Outer_Diameter", "stator_outer_diameter_mm": "Stator_Outer_Diameter", "stator_outer_dia": "Stator_Outer_Diameter", "stator_od": "Stator_Outer_Diameter", "stator_lam_outer_dia": "Stator_Outer_Diameter", "stator_lam_dia_outer": "Stator_Outer_Diameter", "stator_lam_outside_dia": "Stator_Outer_Diameter", "stator_lamination_outer_dia": "Stator_Outer_Diameter", "stator_inner_diameter": "Stator_Inner_Diameter", "stator_inner_dia": "Stator_Inner_Diameter", "stator_lam_inner_dia": "Stator_Inner_Diameter", "stator_id": "Stator_Inner_Diameter", "slot_depth": "Slot_Depth", "slot_depth_mm": "Slot_Depth", "stator_slot_depth": "Slot_Depth", "turns_per_coil": "Turns_per_Coil", "coil_turns": "Turns_per_Coil", "turnspercoil": "Turns_per_Coil", "coil_turns_per_phase": "Turns_per_Coil", "turns_per_coil_per_phase": "Turns_per_Coil", "number_of_turns": "Turns_per_Coil", "wire_diameter": "Wire_Diameter", "wire_diameter_mm": "Wire_Diameter", "wire_dia": "Wire_Diameter", "magnet_wire_diameter": "Wire_Diameter", "coil_wire_diameter": "Wire_Diameter", } def _normalize_variable_name(name: str) -> str: """Map AI-output variable name to exact Motor-CAD variable name.""" if not name: return name # Direct match in registry if name in SCAN_PARAMETERS: return name # Case-insensitive lookup in registry lower = name.lower() for reg_name in SCAN_PARAMETERS: if reg_name.lower() == lower: return reg_name # Mapping table mapped = _VARIABLE_NAME_MAP.get(lower) if mapped: return mapped return name def _extract_range(sv: Dict[str, Any]) -> Tuple[Optional[float], Optional[float], Optional[float]]: """Extract min/max/step from a scan variable dict with flexible field names.""" min_v = None max_v = None step = None # Try various field names for min for key in ["min_value", "min", "minVal", "lower", "start", "from", "minimum"]: if key in sv and sv[key] is not None: try: min_v = float(sv[key]) break except (ValueError, TypeError): pass # Try various field names for max for key in ["max_value", "max", "maxVal", "upper", "stop", "end", "to", "maximum"]: if key in sv and sv[key] is not None: try: max_v = float(sv[key]) break except (ValueError, TypeError): pass # Try step for key in ["step", "step_size", "increment", "delta", "resolution"]: if key in sv and sv[key] is not None: try: step = float(sv[key]) break except (ValueError, TypeError): pass # If range provided as [min, max] if (min_v is None or max_v is None) and "range" in sv: rng = sv["range"] if isinstance(rng, (list, tuple)) and len(rng) >= 2: try: min_v = float(rng[0]) if min_v is None else min_v max_v = float(rng[1]) if max_v is None else max_v except (ValueError, TypeError): pass return min_v, max_v, step def _generate_values(start: float, stop: float, step: float) -> List[float]: """Generate evenly spaced values from start to stop inclusive.""" if step <= 0 or start is None or stop is None: return [] count = int(math.floor((stop - start) / step + 1e-9)) + 1 if count <= 0 or count > 500: return [] vals = [round(start + i * step, 6) for i in range(count)] if vals and abs(vals[-1] - stop) > 1e-9: vals.append(round(stop, 6)) return vals def convert_ai_plan_to_unified(ai_plan: Dict[str, Any], boundary_conditions: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """Convert AI-generated plan format to unified plan schema v2.0. ... """ bc = BoundaryConditions.from_dict(boundary_conditions or {}) # Collect scan variables from either scan_variables or variables field raw_vars = ai_plan.get("scan_variables", []) or ai_plan.get("variables", []) variables = [] for sv in raw_vars: if not isinstance(sv, dict): continue raw_name = sv.get("name", "") if not raw_name: continue # Normalize variable name to Motor-CAD exact name name = _normalize_variable_name(raw_name) # If AI already provided values array, use it directly values = sv.get("values", []) if not isinstance(values, list): values = [] min_v, max_v, step = _extract_range(sv) # If no values but have range, generate if not values and min_v is not None and max_v is not None and step: values = _generate_values(min_v, max_v, step) # If still no values, fall back to rule engine defaults if not values: param = get_parameter(name) if param: rng = recommend_range(name, bc) min_v = rng["start"] max_v = rng["stop"] step = rng["step"] values = _generate_values(min_v, max_v, step) # Get display name and unit from registry if available param = get_parameter(name) tmpl = get_fixed_param_template(name) display_name = sv.get("display_name") or (param.display_name if param else tmpl.get("display_name", raw_name)) name_cn = sv.get("name_cn") or tmpl.get("name_cn") or (param.display_name if param else raw_name) unit = sv.get("unit") or (param.unit if param else tmpl.get("unit", "")) category = sv.get("category") or (param.category if param else tmpl.get("category", "")) category_cn = sv.get("category_cn") or CATEGORY_CN.get(category, category) description = sv.get("description") or tmpl.get("description", "") description_cn = sv.get("description_cn") or tmpl.get("description_cn", "") variables.append({ "name": name, "display_name": display_name, "name_cn": name_cn, "unit": unit, "start": min_v, "stop": max_v, "step": step, "values": values, "category": category, "category_cn": category_cn, "description": description, "description_cn": description_cn, }) # Deduplicate variables by name (keep first) seen = set() unique_vars = [] for v in variables: if v["name"] not in seen: seen.add(v["name"]) unique_vars.append(v) variables = unique_vars # Separate standard registry variables from custom variables standard_vars = [] custom_vars = [] warnings = [] for v in variables: if v["name"] in SCAN_PARAMETERS: standard_vars.append(v) else: custom_vars.append(v) warnings.append(f"Variable '{v['name']}' is not a standard Motor-CAD variable, may need manual adjustment") # If too many standard variables, keep only top 4 (to avoid combinatorial explosion) MAX_VARIABLES = 4 if len(standard_vars) > MAX_VARIABLES: warnings.append(f"Too many scan variables ({len(standard_vars)}), keeping top {MAX_VARIABLES} to avoid combinatorial explosion") standard_vars = standard_vars[:MAX_VARIABLES] # Use standard variables primarily, append custom ones that have explicit values variables = standard_vars for cv in custom_vars: if cv.get("values") and len(cv["values"]) > 0: variables.append(cv) else: warnings.append(f"Skipping custom variable '{cv['name']}' - no valid values") # Hard limit: total variables <= 4 if len(variables) > MAX_VARIABLES: warnings.append(f"Total variables ({len(variables)}) exceeds limit {MAX_VARIABLES}, keeping first {MAX_VARIABLES}") variables = variables[:MAX_VARIABLES] # Recalculate total points total_points = 1 for v in variables: total_points *= len(v.get("values", [])) if v.get("values") else 1 # If still too many points, trim variables one by one until under threshold MAX_POINTS = 200 while total_points > MAX_POINTS and len(variables) > 1: removed = variables.pop() warnings.append(f"Removed variable '{removed['name']}' ({len(removed.get('values', []))} values) to reduce total points below {MAX_POINTS}") total_points = 1 for v in variables: total_points *= len(v.get("values", [])) if v.get("values") else 1 if total_points > MAX_POINTS: warnings.append(f"Total scan points ({total_points}) exceeds {MAX_POINTS}, consider narrowing ranges") # Fixed params: template-driven (ALL parameters from FIXED_PARAM_TEMPLATES, # excluding scan variables). Values inferred from boundary conditions where possible. # AI does not need to output fixed_params - the template is the single source. scan_names = {v["name"].lower() for v in variables} fixed_params = build_default_fixed_params( scan_variable_names=list(scan_names), boundary_conditions=boundary_conditions, ) # Append any custom fixed params the AI explicitly provided that are not # already in the template (keeps AI flexibility for special cases). existing_names = {fp["name"].lower() for fp in fixed_params} for fp in ai_plan.get("fixed_params", []): if not isinstance(fp, dict) or not fp.get("name"): continue fname = fp["name"].lower() if fname in existing_names or fname in scan_names: continue tmpl = get_fixed_param_template(fp["name"]) val = fp.get("value", 0) try: val = float(val) except (TypeError, ValueError): pass fixed_params.append({ "name": fp["name"], "display_name": fp.get("display_name") or tmpl.get("display_name", fp["name"]), "name_cn": fp.get("name_cn") or tmpl.get("name_cn", fp["name"]), "unit": fp.get("unit") or tmpl.get("unit", ""), "value": val, "category": fp.get("category") or tmpl.get("category", "General"), "category_cn": fp.get("category_cn") or CATEGORY_CN.get(tmpl.get("category", "General"), tmpl.get("category", "General")), "description": fp.get("description") or tmpl.get("description", ""), "description_cn": fp.get("description_cn") or tmpl.get("description_cn", ""), }) existing_names.add(fname) # Normalize acceptance_criteria to standard format ac = ai_plan.get("acceptance_criteria", {}) if isinstance(ac, dict): hc = ac.get("hard_constraints") if isinstance(hc, dict): # Nested dict form: {"outer_diameter_mm": "<=110", ...} -> list of # "metric op value" strings so the UI can render them directly. ac = dict(ac) ac["hard_constraints"] = [f"{k} {v}" for k, v in hc.items()] if ac and not isinstance(ac.get("hard_constraints"), list): # AI may return custom format like {rated_torque_Nm_min: 25, ...} hard_constraints = [] for key, val in ac.items(): if isinstance(val, (int, float)): if key.endswith("_min"): metric = key.replace("_min", "").replace("_Nm", "").replace("_percent", "").replace("_N", "").replace("_T", "") hard_constraints.append(f"{metric} >= {val}") elif key.endswith("_max"): metric = key.replace("_max", "").replace("_Nm", "").replace("_percent", "").replace("_N", "").replace("_T", "") hard_constraints.append(f"{metric} <= {val}") if hard_constraints: ac = { "hard_constraints": hard_constraints, "objective_metric": ac.get("objective_metric", "efficiency_pct"), "objective_direction": ac.get("objective_direction", "maximize"), "soft_targets": {}, } # Normalize search_strategy ss = ai_plan.get("search_strategy", {}) if not isinstance(ss, dict): ss = {} # Normalize the method against the platform strategy registry. The model may # emit a non-registered method (e.g. "full_factorial_grid"); fall back to the # deterministic full_factorial default and record a warning instead of letting # the invalid value propagate to downstream validation. raw_method = ss.get("method", "") method = _strategy_normalize(raw_method) if method and not _strategy_registered(method): warnings.append( f"Unknown search strategy method '{raw_method}' - defaulting to full_factorial" ) method = "full_factorial" if not method: method = "full_factorial" if ss and "method" not in ss: ss = { "method": method, "initial_samples": 16, "batch_size": 4, "max_solver_calls": 80, "local_trust_region": False, "objective_metric": "efficiency_pct", "objective_direction": "maximize", } else: ss = dict(ss) ss["method"] = method # Estimate total points total_points = 1 for v in variables: total_points *= len(v.get("values", [])) if v.get("values") else 1 return { "name": ai_plan.get("plan_name", "AI Generated Plan"), "topology": ai_plan.get("topology", "SSSR"), "model_path": ai_plan.get("model_path", ""), "fixed_params": fixed_params, "variables": variables, "cases": [{"id": "default", "name": "Default operating point", "params": {}}], "output_metrics": [ "tavg_nm", "ripple_pct", "efficiency_pct", "total_losses_w", "copper_loss_w", "iron_loss_w", "magnet_loss_w", "back_emf_v", "input_power_w", "output_power_w", "shaft_speed_rpm", ], "search_strategy": ss, "acceptance_criteria": ac if ac else None, "ai_reasoning": ai_plan.get("reasoning", ""), "estimated_points": total_points, "estimated_time_min": total_points * 3, "warnings": warnings, } class AIPlanGenerator: """Generate simulation plans from natural language using AI.""" def __init__(self, l0_engine: Optional[L0PreScreeningEngine] = None): self.ai_client = get_kimi_client() self.l0_engine = l0_engine or L0PreScreeningEngine() self._prompt_template = None def _load_prompt(self) -> str: """Load plan generation prompt template.""" if self._prompt_template is None: prompt_path = PROMPTS_DIR / "plan_generation" / "generate.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: """Fallback default prompt.""" return """\u4f60\u662f\u8f74\u5411\u78c1\u901a\u7535\u673a\u4eff\u771f\u65b9\u6848\u4e13\u5bb6\u3002\u6839\u636e\u7528\u6237\u9700\u6c42\u548c\u8fb9\u754c\u6761\u4ef6\u751f\u6210JSON\u683c\u5f0f\u7684\u4eff\u771f\u65b9\u6848\u3002\n\u56fa\u5b9a\u53c2\u6570\u5df2\u7531\u7cfb\u7edf\u7edf\u4e00\u6a21\u677f\u751f\u6210\uff0c\u4f60\u65e0\u9700\u8f93\u51fafixed_params\uff1b\u4f60\u53ea\u9700\u8f93\u51fa\u4ee5\u4e0b\u5b57\u6bb5\uff1a\n- plan_name: \u65b9\u6848\u540d\u79f0\uff08\u7b80\u77ed\u63cf\u8ff0\uff09\n- topology: \u62d3\u6251\u7ed3\u6784\uff08SSSR/DRSS/SDSR\uff09\n- scan_variables: \u626b\u63cf\u53d8\u91cf\u6570\u7ec4\uff0c\u6bcf\u9879\u542bname\uff08\u7528Motor-CAD\u6807\u51c6\u53d8\u91cf\u540d\uff09\u3001start\u3001stop\u3001step\uff0c\u6700\u591a4\u4e2a\n- search_strategy: \u641c\u7d22\u7b56\u7565\uff08method, objective_metric, objective_direction\u7b49\uff09\n- acceptance_criteria: \u9a8c\u6536\u6807\u51c6\uff08\u786c\u7ea6\u675f\u3001\u6027\u80fd\u76ee\u6807\uff09\n- reasoning: \u4e2d\u6587\u8bbe\u8ba1\u601d\u8def\uff0c\u7ea6200\u5b57\uff0c\u53ea\u8bb2\u5173\u952e\u8bbe\u8ba1\u51b3\u7b56\u548c\u6838\u5fc3\u53c2\u6570\u9009\u62e9\u7406\u7531\uff0c\u7b80\u6d01\u76f4\u63a5\n\u8f93\u51fa\u7eafJSON\uff0c\u4e0d\u8981\u591a\u4f59\u6587\u5b57\u3002""" def generate( self, user_requirement: str, project_context: Optional[Dict[str, Any]] = None, existing_experience: Optional[List[Dict[str, Any]]] = None, ) -> Dict[str, Any]: """Generate a simulation plan from natural language. Args: user_requirement: Natural language description of the simulation need. project_context: Optional project context (topology, existing boundary conditions). existing_experience: Optional similar experience cases for reference. Returns: Dict with generated plan, validation results, and AI reasoning. """ if not self.ai_client.is_configured: raise RuntimeError("KIMI_API_KEY is not configured") # Build context message context_parts = [] if project_context: context_parts.append(f"\u9879\u76ee\u4e0a\u4e0b\u6587\uff1a{json.dumps(project_context, ensure_ascii=False)}") if existing_experience: context_parts.append(f"\u53c2\u8003\u7ecf\u9a8c\u6848\u4f8b\uff08{len(existing_experience)}\u4e2a\uff09\uff1a{json.dumps(existing_experience[:3], ensure_ascii=False)}") user_message = user_requirement if context_parts: user_message += "\n\n" + "\n".join(context_parts) # Call AI system_prompt = self._load_prompt() result = self.ai_client.chat_json( messages=[{"role": "user", "content": user_message}], system_prompt=system_prompt, max_tokens=KIMI_MAX_TOKENS, ) # Parse generated plan plan = result.get("parsed_json", {}) if not plan: # Try to extract and repair JSON from raw content raw = result.get("raw_content", result.get("content", "")) plan = self._try_repair_json(raw) if not plan: plan = {"error": "Failed to parse AI response", "raw_content": raw[:1000]} # Convert to unified format bc = (project_context or {}).get("boundary_conditions") if project_context else None unified_plan = convert_ai_plan_to_unified(plan, boundary_conditions=bc) # Validate with L0 pre-screening validation = self._validate_plan(plan) return { "plan": unified_plan, "raw_ai_plan": plan, "validation": validation, "ai_reasoning": plan.get("reasoning", ""), "usage": result.get("usage", {}), "raw_content": result.get("content", ""), } def _try_repair_json(self, raw: str) -> Dict[str, Any]: """Try to extract and repair JSON from potentially truncated AI output. Handles: - Markdown code fences - Truncated JSON (missing closing braces) - Extra text before/after JSON """ if not raw: return {} text = raw.strip() # Remove markdown code fences if text.startswith("```"): lines = text.split("\n") if lines[0].startswith("```"): lines = lines[1:] if lines and lines[-1].strip() == "```": lines = lines[:-1] text = "\n".join(lines).strip() # Find first { and last } first_brace = text.find("{") last_brace = text.rfind("}") if first_brace == -1: return {} if last_brace == -1 or last_brace < first_brace: # JSON is truncated - try to close it json_str = text[first_brace:] # Count open braces and close them open_braces = json_str.count("{") - json_str.count("}") open_brackets = json_str.count("[") - json_str.count("]") # Remove trailing incomplete content last_comma = max(json_str.rfind(","), json_str.rfind(":")) if last_comma > len(json_str) * 0.8: json_str = json_str[:last_comma] # Close brackets and braces json_str += "]" * max(0, open_brackets) json_str += "}" * max(0, open_braces) else: json_str = text[first_brace:last_brace + 1] try: return json.loads(json_str) except json.JSONDecodeError: # Try one more repair: remove trailing comma before closing try: repaired = json_str.rstrip() while repaired and repaired[-1] in ", \n\t": repaired = repaired[:-1] # Re-count and close open_braces = repaired.count("{") - repaired.count("}") open_brackets = repaired.count("[") - repaired.count("]") repaired += "]" * max(0, open_brackets) repaired += "}" * max(0, open_braces) return json.loads(repaired) except (json.JSONDecodeError, Exception): return {} def _validate_plan(self, plan: Dict[str, Any]) -> Dict[str, Any]: """Validate generated plan with L0 pre-screening. Checks: 1. Boundary conditions feasibility 2. Scan variable ranges within engineering limits 3. Sample point count estimation 4. Required fields presence """ issues = [] warnings = [] checks = [] # Check required fields required_fields = ["plan_name", "topology", "scan_variables"] for field in required_fields: if field not in plan: issues.append(f"Missing required field: {field}") # Validate topology topology = plan.get("topology", "") if topology and topology not in ("SSSR", "DRSS", "SDSR"): warnings.append(f"Unusual topology: {topology}") # Validate boundary conditions with L0 bc = plan.get("boundary_conditions", {}) if bc: l0_report = self.l0_engine.evaluate(bc) checks.append({ "name": "boundary_conditions_l0", "feasible": l0_report.feasible, "passed": l0_report.passed_checks, "total": l0_report.total_checks, "failed_items": [r.name for r in l0_report.results if not r.passed], }) if not l0_report.feasible: issues.append(f"Boundary conditions infeasible: {l0_report.failed_checks} checks failed") # Validate scan variables variables = plan.get("scan_variables", []) if not isinstance(variables, list): variables = [] total_points = 1 normalized_vars = [] for var in variables: if isinstance(var, str): var = {"name": var, "min_value": None, "max_value": None} if not isinstance(var, dict): continue normalized_vars.append(var) name = var.get("name", "unknown") min_v = var.get("min_value") max_v = var.get("max_value") step = var.get("step") if min_v is None or max_v is None: issues.append(f"Variable {name}: missing min/max values") continue if min_v >= max_v: issues.append(f"Variable {name}: min ({min_v}) >= max ({max_v})") if step and step > 0: n_points = int((max_v - min_v) / step) + 1 total_points *= n_points if n_points > 50: warnings.append(f"Variable {name}: {n_points} levels may be too many") plan["scan_variables"] = normalized_vars checks.append({ "name": "scan_variables", "count": len(variables), "estimated_full_factorial_points": total_points, "adaptive_search_recommended": total_points > 100, }) if total_points > 500: warnings.append(f"Full factorial would require {total_points} points - strongly recommend adaptive search") # Validate search strategy search = plan.get("search_strategy", {}) if search: max_calls = search.get("max_solver_calls", 80) if max_calls < total_points and not search.get("method", "").startswith(("constrained", "active")): warnings.append(f"Budget ({max_calls}) < full factorial ({total_points}) but method is not adaptive") overall_valid = len(issues) == 0 return { "valid": overall_valid, "issues": issues, "warnings": warnings, "checks": checks, } def refine_plan( self, original_plan: Dict[str, Any], user_feedback: str, ) -> Dict[str, Any]: """Refine an existing plan based on user feedback. Args: original_plan: The previously generated plan. user_feedback: Natural language feedback for refinement. Returns: Refined plan with validation. """ if not self.ai_client.is_configured: raise RuntimeError("KIMI_API_KEY is not configured") system_prompt = self._load_prompt() + "\n\n\u4f60\u6b63\u5728\u4f18\u5316\u4e00\u4e2a\u5df2\u6709\u7684\u4eff\u771f\u65b9\u6848\u3002\u6839\u636e\u7528\u6237\u53cd\u9988\u8c03\u6574\u65b9\u6848\uff0c\u4fdd\u6301\u5176\u4ed6\u90e8\u5206\u4e0d\u53d8\u3002" user_message = f"\u539f\u6709\u65b9\u6848\uff1a\n{json.dumps(original_plan, ensure_ascii=False, indent=2)}\n\n\u7528\u6237\u53cd\u9988\uff1a{user_feedback}\n\n\u8bf7\u8f93\u51fa\u4f18\u5316\u540e\u7684\u5b8c\u6574\u65b9\u6848JSON\u3002" result = self.ai_client.chat_json( messages=[{"role": "user", "content": user_message}], system_prompt=system_prompt, max_tokens=KIMI_MAX_TOKENS, ) refined_plan = result.get("parsed_json", {}) if not refined_plan: raw = result.get("raw_content", result.get("content", "")) refined_plan = self._try_repair_json(raw) if not refined_plan: refined_plan = {"error": "Failed to parse AI response", "raw_content": raw[:1000]} bc = original_plan.get("boundary_conditions") if isinstance(original_plan, dict) else None unified_plan = convert_ai_plan_to_unified(refined_plan, boundary_conditions=bc) validation = self._validate_plan(refined_plan) return { "plan": unified_plan, "raw_ai_plan": refined_plan, "validation": validation, "ai_reasoning": refined_plan.get("reasoning", ""), "usage": result.get("usage", {}), } # Global singleton _generator: Optional[AIPlanGenerator] = None def get_plan_generator() -> AIPlanGenerator: """Get or create global AIPlanGenerator singleton.""" global _generator if _generator is None: _generator = AIPlanGenerator() return _generator