更新了skill

2026-06-22 16:41:44 +08:00
parent 48b965ec4e
commit 814dbf1681
13 changed files with 845 additions and 13 deletions
@@ -0,0 +1,123 @@
+"""In-process driver for the fpga-imx6-flow LangGraph workflow.
+
+Builds the compiled graph from the project's app package and drives it with a
+persistent SqliteSaver — no `langgraph dev` server. Mirrors the fpga-flow helper
+convention: exit code 0 = done, 1 = error, 2 = interrupt fired.
+
+Commands:
+    start   --project-dir <dir> [--program-mode bitstream|flash] [--bit-path ..]
+            [--flash-path ..] [--thread-id ..] [--real] [--max-iterations N]
+    resume  --project-dir <dir> --resume-json '<json>' [--thread-id ..] [--real]
+    status  --project-dir <dir> [--thread-id ..]
+
+Output: one JSON line on stdout describing status + the pending interrupt (if any).
+Run via the env python directly or `conda run --no-capture-output` to avoid the
+Windows GBK console re-encode crash on Chinese interrupt payloads.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+from pathlib import Path
+
+try:
+    sys.stdout.reconfigure(encoding="utf-8")  # emit UTF-8 even on a GBK console
+except Exception:
+    pass
+
+DEFAULT_PROJECT = r"E:\LangGraph_Design\FPGA-imx6-flow"
+
+
+def _emit(obj: dict, code: int) -> None:
+    print(json.dumps(obj, ensure_ascii=False))
+    sys.exit(code)
+
+
+def _pending_interrupt(snap):
+    for task in snap.tasks:
+        if task.interrupts:
+            return task.interrupts[0].value
+    return None
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description="Drive the fpga-imx6-flow graph in-process.")
+    ap.add_argument("command", choices=["start", "resume", "status"])
+    ap.add_argument("--project-dir", default=DEFAULT_PROJECT)
+    ap.add_argument("--thread-id", default="imx6-skill-1")
+    ap.add_argument("--resume-json", default="")
+    ap.add_argument("--program-mode", default="bitstream", choices=["bitstream", "flash"])
+    ap.add_argument("--bit-path", default="")
+    ap.add_argument("--flash-path", default="")
+    ap.add_argument("--real", action="store_true",
+                    help="drive real hardware (workflow tools_* must be wired; default is MOCK dry-run)")
+    ap.add_argument("--max-iterations", type=int, default=8)
+    ap.add_argument("--max-hw-retry", type=int, default=3)
+    args = ap.parse_args()
+
+    project = Path(args.project_dir)
+    if not (project / "app" / "main_graph.py").exists():
+        _emit({"status": "error",
+               "error": f"app/main_graph.py not found under {project}; check --project-dir"}, 1)
+
+    # MOCK by default so the design loop can be dry-run with no hardware attached.
+    os.environ["FPGA_IMX6_MOCK"] = "0" if args.real else "1"
+    sys.path.insert(0, str(project))
+
+    try:
+        from langgraph.checkpoint.sqlite import SqliteSaver
+        from langgraph.types import Command
+        from app.main_graph import build_graph
+    except Exception as exc:  # noqa: BLE001
+        _emit({"status": "error",
+               "error": f"import failed: {exc}. Run /langgraph-vivado-install to set up the env."}, 1)
+
+    runs = project / "runs"
+    runs.mkdir(exist_ok=True)
+    db = str(runs / "skill_checkpoint.sqlite")
+    cfg = {"configurable": {"thread_id": args.thread_id}}
+
+    with SqliteSaver.from_conn_string(db) as cp:
+        graph = build_graph(checkpointer=cp)
+        try:
+            if args.command == "start":
+                init = {
+                    "iteration": 0,
+                    "program_mode": args.program_mode,
+                    "bit_path": args.bit_path,
+                    "flash_path": args.flash_path,
+                    "max_hw_retry": args.max_hw_retry,
+                    "max_iterations": args.max_iterations,
+                    "history": [],
+                }
+                graph.invoke(init, cfg)
+            elif args.command == "resume":
+                payload = json.loads(args.resume_json) if args.resume_json else {}
+                graph.invoke(Command(resume=payload), cfg)
+            # status: no invoke, just snapshot
+            snap = graph.get_state(cfg)
+        except Exception as exc:  # noqa: BLE001
+            _emit({"status": "error", "error": str(exc)}, 1)
+
+        if snap.next:
+            _emit({
+                "status": "interrupt",
+                "thread_id": args.thread_id,
+                "mock": not args.real,
+                "next": list(snap.next),
+                "iteration": snap.values.get("iteration"),
+                "interrupt": _pending_interrupt(snap),
+            }, 2)
+        else:
+            _emit({
+                "status": "done",
+                "thread_id": args.thread_id,
+                "iteration": snap.values.get("iteration"),
+                "steps": len(snap.values.get("history", [])),
+            }, 0)
+
+
+if __name__ == "__main__":
+    main()