diff --git a/Demo3.md b/Demo3.md index 2832451..83d6d59 100644 --- a/Demo3.md +++ b/Demo3.md @@ -254,12 +254,3 @@ SkillSpec → JSON Schema → LLM tools 那他已经在 **Agent Framework 设计层**,而不是“用 Agent”。 --- - -如果你愿意,下一步我可以直接帮你做: - -1️⃣ **SKILL.md → JSON Schema 自动生成** -2️⃣ **参数校验(Pydantic)** -3️⃣ **Skill Planner(多技能组合)** -4️⃣ **MCP Server 适配(你这套非常合适)** - -你点一个,我直接继续。 diff --git a/Demo4.md b/Demo4.md new file mode 100644 index 0000000..660d259 --- /dev/null +++ b/Demo4.md @@ -0,0 +1,284 @@ +很好,这一步你已经**想清楚“为什么拆”了**,那接下来就该**最小侵入式地改 PocketFlow 项目结构**,而不是推倒重来。 +我按 **“不破坏 PocketFlow 使用习惯 + 增量升级”** 的原则,给你一套**可直接落地的调整方案**。 + +--- + +# 总体目标(你现在要做的事) + +在 **不改变 PocketFlow Node / Flow 编排方式** 的前提下,引入: + +* `brief` → **超轻量 Router** +* `description + parameters` → **工具级调用** +* `body` → **按需加载语义** + +👉 PocketFlow 继续负责 **流程** +👉 你新增的是 **Skill Semantic Layer** + +--- + +# 一、调整 SkillSpec(核心数据结构) + +## 原来(简化版) + +```python +@dataclass +class SkillSpec: + name: str + description: str + parameters: dict + body: str + run: Callable +``` + +## 调整后(渐进式友好) + +```python +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 要**明确支持分层**。 + +```markdown +--- +name: add-calc +brief: 对数字列表执行加法计算 +description: 对两个或多个数字进行加法运算 +parameters: + numbers: + type: array + items: number +required: [numbers] +--- + +## 使用场景 +- 求和 +- 相加 +- 计算总数 + +## 不适用场景 +- 需要乘法、减法 +- 非数值输入 +``` + +📌 约定: + +* `brief`:**10~20 token** +* `description`:**工具级说明** +* body:**可长、但不默认进 prompt** + +--- + +# 三、Skill Loader 调整(PocketFlow 友好版) + +## 1️⃣ 解析 SKILL.md(frontmatter + body) + +```python +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(最小侵入) + +```python +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(强烈推荐) + +```python +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(可选):参数校验 / 二次确认 + +如果你发现 **参数经常乱填**,再加这一层: + +```python +def refine_skill(skill: SkillSpec, query: str): + return llm(f""" +技能:{skill.name} +说明:{skill.description} +参数结构:{skill.parameters} + +用户问题: +{query} + +请生成该技能需要的参数 JSON。 +""") +``` + +--- + +## Stage 3(仅在歧义时):拉 body + +```python +def disambiguate(skills, query): + context = "\n".join( + f"### {s.name}\n{s.body}" + for s in skills + ) + + return llm(f""" +以下是技能详细说明: +{context} + +用户问题: +{query} + +请选择最合适的技能。 +""") +``` + +👉 **90% 情况不会触发** + +--- + +# 五、ExecuteSkillNode 不需要任何修改 ✅ + +```python +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 三节点拆分** + +你现在这个系统,已经非常值得继续往上走了。 diff --git a/src/pipeline/core/llm.py b/src/pipeline/core/llm.py index 1757952..0ef528d 100644 --- a/src/pipeline/core/llm.py +++ b/src/pipeline/core/llm.py @@ -158,6 +158,9 @@ class AsyncLLm: logger.exception("Chat request failed") return "" + async def chat_prompt(self, prompt:str): + return await self.chat([{"role": "user", "content": prompt}]) + async def stream_chat( self, messages: list[dict], diff --git a/src/pipeline/core/nodes.py b/src/pipeline/core/nodes.py index 9c6cb06..33d4841 100644 --- a/src/pipeline/core/nodes.py +++ b/src/pipeline/core/nodes.py @@ -222,7 +222,7 @@ class RerankNode(AsyncBatchNode): # Agent 相关 # ----------------------------- -class DecideNode(AsyncNode): +class DecideSkillNode(AsyncNode): """ 选择技能 """ @@ -231,16 +231,7 @@ class DecideNode(AsyncNode): async def exec_async(self, prep_res): query, skills = prep_res - skill_desc = "\n".join( - [ - f""" -- {x.name} - 描述: {x.description} - 参数: {x.parameters} -""" - for x in skills - ] - ) + skill_desc = "\n".join(f"- {s.name}: {s.brief}" for s in skills) prompt = f""" 你可以使用以下技能: {skill_desc} @@ -248,27 +239,106 @@ class DecideNode(AsyncNode): 用户问题: {query} -请选择一个技能,并返回 JSON: -{{ - "skill": "技能名", - "args": {{ 参数 }} -}} +请选择最可能相关的技能(最多 2 个),返回 JSON: +{{ "candidates": ["skill_name"] }} """ - res = await llm.client.chat([{"role": "user", "content": prompt}]) + res = await llm.client.chat_prompt(prompt) logger.debug(res) return parse_llm_json(res) async def post_async(self, shared, prep_res, exec_res): - shared["selected_skill"] = exec_res + shared["candidates"] = exec_res["candidates"] or [] + return "default" + + +class RefineSkillNode(AsyncNode): + """ + - 在 不读 body 的情况下 + - 尝试生成可用的 skill + args + - 如果失败 → 返回 disambiguate + """ + + async def prep_async(self, shared): + candidates = shared["candidates"] + skills = [s for s in shared["skills"] if s.name in candidates] + return shared["query"], skills + + async def exec_async(self, prep_res): + query, skills = prep_res + skill_desc = "\n".join(f""" +### {s.name} +说明: {s.description} +参数结构: {s.parameters} +""" for s in skills) + + prompt = f""" +以下是候选技能: +{skill_desc} + +用户问题: +{query} + +请选择最合适的技能并生成参数。 +如果无法确定,请返回: +{{ "need_disambiguation": true }} + +否则返回: +{{ "skill": "name", "args": {{...}} }} +""" + res = await llm.client.chat_prompt(prompt) + logger.debug(res) + return parse_llm_json(res) + + async def post_async(self, shared, prep_res, exec_res): + if "need_disambiguation" in exec_res and exec_res["need_disambiguation"]: + return "disambiguate" + if "skill" in exec_res and "args" in exec_res: + shared["decision"] = exec_res + return "default" + return "disambiguate" + + +class DisambiguateSkillNode(AsyncNode): + + async def prep_async(self, shared): + candidates = shared["candidates"] + skills = [s for s in shared["skills"] if s.name in candidates] + return shared["query"], skills + + async def exec_async(self, prep_res): + query, skills = prep_res + skill_desc = "\n".join( + f""" +### {s.name} +说明: {s.description} +参数结构: {s.parameters} + +{s.body} +""" for s in skills) + + prompt = f""" +以下是候选技能: +{skill_desc} + +用户问题: +{query} + +请选择最合适的技能并生成参数。返回 JSON +{{ "skill": "name", "args": {{...}} }} +""" + res = await llm.client.chat_prompt(prompt) + logger.debug(res) + return parse_llm_json(res) + + async def post_async(self, shared, prep_res, exec_res): + shared["decision"] = exec_res return "default" class ExecuteSkillNode(AsyncNode): async def prep_async(self, shared): - if shared["selected_skill"]: - return shared["selected_skill"], shared["skills"] - return "", shared["skills"] + return shared["decision"], shared["skills"] async def exec_async(self, prep_res): decision, skills = prep_res diff --git a/src/pipeline/core/skills.py b/src/pipeline/core/skills.py index 03e0eac..bd90065 100644 --- a/src/pipeline/core/skills.py +++ b/src/pipeline/core/skills.py @@ -6,11 +6,14 @@ import asyncio import yaml from dataclasses import dataclass + @dataclass class Skill: name: str + brief: str description: str parameters: dict + body: str run: callable @@ -25,11 +28,13 @@ async def _load_single_skill(path: str, folder: str) -> Skill | None: content = "" async with aiofiles.open(md, "r", encoding="utf-8") as f: content = await f.read() - frontmatter = content.split("---")[1] - meta = yaml.safe_load(frontmatter) - name = meta.get("name", "") - description = meta.get("description", "") - parameters = meta.get("parameters", "") + frontmatter = content.split("---") + meta = yaml.safe_load(frontmatter[1]) + name = meta.get("name", "") + description = meta.get("description", "") + parameters = meta.get("parameters", "") + brief = meta.get("parameters", "") + body = frontmatter[-1] # 动态加载 run.py(这一步只能 sync) spec = importlib.util.spec_from_file_location(f"{folder}_skill", py) @@ -46,7 +51,14 @@ async def _load_single_skill(path: str, folder: str) -> Skill | None: run = async_run - return Skill(name=name, description=description, parameters=parameters, run=run) + return Skill( + name=name, + brief=brief, + description=description, + parameters=parameters, + body=body, + run=run, + ) async def load_skills(path="./skills") -> list[Skill]: diff --git a/src/skills/search-baidu/SKILL.md b/src/skills/search-baidu/SKILL.md index f9d302d..30700fc 100644 --- a/src/skills/search-baidu/SKILL.md +++ b/src/skills/search-baidu/SKILL.md @@ -1,5 +1,6 @@ --- name: baidu-search +brief: 网页搜索 description: > 当用户问题需要查询互联网公开信息、技术概念解释、 新闻动态、人物或项目信息,且本地知识库无法覆盖时使用。 @@ -20,8 +21,6 @@ parameters: 通过互联网搜索获取公开信息, 并将多个搜索结果整理为可信、简洁的自然语言回答。 ---- - ## 适用场景 - 查询某个概念 / 技术 / 框架是什么 @@ -29,8 +28,6 @@ parameters: - 需要获取较新的公开资料 - 本地知识库未命中或信息不足 ---- - ## 工作流程(SOP) ### 1. 分析用户意图 @@ -40,37 +37,27 @@ parameters: - 现状 / 进展 - 对比或事实查询 ---- - ### 2. 构造搜索关键词 - 使用简洁、明确的中文关键词 - 避免过长句子 - 必要时添加限定词(如:官网、介绍、教程) ---- - ### 3. 获取搜索结果 - 优先关注权威来源 - 多条结果进行交叉参考 ---- - ### 4. 信息整理 - 去除广告和无关内容 - 合并重复信息 - 提炼核心要点 ---- - ### 5. 生成最终回答 - 使用自然语言总结 - 不暴露搜索过程 - 直接回答用户问题 ---- - ## 注意事项 - 搜索结果可能存在噪声,应谨慎判断 - 若信息不充分,应在回答中说明不确定性 -- 不需要解释搜索工具或接口细节 +- 不需要解释搜索工具或接口细节 \ No newline at end of file diff --git a/src/tests/test_nodes.py b/src/tests/test_nodes.py index 3953da4..dca5a71 100644 --- a/src/tests/test_nodes.py +++ b/src/tests/test_nodes.py @@ -122,10 +122,18 @@ async def test_agent(): "skills": await load_skills("./src/skills"), } - decideNode = nodes.DecideNode() + decideNode = nodes.DecideSkillNode() + refineNode = nodes.RefineSkillNode() + disambiguateNode = nodes.DisambiguateSkillNode() runNode = nodes.ExecuteSkillNode() - decideNode >> runNode + decideNode >> refineNode + + refineNode >> runNode + + disambiguateNode >> runNode + + refineNode - "disambiguate" >> disambiguateNode flow = AsyncFlow(decideNode) await flow.run_async(shared)