| """std/intent-planner — the agent-class module (L5 intent-driven editing). |
| |
| Takes a natural-language intent + an asset manifest (.asset.json) and emits an |
| executable edit DAG scoped to the asset's modalities. The planner is pluggable: |
| rule_based_planner is the deterministic default (and what tests use); a real LLM |
| (Hermes harness, any open model) plugs in as llm(ctx) -> plan-dict without any |
| other change. Emitted plans run directly through the standard DAG runner — |
| including human-gate nodes for the steps the user hand-edits in Blender etc. |
| """ |
| from __future__ import annotations |
|
|
| import json |
| import time |
|
|
| from m3d.module import M3DModule |
| from m3d.store import Store |
|
|
| MANIFEST = { |
| "id": "std/intent-planner", |
| "version": "0.1.0", |
| "class": "agent", |
| "modalities": ["intent"], |
| "runtime": {"type": "local-process", "ref": "python:m3d.modules.intent_planner"}, |
| "interface": {"tools": ["plan"]}, |
| "io": { |
| "inputs": [{"name": "intent", "kind": "text"}, {"name": "manifest_uri", "kind": "uri:asset-manifest"}], |
| "outputs": [{"name": "plan", "kind": "workflow"}], |
| }, |
| "license": {"code": "MIT", "model_licenses": [], "usage_profiles": ["personal", "commercial", "redistribution"]}, |
| "health": {"probe": "in-process", "fallbacks": []}, |
| } |
|
|
|
|
| |
|
|
| def rule_based_planner(ctx: dict) -> dict: |
| """Deterministic intent->edit-pathway planner. Maps intent keywords against |
| the manifest's parts; missing-modality intents raise instead of inventing nodes. |
| Swap for a real open LLM call without changing the contract: llm(ctx) -> plan.""" |
| intent: str = ctx["intent"].lower() |
| parts = {p["role"]: p["uri"] for p in ctx.get("manifest", {}).get("parts", [])} |
| nodes, edges = [], [] |
| producers = [] |
|
|
| wants_texture = any(w in intent for w in ("texture", "color", "colour", "material", "paint", "albedo", "tatter")) |
| wants_mesh = any(w in intent for w in ("sculpt", "reshape", "tatter", "mesh", "geometry", "holes")) |
| wants_audio = any(w in intent for w in ("sound", "audio", "footstep", "sfx", "music", "voice", "heavier")) |
|
|
| if wants_texture: |
| if "texture/albedo" not in parts: |
| raise ValueError("intent needs a texture edit but asset manifest has no texture/albedo part") |
| nodes.append({ |
| "id": "tex", |
| "module": "std/mock-texture", |
| "tool": "repaint", |
| "inputs": {"uri": parts["texture/albedo"], "style": "tattered" if "tatter" in intent else "repaint"}, |
| }) |
| producers.append("tex") |
| if wants_mesh: |
| if "mesh" not in parts: |
| raise ValueError("intent needs a mesh edit but asset manifest has no mesh part") |
| nodes.append({ |
| "id": "mesh-edit", |
| "type": "human-gate", |
| "offer": ["approve", "edit-externally", "regenerate"], |
| "recipe": {"editor": "blender", "hint": f"edit {parts['mesh']} per intent"}, |
| }) |
| producers.append("mesh-edit") |
| if wants_audio: |
| if "audio" not in parts: |
| raise ValueError("intent needs an audio edit but asset manifest has no audio part") |
| nodes.append({ |
| "id": "sfx", |
| "module": "std/mock-sfx", |
| "tool": "process", |
| "inputs": {"uri": parts["audio"], "intensity": 1.5}, |
| }) |
| producers.append("sfx") |
|
|
| if not producers: |
| raise ValueError(f"no modality matched intent: {intent!r}") |
|
|
| nodes.append({"id": "review", "type": "human-gate", "offer": ["approve", "edit-externally", "regenerate"]}) |
| nodes.append({ |
| "id": "post", |
| "module": "std/gltf-mcp", |
| "tool": "inspect", |
| "inputs": {"uri": parts.get("mesh", "asset://missing")}, |
| }) |
| for p in producers: |
| edges.append(f"{p}->review") |
| edges.append("review[approve]->post") |
|
|
| return { |
| "id": f"intent-{int(time.time())}", |
| "version": "0.1.0", |
| "license": "CC-BY-4.0", |
| "nodes": nodes, |
| "edges": edges, |
| "report": { |
| "planner": "rule_based_planner", |
| "modalities_touched": [p.rsplit("-", 1)[0] for p in producers], |
| "untouched": [r for r in parts if r not in ("texture/albedo", "mesh", "audio")], |
| }, |
| } |
|
|
|
|
| |
|
|
| def build_module(store: Store, llm=rule_based_planner) -> M3DModule: |
| m = M3DModule(MANIFEST) |
|
|
| @m.tool() |
| def plan(intent: str, manifest_uri: str) -> dict: |
| """Turn a natural-language intent + asset manifest into an executable edit DAG.""" |
| path = store.resolve(manifest_uri) |
| manifest = json.loads(path.read_text()) |
| ctx = {"intent": intent, "manifest": manifest} |
| plan_wf = llm(ctx) |
| if not isinstance(plan_wf, dict) or "nodes" not in plan_wf or "edges" not in plan_wf: |
| return {"ok": False, "error": f"planner returned non-workflow: {plan_wf!r}"} |
| return { |
| "ok": True, |
| "plan": plan_wf, |
| "report": plan_wf.get("report", {}), |
| } |
|
|
| return m |
