ai_pipeline/Demo4.md

5.6 KiB
Raw Blame History

很好,这一步你已经想清楚“为什么拆”了,那接下来就该最小侵入式地改 PocketFlow 项目结构,而不是推倒重来。 我按 “不破坏 PocketFlow 使用习惯 + 增量升级” 的原则,给你一套可直接落地的调整方案


总体目标(你现在要做的事)

不改变 PocketFlow Node / Flow 编排方式 的前提下,引入:

  • brief超轻量 Router
  • description + 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 1Router 用(极短)
    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]
---

## 使用场景
- 求和
- 相加
- 计算总数

## 不适用场景
- 需要乘法、减法
- 非数值输入

📌 约定:

  • brief10~20 token
  • description工具级说明
  • body可长、但不默认进 prompt

三、Skill Loader 调整PocketFlow 友好版)

1 解析 SKILL.mdfrontmatter + 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

背后用的同一套思想


如果你愿意下一步继续进阶

我可以直接帮你做三选一:

  1. 🔌 Skill → OpenAI tools schema 自动生成
  2. 🧠 Skill body 向量化SkillRAG
  3. 🧩 Planner + Router + Executor 三节点拆分

你现在这个系统,已经非常值得继续往上走了。