加"技能执行"示例驱动: 把 skill 真跑一遍 (选模型 / 拼 prompt / 校验输出 / 重试)

- 驱动/技能执行/: oneshot + system 解释器. 读自带 技能/*.json -> 按 tier 选模型 ->
  拼 prompt -> 调 OpenAI 兼容端点 -> 剥 JSON -> 逐条按档校验 -> 落盘 + 写 events
  main 档只锁外层 (plan 是数组 + 每步有 want); small 档锁死 (单值 want + 必须在契约
  清单里 + args 键白名单 + confidence 只许 0/0.5/1)
- 契约清单只读 PG 的 drivers 表拿 (不 import 别的驱动、不翻别人的文件夹);
  也支持 EFI_SKILL_CONTRACTS 显式给一份 (手工跑 / PG 不在时)
- 思考型模型把 max_tokens 吃光时, 把预算翻倍重试一次; 本机端点强制不走代理
- 驱动/Json解码: 补 provides ["Json解码:解析"] -- skill 样例里一直在用这个契约名, 之前没人提供
- 文档同步: README / 文档 02 / 文档 06 / 文档 10 (坑 27~30) / 设计 01 / 设计 05 (补记执行侧)
- .gitignore: 盖住 驱动/*/输出/ (运行时产物不进 git)
This commit is contained in:
lou
2026-09-17 12:39:10 +08:00
parent 8fd8c8411e
commit 2155f65c76
12 changed files with 758 additions and 9 deletions
+1
View File
@@ -3,6 +3,7 @@
运行.efi.json
*/运行.efi.json
**/logs/
**/输出/
__pycache__/
*.pyc
.venv/
+2 -2
View File
@@ -14,7 +14,7 @@
| 引导器 | `UEFI.boot.py` | 体检 6 项 / venv / 包台账 / 内核进程启停记账 / 参数原样透传 |
| 内核 | `内核/` | 常驻总调度:命令消费、驱动契约仲裁(六条)、依赖链巡检、收尸、独一份调度锁 |
| 驱动 | `驱动/<名>/` | **一个文件夹 = 一个驱动**(根目录有 `配置.efi.json` 才认);只经 PG 表与外界说话 |
| 程序(Skill) | 驱动内的 `技能/*.json` | 声明式、模型驱动,**归驱动管**(内核不掺和);样例 `驱动/样板技能/`,规范 `设计/05` |
| 程序(Skill) | 驱动内的 `技能/*.json` | 声明式、模型驱动,**归驱动管**(内核不掺和);只列不执行的样例 `驱动/样板技能/`,真跑一遍的样例 `驱动/技能执行/`,规范 `设计/05` |
驱动之间**零耦合**:不 import、不互调、配置里也不写对方名字 —— 只声明"我要什么 / 我产出什么"(契约名),
谁给、什么顺序、怎么送达全归内核。
@@ -43,7 +43,7 @@ python3 UEFI.boot.py 内核 停止 # 收工 (驱动不动, 下
| `内核/` | 内核本体:`内核.py` 常驻调度 + `扫描.py` 校验契约 + `状态.py` 状态机快照 + `db.py`(**唯一碰 SQL**)+ `进程.py` + `日志.py` + `文本.py` |
| `内核/自测*.py` | 五份**真机**自测(进程 / 内核 / 配置 / db / 日志) |
| `试跑引导器.py` | 端到端验收(11 项,含残留检查) |
| `驱动/` | 四个样板:`Json解码`(oneshot) / `样板常驻`(resident + 提供服务) / `样例消费器`(消费契约) / `样板技能`(skill 归驱动管,main+small 两档) |
| `驱动/` | 五个:`Json解码`(oneshot,提供 `Json解码:解析`) / `样板常驻`(resident + 提供服务) / `样例消费器`(消费契约) / `样板技能`(skill 归驱动管,main+small 两档,只列不执行) / `技能执行`(**示例驱动**:把 skill 真跑一遍 —— 选模型 / 拼 prompt / 校验输出) |
| `设计/` | 设计档案 `01-驱动规范` ~ `05-技能规范`(字段表、DDL、判定表的原文在这里) |
| `文档/` | 使用与开发文档 `00-索引` ~ `10-验收与质量门` |
| `归档/` | 修前的证据存档(如日志系统重做前的原始 `内核.log`) |
+6 -2
View File
@@ -1,8 +1,9 @@
# 02 · 写一个驱动
> 驱动 = **一个文件夹**。内核认不认它只看一件事:根目录有没有 `配置.efi.json`。
> 字段表原文在 `设计/01-驱动规范.md` 第 3 节(那份是规范,这份是**动手做**)。本机已有三个真样板:
> `驱动/Json解码`(oneshot)、`驱动/样板常驻`(resident + 提供服务)、`驱动/样例消费器`(resident + 消费服务)。
> 字段表原文在 `设计/01-驱动规范.md` 第 3 节(那份是规范,这份是**动手做**)。本机已有五个真样板:
> `驱动/Json解码`(oneshot)、`驱动/样板常驻`(resident + 提供服务)、`驱动/样例消费器`(resident + 消费服务)、
> `驱动/样板技能`(带 skill,只列不执行)、`驱动/技能执行`(**示例驱动**:把 skill 真跑一遍)。
## 1. 五分钟:一个最小驱动(oneshot)
@@ -196,6 +197,9 @@ def 汇报(连接, message: str, kind: str = "heartbeat") -> None:
驱动怎么用:启动时读自己目录下的 `技能/`,按 `tier` 选模型、拼 prompt、校验输出、失败重试 —— **全在你手里**。
`驱动/样板技能/技能.py` 是最小示范(只把自带的 skill 列出来打印,不执行)。
要**真跑**(按档选模型 → 拼 prompt → 调模型 → 剥 JSON → 逐条校验输出 → 落盘 + 写 `events`)看
`驱动/技能执行/执行.py` —— 它把同一份 small 档 skill 真喂给本地模型跑了一遍,两个反例(多步请求、信息不足)
都会按 skill 写的规矩回问而不是硬猜。
## 9. 交付一个驱动前的检查表
+1 -1
View File
@@ -125,4 +125,4 @@
| 调用结果没有大小限制 | `result` 是 jsonb;大对象该走"提供方写自己的表 + result 只放引用" | 出现大结果时 |
| 依赖巡检 10s 粒度 | 崩了最多 10s 才被发现/重拉 | 需要更快时(但会更吵) |
| 单 PG 单库 | 没有分片/跨库 | 不在本版范围 |
| 程序/Skill 层未加载 | skill **归驱动管**:声明在 `驱动/<名>/技能/*.json`(两档样例 `驱动/样板技能/`,规范 `设计/05`),驱动自己读自己解释;内核不认识它 | v0.2 才定"谁执行" |
| 程序/Skill 层由驱动执行 | skill **归驱动管**:声明在 `驱动/<名>/技能/*.json`(两档样例 `驱动/样板技能/`,规范 `设计/05`),驱动自己读自己解释;内核不认识它 | 执行样例见 `驱动/技能执行/`(2026-09-17 落地:选模型 / 拼 prompt / 逐条校验 / 预算不够重试一次) |
+4
View File
@@ -114,5 +114,9 @@ python3 内核/自测AST等价.py /tmp/基准.py 内核/内核.py
| 24 | "上游只是没启动"被算成下游"依赖失效" → 天天假警 | 只有上游 `崩了`/`失败` 才算(`停止` 不算) |
| 25 | 自测想"行在就取某列"写成 `行["x"] if 行 else None` → 被判 `reportOptionalSubscript` | 统一走 `值(行, "列名")` 助手 |
| 26 | 一组抛异常拖停整个自测 | `跑组(名称, 动作)` 每组装 try/except + `traceback`,末了出汇总表 |
| 27 | 思考型模型(DeepSeek / Qwen3 系)把 `max_tokens` 全用在思考上 → `content` 回来是**空串**,看着像"模型没答"(2026-09-17 实测:9B 思考 4000+ token 还没完,82s 空手而归) | 驱动把预算**翻倍重试一次**(示例驱动已做);`/no_think` 与 `chat_template_kwargs: {enable_thinking:false}` 只在认它的模型上有效,别指望 |
| 28 | `EFI_LLM_BASE` 指本机(`127.0.0.1`)时,本机 Clash 的 `http_proxy` 会把请求也劫走 → 连不上模型 | 本机端点强制不走代理(`build_opener(ProxyHandler({}))`) |
| 29 | 8G 卡装 7.95GB 的 GLM → `lms load` 报 `SIGABRT`(引擎起来就死,不是 OOM 提示) | `--gpu 0.95` 部分 offload + 用 `-c` 收住上下文;加载前先 `nvidia-smi` 看余量 |
| 30 | 模型输出常带 ` ```json ` 围栏或前后废话 | 剥 JSON 是**驱动**的活(示例驱动 `剥JSON()`:先剥围栏,再退到第一个 `{` 到最后一个 `}`) |
> 新踩到的坑:**先修代码,再补进这张表**(以及技能里的坑表)—— 坑表是这套底座最贵的资产之一。
+1 -1
View File
@@ -218,7 +218,7 @@ __pycache__/
- **驱动 = 确定性代码能力接入层** → 本文件写的这一层,有 `配置.efi.json` 才是驱动。
- **程序(Skill) = 声明式、模型驱动,且归驱动管** → `web.json`({html, css, js})是最早的雏形样本,由
`Json解码` 这类驱动负责解释执行;skill 现在落在 `驱动/<名>/技能/*.json`,分主模型 / 小模型两档,
**内核不认识它**(字段表与两档差别见 `05-技能规范.md`,样例见 `驱动/样板技能/`)。
**内核不认识它**(字段表与两档差别见 `05-技能规范.md`,声明样例 `驱动/样板技能/`,真跑样例 `驱动/技能执行/`)。
驱动 ↔ 内核、驱动 ↔ 驱动之间**不设计协议**,全部经 PostgreSQL(`events` 表当总线,见本文件第 3 节
末与 02 第 3 节)—— 这是老板定的口径,不需要"三选一"。
+9 -2
View File
@@ -54,6 +54,8 @@
| `驱动/样板技能/技能.py` | **只列不执行**:把自带的 skill 读出来打印(档次 / 步骤数 / 样例数 / 边界数) |
| `驱动/样板技能/技能/请求解析-主模型.json` | main 档样例(任务:把一句话解析成 `calls` 的 `want` + `args`,允许多步计划) |
| `驱动/样板技能/技能/请求解析-小模型.json` | small 档样例(同一任务,锁死结构 + 3 条样例含 2 条反例) |
| `驱动/技能执行/配置.efi.json` + `执行.py` | **示例驱动(真的会跑)**:读 `技能/` 下这一档的 skill → 按 `tier` 选模型 → 拼 prompt → 调 OpenAI 兼容端点 → 剥 JSON → **逐条按档校验** → 落盘 + 写 `events`。它把上面那三档"归驱动实现"的事**做了一遍**,写新驱动可以照抄 |
| `驱动/技能执行/技能/*.json` | 与 `样板技能` 同两份(驱动文件夹自包含:拷走整个文件夹就能跑,这也正是"驱动之间零耦合"的用法) |
```bash
./.venv/bin/python 内核/内核.py 启动 样板技能 # 跑一遍
@@ -64,7 +66,7 @@
主模型负责"该调谁、要不要多步",小模型负责"把一句话填进 `want` + `args` 这个固定结构"。
skill 自己**不碰 PG、不发 `calls`**(那是驱动的事),它只是声明。
## 5. 版本边界(v0.1 明确不做)
## 5. 版本边界(内核侧 v0.1 明确不做;驱动侧 2026-09-17 起有参考实现)
| 不做 | 为什么 / 谁做 |
|---|---|
@@ -74,6 +76,11 @@ skill 自己**不碰 PG、不发 `calls`**(那是驱动的事),它只是
| skill 之间互相调用 / 组合 | 不需要:要串就由驱动编排,或让主模型在 `plan` 里排多步 |
| 权限、配额、沙箱 | 跟驱动同等待遇,v0.2 再说 |
**改动这份规范时的连带清单**:`驱动/样板技能/技能/*.json`(样例要跟着改)→
> **补记(2026-09-17)**:上表第 1~3 条原来只写到"归驱动实现",现在 `驱动/技能执行/` 已经把这
> 三件事**真做了一遍**(读 `技能/` 下的文件 / 按 `tier` 选模型 / 拼 prompt / 逐条校验输出结构 /
> 思考型模型把 `max_tokens` 吃光时把预算翻倍重试一次 / 结果落盘 + 写 `events`)。
> 它是"示例驱动",不是规范的一部分 —— **内核侧依旧一行都不碰 skill**,规矩没变,只是多了个能抄的东西。
**改动这份规范时的连带清单**:`驱动/样板技能/技能/*.json` 与 `驱动/技能执行/技能/*.json`(两份样例要一起改)→
`文档/02-写一个驱动.md`("驱动可以带 skill"那节)→ `文档/06-架构与不变量.md`(分层表里 Skill 那行)→
`README.md`(三层结构表)→ 技能 `kernel-driver-framework`。
+2 -1
View File
@@ -5,5 +5,6 @@
"mode": "oneshot",
"entry": "json解码.py",
"interpreter": "system",
"note": "把 web.json ({html,css,js}) 解出来打印 -- 程序/Skill 层的最小样本, 由它来解释执行"
"provides": ["Json解码:解析"],
"note": "把 web.json ({html,css,js}) 解出来打印 -- 程序/Skill 层的最小样本, 由它来解释执行. 契约名 Json解码:解析 是 skill 样例里已经在用的那种\"能力名\""
}
+586
View File
@@ -0,0 +1,586 @@
"""技能执行: 示例驱动 -- 把驱动自带的 skill 真正执行一遍.
[这个示例演示了什么]
1. 底座最下面那层是"确定性代码能力接入层". 驱动把一种外部能力接进来, 这里接的是
"一次模型调用"; 上面两层(引导器 / 内核)完全不知道下面在用模型.
2. **skill 归驱动管**. 读哪份 / 选哪个模型 / prompt 怎么拼 / 输出怎么校验 / 怎么记账,
全在本文件里, 内核不知道 skill 的存在 (依据 设计/01 第 7 节 与 设计/05 技能规范).
3. 两档 skill 的差别落到代码上 -- 同一件事写两份, 校验强度不同:
main : 只锁外层 (plan 是数组 + 每步有 want + confidence 在 0~1)
small: 全锁死 (want 单值 + 必须在契约清单里 + args 键白名单 + confidence 只许 0/0.5/1)
4. 与底座的全部接口只有两处:
环境变量 EFI_DB -- 内核注入的库连接串 (没有也能跑, 只是不汇报)
events 表 -- 汇报的唯一通道, 往表里插一行 (没有协议)
另外它**只读** drivers 表拿契约清单 -- 真实驱动就该这么拿底座里的信息, 而不是去
翻别的驱动文件夹 (驱动之间零耦合, 见 设计/01 第 3 节).
[怎么改]
模型端点 / 模型名 / 跑哪一档 / 喂什么话, 全在 配置.efi.json 的 env 段, 源码不硬编码;
密钥类(EFI_LLM_KEY)不写进配置, 从环境变量继承 (密钥不落盘).
不需要内核也能手工跑一遍: python3 驱动/技能执行/执行.py
"""
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime
from pathlib import Path
from typing import Any, cast
驱动名 = "技能执行"
驱动目录 = Path(__file__).resolve().parent
技能目录 = 驱动目录 / "技能"
输出目录 = 驱动目录 / "输出"
# 模型端点超时: 本地小模型带思考时会慢, 给宽一点; 但绝不无限等 (等不到就如实报错退出)
模型超时秒 = 240.0
# 单次最多让模型吐多少 token; 思考型模型会把预算全用在思考上, 所以 content 空时自动翻倍重试
模型默认最大输出 = 4096
最大输出封顶 = 16384
# 允许的置信度取值 (small 档写死的三档); 用集合判"是不是这三个之一"
小模型置信度 = {0.0, 0.5, 1.0}
# ---------------------------------------------------------------- 与底座打交道
def 取环境(键: str, 默认: str = "") -> str:
"""取一个配置值: 内核把 配置.efi.json 的 env 段注入环境变量, 手工跑就是普通环境变量."""
值 = os.environ.get(键, "").strip()
return 值 if 值 else 默认
def 汇报(消息: str, kind: str = "log", 数据: dict[str, Any] | None = None) -> None:
"""往 events 表插一行 -- 驱动唯一的汇报方式 (没有协议, 内核只读表, 不解析 stdout).
连不上 PG / 没装 psycopg2 都不算失败: 打印一行就够了 (设计 01 第 3 节). 汇报失败
也不许把驱动搞死. 这里刻意每次新开连接: 本驱动是一次性任务, 用完就关最省事.
"""
串 = os.environ.get("EFI_DB", "")
if not 串:
return
try:
import psycopg2 # type: ignore[import-untyped]
except ImportError:
return
try:
连接对象 = psycopg2.connect(串)
连接对象.autocommit = True
游标 = 连接对象.cursor()
游标.execute(
"INSERT INTO events (source, driver, level, kind, message, data)"
" VALUES (%s, %s, 'info', %s, %s, %s::jsonb)",
(驱动名, 驱动名, kind, 消息, json.dumps(数据 or {}, ensure_ascii=False)),
)
游标.close()
连接对象.close()
except Exception as 错: # 汇报失败不该把驱动搞死
print(f"{驱动名}: 汇报失败: {错}", flush=True)
def 读契约清单() -> list[dict[str, Any]]:
"""问底座"现在有哪些契约可以用" -- 只读 PG 的 drivers 表 (内核扫描时写进去的).
真实驱动就该这么拿信息: 不 import 别的驱动, 也不去翻别人的文件夹. 参数列表库里没有
(契约名本身不带参数说明), 一律给空表 -- 对应 skill 里那条"参数列表为空就表示这个契约
不接受参数, args 填 {}".
"""
显式 = 取环境("EFI_SKILL_CONTRACTS")
if 显式:
# 手工跑 / PG 不在时也能演示: 直接给一份清单
try:
读了: Any = json.loads(显式)
except json.JSONDecodeError as 错:
print(f"{驱动名}: EFI_SKILL_CONTRACTS 不是合法 JSON: {错}", flush=True)
return []
if isinstance(读了, list):
return [cast(dict[str, Any], 项) for 项 in cast(list[Any], 读了) if isinstance(项, dict)]
return []
串 = os.environ.get("EFI_DB", "")
if not 串:
print(f"{驱动名}: 没拿到 EFI_DB, 契约清单当空 (手工跑就这样, 不算错)", flush=True)
return []
try:
import psycopg2 # type: ignore[import-untyped]
连接对象 = psycopg2.connect(串)
连接对象.autocommit = True
游标 = 连接对象.cursor()
游标.execute("SELECT provides FROM drivers WHERE provides IS NOT NULL ORDER BY name")
行集 = 游标.fetchall()
游标.close()
连接对象.close()
except Exception as 错:
print(f"{驱动名}: 读 drivers 表失败 (当没有契约): {错}", flush=True)
return []
清单: list[dict[str, Any]] = []
for 行 in cast(list[Any], 行集):
契约们 = 行[0]
if not isinstance(契约们, list):
continue
for 契约 in cast(list[Any], 契约们):
文本 = str(契约).strip()
if 文本:
清单.append({"want": 文本, "参数": []})
return 清单
# ---------------------------------------------------------------- 解释 skill
def 取对象(值: Any) -> dict[str, Any]:
"""值确实是 dict 就给 dict, 否则空 dict -- 窄化只做一次, 别在代码里到处 cast."""
if isinstance(值, dict):
return cast(dict[str, Any], 值)
return {}
def 字符串表(值: Any) -> list[str]:
"""把 json 里读出来的 list 收成 list[str] (脏数据不炸: 非字符串项直接丢)."""
if not isinstance(值, list):
return []
出: list[str] = []
for 项 in cast(list[Any], 值):
if isinstance(项, str) and 项.strip():
出.append(项)
return 出
def 编号(项们: list[str]) -> str:
"""1. xxx / 2. xxx 排下来 (步骤和红线都用这个, 数字让模型逐步跟)."""
return "\n".join(f" {序}. {文}" for 序, 文 in enumerate(项们, 1))
def 读技能(档: str) -> tuple[str, dict[str, Any]] | None:
"""按 tier 挑一份 skill 读出来; 找不到 / 读不了就说清为什么 (不静默跳过).
skill 放 技能/*.json, 一个 skill 一个文件, 文件名不参与语义 (name 字段才是名字).
"""
if not 技能目录.is_dir():
print(f"{驱动名}: 没有 技能/ 目录, 没东西可执行", flush=True)
return None
有档: list[str] = []
for 路径 in sorted(技能目录.glob("*.json")):
try:
内容: Any = json.loads(路径.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as 错:
print(f"{驱动名}: 读不了 {路径.name}: {错}", flush=True)
continue
if not isinstance(内容, dict):
print(f"{驱动名}: {路径.name} 顶层必须是对象, 跳过", flush=True)
continue
对象 = cast(dict[str, Any], 内容)
现档 = str(对象.get("tier") or "")
有档.append(现档)
if 现档 == 档:
return 路径.name, 对象
print(f"{驱动名}: 没有 tier={档} 的 skill (现有档: {sorted(有档)})", flush=True)
return None
def 拼系统提示(技能: dict[str, Any]) -> str:
"""把 skill 声明拼成给模型的系统提示 -- 这就是"驱动解释 skill"的落点.
骨架只有一份: 两份 skill 的结构一样, 差别全在文件里写的步骤 / 输出结构 / 边界 / 样例,
所以**换档只是换一份文件, 换模型也不用改这里**.
"""
段: list[str] = []
段.append(f"你是 {驱动名} 里的解析器. 任务: {str(技能.get('description') or '')}")
触发 = 字符串表(技能.get("when"))
if 触发:
段.append("什么情况下用:\n" + 编号(触发))
步骤 = 字符串表(技能.get("steps"))
if 步骤:
段.append("执行步骤:\n" + 编号(步骤))
输出 = 取对象(技能.get("output"))
if 输出:
模式 = 取对象(输出.get("schema"))
if 模式:
字段行 = "\n".join(f" {键}: {值}" for 键, 值 in 模式.items())
段.append("输出结构 (必须是 JSON):\n" + 字段行)
说明 = str(输出.get("说明") or "")
if 说明:
段.append(说明)
样例 = 技能.get("examples")
if isinstance(样例, list) and cast(list[Any], 样例):
块: list[str] = []
for 项 in cast(list[Any], 样例):
对 = 取对象(项)
块.append("输入: " + json.dumps(对.get("输入"), ensure_ascii=False))
块.append("输出: " + json.dumps(对.get("输出"), ensure_ascii=False))
段.append("示例:\n" + "\n".join(块))
边界 = 字符串表(技能.get("boundaries"))
if 边界:
段.append("红线 (违反即错):\n" + 编号(边界))
段.append("只输出 JSON 本体: 前后不要任何解释文字, 不要 markdown 代码块标记.")
if 取环境("EFI_LLM_NO_THINK"):
# 带思考的模型(DeepSeek / Qwen3 系)会把 max_tokens 全用在思考上, content 回来是空串.
# Qwen3 系的软开关 /no_think 放在**提示末尾**才认, 比塞进请求体的字段更通用 (2026-09-17 实测).
段.append("/no_think")
return "\n\n".join(段)
def 拼用户输入(技能: dict[str, Any], 原话: str, 清单: list[dict[str, Any]]) -> str:
"""按这份 skill 自己声明的 input.schema 结构, 把真实输入拼成 JSON.
两档的输入字段名不一样 (main 要"可用契约"是字符串表, small 要"契约清单"带参数名),
这里**照 skill 写的 schema 填**, 不写死字段名 -- 又是一处"驱动解释 skill"而不是"驱动写死 skill".
"""
模式 = 取对象(取对象(技能.get("input")).get("schema"))
载荷: dict[str, Any] = {}
for 键 in 模式:
if 键 == "用户原话":
载荷[键] = 原话
elif 键 == "可用契约":
载荷[键] = [str(项.get("want") or "") for 项 in 清单]
elif 键 == "契约清单":
载荷[键] = 清单
else:
载荷[键] = None
if not 载荷: # skill 没写 input.schema 就退回一个最小结构, 不让模型空手答
载荷 = {"用户原话": 原话, "契约清单": 清单}
return json.dumps(载荷, ensure_ascii=False, indent=2)
# ---------------------------------------------------------------- 调模型
def 建打开器(端点: str) -> urllib.request.OpenerDirector:
"""本机端点直连: 本机装了 Clash 这类代理时, http_proxy 会把 127.0.0.1 的请求也劫走."""
主机 = urllib.parse.urlsplit(端点).hostname or ""
if 主机 in ("127.0.0.1", "localhost", "::1"):
return urllib.request.build_opener(urllib.request.ProxyHandler({}))
return urllib.request.build_opener()
def 调模型(端点: str, 模型: str, 系统提示: str, 用户输入: str, 最大输出: int) -> tuple[bool, str, bool]:
"""调一次 OpenAI 兼容的 /chat/completions (只用标准库, 驱动不引第三方依赖).
返回 (成没成, 模型原文 或 错误原文, 是不是"思考把输出吃光了"). 失败**不吞错**,
也不编一个假结果出来; 最后那个标志位让调用方决定要不要加大预算重试一次.
"""
请求体: dict[str, Any] = {
"model": 模型,
"messages": [
{"role": "system", "content": 系统提示},
{"role": "user", "content": 用户输入},
],
"temperature": 0,
"max_tokens": 最大输出,
"stream": False,
}
if 取环境("EFI_LLM_NO_THINK"):
# 带思考的模型(DeepSeek / Qwen3 系)会把 max_tokens 全用在思考上, content 回来是空串.
# 这是 Qwen3 系的通用写法; 换成不认这个字段的端点(如 OpenAI 官方)就把配置里那行删掉.
请求体["chat_template_kwargs"] = {"enable_thinking": False}
头 = {"Content-Type": "application/json"}
密钥 = 取环境("EFI_LLM_KEY")
if 密钥:
头["Authorization"] = f"Bearer {密钥}"
请求 = urllib.request.Request(
url=端点.rstrip("/") + "/chat/completions",
data=json.dumps(请求体, ensure_ascii=False).encode("utf-8"),
headers=头,
method="POST",
)
打开器 = 建打开器(端点)
try:
with 打开器.open(请求, timeout=模型超时秒) as 应答:
原文 = 应答.read().decode("utf-8", errors="replace")
except urllib.error.HTTPError as 错:
try:
详情 = 错.read().decode("utf-8", errors="replace")
except OSError:
详情 = ""
return False, f"模型端点回了 HTTP {错.code}: {详情[:400]}", False
except (urllib.error.URLError, TimeoutError, OSError) as 错:
return False, f"连不上模型端点 {端点}: {错}", False
try:
包: Any = json.loads(原文)
except json.JSONDecodeError as 错:
return False, f"端点回的不是 JSON: {错} / 原文 {原文[:200]}", False
选择 = 取对象(包).get("choices")
if not isinstance(选择, list) or not cast(list[Any], 选择):
return False, f"端点没给 choices: {原文[:300]}", False
首项 = 取对象(cast(list[Any], 选择)[0])
内容 = 取对象(首项.get("message")).get("content")
if not isinstance(内容, str) or not 内容.strip():
# 带思考的模型把 max_tokens 全用在思考上时, content 就是空的 (2026-09-17 真踩到)
return False, (
"模型 content 是空的 (带思考的模型把 max_tokens 用在思考上了; "
"配置里开 EFI_LLM_NO_THINK=1 关掉思考, 或者把 max_tokens / 上下文调大): "
+ 原文[:300]
), True
return True, 内容, False
def 剥JSON(文本: str) -> tuple[bool, Any]:
"""从模型原文里抠出 JSON 对象 -- 模型爱加 ``` 围栏或前后废话, 清洗是驱动该干的活."""
清洗 = 文本.strip()
if 清洗.startswith("```"):
换行 = 清洗.find("\n")
清洗 = 清洗[换行 + 1 :] if 换行 >= 0 else 清洗
右围栏 = 清洗.rfind("```")
if 右围栏 >= 0:
清洗 = 清洗[:右围栏]
清洗 = 清洗.strip()
try:
return True, json.loads(清洗)
except json.JSONDecodeError:
pass
# 兜底: 取第一个 { 到最后一个 } (前后还有废话时用)
左 = 清洗.find("{")
右 = 清洗.rfind("}")
if 左 >= 0 and 右 > 左:
try:
return True, json.loads(清洗[左 : 右 + 1])
except json.JSONDecodeError as 错:
return False, f"剥出来的片段不是合法 JSON: {错}"
return False, "模型输出里没有 JSON 对象"
# ---------------------------------------------------------------- 校验输出
def 校验输出(档: str, 结果: Any, 清单: list[dict[str, Any]]) -> list[tuple[bool, str]]:
"""按这一档的规矩逐条核对模型输出 (每条给"判据 + 结论", 可解释、可审计).
main : 只锁外层 -- plan 是数组 + 每步有 want + confidence 在 0~1
small: 全锁死 -- want 单值 + 必须在契约清单里 + args 键白名单 + confidence 只许 0/0.5/1
"""
判据: list[tuple[bool, str]] = []
对象 = 取对象(结果)
if not 对象:
判据.append((False, "顶层是 JSON 对象"))
return 判据
判据.append((True, "顶层是 JSON 对象"))
if 档 == "small":
判据.extend(_校小模型(对象, 清单))
else:
判据.extend(_校主模型(对象))
return 判据
def _校小模型(对象: dict[str, Any], 清单: list[dict[str, Any]]) -> list[tuple[bool, str]]:
"""small 档的校验: 结构全锁死, 差一格都算不过 (这一档靠结构稳, 不靠模型聪明)."""
判据: list[tuple[bool, str]] = []
可用 = {str(项.get("want") or ""): 字符串表(项.get("参数")) for 项 in 清单}
if "want" not in 对象:
判据.append((False, "有 want 字段"))
return 判据
要的 = 对象.get("want")
if 要的 is None:
判据.append((True, "want 是 null (回问, 合法)"))
elif isinstance(要的, str):
判据.append((True, "want 是字符串"))
if 要的 in 可用:
判据.append((True, f"want 在契约清单里 ({要的})"))
elif not 可用:
判据.append((False, f"want={要的} 但契约清单是空的 (不许自造契约名)"))
else:
判据.append((False, f"want={要的} 不在契约清单里 ({sorted(可用)})"))
else:
判据.append((False, "want 必须是字符串或 null"))
参数 = 对象.get("args")
if isinstance(参数, dict):
参数对象 = cast(dict[str, Any], 参数)
判据.append((True, "args 是对象"))
白名单 = 可用.get(str(要的 or ""), [])
越界 = [键 for 键 in 参数对象 if 键 not in 白名单]
if 越界:
判据.append((False, f"args 里有多余的键 {越界} (白名单 {白名单})"))
else:
判据.append((True, f"args 的键都在白名单里 {白名单}"))
else:
判据.append((False, "args 必须是对象"))
置信 = 对象.get("confidence")
if isinstance(置信, bool) or not isinstance(置信, (int, float)):
判据.append((False, "confidence 必须是 0 / 0.5 / 1 之一"))
elif float(置信) in 小模型置信度:
判据.append((True, f"confidence={float(置信)} 在三档之内"))
else:
判据.append((False, f"confidence={置信} 不在 {{0, 0.5, 1}} 里"))
if 要的 is None:
回问 = 对象.get("need_user")
if isinstance(回问, str) and 回问.strip():
判据.append((True, "want 为 null 时给了 need_user"))
else:
判据.append((False, "want 为 null 却没给 need_user"))
return 判据
def _校主模型(对象: dict[str, Any]) -> list[tuple[bool, str]]:
"""main 档的校验: 只锁外层结构 (plan 是数组、每步有 want),细节交给模型."""
判据: list[tuple[bool, str]] = []
计划 = 对象.get("plan")
if not isinstance(计划, list):
判据.append((False, "plan 是数组"))
return 判据
步们 = cast(list[Any], 计划)
判据.append((True, f"plan 是数组 ({len(步们)} 步)"))
坏步: list[int] = []
for 序, 步 in enumerate(步们, 1):
步对象 = 取对象(步)
要的 = 步对象.get("want")
if not isinstance(要的, str) or not 要的.strip():
坏步.append(序)
if 坏步:
判据.append((False, f"第 {坏步} 步缺 want (每步都必须有)"))
else:
判据.append((True, "每一步都有 want"))
# plan 空 + 也不回问 = 既没规划也没说缺什么, 等于什么都没答 (2026-09-17 实测抓到)
回问 = 对象.get("need_user")
有话说 = isinstance(回问, str) and bool(回问.strip())
if 步们 or 有话说:
判据.append((True, "plan 非空, 或者给了 need_user (有结论)"))
else:
判据.append((False, "plan 是空的, 又没有 need_user (等于没答)"))
置信 = 对象.get("confidence")
if isinstance(置信, bool) or not isinstance(置信, (int, float)):
判据.append((False, "confidence 必须是 0~1 的数"))
elif 0.0 <= float(置信) <= 1.0:
判据.append((True, f"confidence={float(置信)} 在 0~1"))
else:
判据.append((False, f"confidence={置信} 超出 0~1"))
return 判据
# ---------------------------------------------------------------- 主流程
def 现在文本() -> str:
"""带时区的时刻 (项目里一律带 +08:00, 事后看日志不用猜时区)."""
return datetime.now().astimezone().isoformat(timespec="seconds")
def 落盘(记录: dict[str, Any]) -> Path:
"""把这一趟的原样记下来 (skill / 输入 / 提示 / 模型原文 / 校验结论), 方便事后审计."""
输出目录.mkdir(parents=True, exist_ok=True)
路径 = 输出目录 / f"结果-{datetime.now().strftime('%Y%m%d-%H%M%S')}.json"
路径.write_text(json.dumps(记录, ensure_ascii=False, indent=2), encoding="utf-8")
return 路径
def 主() -> int:
"""跑一趟: 读 skill -> 拿契约清单 -> 拼 prompt -> 调模型 -> 剥 JSON -> 校验 -> 记账."""
档 = 取环境("EFI_SKILL_TIER", "small")
原话 = 取环境("EFI_SKILL_INPUT")
端点 = 取环境("EFI_LLM_BASE")
模型 = 取环境("EFI_LLM_MODEL_MAIN" if 档 == "main" else "EFI_LLM_MODEL_SMALL")
print(f"{驱动名}: 示例驱动 (skill 归驱动管, 内核不知道它的存在)", flush=True)
读到的 = 读技能(档)
if 读到的 is None:
汇报(f"{驱动名} 执行失败: 没有 tier={档} 的 skill", kind="error")
return 1
文件名, 技能 = 读到的
步数 = len(字符串表(技能.get("steps")))
样例数 = len(技能.get("examples") or [])
边界数 = len(字符串表(技能.get("boundaries")))
print(f"{驱动名}: skill [{档}] {技能.get('name') or 文件名} (步骤 {步数} / 样例 {样例数} / 边界 {边界数})", flush=True)
if not 原话:
print(f"{驱动名}: 没给 EFI_SKILL_INPUT, 不知道要解析什么", flush=True)
汇报(f"{驱动名} 执行失败: 没给输入", kind="error")
return 1
if not 端点:
print(f"{驱动名}: 没给 EFI_LLM_BASE (模型端点), 没法执行 -- 看 配置.efi.json 的 env 段", flush=True)
汇报(f"{驱动名} 执行失败: 没配模型端点", kind="error")
return 1
清单 = 读契约清单()
来源 = "环境变量" if 取环境("EFI_SKILL_CONTRACTS") else "PG 的 drivers 表"
print(f"{驱动名}: 契约清单 {len(清单)} 条 (来源 {来源}): {[项['want'] for 项 in 清单]}", flush=True)
系统提示 = 拼系统提示(技能)
用户输入 = 拼用户输入(技能, 原话, 清单)
print(f"{驱动名}: 模型 {模型} @ {端点} 输入: {原话}", flush=True)
print(f"{驱动名}: 提示拼好 {len(系统提示)} 字, 调模型中 (超时 {int(模型超时秒)} 秒)...", flush=True)
起 = time.monotonic()
# max_tokens 可以配置; 没配就用默认. 思考型模型会把预算全用在思考上 -> content 空,
# 那就**加大预算重试一次**(设计 05 把"重试"明确划给驱动实现), 还空就如实报错退出.
预算文本 = 取环境("EFI_LLM_MAX_TOKENS")
最大输出 = int(预算文本) if 预算文本.isdigit() and int(预算文本) > 0 else 模型默认最大输出
成, 原文, 被吃光 = 调模型(端点, 模型, 系统提示, 用户输入, 最大输出)
while (not 成) and 被吃光 and 最大输出 < 最大输出封顶:
最大输出 = min(最大输出 * 2, 最大输出封顶)
print(f"{驱动名}: 思考把输出预算吃光了, max_tokens 提到 {最大输出} 重试一次...", flush=True)
成, 原文, 被吃光 = 调模型(端点, 模型, 系统提示, 用户输入, 最大输出)
用时 = time.monotonic() - 起
if not 成:
print(f"{驱动名}: 调模型失败 ({用时:.1f}s): {原文}", flush=True)
汇报(f"{驱动名} 调模型失败: {原文}", kind="error", 数据={"模型": 模型, "端点": 端点})
return 1
print(f"{驱动名}: 模型回了 {len(原文)} 字, 用时 {用时:.1f}s", flush=True)
剥好, 结果 = 剥JSON(原文)
判据: list[tuple[bool, str]] = []
if not 剥好:
判据.append((False, str(结果)))
else:
print(f"{驱动名}: 模型输出 = {json.dumps(结果, ensure_ascii=False)}", flush=True)
判据 = 校验输出(档, 结果, 清单)
print(f"{驱动名}: 校验 --", flush=True)
for 过没过, 说明 in 判据:
print(f" [{'通过' if 过没过 else '不过'}] {说明}", flush=True)
通过数 = sum(1 for 过没过, _ in 判据 if 过没过)
记录: dict[str, Any] = {
"技能": str(技能.get("name") or 文件名),
"档": 档,
"模型": 模型,
"端点": 端点,
"输入": 原话,
"契约清单": 清单,
"系统提示": 系统提示,
"用户输入": 用户输入,
"模型原文": 原文,
"模型输出": 结果 if 剥好 else None,
"校验": [{"判据": 说明, "通过": 过没过} for 过没过, 说明 in 判据],
"用时秒": round(用时, 1),
"时刻": 现在文本(),
}
路径 = 落盘(记录)
print(f"{驱动名}: 原样记到 {路径}", flush=True)
全过 = bool(判据) and 通过数 == len(判据)
if 全过:
汇报(
f"{驱动名}: {档} 档执行成功 ({技能.get('name') or 文件名}, {通过数}/{len(判据)} 判据通过)",
kind="produce",
数据={"模型": 模型, "档": 档, "结果": 结果, "用时秒": round(用时, 1)},
)
print(f"{驱动名}: 收工 -- {通过数}/{len(判据)} 条判据全过", flush=True)
return 0
汇报(
f"{驱动名}: {档} 档输出没过校验 ({通过数}/{len(判据)})",
kind="error",
数据={"模型": 模型, "档": 档, "结果": 结果, "用时秒": round(用时, 1)},
)
print(f"{驱动名}: 收工 -- 只过了 {通过数}/{len(判据)} 条, 退出码 1 (不装成功)", flush=True)
return 1
if __name__ == "__main__":
sys.exit(主())
@@ -0,0 +1,55 @@
{
"efi": 1,
"name": "请求解析-主模型",
"version": "0.1",
"tier": "main",
"description": "把用户一句自然语言请求, 解析成驱动调用 (want 契约名 + args 参数); 需要时拆成多步计划.",
"when": [
"用户的话里有明确动作意图 (取什么 / 解什么 / 算什么)",
"要判断该调哪个驱动、参数怎么填",
"信息不够时敢回问, 不硬猜"
],
"input": {
"说明": "用户原话 + 当前可用契约清单 (由内核扫描产出, 不写死在 skill 里)",
"schema": {
"用户原话": "string",
"可用契约": ["string"]
}
},
"steps": [
"读懂用户到底要什么 (要数据、要结论、还是要动作)",
"在可用契约里挑最贴的; 一个不够就串成多步, 顺序写清谁先谁后",
"参数填成 JSON; 拿不准的宁可留 null 并说明, 不要编一个看起来合理的值"
],
"output": {
"format": "json",
"schema": {
"plan": [{ "want": "契约名", "args": {}, "为什么": "一句话" }],
"need_user": "string|null -- 信息不够时写要问什么, 此时 plan 留空",
"confidence": "0~1"
},
"说明": "主模型档只锁外层结构: plan 是数组 (一步或多步都行), 每步必须有 want; 参数细节不预设, 由模型按契约语义自己定."
},
"examples": [
{
"输入": {
"用户原话": "把那份 json 里的字段名都列出来",
"可用契约": ["Json解码:解析", "样板:心跳"]
},
"输出": {
"plan": [{ "want": "Json解码:解析", "args": { "文件": "上次那份" }, "为什么": "只有它能读 json" }],
"need_user": null,
"confidence": 0.8
}
}
],
"boundaries": [
"契约清单里没有能满足的 -> 不要硬编一个 want, 用 need_user 说明缺什么",
"用户话里有歧义 -> 回问一句, 不要两件事都做",
"输出必须是能直接被 json.loads 的 JSON, 前后不加解释文字"
],
"notes": [
"这一档靠模型自己规划, 所以步骤写得抽象; 换个更强的模型不用改这份文件.",
"给云端强模型或本地 27B 以上跑 (plan 允许多步, 上下文要够长)."
]
}
@@ -0,0 +1,74 @@
{
"efi": 1,
"name": "请求解析-小模型",
"version": "0.1",
"tier": "small",
"description": "把用户一句话填进固定结构: 只从给定契约清单里选一个 want, 并填 args; 不许自己发明契约名.",
"when": [
"只有一句话、一个动作, 不需要规划多步",
"契约清单和参数名都已给定 (抽取任务, 不是规划任务)",
"要求输出严格 JSON, 能直接被程序读"
],
"input": {
"说明": "用户原话 + 契约清单 (每条带它接受的参数名); 都由驱动给, 不给就不许答",
"schema": {
"用户原话": "string",
"契约清单": [{ "want": "string", "参数": ["string"] }]
}
},
"steps": [
"第 1 步: 读用户原话, 只找动词和名词 (要做什么 / 对什么做), 别的都别看",
"第 2 步: 在契约清单里逐条比对, 选中唯一一条 want; 找不到就填 null 并跳到第 5 步",
"第 3 步: 对选中的 want, 只允许填它 参数 列表里出现的参数名, 不许新增参数名",
"第 4 步: 参数值只能来自用户原话里出现过的词; 原话里没有的参数一律填 null (不许猜、不许举例、不许编路径)",
"第 5 步: 只输出 JSON, 四个字段一个都不能少 (want / args / need_user / confidence)"
],
"output": {
"format": "json",
"schema": {
"want": "string|null -- 必须来自契约清单, 不许自造",
"args": "object -- 键必须来自该契约的 参数 列表; 缺的值填 null",
"need_user": "string|null -- 只有 want=null 时填一句要问的话",
"confidence": "number -- 只能是 0 / 0.5 / 1 三个值之一"
},
"说明": "小模型档把结构锁死: 单值 want (不是数组)、args 键白名单、置信度只许三档、不许输出多余字段."
},
"examples": [
{
"输入": {
"用户原话": "把 demo.json 解出来",
"契约清单": [{ "want": "Json解码:解析", "参数": ["文件"] }]
},
"输出": { "want": "Json解码:解析", "args": { "文件": "demo.json" }, "need_user": null, "confidence": 1 }
},
{
"输入": {
"用户原话": "帮我看看那个东西怎么样了",
"契约清单": [{ "want": "Json解码:解析", "参数": ["文件"] }]
},
"输出": { "want": null, "args": {}, "need_user": "你说的哪个文件?", "confidence": 1 }
},
{
"输入": {
"用户原话": "先解 demo.json 再算一下字数",
"契约清单": [{ "want": "Json解码:解析", "参数": ["文件"] }]
},
"输出": {
"want": null,
"args": {},
"need_user": "这句要两步 (解 + 算字数), 我一次只做一步: 先解哪个文件?",
"confidence": 0.5
}
}
],
"boundaries": [
"契约清单没给 -> 直接 need_user 要清单, 不要凭常识编契约名",
"用户原话里没出现过的参数值 -> 填 null, 不要填示例值、不要猜路径",
"多步请求 (出现 先...再...) -> 这一档只做第一步, 剩下写进 need_user, 不要自己串",
"输出前后不许有任何解释文字, markdown 代码块标记也算多余"
],
"notes": [
"这一档靠结构稳, 不靠模型聪明: 所有判断都落在'与原话字面比对'上, 所以 4B~9B 也跑得稳.",
"步骤编号写死是让模型逐步跟; 改成大段描述就退化 (老板口径: 结构稳不赖模型智商)."
]
}
+17
View File
@@ -0,0 +1,17 @@
{
"efi": 1,
"name": "技能执行",
"runtime": "python",
"mode": "oneshot",
"entry": "执行.py",
"interpreter": "system",
"env": {
"EFI_SKILL_TIER": "small",
"EFI_SKILL_INPUT": "给我来一次心跳",
"EFI_LLM_BASE": "http://127.0.0.1:43953/v1",
"EFI_LLM_MODEL_MAIN": "prism-ml/bonsai-27b",
"EFI_LLM_MODEL_SMALL": "zai-org/glm-4.6v-flash",
"EFI_LLM_NO_THINK": "1"
},
"note": "示例驱动: 把驱动自带的 skill 真正执行一遍 -- 按 tier 选模型 / 拼 prompt / 调模型 / 校验输出结构 (内核不知道 skill 的存在). EFI_LLM_NO_THINK 只对带思考的本地模型有意义, 用官方 API 时删掉这行"
}