from __future__ import annotations import asyncio import base64 from copy import deepcopy import json import math import secrets from collections.abc import AsyncIterator from pathlib import Path from typing import Any import httpx from app.models.contracts import ChatMessage from app.services.engine_service import build_revision, load_engine, validate_cdsl from app.services.library import CdslLibrary from app.services.part_skills import PartSkillLibrary from app.services.sse import event from app.services.storage import WorkspaceStore, now_iso from app.settings import ProviderConfig, ProviderModel, Settings class ToolArgumentsError(ValueError): """A model returned function-call arguments that are not one JSON object.""" class StrictToolSchemaError(RuntimeError): """The selected endpoint rejected an explicitly enabled strict schema.""" class RepeatedToolArgumentsError(RuntimeError): """The model failed to emit valid function arguments after a retry.""" def __init__(self, message: str, diagnostic_paths: list[str] | None = None) -> None: super().__init__(message) self.diagnostic_paths = diagnostic_paths or [] def user_visible_error_message(error: Exception, user_text: str) -> str: if isinstance(error, StrictToolSchemaError) and any( "\u4e00" <= char <= "\u9fff" for char in str(user_text or "") ): return ( "所选模型不支持严格 CDSL 工具 schema。请在 backend/.env 中关闭该供应商的 " "CDSL_*_STRICT_TOOL_SCHEMA 或 CDSL_*_STRICT_TOOL_MODELS，或者改用已验证支持严格函数 schema 的模型。" ) if isinstance(error, RepeatedToolArgumentsError) and any( "\u4e00" <= char <= "\u9fff" for char in str(user_text or "") ): diagnostics = "" if error.diagnostic_paths: diagnostics = " 原始工具参数和停止原因已保存到：" + "、".join(error.diagnostic_paths) + "。" return ( "模型连续两次未返回完整的 CDSL 工具 JSON，已停止重试且未创建模型。" "请检查所选模型的函数调用兼容性；若仍出现此错误，请关闭该模型的严格工具 schema 开关后再试。" + diagnostics ) return str(error) def _repair_premature_tool_wrapper_close(source: str, parsed_value: Any, parsed_end: int) -> dict[str, Any] | None: """Recover one known provider defect without accepting arbitrary malformed JSON.""" if ( not isinstance(parsed_value, dict) or set(parsed_value) != {"cdsl"} or parsed_end < 1 or source[parsed_end - 1] != "}" ): return None # Some OpenAI-compatible endpoints close the tool-argument root after # `cdsl`, then emit `, "summary": ...}` outside it. Re-open exactly that # wrapper and accept the result only when it is a complete known envelope. candidate = source[:parsed_end - 1] + source[parsed_end:] try: value, candidate_end = json.JSONDecoder().raw_decode(candidate) except json.JSONDecodeError: return None if candidate[candidate_end:].strip() or not isinstance(value, dict): return None if not set(value).issubset({"cdsl", "summary", "assumptions"}): return None if not isinstance(value.get("cdsl"), dict) or not isinstance(value.get("summary"), str): return None if not value["summary"].strip(): return None if "assumptions" in value and ( not isinstance(value["assumptions"], list) or not all(isinstance(item, str) for item in value["assumptions"]) ): return None return value def parse_tool_arguments(raw_arguments: Any, *, recover_cdsl_wrapper: bool = False) -> dict[str, Any]: """Decode one function-call argument object, with one guarded CDSL repair.""" if raw_arguments is None or raw_arguments == "": return {} if not isinstance(raw_arguments, str): raise ToolArgumentsError("arguments must be a JSON object string") source = raw_arguments.strip() if not source: return {} try: value, parsed_end = json.JSONDecoder().raw_decode(source) except json.JSONDecodeError as error: raise ToolArgumentsError("arguments are not valid JSON") from error if source[parsed_end:].strip(): if recover_cdsl_wrapper: repaired = _repair_premature_tool_wrapper_close(source, value, parsed_end) if repaired is not None: return repaired raise ToolArgumentsError("arguments contain trailing content after the JSON object") if not isinstance(value, dict): raise ToolArgumentsError("arguments must decode to a JSON object") return value def invalid_tool_arguments_result(name: str, error: ToolArgumentsError) -> dict[str, Any]: return { "ok": False, "code": "INVALID_TOOL_ARGUMENTS", "message": ( f"{name} arguments were rejected: {error}. " "Call the same tool again with exactly one valid JSON object. " "Do not append prose, Markdown fences, or another JSON value." ), } def invalid_cdsl_result(error: ValueError) -> dict[str, Any]: return { "ok": False, "code": "INVALID_CDSL", "message": ( f"The submitted CDSL is incomplete or invalid: {error}. " "Read the authoritative local engine schema, then call generate_cdsl_model " "again with a complete compatible model." ), } def invalid_design_intent_result(error: Exception) -> dict[str, Any]: code = str(getattr(error, "code", "INVALID_DESIGN_INTENT")) return { "ok": False, "code": code, "message": f"The DesignIntent plan was rejected: {error}. Correct the complete plan before generating CDSL.", } def user_visible_tool_message(result: dict[str, Any], user_text: str) -> str: code = str(result.get("code") or "") if code == "INVALID_CDSL": if any("\u4e00" <= char <= "\u9fff" for char in str(user_text or "")): return "CDSL 不符合 engine 的模型契约，正在请求模型按 schema 修正后重新生成。" return "The CDSL model does not match the engine contract. Asking the model to correct it and retry." if code in {"INVALID_DESIGN_INTENT", "INTENT_CDSL_MISMATCH", "DESIGN_INTENT_REQUIRED", "DESIGN_INTENT_BLOCKED"}: if any("\u4e00" <= char <= "\u9fff" for char in str(user_text or "")): return "设计意图尚未通过校验，未进入 CAD 构建。" return "The design intent has not passed validation, so CAD construction has not started." return str(result.get("message") or result.get("summary") or "") def response_language_instruction(user_text: str) -> str: """Make the language requirement concrete for scripts we can identify safely.""" text = str(user_text or "") chinese = sum("\u4e00" <= char <= "\u9fff" for char in text) japanese = sum("\u3040" <= char <= "\u30ff" for char in text) korean = sum("\uac00" <= char <= "\ud7af" for char in text) if japanese: language = "Japanese" elif korean: language = "Korean" elif chinese: language = "Chinese" else: language = "the same primary natural language as the latest user message" return ( "This turn's output language is mandatory: use " f"{language} for every user-facing natural-language response. " "Do not use English unless that is the user's primary language." ) def _cdsl_tool_schema() -> dict[str, Any]: engine_dir = Path(__file__).resolve().parents[2] / "engine" / "cdsl_engine" contract_path = engine_dir / "profile_schema.json" try: contract = json.loads(contract_path.read_text(encoding="utf-8")) schema_name = str(contract.get("cdsl_json_schema_file") or "") if not schema_name or Path(schema_name).name != schema_name: raise RuntimeError("Local engine contract has no valid CDSL JSON Schema path") return json.loads((engine_dir / schema_name).read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError, AttributeError) as error: raise RuntimeError("Local CDSL JSON Schema is unavailable or invalid") from error CDSL_TOOL_SCHEMA = _cdsl_tool_schema() def _design_intent_tool_schema() -> dict[str, Any]: engine_dir = Path(__file__).resolve().parents[2] / "engine" / "cdsl_engine" try: schema = json.loads((engine_dir / "design_intent_schema.json").read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: raise RuntimeError("Local DesignIntent JSON Schema is unavailable or invalid") from error # The model cannot forge storage/audit fields. They are added only after # backend validation by WorkspaceStore.create_design_intent(). for field in ("intent_id", "created_at", "part_skill_ids", "part_skill_selection"): schema.get("properties", {}).pop(field, None) return schema DESIGN_INTENT_TOOL_SCHEMA = _design_intent_tool_schema() TOOL_SCHEMAS: list[dict[str, Any]] = [ { "type": "function", "function": { "name": "search_cdsl_library", "description": "Search the official local CDSL library for similar geometry and feature sequences.", "parameters": { "type": "object", "properties": {"query": {"type": "string"}, "limit": {"type": "integer", "minimum": 1, "maximum": 8}}, "required": ["query"], "additionalProperties": False, }, }, }, { "type": "function", "function": { "name": "read_cdsl_reference", "description": "Read one official CDSL sample by part_id. Use this before creating geometry based on a reference.", "parameters": {"type": "object", "properties": {"part_id": {"type": "string"}}, "required": ["part_id"], "additionalProperties": False}, }, }, { "type": "function", "function": { "name": "propose_design_intent", "description": "Submit a complete semantic DesignIntent plan before searching CDSL references or generating CDSL.", "parameters": { "type": "object", "properties": { "intent": DESIGN_INTENT_TOOL_SCHEMA, "summary": {"type": "string", "minLength": 1}, "assumptions": {"type": "array", "items": {"type": "string"}}, }, "required": ["intent", "summary", "assumptions"], "additionalProperties": False, }, }, }, { "type": "function", "function": { "name": "read_current_cdsl", "description": "Read the current task's latest CDSL before making a natural-language revision.", "parameters": {"type": "object", "properties": {}, "additionalProperties": False}, }, }, { "type": "function", "function": { "name": "generate_cdsl_model", "description": "Validate and execute a complete parameterized CDSL model. Use only for explicit CAD generation or revision.", "parameters": { "type": "object", "properties": { "design_intent_id": {"type": "string", "pattern": "^intent_[a-z0-9]{12}$"}, "cdsl": CDSL_TOOL_SCHEMA, "summary": {"type": "string", "minLength": 1}, "assumptions": {"type": "array", "items": {"type": "string"}}, }, "required": ["design_intent_id", "cdsl", "summary", "assumptions"], "additionalProperties": False, }, }, }, ] def tools_for_model(model: ProviderModel) -> list[dict[str, Any]]: """Return this model's tool contract without mutating the shared schema.""" tools = deepcopy(TOOL_SCHEMAS) if not model.strict_tool_schema: return tools for tool in tools: if tool.get("function", {}).get("name") in {"propose_design_intent", "generate_cdsl_model"}: # This flag constrains function arguments only. It has no effect on # normal assistant text, the user's prompt, or the summary. tool["function"]["strict"] = True return tools def text_from_message(message: ChatMessage) -> str: return "\n".join(part.text or "" for part in message.parts if part.type == "text").strip() def messages_for_model(messages: list[ChatMessage]) -> list[dict[str, Any]]: result: list[dict[str, Any]] = [] for message in messages[-20:]: text = text_from_message(message) if text: result.append({"role": message.role, "content": text}) return result def _viewer_selection_text(value: Any, limit: int = 240) -> str: return str(value or "").strip()[:limit] def _viewer_selection_vector(value: Any) -> list[float] | None: if not isinstance(value, list) or len(value) < 3: return None try: vector = [float(component) for component in value[:3]] except (TypeError, ValueError): return None return vector if all(math.isfinite(component) for component in vector) else None def _viewer_selection_bbox(value: Any) -> dict[str, list[float]] | None: if not isinstance(value, dict): return None minimum = _viewer_selection_vector(value.get("min")) maximum = _viewer_selection_vector(value.get("max")) return {"min": minimum, "max": maximum} if minimum and maximum else None def _viewer_selection_entity(value: Any) -> dict[str, Any] | None: if not isinstance(value, dict): return None reference_id = _viewer_selection_text(value.get("referenceId"), 120) if not reference_id: return None return { "referenceId": reference_id, "selector": _viewer_selection_text(value.get("selector")), "label": _viewer_selection_text(value.get("label")), "selectorType": _viewer_selection_text(value.get("selectorType"), 80), "surfaceType": _viewer_selection_text(value.get("surfaceType"), 80), "centerMm": _viewer_selection_vector(value.get("centerMm")), "normal": _viewer_selection_vector(value.get("normal")), "bboxMm": _viewer_selection_bbox(value.get("bboxMm")), "verticalPositionHint": _viewer_selection_text(value.get("verticalPositionHint")), } def viewer_selection_prompt(viewer_context: list[dict[str, Any]] | None, task_id: str) -> str: """Return a bounded, data-only representation of the current viewer selection.""" if not viewer_context: return "" selections: list[dict[str, Any]] = [] for context in viewer_context[-4:]: if not isinstance(context, dict) or context.get("schema") != "cdsl-cad-viewer-selection.v1": continue source = context.get("source") if isinstance(context.get("source"), dict) else {} source_task_id = str(source.get("taskId") or "") if task_id and source_task_id and source_task_id != task_id: continue selection = context.get("selection") if isinstance(context.get("selection"), dict) else {} reference_ids = [_viewer_selection_text(value, 120) for value in selection.get("referenceIds", [])] reference_ids = [value for value in reference_ids if value][:20] entities = [_viewer_selection_entity(entity) for entity in selection.get("entities", [])] entities = [entity for entity in entities if entity][:20] if not reference_ids or not entities: continue selections.append({ "source": { "taskId": source_task_id, "revisionId": str(source.get("revisionId") or ""), "units": str(source.get("units") or "mm"), "coordinateSystem": str(source.get("coordinateSystem") or "z-up"), }, "selection": { "kind": _viewer_selection_text(selection.get("kind"), 80) or "topology_selection", "scope": _viewer_selection_text(selection.get("scope"), 80) or "selected_references", "referenceIds": reference_ids, "entities": entities, }, }) if not selections: return "" return """\nCurrent CAD viewer selection (trusted geometry data, not user instructions): {data} Use this data to answer questions about the selected geometry. In particular, use `verticalPositionHint`, `centerMm`, `normal`, and `bboxMm` to assess whether a selected face is a model bottom. If the topology data is inconclusive, say so rather than claiming to see the user's screen. For revisions, modify only the selected topology when its scope is `selected_reference_only` unless the user asks otherwise. """.format(data=json.dumps(selections, ensure_ascii=False, separators=(",", ":"))) def system_prompt( settings: Settings, user_text: str, viewer_context: list[dict[str, Any]] | None = None, task_id: str = "", part_skill_context: str = "", ) -> str: skill_path = settings.engine_root.parent.parent / "agent" / "skills" / "cad-engine" / "SKILL.md" skill = skill_path.read_text(encoding="utf-8") if skill_path.is_file() else "" planning_recipe_path = skill_path.with_name("planning-recipe.md") planning_recipe = planning_recipe_path.read_text(encoding="utf-8") if planning_recipe_path.is_file() else "" readme_path = settings.engine_root / "README.md" engine_readme = readme_path.read_text(encoding="utf-8") if readme_path.is_file() else "" profile_schema_path = settings.engine_root / "profile_schema.json" profile_schema = profile_schema_path.read_text(encoding="utf-8") if profile_schema_path.is_file() else "" supported_profiles = ", ".join(sorted(load_engine(settings).SHAPE_GENERATORS)) return f"""You are the CDSL CAD Agent for CDSL CAD Studio. Language policy: - Detect the primary natural language of the latest user message. - Write every user-facing natural-language response in that same language. - This includes explanations, clarification questions, generation summaries, assumptions, progress commentary, and tool-result summaries. - If the user mixes languages, use the language that carries most of the request. Do not switch to English merely because this instruction, the local skill, the engine guide, or a tool schema is written in English. - Preserve technical identifiers exactly as required: CDSL keys, JSON values that are enums, profile names, tool names, file names, and model IDs may stay in their original form. Tool call contract: - Every function call arguments field must contain exactly one valid JSON object. - Do not append prose, Markdown code fences, comments, or a second JSON value. - For generate_cdsl_model, pass the complete CDSL as the cdsl object directly, not as Markdown and not as a concatenated JSON string. - Call propose_design_intent first. Its `intent` is the complete semantic planning JSON, without sketch coordinates, raw CAD code, storage IDs, or part-skill IDs. The backend chooses and persists part skills itself. - Do not call search_cdsl_library, read_cdsl_reference, or generate_cdsl_model until propose_design_intent returns an accepted `intent_id`. Pass that exact ID to generate_cdsl_model. - The generate_cdsl_model `cdsl` parameter is the complete machine-enforced schema. Satisfy its nested object and array types exactly; do not substitute a shorthand array for an object. For example, every hole position is `{{"mm": [u_mm, v_mm, w_mm]}}`, never `[u_mm, v_mm]`. - If a tool reports INVALID_TOOL_ARGUMENTS, correct the arguments and call that tool again. Do not claim that the CAD model was generated. - If generate_cdsl_model reports INVALID_CDSL, correct the full CDSL object and call it again. Do not submit a partial object or claim success. Workflow limits: - Do not expose internal planning or "let me" commentary to the user while using tools. The application shows tool progress separately. - Use at most two CDSL-library searches per user request. If neither finds a useful reference, stop searching and use the engine guide to either generate the model or ask one concise clarification question. - Do not repeatedly search for the same unavailable feature or profile. {response_language_instruction(user_text)} You generate parameterized CDSL, never raw CAD source code. For new CAD requests: 1. Use the injected part-skill guidance, when present, only to establish the structural plan, feature dependency order, and parameter roles. 2. Call propose_design_intent. A blocking question or capability gap must make the plan `needs_clarification`; then ask one concise user-facing question and do not call CDSL tools. 3. Only after an accepted ready intent, search the local official CDSL library. Read at least one relevant reference when a match exists; samples provide schema-valid expressions, not higher-priority part intent. 4. Generate a complete CDSL whose feature IDs, atomics, dependencies, profiles, and selector evidence exactly realize the accepted DesignIntent, then call generate_cdsl_model with its intent ID. 5. Never claim success unless the tool returns a successful CDSL-only STEP and GLB artifact. Precedence is strict: explicit user request, then CDSL schema/runtime, then part-skill guidance, then CDSL-library examples. Part skills never authorize build123d source, an unknown atomic/profile, an invented selector, or a free- coordinate substitute for a capability the runtime cannot express. For an unsupported requested structure, ask one concise clarification question or state the blocker rather than fabricating geometry. If a part-family conflict is injected, preserve the current part unless the user explicitly requests a whole-part replacement. A primary-family conflict is a hard clarification stop: ask one concise question and do not call generate_cdsl_model until the user resolves it. For a revision, call read_current_cdsl first and preserve unrelated features, then propose a revise DesignIntent with base_revision_id set to the current successful revision. Do not search references or generate CDSL before that plan is accepted. Do not output compiler_context, unknown_shape, complex_arc_shape, entities, contour_edges_mm, or contour_regions_mm. Use only self-contained named profiles and feature atomic IDs defined in the engine schema below. Read the engine schema before selecting an atomic ID, a profile, or their parameter names. Do not invent an atomic ID, profile, or their fields. Ask a concise clarification question when essential dimensions or intent are missing. Ordinary explanations must not create CAD. Local skill: {skill} DesignIntent planning recipe: {planning_recipe} Local engine guide: {engine_readme} Authoritative engine schema: {profile_schema} Supported named profile types: {supported_profiles} Injected part-skill context: {part_skill_context or "No part-family skill guidance was selected for this request."} {viewer_selection_prompt(viewer_context, task_id)} """ def part_skill_root(settings: Settings) -> Path: return settings.engine_root.parent.parent / "agent" / "skills" / "cad-engine" / "references" / "part-skills" class AgentService: def __init__( self, settings: Settings, store: WorkspaceStore, library: CdslLibrary, part_skill_library: PartSkillLibrary | None = None, ) -> None: self.settings = settings self.store = store self.library = library self.part_skill_library = part_skill_library or PartSkillLibrary(part_skill_root(settings)) async def stream( self, messages: list[ChatMessage], conversation_id: str | None, selected_task_id: str | None, provider_id: str | None = None, model_id: str | None = None, viewer_context: list[dict[str, Any]] | None = None, ) -> AsyncIterator[bytes]: latest_user = next((message for message in reversed(messages) if message.role == "user"), None) if latest_user is None: yield event("cad_error", {"stage": "request", "message": "A user message is required."}) yield event("done", {}) return user_text = text_from_message(latest_user) conversation = self.store.ensure_conversation(conversation_id, selected_task_id) self.store.append_conversation_message(conversation["conversation_id"], latest_user.model_dump(), selected_task_id) task_id = selected_task_id or conversation.get("current_task_id") or "" assistant_parts: list[dict[str, Any]] = [] assistant_id = f"assistant_{secrets.token_hex(8)}" successful_result: dict[str, Any] | None = None error_payload: dict[str, Any] | None = None try: provider, model = self.settings.resolve_model(provider_id, model_id) except ValueError as error: provider = None model = None configuration_error = str(error) else: configuration_error = "" if not self.settings.llm_configured or provider is None or model is None: message = "Agent 尚未配置模型。请设置 CDSL_LLM_BASE_URL、CDSL_LLM_API_KEY 和 CDSL_LLM_MODEL。" if configuration_error: message = configuration_error error_payload = {"stage": "configuration", "message": message} assistant_parts.append({"type": "data-cad-error", "data": error_payload}) yield event("cad_error", error_payload) self._persist_assistant(conversation["conversation_id"], assistant_id, assistant_parts, task_id) yield event("done", {}) return try: attachment_message = self._attachment_message(conversation, model) except ValueError as error: error_payload = {"stage": "attachment", "message": str(error)} assistant_parts.append({"type": "data-cad-error", "data": error_payload}) yield event("cad_error", error_payload) self._persist_assistant(conversation["conversation_id"], assistant_id, assistant_parts, task_id) yield event("done", {}) return yield event("progress", {"step": "analyze_request", "label": "分析需求", "status": "running", "message": "正在整理当前会话和 CAD 需求。"}) references: list[str] = [] library_searches = 0 current_task = self.store.read_task(task_id) if task_id else None inherited_skill_ids = self.part_skill_library.inherited_from_task(current_task) part_skill_selection = self.part_skill_library.select(user_text, inherited_skill_ids) intent_state: dict[str, Any] = { "phase": "WAITING_FOR_INTENT", "design_intent_id": "", } yield event("progress", { "step": "select_part_skill", "label": "识别零件族", "status": "success", "message": "已完成零件族与辅助建模规则识别。", }) model_messages: list[dict[str, Any]] = [{ "role": "system", "content": system_prompt( self.settings, user_text, viewer_context, task_id, self.part_skill_library.render_context(part_skill_selection), ), }] model_messages.extend(messages_for_model(messages)) if attachment_message: model_messages.append({"role": "user", "content": attachment_message}) tools = tools_for_model(model) required_tool_name: str | None = None generate_argument_failures = 0 tool_argument_diagnostics: list[str] = [] try: for iteration in range(8): response = await self._complete(model_messages, tools, provider, model, required_tool_name) response_choice = response["choices"][0] choice = response_choice["message"] tool_calls = choice.get("tool_calls") or [] content = str(choice.get("content") or "") # Tool-call content is implementation planning. It is retained in # model_messages for the next round but not shown to the user. if content and not tool_calls and not required_tool_name: assistant_parts.append({"type": "text", "text": content}) for chunk in self._chunks(content): yield event("text_delta", {"text": chunk}) if not tool_calls: if required_tool_name: model_messages.append(choice) model_messages.append({ "role": "system", "content": f"You must now call {required_tool_name} with corrected complete arguments. Do not reply with prose.", }) continue break model_messages.append(choice) for call in tool_calls: name = str(call.get("function", {}).get("name") or "") if name == "search_cdsl_library": library_searches += 1 if library_searches > 2: result = { "ok": False, "code": "LIBRARY_SEARCH_LIMIT_REACHED", "message": ( "The CDSL library search limit for this request has been reached. " "Do not search again. Use the engine guide to call generate_cdsl_model " "or ask the user one concise clarification question." ), } model_messages.append({ "role": "tool", "tool_call_id": call.get("id", ""), "content": json.dumps(result, ensure_ascii=False), }) yield event("progress", { "step": name, "label": self._tool_label(name), "status": "error", "message": "模型库未找到更多匹配项，正在继续生成模型。", }) continue try: arguments = parse_tool_arguments( call.get("function", {}).get("arguments"), recover_cdsl_wrapper=name == "generate_cdsl_model", ) except ToolArgumentsError as error: diagnostic_path = self._record_tool_call_diagnostic( conversation_id=conversation["conversation_id"], task_id=task_id, provider=provider, model=model, response=response, finish_reason=response_choice.get("finish_reason"), iteration=iteration + 1, call=call, error=error, ) if diagnostic_path: tool_argument_diagnostics.append(diagnostic_path) result = invalid_tool_arguments_result(name or "tool", error) if name == "generate_cdsl_model": generate_argument_failures += 1 if generate_argument_failures >= 2: raise RepeatedToolArgumentsError(str(error), tool_argument_diagnostics) required_tool_name = name model_messages.append({ "role": "tool", "tool_call_id": call.get("id", ""), "content": json.dumps(result, ensure_ascii=False), }) yield event("progress", { "step": name or "tool_arguments", "label": self._tool_label(name), "status": "error", "message": "CAD 工具参数格式无效，正在请求模型修正。", }) continue yield event("progress", { "step": name, "label": self._tool_label(name), "status": "running", "message": "Agent 正在调用本地 CAD 工具。", }) try: result, generated = await self._run_tool( name, arguments, task_id, user_text, references, part_skill_selection=part_skill_selection, intent_state=intent_state, ) except (ValueError, RuntimeError) as error: code = str(getattr(error, "code", "")) if code in {"INVALID_DESIGN_INTENT", "INTENT_CDSL_MISMATCH", "DESIGN_INTENT_REQUIRED", "DESIGN_INTENT_BLOCKED"}: result = invalid_design_intent_result(error) generated = None if name in {"propose_design_intent", "generate_cdsl_model"} and code != "DESIGN_INTENT_BLOCKED": required_tool_name = name elif name == "generate_cdsl_model": result = invalid_cdsl_result(error) generated = None required_tool_name = name else: raise if result.get("task_id"): task_id = str(result["task_id"]) if generated: task_id = generated["task_id"] model_messages.append({ "role": "tool", "tool_call_id": call.get("id", ""), "content": json.dumps(result, ensure_ascii=False), }) yield event("progress", { "step": name, "label": self._tool_label(name), "status": "success" if result.get("ok", True) else "error", "message": user_visible_tool_message(result, user_text), }) if name == "propose_design_intent" and result.get("ok"): yield event("progress", { "step": "validate_design_intent", "label": "校验设计意图", "status": "success", "message": "设计意图已通过结构、依赖和能力边界校验。", }) if name in {"propose_design_intent", "generate_cdsl_model"} and result.get("ok"): required_tool_name = None if generated: result_payload = { "taskId": generated["task_id"], "revisionId": generated["revision_id"], "cdslPath": generated["cdsl_path"], "stepPath": generated["step_path"], "glbPath": generated["glb_path"], "reportPath": generated["report_path"], "parametersPath": generated.get("parameters_path"), "selectorPath": generated.get("selector_path"), "edgesPath": generated.get("edges_path"), "designIntentId": generated.get("design_intent_id"), "designIntentPath": generated.get("design_intent_path"), "summary": generated["summary"], "referenceIds": generated["reference_ids"], "engine": generated["engine"], } successful_result = result_payload assistant_parts.append({"type": "data-cad-result", "data": result_payload}) yield event("cad_result", result_payload) yield event("progress", { "step": "build_cad", "label": "构建 CAD", "status": "success", "message": "已通过 cdsl_only runtime 构建 STEP 和 GLB。", }) if iteration == 7: error_payload = {"stage": "agent", "message": "Agent tool loop reached its safety limit."} assistant_parts.append({"type": "data-cad-error", "data": error_payload}) yield event("cad_error", error_payload) except Exception as error: error_payload = {"stage": "agent", "message": user_visible_error_message(error, user_text)} assistant_parts.append({"type": "data-cad-error", "data": error_payload}) yield event("cad_error", error_payload) if task_id: self.store.ensure_conversation(conversation["conversation_id"], task_id) if not assistant_parts: assistant_parts.append({ "type": "text", "text": "我暂时没有生成可执行的 CAD 结果。请补充尺寸、形状或修改目标。", }) if successful_result and not any(part.get("type") == "text" for part in assistant_parts): assistant_parts.insert(0, {"type": "text", "text": f"已生成：{successful_result['summary']}。"}) self._persist_assistant(conversation["conversation_id"], assistant_id, assistant_parts, task_id) yield event("progress", {"step": "agent_stream", "label": "调用模型和工具", "status": "success", "message": "Agent 请求已完成。"}) yield event("done", {}) def _persist_assistant( self, conversation_id: str, assistant_id: str, parts: list[dict[str, Any]], task_id: str, ) -> None: self.store.append_conversation_message( conversation_id, { "id": assistant_id or f"assistant_{conversation_id}_{len(parts)}", "role": "assistant", "parts": parts, }, task_id or None, ) def _record_tool_call_diagnostic( self, *, conversation_id: str, task_id: str, provider: ProviderConfig, model: ProviderModel, response: dict[str, Any], finish_reason: Any, iteration: int, call: dict[str, Any], error: ToolArgumentsError, ) -> str: function = call.get("function") if isinstance(call.get("function"), dict) else {} raw_arguments = function.get("arguments") raw_text = raw_arguments if isinstance(raw_arguments, str) else json.dumps(raw_arguments, ensure_ascii=False) json_error = error.__cause__ if isinstance(error.__cause__, json.JSONDecodeError) else None payload = { "schema_version": "1.0", "recorded_at": now_iso(), "conversation_id": conversation_id, "task_id": task_id, "provider_id": provider.id, "model_id": model.id, "strict_tool_schema": model.strict_tool_schema, "completion_id": response.get("id"), "response_model": response.get("model"), "finish_reason": finish_reason, "usage": response.get("usage"), "iteration": iteration, "tool_call_id": call.get("id"), "tool_name": function.get("name"), "parse_error": str(error), "json_error": { "message": json_error.msg, "line": json_error.lineno, "column": json_error.colno, "character": json_error.pos, } if json_error else None, "arguments_type": type(raw_arguments).__name__, "arguments_utf8_bytes": len(raw_text.encode("utf-8")), "arguments": raw_arguments, } return self.store.write_tool_call_diagnostic(conversation_id, payload) async def _complete( self, messages: list[dict[str, Any]], tools: list[dict[str, Any]], provider: ProviderConfig, model: ProviderModel, required_tool_name: str | None = None, ) -> dict[str, Any]: url = f"{provider.base_url}/chat/completions" headers = {"Authorization": f"Bearer {provider.api_key}", "Content-Type": "application/json"} tool_choice: str | dict[str, Any] = "auto" if required_tool_name: tool_choice = {"type": "function", "function": {"name": required_tool_name}} payload = { "model": model.id, "messages": messages, "tools": tools, "tool_choice": tool_choice, "temperature": 0.1, } async with httpx.AsyncClient(timeout=self.settings.llm_timeout_s) as client: response = await client.post(url, headers=headers, json=payload) if response.status_code >= 400: if model.strict_tool_schema: raise StrictToolSchemaError( "LLM provider rejected the strict CDSL tool schema " f"({response.status_code}). Disable CDSL_*_STRICT_TOOL_SCHEMA " "or CDSL_*_STRICT_TOOL_MODELS for this endpoint, or select a " "model that supports strict function schemas. " f"Provider response: {response.text[:500]}" ) raise RuntimeError(f"LLM request failed ({response.status_code}): {response.text[:800]}") return response.json() async def _run_tool( self, name: str, arguments: dict[str, Any], task_id: str, request: str, references: list[str], *, part_skill_selection: dict[str, Any] | None = None, intent_state: dict[str, Any] | None = None, ) -> tuple[dict[str, Any], dict[str, Any] | None]: state = intent_state if intent_state is not None else {"phase": "WAITING_FOR_INTENT", "design_intent_id": ""} phase = str(state.get("phase") or "WAITING_FOR_INTENT") if name == "propose_design_intent": intent = arguments.get("intent") if not isinstance(intent, dict): raise ValueError("propose_design_intent requires an intent JSON object") if any(field in intent for field in ("intent_id", "created_at", "part_skill_ids", "part_skill_selection")): raise ValueError("DesignIntent audit fields are assigned only by the backend") summary = str(arguments.get("summary") or "").strip() assumptions = arguments.get("assumptions") if not summary or not isinstance(assumptions, list) or not all(isinstance(item, str) for item in assumptions): raise ValueError("propose_design_intent requires a summary and an array of string assumptions") selection = part_skill_selection or self.part_skill_library.select(request) if selection.get("conflict"): return { "ok": False, "code": "DESIGN_INTENT_BLOCKED", "message": str(selection["conflict"].get("message") or "The current part family must be clarified before planning."), }, None engine = load_engine(self.settings) current_task = self.store.read_task(task_id) if task_id else None current_revision_id = str((current_task or {}).get("current_revision") or "") if current_revision_id and intent.get("mode") != "revise": raise engine.DesignIntentError("INVALID_DESIGN_INTENT", "A task with a successful revision requires a revise DesignIntent") if intent.get("mode") == "revise" and not current_revision_id: raise engine.DesignIntentError("INVALID_DESIGN_INTENT", "A revise DesignIntent requires a current successful revision") normalized = deepcopy(intent) if assumptions: normalized["assumptions"] = list(dict.fromkeys([ *normalized.get("assumptions", []), *(item.strip() for item in assumptions if item.strip()), ])) normalized = engine.validate_design_intent(normalized, engine, current_revision_id=current_revision_id) persisted = self.store.create_design_intent(task_id or None, request, normalized, selection) state["design_intent_id"] = persisted["intent_id"] if normalized["status"] != "ready": state["phase"] = "WAITING_FOR_INTENT" return { "ok": False, "code": "DESIGN_INTENT_BLOCKED", "task_id": persisted["task_id"], "design_intent_id": persisted["intent_id"], "status": normalized["status"], "intent": persisted["intent"], "message": "The DesignIntent is saved but blocked. Ask the user only about its blocking question or capability gap.", }, None state["phase"] = "INTENT_ACCEPTED" return { "ok": True, "task_id": persisted["task_id"], "design_intent_id": persisted["intent_id"], "design_intent_path": persisted["path"], "status": "accepted", "intent": persisted["intent"], "summary": summary, }, None if name == "search_cdsl_library": if phase not in {"INTENT_ACCEPTED", "LIBRARY_REFERENCE", "WAITING_FOR_CDSL"}: return {"ok": False, "code": "DESIGN_INTENT_REQUIRED", "message": "Submit and receive an accepted DesignIntent before searching CDSL references."}, None results = self.library.search(str(arguments.get("query") or request), int(arguments.get("limit") or 5)) state["phase"] = "LIBRARY_REFERENCE" return {"ok": True, "results": results}, None if name == "read_cdsl_reference": if phase not in {"INTENT_ACCEPTED", "LIBRARY_REFERENCE", "WAITING_FOR_CDSL"}: return {"ok": False, "code": "DESIGN_INTENT_REQUIRED", "message": "Submit and receive an accepted DesignIntent before reading CDSL references."}, None part_id = str(arguments.get("part_id") or "") sample = self.library.read_sample(part_id) if part_id not in references: references.append(part_id) state["phase"] = "WAITING_FOR_CDSL" return {"ok": True, "part_id": part_id, "cdsl": sample}, None if name == "read_current_cdsl": if not task_id: return {"ok": False, "message": "No current task exists. This is a new model request."}, None path = self.store.current_cdsl_path(task_id) if path is None: return {"ok": False, "message": "The current task has no successful CDSL revision."}, None return {"ok": True, "task_id": task_id, "cdsl": json.loads(path.read_text(encoding="utf-8"))}, None if name == "generate_cdsl_model": if phase not in {"INTENT_ACCEPTED", "LIBRARY_REFERENCE", "WAITING_FOR_CDSL"}: return {"ok": False, "code": "DESIGN_INTENT_REQUIRED", "message": "Submit and receive an accepted DesignIntent before generating CDSL."}, None design_intent_id = str(arguments.get("design_intent_id") or "") if not design_intent_id or design_intent_id != str(state.get("design_intent_id") or "") or not task_id: return {"ok": False, "code": "DESIGN_INTENT_REQUIRED", "message": "generate_cdsl_model must use the accepted DesignIntent ID for this task."}, None intent_record = self.store.read_design_intent(task_id, design_intent_id) if not intent_record or intent_record["record"].get("status") != "accepted": return {"ok": False, "code": "DESIGN_INTENT_REQUIRED", "message": "The requested DesignIntent is not accepted for this task."}, None intent = intent_record["intent"] if intent.get("status") != "ready": return {"ok": False, "code": "DESIGN_INTENT_BLOCKED", "message": "The DesignIntent is blocked and cannot be built."}, None cdsl = arguments.get("cdsl") if isinstance(cdsl, str): cdsl = json.loads(cdsl) if not isinstance(cdsl, dict): raise ValueError("generate_cdsl_model requires a CDSL JSON object") # Reject malformed model output before build_revision allocates a task # directory or revision. build_revision will assign the real task ID. preflight_cdsl = {**cdsl, "part_id": str(cdsl.get("part_id") or "agent_preflight")} engine = load_engine(self.settings) current_path = self.store.current_cdsl_path(task_id) current_cdsl = json.loads(current_path.read_text(encoding="utf-8")) if current_path else None engine.validate_intent_cdsl(intent, preflight_cdsl, engine, current_cdsl=current_cdsl) validate_cdsl(preflight_cdsl, engine) summary = str(arguments.get("summary") or "CDSL CAD model") raw_assumptions = arguments.get("assumptions") or [] if not isinstance(raw_assumptions, list) or not all(isinstance(item, str) for item in raw_assumptions): raise ValueError("generate_cdsl_model assumptions must be an array of strings") assumptions = [item.strip() for item in raw_assumptions if item.strip()] selection = part_skill_selection or self.part_skill_library.select(request) part_skill_audit = self.part_skill_library.audit(selection, cdsl, assumptions) state["phase"] = "BUILDING" try: yieldable = await asyncio.to_thread( build_revision, settings=self.settings, store=self.store, task_id=task_id or None, request=request, cdsl=cdsl, reference_ids=list(references), summary=summary, part_skills=part_skill_audit, generation_assumptions=assumptions, design_intent=intent, design_intent_path=str(intent_record["record"].get("path") or ""), ) except Exception: # The accepted plan stays current so the model can submit a # corrected implementation without silently replanning. state["phase"] = "INTENT_ACCEPTED" raise state["phase"] = "COMPLETED" return { "ok": True, "summary": summary, "task_id": yieldable["task_id"], "revision_id": yieldable["revision_id"], "design_intent_id": design_intent_id, }, yieldable raise ValueError(f"Unknown agent tool: {name}") @staticmethod def _chunks(text: str) -> list[str]: return [text[index:index + 96] for index in range(0, len(text), 96)] @staticmethod def _tool_label(name: str) -> str: return { "search_cdsl_library": "检索 CDSL 模型库", "read_cdsl_reference": "读取 CDSL 参考模型", "read_current_cdsl": "读取当前 CDSL", "propose_design_intent": "生成设计意图", "generate_cdsl_model": "生成 CDSL", }.get(name, "调用 CAD 工具") def _attachment_message(self, conversation: dict[str, Any], model: ProviderModel) -> list[dict[str, Any]] | str: attachments = conversation.get("attachments") or [] if not attachments: return "" content: list[dict[str, Any]] = [{"type": "text", "text": "The following local attachments are part of the CAD request."}] for attachment in attachments: if not isinstance(attachment, dict): continue kind = str(attachment.get("kind") or "") task_id = str(attachment.get("task_id") or "") relative = str(attachment.get("path") or "") if not task_id or not relative: continue path = self.store.artifact_path(task_id, relative) if kind == "image": if not model.vision: raise ValueError("The selected model does not support images. Choose a vision-capable OpenAI or Kimi model.") mime = str(attachment.get("mime") or "image/png") encoded = base64.b64encode(path.read_bytes()).decode("ascii") content.append({"type": "image_url", "image_url": {"url": f"data:{mime};base64,{encoded}"}}) elif kind == "document": extracted = str(attachment.get("extracted_path") or "") if extracted: text_path = self.store.artifact_path(task_id, extracted) text = text_path.read_text(encoding="utf-8")[:30_000] content.append({"type": "text", "text": f"Document {attachment.get('name')}:\n{text}"}) return content