5.6 KiB
5.6 KiB
很好,这一步你已经想清楚“为什么拆”了,那接下来就该最小侵入式地改 PocketFlow 项目结构,而不是推倒重来。 我按 “不破坏 PocketFlow 使用习惯 + 增量升级” 的原则,给你一套可直接落地的调整方案。
总体目标(你现在要做的事)
在 不改变 PocketFlow Node / Flow 编排方式 的前提下,引入:
brief→ 超轻量 Routerdescription + parameters→ 工具级调用body→ 按需加载语义
👉 PocketFlow 继续负责 流程 👉 你新增的是 Skill Semantic Layer
一、调整 SkillSpec(核心数据结构)
原来(简化版)
@dataclass
class SkillSpec:
name: str
description: str
parameters: dict
body: str
run: Callable
调整后(渐进式友好)
from dataclasses import dataclass
from typing import Callable, Any, Optional
@dataclass
class SkillSpec:
name: str
# Stage 1:Router 用(极短)
brief: str
# Stage 2:工具说明
description: str
parameters: dict
# Stage 3:按需加载
body: Optional[str]
run: Callable[[dict], Any]
📌 这是全套调整的“锚点”
二、调整 SKILL.md 规范(一次到位)
你现在的 SKILL.md 要明确支持分层。
---
name: add-calc
brief: 对数字列表执行加法计算
description: 对两个或多个数字进行加法运算
parameters:
numbers:
type: array
items: number
required: [numbers]
---
## 使用场景
- 求和
- 相加
- 计算总数
## 不适用场景
- 需要乘法、减法
- 非数值输入
📌 约定:
brief:10~20 tokendescription:工具级说明- body:可长、但不默认进 prompt
三、Skill Loader 调整(PocketFlow 友好版)
1️⃣ 解析 SKILL.md(frontmatter + body)
import re
import yaml
def parse_skill_md(text: str):
pattern = r"^---\s*\n(.*?)\n---\s*\n(.*)$"
m = re.match(pattern, text, re.S)
if not m:
raise ValueError("SKILL.md 缺少 YAML frontmatter")
meta = yaml.safe_load(m.group(1))
body = m.group(2).strip()
return meta, body
2️⃣ load_skills(最小侵入)
import importlib
from pathlib import Path
from skills.base import SkillSpec
def load_skills(root="skills"):
skills = []
for p in Path(root).iterdir():
if not p.is_dir():
continue
md = p / "SKILL.md"
if not md.exists():
continue
meta, body = parse_skill_md(md.read_text())
module = importlib.import_module(f"skills.{p.name}.skill")
skills.append(
SkillSpec(
name=meta["name"],
brief=meta.get("brief", meta["description"]),
description=meta["description"],
parameters=meta.get("parameters", {}),
body=body,
run=module.run,
)
)
return skills
👉 兼容旧 skill:没有 brief 也能跑
四、调整 DecideNode(真正体现“拆分价值”)
Stage 1:只用 brief(强烈推荐)
class DecideNode(Node):
def prep(self, shared):
return shared["query"], shared["skills"]
def exec(self, data):
query, skills = data
index = "\n".join(
f"- {s.name}: {s.brief}"
for s in skills
)
return llm(f"""
你可以使用以下技能(简要):
{index}
用户问题:
{query}
请选择最合适的技能名称,返回 JSON:
{{ "skill": "skill_name" }}
""")
📌 这一轮 token 极低,skill 再多也扛得住
Stage 2(可选):参数校验 / 二次确认
如果你发现 参数经常乱填,再加这一层:
def refine_skill(skill: SkillSpec, query: str):
return llm(f"""
技能:{skill.name}
说明:{skill.description}
参数结构:{skill.parameters}
用户问题:
{query}
请生成该技能需要的参数 JSON。
""")
Stage 3(仅在歧义时):拉 body
def disambiguate(skills, query):
context = "\n".join(
f"### {s.name}\n{s.body}"
for s in skills
)
return llm(f"""
以下是技能详细说明:
{context}
用户问题:
{query}
请选择最合适的技能。
""")
👉 90% 情况不会触发
五、ExecuteSkillNode 不需要任何修改 ✅
class ExecuteSkillNode(Node):
def exec(self, data):
decision, skills = data
skill = next(s for s in skills if s.name == decision["skill"])
return skill.run(decision["args"])
这正是你这套设计最优雅的地方。
六、这套改造对 PocketFlow 的影响
| 维度 | 影响 |
|---|---|
| Node / Flow | ❌ 不变 |
| 执行模型 | ❌ 不变 |
| Skill 扩展 | ✅ 大幅增强 |
| Token 成本 | ✅ 可控 |
| MCP / tools 对齐 | ✅ 顺滑 |
👉 PocketFlow 被你“语义升级”,但没被你“污染”
七、你现在这个阶段的“最佳状态判断”
你现在已经做到:
Skill = 语义声明 + 运行实体 + 渐进式暴露
这已经是:
- LangGraph
- OpenAI tools
- MCP
背后用的同一套思想。
如果你愿意下一步继续进阶
我可以直接帮你做三选一:
- 🔌 Skill → OpenAI tools schema 自动生成
- 🧠 Skill body 向量化(SkillRAG)
- 🧩 Planner + Router + Executor 三节点拆分
你现在这个系统,已经非常值得继续往上走了。