sync: WIN10-I9-64G 2026-09-19
This commit is contained in:
@@ -0,0 +1,215 @@
|
||||
---
|
||||
name: kicad-flow
|
||||
description: >
|
||||
从一份需求 md 出发,驱动 KiCad 原理图的完整设计流程:画 drawio 电路框图、跟用户
|
||||
确认功能划分、整理参考资料、必要时做原理仿真、画 KiCad 原理图、校验 FPGA/MCU 的
|
||||
管脚与接口、加封装、回写 drawio、三道独立 agent 复核(一致性/排版/需求符合性),
|
||||
最后导入 PCB。驱动 E:\LangGraph_Design\kicad-flow 里的 LangGraph 工作流(27 节点 /
|
||||
8 个判定回环),进程内 + SQLite 持久化,不用 langgraph dev。图在每个要动 MCP、
|
||||
要派 agent、要人拍板的地方 interrupt 回到 Claude 会话,由本 skill 执行真实动作再
|
||||
回填推进。数据库落在被设计的那个 KiCad 工程下,不在图工程目录下。
|
||||
|
||||
只要用户提到 画原理图、设计一块板、KiCad 原理图设计、从需求做电路、电路框图、
|
||||
kicad-flow 工作流、加封装/封装补齐、原理图排版检查、原理图与框图对不上、
|
||||
原理图导入 PCB、或者说 "启动 kicad-flow"、"跑原理图流程"、"接着上次那块板"、
|
||||
"design a schematic in KiCad"、"start the KiCad schematic flow",就必须用这个
|
||||
skill —— 哪怕他们没说 skill、没提 LangGraph,只说了 "帮我把这块板的原理图画出来"、
|
||||
"这板子画到一半,接着弄"。
|
||||
|
||||
注意区分:把已有的 Altium 工程搬到 KiCad 用 altium-kicad skill,不是这个。
|
||||
---
|
||||
|
||||
# KiCad 原理图设计流程
|
||||
|
||||
## 这个 skill 是什么
|
||||
|
||||
一块板从需求到原理图 = 4 个阶段:**框图与资料 → 画原理图并校验平台 → 加封装 →
|
||||
回写与三道复核 → 导入 PCB**。这些步骤有严格先后、有失败要退回重做的回环、
|
||||
单次会话跑不完(用户要在 drawio、KiCad、Vivado 里手动看很多次)。所以状态由
|
||||
LangGraph 图持久化在 SQLite 里,你负责把图停下来的每一步翻译成真实动作,
|
||||
做完回填结论让它往下走。
|
||||
|
||||
**MCP 边界(必须理解,否则会白折腾)**:kicad、drawio、simetrix、vivado-mcp、
|
||||
cubemx 这些 MCP 跑在**你这个 Claude 会话进程里**,不在 LangGraph 的 Python 进程里。
|
||||
「派单独 agent 复核」同理,只有你能派。所以图不可能自己去画原理图、自己去数封装 ——
|
||||
它只做编排,凡是要碰真实世界的地方一律 `interrupt()` 交回给你。这是设计,不是缺陷。
|
||||
|
||||
## 环境
|
||||
|
||||
| 项 | 值 |
|
||||
|---|---|
|
||||
| 图工程根 | `~\skills_env.md` 的 `KICAD_FLOW_ROOT`,本机 `E:\LangGraph_Design\kicad-flow`(可用 `--graph-root` 覆盖) |
|
||||
| Python | conda env `langgraph`(`~\skills_env.md` 的 `CONDA_LANGGRAPH_PYTHON`) |
|
||||
| checkpoint | **`<被设计的 KiCad 工程>\.kicad-flow\checkpoint.sqlite`** |
|
||||
| 证据/报告 | `<工程>\.kicad-flow\runs\<thread>\` |
|
||||
| thread 命名 | 由工程目录名推导,`E:\Kicad_design\MyBoard_V1p0` → `kf-myboard-v1p0` |
|
||||
| 驱动脚本 | `%USERPROFILE%\.claude\skills\kicad-flow\scripts\kf_flow.py` |
|
||||
|
||||
**数据库不在图工程目录下**,每块板一个库,跟着板子走、跟着板子备份。
|
||||
(altium-kicad 是反的,它的库在图工程根下。别照抄那边的路径。)
|
||||
|
||||
需要的 MCP:**kicad**(主力)、**drawio**(框图)、**simetrix**(仅当要原理仿真)、
|
||||
**vivado-mcp**(仅当板上有 FPGA)、**cubemx**(仅当是 STM32,用来交叉验证 pinout)。
|
||||
|
||||
## 驱动命令
|
||||
|
||||
统一用 `kf_flow.py`。**退出码 2 = 停在 interrupt 等你回填**(不是错误),0 = 图跑完,1 = 出错。
|
||||
|
||||
```powershell
|
||||
$KF = "$env:USERPROFILE\.claude\skills\kicad-flow\scripts\kf_flow.py"
|
||||
# $PY 取 ~\skills_env.md 的 CONDA_LANGGRAPH_PYTHON,别写死本机路径
|
||||
$PY = (Select-String -Path "$env:USERPROFILE\skills_env.md" `
|
||||
-Pattern '^\s*CONDA_LANGGRAPH_PYTHON\s*=\s*(.+?)\s*$').Matches[0].Groups[1].Value
|
||||
|
||||
# 新建 / 续跑(同一个 --prj-dir 重复跑就是续跑,不会重置)
|
||||
& $PY $KF start --prj-dir "E:\Kicad_design\MyBoard"
|
||||
|
||||
# 看现在停在哪
|
||||
& $PY $KF status --prj-dir "E:\Kicad_design\MyBoard"
|
||||
|
||||
# 回填结论、推进
|
||||
& $PY $KF resume --prj-dir "E:\Kicad_design\MyBoard" --set partition_ok=true
|
||||
|
||||
# 列线程 / 清掉某线程重来
|
||||
& $PY $KF list --prj-dir "E:\Kicad_design\MyBoard"
|
||||
& $PY $KF reset --prj-dir "E:\Kicad_design\MyBoard"
|
||||
```
|
||||
|
||||
### 回填一律用 `--set`,不要自己拼 JSON
|
||||
|
||||
`--set key=value` 免转义:`true/false`→bool,纯数字→int,`[`/`{`开头→按 JSON 解析,
|
||||
其余当字符串。**含 Windows 路径的值必须走 `--set`**:
|
||||
|
||||
```powershell
|
||||
& $PY $KF resume --prj-dir "E:\Kicad_design\MyBoard" `
|
||||
--set drawio_path="E:\Kicad_design\MyBoard\Notes\MyBoard.drawio"
|
||||
```
|
||||
|
||||
> ⚠️ **不要绕过 kf_flow.py 去用图工程里的 `run.py '<JSON>'`。** 那条路要把 JSON 当
|
||||
> 位置参数传,Windows 路径里的 `\` 在 shell 和 JSON 两层转义之间必然出事 ——
|
||||
> 实测会报 `Invalid \escape`,只能把反斜杠全写成 `\u005c` 才过得去,纯粹浪费时间。
|
||||
> `--set` 不经过 JSON 解析器,没这个问题。
|
||||
|
||||
值里有空格就整体加引号:`--set "report_path=E:\My Board\report.md"`。
|
||||
一次要填多个键就多写几个 `--set`。确实需要嵌套结构(数组/对象)时用 `--json`,
|
||||
两者可混用。
|
||||
|
||||
`start` 可以调 9 个回环的重试上限,超限走 safety_end:
|
||||
`--max-partition-retry` `--max-material-retry` `--max-layout-retry` `--max-vivado-retry`
|
||||
`--max-mcu-retry` `--max-footprint-retry` `--max-drawio-sync-retry`
|
||||
`--max-recheck-retry` `--max-requirement-retry`。
|
||||
|
||||
> **回填不能是空的**。langgraph 把 falsy 的 resume 当作「没提供」,会在同一个节点再
|
||||
> 中断一次,看起来像卡住。脚本已经帮你拦了,但别指望它替你想该填什么。
|
||||
|
||||
`[INTERRUPT]` 的 JSON 里 `action` 字段告诉你该做哪一步。逐个说见
|
||||
[references/nodes.md](references/nodes.md)。
|
||||
|
||||
## 开工前先问清楚
|
||||
|
||||
**目标 KiCad 工程根目录的绝对路径**(没有默认值,猜错会把库和板子写到别处),
|
||||
以及**需求 md 在哪**(约定在 `<工程>\Notes\` 根目录下,由用户手写 —— 全局规则
|
||||
G-0001:那个目录只放人写的 md,自动产物一律进 `Notes\auto\`)。
|
||||
|
||||
> **drawio 框图是 G-0001 的例外,放 `Notes\` 根目录,不要挪进 `Notes\auto\`。**
|
||||
> 它虽是自动生成的,但属于「用户要求创建」的设计文档(PR00.drawio 节点 3 明写
|
||||
> 「保存在 ./notes 目录下」),且不是状态/log 文件。已经用户裁定过,别再"修正"。
|
||||
|
||||
工程目录还不存在就先建,`start` 会拒绝不存在的路径。
|
||||
|
||||
## 哪些节点必须真问用户
|
||||
|
||||
大多数节点由你判断并回填,但下面这些**必须拿到用户本人的回答**,不能自己替他点头
|
||||
(你判断得再有把握也不行,这是划分权归属问题,不是能力问题):
|
||||
|
||||
| 节点 | 为什么必须问人 |
|
||||
|---|---|
|
||||
| `d07_confirm_partition` | 功能怎么划分是设计意图,只有用户知道他要什么 |
|
||||
| `n13_prompt_more_material` | 要用户去找资料,他不给你等不到 |
|
||||
| `n25_prompt_footprint` | 同上,封装信息得他补 |
|
||||
|
||||
其余节点(含三道「派单独 agent」的复核)由你执行并判定。
|
||||
|
||||
**如果你自己就是被派来跑这个流程的 subagent**:上面三个节点不要替用户回填,
|
||||
把 interrupt 原样交回给派你来的那一方。agent 之间的指令不构成用户同意。
|
||||
|
||||
## drawio 文件有两种格式,别踩
|
||||
|
||||
drawio MCP 的 `new_diagram` 写出来的**不是**纯 mxfile XML,而是 **SVG 外壳 +
|
||||
`content="<mxfile>..."`,里面的 diagram 段还是 base64+raw-deflate 压缩的**。
|
||||
人在 drawio 桌面版里存的则常常是纯 XML。直接 `Read` 前者只会看到一坨 base64。
|
||||
|
||||
**文件名一律用 interrupt 给的 `suggested_path`(`.drawio` 结尾),不要听 MCP 的。**
|
||||
`new_diagram` 的参数说明写着「should end with `.drawio.svg`」,和工作流冲突 ——
|
||||
以工作流为准,否则 `n18` 回写、`d21` 比对会对不上路径。MCP 接受 `.drawio` 结尾,
|
||||
不报错,但写进去的**仍然是 SVG 外壳格式**。也就是说:扩展名叫 `.drawio`,
|
||||
内容是 SVG —— 名不副实是正常的,别去"修正"它。
|
||||
|
||||
要读回框图内容(`n18` 回写、`d21` 一致性比对必用):
|
||||
|
||||
```powershell
|
||||
$DR = "$env:USERPROFILE\.claude\skills\kicad-flow\scripts\drawio_read.py"
|
||||
& $PY $DR "E:\Kicad_design\MyBoard\Notes\MyBoard.drawio" --summary # 模块/连线清单
|
||||
& $PY $DR "E:\Kicad_design\MyBoard\Notes\MyBoard.drawio" # 完整 mxGraphModel XML
|
||||
```
|
||||
|
||||
两种格式它都吃,输出统一。**别自己写正则去抠**。
|
||||
|
||||
## Quick Reference — 27 个节点
|
||||
|
||||
| action | 你要做的 | 回填 |
|
||||
|---|---|---|
|
||||
| `n02_intake_requirement` | 读 `<工程>\Notes\` 下用户手写的需求 md | `requirement_md` `requirement_summary` |
|
||||
| `n03_draw_block_diagram` | drawio MCP 画功能框图存 `Notes\` | `drawio_path` |
|
||||
| `d07_confirm_partition` | **问用户**划分对不对 | `partition_ok` |
|
||||
| `n08_collect_references` | 登记开发板资料/参考设计/datasheet/子电路 md | `references` |
|
||||
| `d04_have_reference_or_sim` | 每个模块都有依据? | `have_reference_or_sim` |
|
||||
| `d05_need_simulation` | 缺依据的靠仿真还是等用户补料? | `need_simulation` |
|
||||
| `n09_run_spice_sim` | simetrix MCP 跑仿真 | `sim_results` |
|
||||
| `n13_prompt_more_material` | **要用户**补材料 | `supplied` |
|
||||
| `n10_draw_schematic` | kicad MCP 画原理图,drawio 插图放第一页 | `schematic_path` |
|
||||
| `d11_layout_issue` | 查排版(标签/字体/器件/线段重叠、越框) | `layout_issue`(**true=有问题**) |
|
||||
| `d12_has_fpga` | 板上有 FPGA? | `has_fpga` |
|
||||
| `n14_build_vivado_probe` | 建验证性 Vivado 工程(管脚/bank/时钟/差分对,Zynq 另查 MIO-EMIO/QSPI/eMMC/JTAG) | `vivado_prj_dir` |
|
||||
| `d15_bitstream_ok` | 出得来 bitstream? | `bitstream_ok` |
|
||||
| `d16_has_mcu` | 板上有 MCU? | `has_mcu` |
|
||||
| `d17_mcu_io_correct` | 对 datasheet 核 SPI/I2C/PWM 专用脚 | `mcu_io_ok`(**false 会退回重画原理图**) |
|
||||
| `n23_assign_footprints` | 加封装(拷参考设计 → 网上下 → 自己画) | `assigned` |
|
||||
| `d24_all_footprints` | **派 agent** 查封装是否齐 | `footprints_complete` `missing` |
|
||||
| `d26_ref_has_footprint` | 参考资料里还找得到封装吗? | `ref_has_footprint` |
|
||||
| `n25_prompt_footprint` | **要用户**补封装 | `supplied` |
|
||||
| `n18_sync_drawio_back` | 按原理图回写 drawio,重贴第一页插图 | `drawio_path` `synced` |
|
||||
| `d21_sch_matches_drawio` | **派 agent** 比原理图与 drawio | `sch_matches_drawio` |
|
||||
| `n19_recheck_layout` | 再修一遍排版 | `fixed` |
|
||||
| `d20_layout_clean` | **派 agent** 复核排版 | `layout_clean`(**true=干净,前进**) |
|
||||
| `d22_meets_requirement` | **派 agent** 比原始需求 | `meets_requirement` `gaps` |
|
||||
| `n98_collect_fixes` | 把差距整理成可执行修改方案 | `fix_plan` |
|
||||
| `n93_schematic_done` | 无 interrupt,图自己出报告 | — |
|
||||
| `n101_import_to_pcb` | `sync_schematic_to_board` 导入 PCB | `pcb_path` `imported` |
|
||||
|
||||
两个**极易填反**的布尔,填反了流程会朝相反方向跑:
|
||||
|
||||
- `d11_layout_issue`:**true = 有排版问题**(回去重排)。
|
||||
- `d20_layout_clean`:**true = 排版干净**(往前走)。
|
||||
|
||||
这俩语义相反是照着 PR00.drawio 原图来的,不是 bug。
|
||||
|
||||
## 常见错误
|
||||
|
||||
| 现象 | 原因 / 处理 |
|
||||
|---|---|
|
||||
| `Invalid \escape` | 你在用 `run.py '<JSON>'` 拼路径。改用 `kf_flow.py resume --set` |
|
||||
| 中文 hint 是乱码 | 你绕过了 kf_flow.py。它自己转 UTF-8,直接跑 run.py 才需要 `PYTHONIOENCODING=utf-8` |
|
||||
| 同一个节点反复中断 | 回填是 falsy(空 dict/`false`/`""`)。至少给一个非空键 |
|
||||
| `Read` drawio 只看到 base64 | SVG 外壳+压缩格式。用 `drawio_read.py` |
|
||||
| `拒绝启动:...会落在本工作流设计目录内` | `--prj-dir` 指到 `E:\LangGraph_Design\kicad-flow` 了。要指目标板子的工程目录 |
|
||||
| `KF_PRJ_DIR 指向的目录不存在` | 先把工程目录建出来 |
|
||||
| `[SAFETY-END]` | 某回环超上限。先解决报告里指出的卡点,再 `start --restart`,或 `--max-*-retry` 调高 |
|
||||
| 改了图工程的代码想验证 | 图工程根下 `python smoke_test.py`,6 项用例覆盖落点/闸门/回环/护栏 |
|
||||
|
||||
## 相关
|
||||
|
||||
- 逐节点的详细执行指引:[references/nodes.md](references/nodes.md)
|
||||
- 把已有 Altium 工程搬到 KiCad:用 **altium-kicad** skill,不是这个
|
||||
- 流程图原图与设计说明:图工程根的 `Notes\PR00.drawio`、
|
||||
`docs\superpowers\specs\2026-09-19-kicad-flow-design.md`
|
||||
@@ -0,0 +1,296 @@
|
||||
# kicad-flow 逐节点执行指引
|
||||
|
||||
`action` 字段对应下面的小节。括号里的数字是 `Notes\PR00.drawio` 的 cell id。
|
||||
回填命令一律省略前缀,完整形式是:
|
||||
|
||||
```powershell
|
||||
& $PY $KF resume --prj-dir "<工程目录>" --set <下面写的>
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 阶段一:框图与资料
|
||||
|
||||
### `n02_intake_requirement`(2)— 读需求
|
||||
|
||||
需求 md 在 `<工程>\Notes\` 根目录下,由用户手写。payload 的 `candidates` 已经把
|
||||
该目录下的 md 列好了;多于一份就问用户哪份是需求。
|
||||
|
||||
读完把要点摘成一段 —— 这段会一路带到 `d22`,作为「是否符合原始需求」的判据,
|
||||
所以要**摘需求本身,不要摘你打算怎么做**。
|
||||
|
||||
回填:`--set requirement_md="<绝对路径>" --set requirement_summary="<摘要>"`
|
||||
|
||||
### `n03_draw_block_diagram`(3)— 画电路框图
|
||||
|
||||
用 drawio MCP(`new_diagram` → `add_nodes` → `link_nodes`)画功能框图,存
|
||||
`<工程>\Notes\`。payload 的 `suggested_path` 给了建议落点。
|
||||
|
||||
框图的用途是**划分出可独立验证的功能模块**,不是画电路图。每个模块块里写清:
|
||||
模块名、关键器件与型号、关键参数、可独立验证的判据。连线上标信号名与电平。
|
||||
|
||||
返工时(`retry > 0`)payload 里有上一轮用户的意见,照着改,别重画一版一样的。
|
||||
|
||||
> 文件名用 payload 的 `suggested_path`(`.drawio`),**不要**听 `new_diagram`
|
||||
> 参数说明里的「should end with `.drawio.svg`」—— 那会让后面 `n18`/`d21` 对不上路径。
|
||||
> 写出来的文件是 SVG 外壳 + 压缩的 mxfile(哪怕扩展名是 `.drawio`),直接 `Read`
|
||||
> 看不了。要核对内容用 `drawio_read.py --summary`。
|
||||
|
||||
回填:`--set drawio_path="<绝对路径>"`
|
||||
|
||||
### `d07_confirm_partition`(7)— 用户确认划分
|
||||
|
||||
**必须问用户本人。** 把每个模块讲一遍(模块名 + 它负责什么 + 怎么独立验证),
|
||||
请他确认划分是否正确。
|
||||
|
||||
用户说不对就 `--set partition_ok=false`,图回到 `n03` 重画(最多 3 次)。
|
||||
把他的具体意见也带上:`--set partition_ok=false --set feedback="电源划分太粗"`。
|
||||
|
||||
回填:`--set partition_ok=true`
|
||||
|
||||
### `n08_collect_references`(8)— 参考资料整理
|
||||
|
||||
逐模块登记四类资料:① 开发板资料 ② 参考设计(KiCad/Altium 工程)③ 器件 datasheet
|
||||
④ 子电路设计描述 md。
|
||||
|
||||
嵌套结构用 `--json`:
|
||||
|
||||
```powershell
|
||||
& $PY $KF resume --prj-dir "<工程>" --json '{"references":[{"module":"PWR","kind":"refdesign","path":"E:\\ref\\foo","note":"LDO 部分可整段照抄"}]}'
|
||||
```
|
||||
|
||||
`kind` 取 `board` / `refdesign` / `datasheet` / `spec`。
|
||||
|
||||
这是回环节点:`n09`(仿真完)和 `n13`(用户补料完)都会回到这里再整理一轮,
|
||||
payload 的 `retry` 和 `sim_results` 告诉你上一轮发生了什么,**把新增的并进来**,
|
||||
不要只报新的。
|
||||
|
||||
### `d04_have_reference_or_sim`(4)— 每个模块都有依据吗
|
||||
|
||||
逐个比对框图模块与已登记资料:每个模块是否都能找到**可直接照抄的参考设计**,
|
||||
或**已有可参考的仿真结果**?只要有一个模块两者皆无就判 false。
|
||||
|
||||
判 true 直接进 `n10` 开画。判 false 进 `d05` 决定走仿真还是要料。
|
||||
|
||||
回填:`--set have_reference_or_sim=true`
|
||||
(判 false 时把缺依据的模块带上:`--json '{"have_reference_or_sim":false,"uncovered":["LPF","ANT"]}'`)
|
||||
|
||||
### `d05_need_simulation`(5)— 靠仿真还是靠补料
|
||||
|
||||
缺依据的模块里,哪些是**仿真能定下来的**(滤波器、环路补偿、阻抗匹配、
|
||||
分压与容差),哪些是**只有用户才知道的**(选型偏好、接口协议、机械约束)?
|
||||
|
||||
前者 `--set need_simulation=true` 走 `n09`,后者 `false` 走 `n13`。
|
||||
|
||||
两类都有就先走仿真(能自己解决的先解决),下一轮回到 `d04` 时再判 false 去要料。
|
||||
|
||||
### `n09_run_spice_sim`(9)— 原理仿真
|
||||
|
||||
用 simetrix MCP:`open_schematic`/`add_component` 建电路 → `set_analysis_lines`
|
||||
设分析 → `run_simulation` → `measure_waveform` 量指标 → `export_waveform_csv` 存证。
|
||||
|
||||
**仿真结论要能直接支撑取值**(「R=10k 时纹波 12mV,满足 <20mV」),
|
||||
不是「仿了,看着还行」。
|
||||
|
||||
回填:`--json '{"sim_results":[{"module":"LPF","conclusion":"fc=1.2kHz,阻带 -42dB","evidence":"<csv 路径>"}]}'`
|
||||
|
||||
做完自动回 `n08`(`material_retry` +1,上限 5)。
|
||||
|
||||
### `n13_prompt_more_material`(13)— 要用户补料
|
||||
|
||||
**必须问用户本人。** 明确说清:哪些模块既没有可照抄的参考设计、也不适合靠仿真定,
|
||||
需要他补哪一类材料。**要具体到模块和材料类型**,别只说「资料不够」。
|
||||
|
||||
用户补完:`--set supplied=true --set note="补了 NTC 的 datasheet 和选型表"`
|
||||
|
||||
做完自动回 `n08`(`material_retry` +1,上限 5)。
|
||||
|
||||
---
|
||||
|
||||
## 阶段二:原理图与平台校验
|
||||
|
||||
### `n10_draw_schematic`(10)— 画 KiCad 原理图
|
||||
|
||||
用 kicad MCP 按框图逐模块画。能照抄的照抄,仿真定的取值照搬。
|
||||
**drawio 框图作为插图放在原理图第一页** —— 这是流程的硬要求,`d21` 会查。
|
||||
|
||||
`batch_add_and_connect` / `batch_add_components` / `batch_connect` 比逐个加快得多。
|
||||
|
||||
这是三条回环的汇合点,payload 会告诉你这轮为什么回来:
|
||||
- `fix_plan` 非空 → `d22` 判了不符合需求,**逐条按方案返工**
|
||||
- `layout_retry` 涨了 → `d11` 判了排版有问题
|
||||
- `mcu_retry` 涨了 → `d17` 判了 MCU IO 不对,管脚要改
|
||||
|
||||
回填:`--set schematic_path="<.kicad_sch 绝对路径>"`
|
||||
|
||||
### `d11_layout_issue`(11)— 排版检查
|
||||
|
||||
三条判据:① 信号标签与线段/字体/器件重叠 ② 字体与器件、线段重叠 ③ 电路落在图框外。
|
||||
|
||||
工具:`find_overlapping_elements`、`find_wires_crossing_symbols`、
|
||||
`find_wires_crossing_symbols`、`get_schematic_view` 逐页看、`get_board_extents` 对框。
|
||||
|
||||
> **这里 true = 有问题**(回 `n10` 重排,上限 5 次)。和后面 `d20` 正好相反,别填反。
|
||||
|
||||
回填:`--set layout_issue=false`
|
||||
|
||||
### `d12_has_fpga`(12)/ `d16_has_mcu`(16)— 板上有没有
|
||||
|
||||
看框图和原理图。有就走对应的平台校验支路,没有就跳过。
|
||||
`--set has_fpga=true --set part="xc7a200t"`
|
||||
|
||||
### `n14_build_vivado_probe`(14)— 建验证性 Vivado 工程
|
||||
|
||||
用 vivado-mcp 建一个**简单但接口完整**的工程,目的是验证原理图上的管脚约定站得住,
|
||||
不是实现功能。必须覆盖:
|
||||
|
||||
1. 管脚分配
|
||||
2. 管脚所在**供电 bank**(IO 电平与 bank 电压是否自洽)
|
||||
3. 时钟输入/输出,以及内部 PLL 的使用要求(时钟必须进时钟专用脚)
|
||||
4. 差分对走线要求
|
||||
|
||||
Zynq 系列另须核对:MIO/EMIO 分配是否合规、QSPI/eMMC/JTAG 配置是否正确。
|
||||
|
||||
`xdc_lint` 和 `verify_io_placement_tool` 先跑一遍,能省一轮实现。
|
||||
|
||||
回填:`--set vivado_prj_dir="<工程目录>"`
|
||||
|
||||
### `d15_bitstream_ok`(15)— 出得来 bitstream 吗
|
||||
|
||||
`run_synthesis` → `run_implementation` → `generate_bitstream`,再看
|
||||
`get_io_report` 和 `get_critical_warnings`。bitstream 出来**且** IO 摆放合法才算过。
|
||||
|
||||
失败回 `n14` 改约束或改管脚(上限 3 次)。把失败原因带上:
|
||||
`--json '{"bitstream_ok":false,"errors":["Bank 34 电平冲突"]}'`
|
||||
|
||||
### `d17_mcu_io_correct`(17)— MCU IO 核对
|
||||
|
||||
对着 datasheet 逐脚核原理图上的 IO:SPI/I2C/PWM 等**专用引脚**是否用在支持该复用的
|
||||
脚上、上拉/开漏要求、电平域、启动引脚(BOOT/NRST)有没有被占用。
|
||||
STM32 可用 cubemx MCP 的 `get_pin_functions` / `get_pinout` 交叉验证。
|
||||
|
||||
> ⚠️ **判 false 会一路退回 `n10` 重画原理图**,FPGA/MCU 校验整段重跑(上限 3 次)。
|
||||
> 这是原图 edge 88 的明确画法,不是 bug。所以判之前把问题一次找全,
|
||||
> 别分两轮触发两次长回环。
|
||||
|
||||
回填:`--set mcu_io_ok=true`
|
||||
|
||||
---
|
||||
|
||||
## 阶段三:封装
|
||||
|
||||
### `n23_assign_footprints`(23)— 加封装
|
||||
|
||||
**动手前先查 bmem 全局记忆里关于封装描述格式的条目** —— 那里记着踩过的坑,
|
||||
格式写错封装挂不上。
|
||||
|
||||
来源优先级(payload 的 `priority` 里也有):
|
||||
1. 从参考设计(KiCad/Altium 工程)直接拷贝
|
||||
2. 按型号或 datasheet 从网上下载并安装
|
||||
3. 按型号或 datasheet 自己画(`create_footprint`)
|
||||
|
||||
`search_footprints` / `list_library_footprints` 找库内现成的,
|
||||
`search_jlcpcb_parts` 可顺带确认可采购性。
|
||||
|
||||
返工轮(`retry > 0`)payload 的 `missing` 就是这轮要处理的器件,别全部重来。
|
||||
|
||||
回填:`--set assigned=12 --set note="9 个从参考工程拷,3 个用标准库"`
|
||||
|
||||
### `d24_all_footprints`(24)— 派 agent 查封装是否齐
|
||||
|
||||
**派一个单独的 agent**,让它独立核查每个器件的 Footprint 字段:是否都已填、
|
||||
指向的库里是否真的存在该封装、引脚数是否与符号对得上。
|
||||
|
||||
独立 agent 是为了不被主线的既有判断带偏 —— 别自己顺手查了就填。
|
||||
|
||||
回填:`--json '{"footprints_complete":false,"missing":[{"ref":"U3","part":"XYZ-123"}]}'`
|
||||
|
||||
### `d26_ref_has_footprint`(26)— 参考资料里还有吗
|
||||
|
||||
针对 `missing` 里的器件再翻一遍参考资料:参考工程的封装库、datasheet 的推荐焊盘图、
|
||||
开发板资料。找得到就 true(回 `n23` 继续加),找不到就 false(去要用户补)。
|
||||
|
||||
每进一次这里 `footprint_retry` +1,上限 5。
|
||||
|
||||
### `n25_prompt_footprint`(25)— 要用户补封装
|
||||
|
||||
**必须问用户本人。** 说清哪些器件既拷不到也按 datasheet 定不了,请他给厂商封装库、
|
||||
推荐焊盘图,或明确指定用哪个标准封装。
|
||||
|
||||
回填:`--set supplied=true --set note="给了 TI 的 QFN-24 封装库"`
|
||||
|
||||
---
|
||||
|
||||
## 阶段四:回写与三道复核
|
||||
|
||||
### `n18_sync_drawio_back`(18)— 按原理图回写 drawio
|
||||
|
||||
原理图画完后,实现常与最初框图有出入。用 drawio MCP 按**当前原理图**更新框图
|
||||
(补上实际用到的器件/接口/信号名,删掉没实现的),然后把更新后的框图
|
||||
**重新贴回原理图第一页**的插图位置。
|
||||
|
||||
先用 `drawio_read.py --summary` 看现有框图内容,再决定改哪里。
|
||||
|
||||
回填:`--set drawio_path="<路径>" --set synced=true`
|
||||
|
||||
### `d21_sch_matches_drawio`(21)— 派 agent 比对一致性
|
||||
|
||||
**派一个单独的 agent**,只给它两份文件(drawio 与原理图),让它判断:
|
||||
框图里每个模块在原理图里都有对应实现?模块间连接关系与信号名对得上?
|
||||
第一页插图是不是最新那版 drawio?
|
||||
|
||||
告诉它用 `drawio_read.py --summary` 读框图,否则它会对着 base64 发呆。
|
||||
|
||||
不一致回 `n18` 重写(上限 3 次):
|
||||
`--json '{"sch_matches_drawio":false,"diffs":["框图有 EEPROM,原理图里没画"]}'`
|
||||
|
||||
### `n19_recheck_layout`(19)— 再修一遍排版
|
||||
|
||||
回写 drawio、贴插图之后排版会再次被打乱。按同样三条判据再过一遍并**动手修**:
|
||||
挪标签、挪文字、把越界电路收回图框,必要时 `autoplace_schematic_fields`。
|
||||
|
||||
回填:`--set fixed=3 --set note="挪了 3 处标签"`
|
||||
|
||||
### `d20_layout_clean`(20)— 派 agent 复核排版
|
||||
|
||||
**派一个单独的 agent** 独立复核排版。
|
||||
|
||||
> **这里 true = 干净、往前走**(和 `d11` 正好相反)。
|
||||
> 原图节点文字写的是「是否有排版问题」,但边标签是「是→前进」,
|
||||
> 二者矛盾,已由用户裁定以边标签为准。填之前想一秒。
|
||||
|
||||
仍有问题回 `n19` 继续修(上限 3 次)。
|
||||
|
||||
回填:`--set layout_clean=true`
|
||||
|
||||
### `d22_meets_requirement`(22)— 派 agent 比对原始需求
|
||||
|
||||
**派一个单独的 agent**,只给它**用户最初的需求 md** 和当前原理图 ——
|
||||
**不要给它中间过程的结论**,避免被带偏。让它逐模块判断:
|
||||
|
||||
- 需求里要的功能是否都实现了?
|
||||
- 实现方式是否满足需求写明的指标/接口/电平/电源约束?
|
||||
- 有没有需求里没要、却擅自加进来的东西?
|
||||
|
||||
不符合会走 `n98` 整理修改方案再回 `n10` 重做(上限 3 次):
|
||||
`--json '{"meets_requirement":false,"gaps":["PWR:需求要过流保护,原理图没做"]}'`
|
||||
|
||||
### `n98_collect_fixes`(98)— 整理修改方案
|
||||
|
||||
把差距整理成**可执行**的方案,每条写清:改哪个模块、改什么、改成什么样、
|
||||
依据是需求里的哪一句。这份方案会带回 `n10` 指导返工。
|
||||
|
||||
回填:`--json '{"fix_plan":[{"module":"PWR","change":"加 PTC + 过流比较器","basis":"需求 §2.3 短路保护"}]}'`
|
||||
|
||||
### `n93_schematic_done`(93)— 无 interrupt
|
||||
|
||||
图自己写完成报告到 `<工程>\.kicad-flow\runs\<thread>\`,直接进 `n101`。
|
||||
|
||||
### `n101_import_to_pcb`(101)— 导入 PCB
|
||||
|
||||
`sync_schematic_to_board` 把原理图导入 PCB(器件带封装落板)。导入后确认器件数与
|
||||
网表数与原理图一致(`get_component_list`、`get_nets_list`)。
|
||||
|
||||
PCB 的布局/布线/DRC **不属于本流程**,导入成功即收工。
|
||||
|
||||
回填:`--set pcb_path="<.kicad_pcb 路径>" --set imported=true`
|
||||
@@ -0,0 +1,112 @@
|
||||
"""drawio_read.py — 把任意 drawio 文件规整成纯 mxGraphModel XML 打到 stdout。
|
||||
|
||||
为什么需要它:drawio MCP 的 new_diagram 写出来的**不是**纯 mxfile XML,而是
|
||||
SVG 外壳 + `content="<mxfile>..."`,里面的 diagram 段还是 base64+raw-deflate
|
||||
压缩的。人手在 drawio 桌面版里存的则常常是纯 XML。两种都要能读,否则
|
||||
n18(按原理图回写 drawio)和 d21(原理图与 drawio 一致性比对)根本没法比。
|
||||
|
||||
用法:
|
||||
python drawio_read.py <drawio 文件> # 打印 mxGraphModel XML
|
||||
python drawio_read.py <drawio 文件> --summary # 只打印模块/连线清单
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import base64
|
||||
import html
|
||||
import re
|
||||
import sys
|
||||
import urllib.parse
|
||||
import zlib
|
||||
from pathlib import Path
|
||||
|
||||
for _s in (sys.stdout, sys.stderr):
|
||||
try:
|
||||
_s.reconfigure(encoding="utf-8", errors="replace")
|
||||
except (AttributeError, ValueError):
|
||||
pass
|
||||
|
||||
|
||||
def _inflate(payload: str) -> str:
|
||||
"""base64 + raw-deflate + urlencode 三层解开(drawio 的压缩 diagram 段)。"""
|
||||
blob = zlib.decompress(base64.b64decode(payload), -15)
|
||||
return urllib.parse.unquote(blob.decode("utf-8"))
|
||||
|
||||
|
||||
def to_xml(path: Path) -> str:
|
||||
"""返回纯 mxGraphModel/mxfile XML,无论源文件是哪种封装。"""
|
||||
raw = path.read_text(encoding="utf-8", errors="replace")
|
||||
|
||||
# 形态 1:SVG 外壳,真内容在 content 属性里
|
||||
if not raw.lstrip().startswith(("<mxfile", "<mxGraphModel")):
|
||||
m = re.search(r'content="([^"]*)"', raw)
|
||||
if not m:
|
||||
raise SystemExit(f"{path} 既不是 mxfile,也找不到 SVG 的 content 属性")
|
||||
raw = html.unescape(m.group(1))
|
||||
|
||||
# 形态 2:diagram 段被压缩
|
||||
d = re.search(r"<diagram[^>]*>\s*([A-Za-z0-9+/=]+)\s*</diagram>", raw)
|
||||
if d:
|
||||
try:
|
||||
return _inflate(d.group(1))
|
||||
except (zlib.error, ValueError) as exc:
|
||||
raise SystemExit(f"{path} 的 diagram 段解压失败:{exc}")
|
||||
|
||||
# 形态 3:已经是明文 XML
|
||||
return raw
|
||||
|
||||
|
||||
_CELL = re.compile(r'<mxCell\b[^>]*?id="([^"]*)"[^>]*?>', re.S)
|
||||
_VAL = re.compile(r'value="([^"]*)"')
|
||||
_SRC = re.compile(r'source="([^"]*)"')
|
||||
_TGT = re.compile(r'target="([^"]*)"')
|
||||
|
||||
|
||||
def summarize(xml: str) -> str:
|
||||
"""列出模块与连线,供 d21 做原理图/框图一致性比对时逐条核。"""
|
||||
names: dict[str, str] = {}
|
||||
edges: list[tuple[str, str, str]] = []
|
||||
for m in re.finditer(r"<mxCell\b.*?(?:/>|</mxCell>)", xml, re.S):
|
||||
tag = m.group(0)
|
||||
cid = re.search(r'id="([^"]*)"', tag)
|
||||
if not cid:
|
||||
continue
|
||||
val = _VAL.search(tag)
|
||||
label = html.unescape(val.group(1)) if val else ""
|
||||
label = re.sub(r"<[^>]+>", " ", label) # 去掉 value 里的 HTML 标签
|
||||
label = re.sub(r"\s*[\r\n]+\s*", " / ", label) # 多行 value 压成一行
|
||||
label = re.sub(r"\s{2,}", " ", label).strip()
|
||||
if 'vertex="1"' in tag:
|
||||
if "edgeLabel" in tag: # 边上的 Yes/No 标签不是模块
|
||||
continue
|
||||
names[cid.group(1)] = label
|
||||
elif 'edge="1"' in tag:
|
||||
s, t = _SRC.search(tag), _TGT.search(tag)
|
||||
edges.append((s.group(1) if s else "?", t.group(1) if t else "?", label))
|
||||
|
||||
def short(cid: str) -> str:
|
||||
"""连线清单里只用模块名首段,否则整行被模块说明撑爆。"""
|
||||
return names.get(cid, cid).split(" / ")[0]
|
||||
|
||||
lines = [f"模块 {len(names)} 个:"]
|
||||
lines += [f" [{k}] {v}" for k, v in names.items()]
|
||||
lines.append(f"连线 {len(edges)} 条:")
|
||||
lines += [f" {short(s)} -> {short(t)}" + (f" ({lbl})" if lbl else "")
|
||||
for s, t, lbl in edges]
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description="把任意 drawio 文件规整成 mxGraphModel XML")
|
||||
ap.add_argument("path", type=Path)
|
||||
ap.add_argument("--summary", action="store_true", help="只打印模块/连线清单")
|
||||
args = ap.parse_args()
|
||||
if not args.path.is_file():
|
||||
raise SystemExit(f"文件不存在:{args.path}")
|
||||
xml = to_xml(args.path)
|
||||
print(summarize(xml) if args.summary else xml)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,361 @@
|
||||
"""kf_flow.py — 驱动 kicad-flow 的 LangGraph 图(进程内,无 langgraph dev)。
|
||||
|
||||
图本体在 ~/skills_env.md 的 KICAD_FLOW_ROOT(app/main_graph.py),本脚本只做三件事:
|
||||
1) 用 SqliteSaver 打开持久化 checkpoint,按 thread_id 续跑;
|
||||
2) 把当前挂起的 interrupt 以结构化 JSON 打印给 Claude 会话;
|
||||
3) 把 Claude 执行完真实动作后的结论 Command(resume=...) 回填、推进到下一个 interrupt。
|
||||
|
||||
**数据库落点**:与 altium-kicad 不同,kicad-flow 的 checkpoint 库落在
|
||||
<被调用的 KiCad 工程>/.kicad-flow/checkpoint.sqlite,不在图工程根下。所以每个板子
|
||||
一个库,`list` 也是按工程列线程。落点由图自己的 app/config.py 闸门强制校验。
|
||||
|
||||
为什么不直接用工程里的 run.py:run.py 的 resume 只接受一个位置参数 JSON 字符串,
|
||||
在 PowerShell 里转义引号极易出错。本脚本用 --set key=value 免转义回填。
|
||||
|
||||
退出码:0 = 图已结束(done);2 = 停在 interrupt 等回填(waiting);1 = 出错。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sqlite3
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Windows 控制台默认 GBK,interrupt 里的中文 hint 会变乱码看不清。自己转 UTF-8,
|
||||
# 别指望调用方每次都记得带 PYTHONIOENCODING。
|
||||
for _s in (sys.stdout, sys.stderr):
|
||||
try:
|
||||
_s.reconfigure(encoding="utf-8", errors="replace")
|
||||
except (AttributeError, ValueError):
|
||||
pass
|
||||
|
||||
|
||||
def _env_lookup(key: str, default: str | None = None) -> str:
|
||||
"""环境变量优先,其次 ~/skills_env.md 的 `KEY = value` 行(机器相关参数的真源)。"""
|
||||
v = os.environ.get(key)
|
||||
if v:
|
||||
return v
|
||||
path = os.path.join(os.path.expanduser("~"), "skills_env.md")
|
||||
try:
|
||||
with open(path, encoding="utf-8") as f:
|
||||
for line in f:
|
||||
m = re.match(rf"\s*{key}\s*=\s*(.+?)(?:\s+#.*)?$", line)
|
||||
if m:
|
||||
return m.group(1).strip()
|
||||
except OSError:
|
||||
pass
|
||||
if default is None:
|
||||
sys.exit(f"missing {key}: add it to ~/skills_env.md or set it as an env var")
|
||||
return default
|
||||
|
||||
|
||||
DEFAULT_GRAPH_ROOT = _env_lookup("KICAD_FLOW_ROOT", r"E:\LangGraph_Design\kicad-flow")
|
||||
|
||||
#: 与 app/state.py 的 max_*_retry 一一对应
|
||||
RETRY_LIMITS = {
|
||||
"partition": 3, "material": 5, "layout": 5, "vivado": 3, "mcu": 3,
|
||||
"footprint": 5, "drawio-sync": 3, "recheck": 3, "requirement": 3,
|
||||
}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- utils
|
||||
def _out(*parts: str) -> None:
|
||||
print(*parts, flush=True)
|
||||
|
||||
|
||||
def _fatal(msg: str) -> None:
|
||||
_out(f"[ERROR] {msg}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def slug_for(prj_dir: str) -> str:
|
||||
"""E:\\Kicad_design\\MyBoard_V1p0 -> kf-myboard-v1p0"""
|
||||
name = Path(prj_dir.rstrip("\\/")).name or "run"
|
||||
name = re.sub(r"[^0-9A-Za-z]+", "-", name).strip("-").lower()
|
||||
return f"kf-{name or 'run'}"
|
||||
|
||||
|
||||
def coerce(raw: str):
|
||||
"""--set 的值做最小类型推断:true/false→bool,纯数字→int,[ 或 { 开头→JSON。"""
|
||||
s = raw.strip()
|
||||
low = s.lower()
|
||||
if low in ("true", "yes"):
|
||||
return True
|
||||
if low in ("false", "no"):
|
||||
return False
|
||||
if re.fullmatch(r"-?\d+", s):
|
||||
return int(s)
|
||||
if s[:1] in ("[", "{"):
|
||||
try:
|
||||
return json.loads(s)
|
||||
except json.JSONDecodeError:
|
||||
return raw
|
||||
return raw
|
||||
|
||||
|
||||
def pending_interrupt(snap):
|
||||
for task in snap.tasks:
|
||||
if task.interrupts:
|
||||
return task.interrupts[0].value
|
||||
return None
|
||||
|
||||
|
||||
def load_graph_root(graph_root: Path):
|
||||
"""把图工程根加进 sys.path,返回 (build_graph, config 模块)。"""
|
||||
if not (graph_root / "app" / "main_graph.py").is_file():
|
||||
_fatal(f"图工程根不对:{graph_root} 下没有 app\\main_graph.py。"
|
||||
f"用 --graph-root 指定,设环境变量 KICAD_FLOW_ROOT,"
|
||||
f"或改 ~/skills_env.md 的 KICAD_FLOW_ROOT。")
|
||||
sys.path.insert(0, str(graph_root))
|
||||
from app.main_graph import build_graph # noqa: E402
|
||||
from app import config # noqa: E402
|
||||
return build_graph, config
|
||||
|
||||
|
||||
def init_state(layout: dict, thread_id: str, limits: dict[str, int]) -> dict:
|
||||
state = {**layout, "thread_id": thread_id,
|
||||
"references": [], "sim_results": [], "history": []}
|
||||
for name, default in RETRY_LIMITS.items():
|
||||
key = name.replace("-", "_")
|
||||
state[f"{key}_retry"] = 0
|
||||
state[f"max_{key}_retry"] = limits.get(name) or default
|
||||
return state
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- report
|
||||
def report(graph, thread: dict, before_len: int) -> int:
|
||||
"""打印当前状态;返回进程退出码。"""
|
||||
snap = graph.get_state(thread)
|
||||
values = snap.values or {}
|
||||
history = values.get("history", [])
|
||||
walked = [h.get("node") for h in history[before_len:]]
|
||||
if walked:
|
||||
_out("[NODES] " + " -> ".join(str(n) for n in walked))
|
||||
|
||||
_out(f"[THREAD] {thread['configurable']['thread_id']}")
|
||||
if values.get("prj_dir"):
|
||||
_out(f"[PATHS] prj_dir = {values['prj_dir']}")
|
||||
_out(f"[PATHS] db = {values.get('flow_home', '')}\\checkpoint.sqlite")
|
||||
_out(f"[PATHS] evidence = {values.get('runs_dir', '')}")
|
||||
for key, label in (("drawio_path", "drawio"), ("schematic_path", "sch"),
|
||||
("pcb_path", "pcb")):
|
||||
if values.get(key):
|
||||
_out(f"[FILES] {label:8s} = {values[key]}")
|
||||
|
||||
retries = {k: v for k, v in values.items()
|
||||
if k.endswith("_retry") and not k.startswith("max_") and v}
|
||||
if retries:
|
||||
_out("[RETRY] " + json.dumps(retries))
|
||||
|
||||
if snap.next:
|
||||
payload = pending_interrupt(snap)
|
||||
_out(f"[WAITING] next node: {', '.join(snap.next)}")
|
||||
if payload is not None:
|
||||
_out("[INTERRUPT] " + json.dumps(payload, ensure_ascii=False, indent=2))
|
||||
else:
|
||||
_out("[INTERRUPT] (无 payload — 直接 resume 推进)")
|
||||
return 2
|
||||
|
||||
if values.get("halted"):
|
||||
_out(f"[SAFETY-END] {values.get('halted_reason', '')}")
|
||||
_out(f"[REPORT] {values.get('report_path', '')}")
|
||||
_out("先解决这个卡点,再用 `start --restart` 重跑,或用 --max-*-retry 调高上限。")
|
||||
return 0
|
||||
|
||||
_out("[DONE] 图已跑完:原理图设计完成并已导入 PCB。")
|
||||
_out(f"[REPORT] {values.get('report_path', '(未生成)')}")
|
||||
_out(f"[PCB] {values.get('pcb_path', '(未回填)')}")
|
||||
return 0
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- commands
|
||||
def _open(args):
|
||||
"""解析工程路径 + 落点,返回 (build_graph, db_path, thread, layout)。"""
|
||||
build_graph, config = load_graph_root(Path(args.graph_root))
|
||||
prj_raw = getattr(args, "prj_dir", None)
|
||||
if not prj_raw:
|
||||
_fatal("缺少 --prj-dir(目标 KiCad 工程根目录)。kicad-flow 的数据库就建在它下面,"
|
||||
"没有默认值,猜错会把库写到别处。")
|
||||
try:
|
||||
prj = config.resolve_project_dir(prj_raw)
|
||||
thread_id = args.thread or slug_for(str(prj))
|
||||
layout = config.prepare_project_layout(prj, thread_id)
|
||||
except config.ProjectDirError as exc:
|
||||
_fatal(str(exc))
|
||||
return build_graph, Path(layout["db_path"]), \
|
||||
{"configurable": {"thread_id": thread_id}}, layout
|
||||
|
||||
|
||||
def cmd_start(args):
|
||||
from langgraph.checkpoint.sqlite import SqliteSaver
|
||||
build_graph, db_path, thread, layout = _open(args)
|
||||
|
||||
if args.restart:
|
||||
n = wipe_thread(db_path, thread["configurable"]["thread_id"])
|
||||
_out(f"[RESTART] 清掉 {n} 条 checkpoint 记录")
|
||||
|
||||
limits = {name: getattr(args, f"max_{name.replace('-', '_')}_retry", None)
|
||||
for name in RETRY_LIMITS}
|
||||
with SqliteSaver.from_conn_string(str(db_path)) as cp:
|
||||
graph = build_graph(checkpointer=cp)
|
||||
snap = graph.get_state(thread)
|
||||
before = len((snap.values or {}).get("history", []))
|
||||
if snap.next or snap.values:
|
||||
_out("[RESUME] 该工程已有进行中的线程,直接续跑(不重置)")
|
||||
graph.invoke(None, thread)
|
||||
else:
|
||||
graph.invoke(init_state(layout, thread["configurable"]["thread_id"], limits),
|
||||
thread)
|
||||
return report(graph, thread, before)
|
||||
|
||||
|
||||
def cmd_status(args):
|
||||
from langgraph.checkpoint.sqlite import SqliteSaver
|
||||
build_graph, db_path, thread, _ = _open(args)
|
||||
with SqliteSaver.from_conn_string(str(db_path)) as cp:
|
||||
graph = build_graph(checkpointer=cp)
|
||||
snap = graph.get_state(thread)
|
||||
return report(graph, thread, len((snap.values or {}).get("history", [])))
|
||||
|
||||
|
||||
def cmd_resume(args):
|
||||
from langgraph.checkpoint.sqlite import SqliteSaver
|
||||
from langgraph.types import Command
|
||||
build_graph, db_path, thread, _ = _open(args)
|
||||
|
||||
payload: dict = {}
|
||||
if args.json:
|
||||
try:
|
||||
parsed = json.loads(args.json)
|
||||
except json.JSONDecodeError as exc:
|
||||
_fatal(f"--json 不是合法 JSON:{exc}")
|
||||
if not isinstance(parsed, dict):
|
||||
_fatal("--json 必须是一个对象")
|
||||
payload.update(parsed)
|
||||
for item in args.set or []:
|
||||
if "=" not in item:
|
||||
_fatal(f"--set 要写成 KEY=VALUE,收到:{item}")
|
||||
k, v = item.split("=", 1)
|
||||
payload[k.strip()] = coerce(v)
|
||||
|
||||
if not payload:
|
||||
_fatal("回填内容不能为空 —— langgraph 把 falsy 的 resume 当作『没提供』,"
|
||||
"会在同一个节点再中断一次,看起来像卡住。至少给一个键。")
|
||||
|
||||
with SqliteSaver.from_conn_string(str(db_path)) as cp:
|
||||
graph = build_graph(checkpointer=cp)
|
||||
snap = graph.get_state(thread)
|
||||
if not snap.next:
|
||||
_fatal("当前没有挂起的 interrupt,无需回填。先跑 status 看状态。")
|
||||
before = len((snap.values or {}).get("history", []))
|
||||
_out(f"[RESUME] {json.dumps(payload, ensure_ascii=False)}")
|
||||
graph.invoke(Command(resume=payload), thread)
|
||||
return report(graph, thread, before)
|
||||
|
||||
|
||||
def cmd_list(args):
|
||||
"""列出某个工程库里的所有线程(DB 按工程隔离,所以要给 --prj-dir)。"""
|
||||
_, config = load_graph_root(Path(args.graph_root))
|
||||
try:
|
||||
prj = config.resolve_project_dir(args.prj_dir)
|
||||
except config.ProjectDirError as exc:
|
||||
_fatal(str(exc))
|
||||
db = config.db_path(prj)
|
||||
if not db.is_file():
|
||||
_out(f"[EMPTY] {prj} 下还没有 kicad-flow 库({db} 不存在)")
|
||||
return 0
|
||||
con = sqlite3.connect(str(db))
|
||||
try:
|
||||
rows = con.execute(
|
||||
"SELECT thread_id, COUNT(*) FROM checkpoints GROUP BY thread_id"
|
||||
).fetchall()
|
||||
except sqlite3.DatabaseError as exc:
|
||||
_fatal(f"读 {db} 失败:{exc}")
|
||||
finally:
|
||||
con.close()
|
||||
if not rows:
|
||||
_out("[EMPTY] 库里没有线程")
|
||||
return 0
|
||||
_out(f"[DB] {db}")
|
||||
for tid, n in rows:
|
||||
_out(f" {tid:<28} {n} 条 checkpoint")
|
||||
return 0
|
||||
|
||||
|
||||
def wipe_thread(db_path: Path, thread_id: str) -> int:
|
||||
if not db_path.is_file():
|
||||
return 0
|
||||
con = sqlite3.connect(str(db_path))
|
||||
total = 0
|
||||
try:
|
||||
for table in ("checkpoints", "writes"):
|
||||
try:
|
||||
cur = con.execute(f"DELETE FROM {table} WHERE thread_id = ?", (thread_id,))
|
||||
total += cur.rowcount or 0
|
||||
except sqlite3.DatabaseError:
|
||||
pass
|
||||
con.commit()
|
||||
finally:
|
||||
con.close()
|
||||
return total
|
||||
|
||||
|
||||
def cmd_reset(args):
|
||||
_, config = load_graph_root(Path(args.graph_root))
|
||||
try:
|
||||
prj = config.resolve_project_dir(args.prj_dir)
|
||||
except config.ProjectDirError as exc:
|
||||
_fatal(str(exc))
|
||||
thread_id = args.thread or slug_for(str(prj))
|
||||
n = wipe_thread(config.db_path(prj), thread_id)
|
||||
_out(f"[RESET] {thread_id}:清掉 {n} 条记录")
|
||||
return 0
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- main
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(prog="kf_flow", description="驱动 kicad-flow 的 LangGraph 图")
|
||||
ap.add_argument("--graph-root", default=os.environ.get("KICAD_FLOW_ROOT", DEFAULT_GRAPH_ROOT),
|
||||
help="图工程根(默认取 ~/skills_env.md 的 KICAD_FLOW_ROOT)")
|
||||
sub = ap.add_subparsers(dest="cmd", required=True)
|
||||
|
||||
p = sub.add_parser("start", help="新建或续跑某个板子的原理图设计流程")
|
||||
p.add_argument("--prj-dir", help="目标 KiCad 工程根目录(数据库就建在它下面)")
|
||||
p.add_argument("--thread", help="线程 id(默认由 prj-dir 目录名推导)")
|
||||
p.add_argument("--restart", action="store_true", help="清空该线程重新开始")
|
||||
for name, default in RETRY_LIMITS.items():
|
||||
p.add_argument(f"--max-{name}-retry", type=int,
|
||||
help=f"{name} 回环重试上限(默认 {default})")
|
||||
p.set_defaults(func=cmd_start)
|
||||
|
||||
p = sub.add_parser("status", help="看当前停在哪个 interrupt")
|
||||
p.add_argument("--prj-dir")
|
||||
p.add_argument("--thread")
|
||||
p.set_defaults(func=cmd_status)
|
||||
|
||||
p = sub.add_parser("resume", help="回填结论并推进到下一个 interrupt")
|
||||
p.add_argument("--prj-dir")
|
||||
p.add_argument("--thread")
|
||||
p.add_argument("--set", action="append", metavar="KEY=VALUE",
|
||||
help="免转义回填;true/false→bool,纯数字→int,[ 或 { 开头→JSON")
|
||||
p.add_argument("--json", help="直接给 JSON 对象(与 --set 可混用)")
|
||||
p.set_defaults(func=cmd_resume)
|
||||
|
||||
p = sub.add_parser("list", help="列出该工程库里的所有线程")
|
||||
p.add_argument("--prj-dir")
|
||||
p.set_defaults(func=cmd_list)
|
||||
|
||||
p = sub.add_parser("reset", help="删除某个线程的 checkpoint")
|
||||
p.add_argument("--prj-dir")
|
||||
p.add_argument("--thread")
|
||||
p.set_defaults(func=cmd_reset)
|
||||
|
||||
args = ap.parse_args()
|
||||
return args.func(args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user