Add model-reviewed CAD experience pipeline

This commit is contained in:
Jerry
2026-07-23 14:08:17 +08:00
parent 32ee79e0d1
commit 304911a69f
274 changed files with 6546 additions and 36578 deletions
+389 -4
View File
@@ -66,6 +66,32 @@ FOCUSED_SKILLS: tuple[tuple[str, tuple[str, ...]], ...] = (
("bambu-labs", ("bambu", "拓竹")),
)
EXPERIENCE_FAMILY_RULES: tuple[tuple[str, tuple[str, ...]], ...] = (
(
"flanged_hub_adapter",
("flanged hub", "法兰轮毂", "法兰轴套", "法兰适配器", "带法兰轴套"),
),
("connecting_rod", ("connecting rod", "连杆")),
("gear", ("gear", "齿轮")),
("shaft", ("shaft", "轴类", "传动轴")),
("flange", ("flange", "法兰")),
("bushing", ("bushing", "sleeve", "轴套", "衬套")),
)
EXPERIENCE_FEATURE_RULES: tuple[tuple[str, tuple[str, ...]], ...] = (
("base_flange", ("flange", "法兰", "底盘")),
("hollow_sleeve", ("hub", "sleeve", "bushing", "轴套", "衬套", "轮毂")),
("central_passage", ("through bore", "center bore", "通孔", "中心孔", "内孔")),
("counterbore", ("counterbore", "沉孔", "台阶孔")),
(
"repeated_axial_hole_pattern",
("bolt circle", "hole pattern", "螺丝孔", "螺栓孔", "孔阵列"),
),
("conical_transition", ("cone", "taper", "锥", "斜面过渡")),
("gear_teeth", ("gear", "teeth", "齿轮", "齿")),
("coaxial_cylindrical_stack", ("stepped shaft", "阶梯轴", "同轴", "轴肩")),
)
def contains_any(text: str, terms: tuple[str, ...]) -> bool:
return any(term in text for term in terms)
@@ -80,6 +106,24 @@ def normalize_formats(formats: list[str]) -> list[str]:
return result
def infer_experience_query(request: str) -> tuple[str | None, list[str]]:
text = request.lower()
family = next(
(
name
for name, terms in EXPERIENCE_FAMILY_RULES
if contains_any(text, terms)
),
None,
)
features = [
name
for name, terms in EXPERIENCE_FEATURE_RULES
if contains_any(text, terms)
]
return family, features
def cadam_root() -> Path:
configured = os.environ.get("CADAM_ROOT", "").strip()
if configured:
@@ -109,10 +153,251 @@ def probe_backends() -> dict[str, dict[str, object]]:
}
EXPERIENCE_FORBIDDEN_KEYS = {
"parameters",
"parameter_examples",
"coordinates",
"coordinate",
"center",
"location",
"axis_origin",
"source_sha256",
"source_path",
"face_refs",
"surface_ids",
"evidence",
}
MAJOR_EDIT_TERMS = (
"整体重做",
"完全重做",
"重新设计",
"改成另一种",
"大改",
"拓扑重建",
"major redesign",
"rebuild the whole",
)
LOCAL_FEATURE_EDIT_TERMS = (
"增加孔",
"添加孔",
"删除孔",
"增加槽",
"添加槽",
"删除槽",
"沉孔",
"倒角",
"圆角",
"add hole",
"remove hole",
"add slot",
"counterbore",
"fillet",
"chamfer",
)
PARAMETER_EDIT_TERMS = (
"孔径",
"直径",
"厚度",
"长度",
"宽度",
"高度",
"间距",
"齿数",
"模数",
"改为",
"调整",
"diameter",
"thickness",
"length",
"width",
"height",
"spacing",
"change",
"resize",
)
def _find_forbidden_experience_key(value: object) -> str | None:
if isinstance(value, dict):
for key, child in value.items():
if key in EXPERIENCE_FORBIDDEN_KEYS:
return key
found = _find_forbidden_experience_key(child)
if found:
return found
elif isinstance(value, list):
for child in value:
found = _find_forbidden_experience_key(child)
if found:
return found
return None
def default_experience_library() -> Path | None:
configured = os.environ.get("CAD_EXPERIENCE_LIBRARY", "").strip()
if configured:
return Path(configured).expanduser()
workspace_candidate = SKILL_ROOT.parents[2] / "cad-experience-library" / "library.json"
return workspace_candidate if workspace_candidate.is_file() else None
def classify_edit_context(
request: str,
edit_source: Path | None,
existing_model: bool,
) -> dict[str, object] | None:
if not (edit_source or existing_model):
return None
source = edit_source.expanduser().resolve() if edit_source else None
if source is not None:
parser_root = (
SKILL_ROOT.parents[2] / "cad-experience-plugin" / "parser"
).resolve()
if source == parser_root or parser_root in source.parents:
raise ValueError(
"cad-router cannot use cad-experience-plugin/parser input or "
"output as an edit source; provide the original model or "
"editable generator outside the distillation staging area"
)
suffix = source.suffix.lower() if source else ""
text = request.lower()
if suffix in {".py", ".scad", ".js", ".mjs"}:
mode = "native_parameter_edit"
reason = "An editable generator exists; update its named parameters or feature source."
elif contains_any(text, MAJOR_EDIT_TERMS):
mode = "experience_guided_semantic_reconstruction"
reason = "The requested topology change is large; rebuild from the specified source model using only promoted experience."
elif source and suffix in {".step", ".stp"} and contains_any(
text, LOCAL_FEATURE_EDIT_TERMS
):
mode = "direct_step_local_feature_edit"
reason = "Inspect the specified STEP and rebuild only the affected local feature."
elif source and suffix in {".step", ".stp"} and contains_any(
text, PARAMETER_EDIT_TERMS
):
mode = "direct_step_feature_rebuild"
reason = "STEP has no native feature history; inspect the source and reconstruct the affected dimensional feature."
elif source and suffix in {".step", ".stp"}:
mode = "inspect_existing_step_then_modify"
reason = "Inspect the explicitly supplied STEP before choosing local editing or reconstruction."
else:
mode = "existing_source_required"
reason = "An edit was requested without a resolvable editable source or private case."
return {
"operation": "modify",
"edit_source": str(source) if source else None,
"modification_mode": mode,
"reason": reason,
"allowed_knowledge_sources": [
"the explicitly supplied original model or editable generator",
"the promoted generalized CAD experience library",
"the current user request",
],
"forbidden_knowledge_sources": [
"cad-experience-plugin/parser/input",
"cad-experience-plugin/parser/output",
"private case JSON",
],
"preservation_contract": [
"Preserve the source coordinate frame unless the request explicitly changes it.",
"Preserve unmodified interfaces, datums, and feature relationships.",
"Prefer the native generator over inferred STEP reconstruction when both exist.",
"Use generalized experience only as method guidance, never as replacement source geometry.",
],
}
def load_experience_library(
path: Path | None,
family: str | None = None,
features: list[str] | None = None,
) -> dict[str, object] | None:
"""Load promoted generalized methods; reject case parameters and coordinates."""
candidate = path or default_experience_library()
if candidate is None:
return None
resolved = candidate.expanduser().resolve()
try:
payload = json.loads(resolved.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise ValueError(f"Cannot read generalized CAD experience library: {resolved}: {exc}") from exc
if payload.get("schema_version") != "2.0":
raise ValueError(f"Unsupported generalized CAD experience schema in {resolved}")
if payload.get("library_kind") != "generalized_cad_experience":
raise ValueError(f"Not a generalized CAD experience library: {resolved}")
policy = payload.get("policy")
if not isinstance(policy, dict):
raise ValueError(f"CAD experience library has no policy: {resolved}")
if (
policy.get("instance_parameters_allowed") is not False
or policy.get("absolute_coordinates_allowed") is not False
or policy.get("single_case_promotion_allowed") is not False
):
raise ValueError("CAD experience library policy permits instance-answer leakage")
if (
policy.get("llm_semantic_review_required") is not True
or policy.get("draft_evidence_verified") is not True
or policy.get("router_consumable") is not True
):
raise ValueError(
"CAD experience library was not produced by LLM semantic review "
"plus deterministic evidence verification"
)
forbidden = _find_forbidden_experience_key(payload)
if forbidden:
raise ValueError(f"CAD experience library contains forbidden instance key: {forbidden}")
requested_features = {
item.strip().lower().replace(" ", "_")
for item in (features or [])
if item.strip()
}
normalized_family = family.strip().lower().replace(" ", "_") if family else None
selected: list[object] = []
for item in payload.get("experiences", []):
if not isinstance(item, dict):
continue
scope = set(item.get("scope", []))
if "global" not in scope:
if not normalized_family or normalized_family not in scope:
continue
required = set(item.get("when", {}).get("features", []))
if required and not required.issubset(requested_features):
continue
selected.append(item)
return {
"schema_version": "2.0",
"context_kind": "generalized_cad_experience_query",
"family": normalized_family,
"requested_features": sorted(requested_features),
"experiences": selected,
"policy": {
"contains_instance_parameters": False,
"contains_absolute_coordinates": False,
},
}
def route(args: argparse.Namespace) -> dict[str, object]:
request = args.request.strip()
text = request.lower()
outputs = normalize_formats(args.output)
inferred_family, inferred_features = infer_experience_query(request)
experience_family = args.experience_family or inferred_family
experience_features = args.experience_feature or inferred_features
edit_context = classify_edit_context(
request,
args.edit_source,
args.existing_model,
)
experience_context = load_experience_library(
args.experience_library,
experience_family,
experience_features,
)
needs_source_classification = bool(edit_context and not experience_family)
scores = {
"build123d": BackendScore("build123d", 35, ["+35 general STEP-first default"]),
"simplecadapi": BackendScore("simplecadapi", 20, ["+20 specialist mechanical baseline"]),
@@ -130,7 +415,7 @@ def route(args: argparse.Namespace) -> dict[str, object]:
if contains_any(text, KEYWORD_GROUPS["replay"]):
scores["simplecadapi"].add(60, "replay, semantic graph, or FreeCAD requirement")
if args.existing_model:
if edit_context:
scores["build123d"].add(55, "existing model inspection or source modification")
scores["cadam"].add(-50, "poor fit for imported editable STEP")
if args.assembly:
@@ -161,6 +446,9 @@ def route(args: argparse.Namespace) -> dict[str, object]:
if any(fmt in {"stl", "3mf"} for fmt in outputs):
scores["cadam"].add(12, "print mesh requested")
scores["build123d"].add(5, "secondary mesh export is available")
if experience_context and experience_context["experiences"]:
scores["build123d"].add(18, "generalized experience can guide parametric feature composition")
scores["simplecadapi"].add(18, "generalized experience can guide semantic feature composition")
forced_backend = args.backend if args.backend != "auto" else None
ranked = sorted(scores.values(), key=lambda item: (-item.score, item.name))
@@ -172,7 +460,7 @@ def route(args: argparse.Namespace) -> dict[str, object]:
selected = ranked[0]
fallback = [item.name for item in ranked[1:]]
profiles: list[str] = ["caddesigner_ecip"]
profiles: list[str] = []
if contains_any(text, KEYWORD_GROUPS["visual_complexity"]):
profiles.extend(["requirement_refinement", "visual_repair"])
if not any(char.isdigit() for char in request) and contains_any(
@@ -211,11 +499,83 @@ def route(args: argparse.Namespace) -> dict[str, object]:
"downstream_skills": downstream,
"requested_outputs": outputs,
"source_of_truth": backend_info["source_of_truth"],
"experience_context": experience_context,
"design_plan": {
"plan_kind": "backend_neutral_parametric_design_plan",
"operation": "modify" if edit_context else "create",
"part_family": experience_family or "unclassified",
"requested_feature_roles": sorted(set(experience_features)),
"selected_backend": selected.name,
"source_of_truth": backend_info["source_of_truth"],
"generalized_methods": [
{
key: item[key]
for key in (
"id",
"kind",
"guidance",
"check",
"repair",
"confidence",
)
if key in item
}
for item in (
experience_context.get("experiences", [])
if experience_context
else []
)
if isinstance(item, dict)
],
"experience_status": (
"pending_source_classification"
if needs_source_classification
else (
"matched"
if experience_context and experience_context.get("experiences")
else "no_promoted_method_matched"
)
),
"experience_query_plan": {
"requires_source_inspection": needs_source_classification,
"family": experience_family,
"feature_roles": sorted(set(experience_features)),
"retry_after_inspection": needs_source_classification,
"retry_arguments": (
"--experience-family <inspected-family> "
"--experience-feature <inspected-feature>"
if needs_source_classification
else None
),
"knowledge_boundary": (
"Classify only from the explicitly supplied source model; "
"never from parser input/output or private case JSON."
),
},
"edit_context": edit_context,
"execution_steps": (
[
"Load only the explicitly supplied source model or native generator.",
"Inspect the source model to classify its family and feature roles when the request does not name them, then re-query promoted experience.",
"Map requested changes to named parameters or affected semantic feature roles.",
"Preserve unchanged interfaces and rebuild only the smallest responsible feature region.",
"Use promoted generalized experience to guide methods; never read parser input/output.",
"Use semantic reconstruction for topology changes that cannot preserve the prior feature structure.",
"Validate changed dimensions plus invariants inherited from the previous model.",
]
if edit_context
else [
"Resolve explicit user dimensions and preserve them as authoritative parameters.",
"Translate requested feature roles and matched generalized methods into backend operations.",
"Generate the editable source of truth and STEP-first artifact when requested.",
"Validate topology, dimensions, feature relationships, and export integrity.",
]
),
},
"project_contributions": {
"text-to-cad": "core Skill host, build123d CAD backend, validation, artifacts, and CAD Viewer handoff",
"text-to-cad": "core Skill host, build123d CAD backend, refinement, visual repair, validation, artifacts, and CAD Viewer handoff",
"SimpleCADAPI": "specialized mechanical backend with replayable graphs and semantic modeling",
"CADAM": "OpenSCAD-WASM execution for printable and pattern-heavy SCAD models",
"CADDesigner-Code": "ECIP brief expansion, execution feedback, and bounded visual repair workflow",
},
"availability": probe_backends() if args.probe else None,
}
@@ -236,6 +596,14 @@ def explain(result: dict[str, object]) -> str:
downstream = result["downstream_skills"]
if downstream:
lines.append(f"Downstream skills: {', '.join('$' + name for name in downstream)}")
experience_context = result.get("experience_context")
if experience_context:
lines.append(
"Experience: "
f"{experience_context.get('family') or 'global'} "
f"({len(experience_context.get('experiences', []))} generalized methods, "
"no case parameters or coordinates)"
)
lines.append("Ranking:")
for item in result["backend_scores"]:
lines.append(f" {item['backend']}: {item['score']}")
@@ -260,9 +628,26 @@ def build_parser() -> argparse.ArgumentParser:
default="unspecified",
)
parser.add_argument("--existing-model", action="store_true", help="The task modifies or inspects an existing model.")
parser.add_argument(
"--edit-source",
type=Path,
help="Existing generator, STEP/STP, or private case JSON to modify.",
)
parser.add_argument("--assembly", action="store_true", help="The requested result is an assembly.")
parser.add_argument("--browser-controls", action="store_true", help="Interactive browser parameter controls are required.")
parser.add_argument("--probe", action="store_true", help="Probe current-interpreter and OpenSCAD availability.")
parser.add_argument(
"--experience-library",
type=Path,
help="Schema 2.0 generalized CAD experience library; case JSON is rejected.",
)
parser.add_argument("--experience-family", help="Optional generalized experience family filter.")
parser.add_argument(
"--experience-feature",
action="append",
default=[],
help="Requested semantic feature role; repeatable.",
)
parser.add_argument("--explain", action="store_true", help="Print a human-readable decision instead of JSON.")
parser.add_argument("--manifest", type=Path, help="Also write the route decision as JSON to this explicit path.")
return parser