diff --git a/Demo2.md b/Demo2.md new file mode 100644 index 0000000..b2ea360 --- /dev/null +++ b/Demo2.md @@ -0,0 +1,278 @@ +这个 Demo **方向是完全正确的**,已经是「工程化 Agent」而不是玩具了 👍 +下面我不重复你已有的代码,而是**站在“长期扩展 + 降低心智负担”的角度**,给你一版**更简洁、更可扩展的优化方案**,并解释**为什么这样改**。 + +我会分 4 个层次来优化: + +--- + +## 一、当前版本的“真实痛点” + +你现在这套实现 **能跑,但会在 3 个地方开始变重**: + +### 1️⃣ Skill Loader 太“反射式” + +* import + dir 扫描 +* Skill 没有统一接口约束 +* 未来参数校验 / schema 很难加 + +### 2️⃣ DecideSkillNode 过重 + +* Prompt 拼接 +* 技能选择 +* 参数结构约定 + 👉 **职责开始膨胀** + +### 3️⃣ SkillCallNode 依赖技能细节 + +* Node 知道 `skill.call(args)` +* 后面想换 MCP / HTTP Tool / Function Call 会很痛 + +### 4️⃣ Agent Flow 不可复用 + +* 每个 Agent 都要重新写 Decide → Call → Answer + +--- + +## 二、核心优化思想(很重要) + +> **把“技能”从 Agent 里抽出来,变成“能力注册表 + 统一调用协议”** + +最终目标是: + +```text +PocketFlow 只管流程 +Agent 只管决策 +Skill 只管能力 +Tool 只管执行 +``` + +--- + +## 三、优化后的核心设计(精简但更强) + +### ✅ 关键变化一:引入 SkillSpec(统一技能协议) + +### `skills/base.py` + +```python +from dataclasses import dataclass +from typing import Callable, Any + +@dataclass +class SkillSpec: + name: str + description: str + run: Callable[[dict], Any] +``` + +👉 **Skill 现在只是一个数据对象,不是类** + +* 更轻 +* 更好序列化 +* 更像 MCP / OpenAI function + +--- + +## 四、Skill 定义:去 class 化(非常重要) + +### ✨ 优化前 + +```python +class AddCalcSkill: + name = "add-calc" + def call(self, args): ... +``` + +### ✨ 优化后(推荐) + +#### `skills/add_calc/skill.py` + +```python +from skills.base import SkillSpec +from .tool import add + +def run(args: dict): + numbers = args["numbers"] + return {"result": add(numbers)} + +skill = SkillSpec( + name="add-calc", + description="执行多个数字的加法运算", + run=run +) +``` + +📌 好处: + +* Skill = **声明式能力** +* Tool 可自由替换 +* MCP / HTTP / RPC 非常容易接 + +--- + +## 五、Skill Loader:极简 + 稳定 + +### `agent/skill_loader.py` + +```python +import importlib +from pathlib import Path + +def load_skills(root="skills"): + specs = [] + for p in Path(root).iterdir(): + if not p.is_dir(): + continue + try: + mod = importlib.import_module(f"skills.{p.name}.skill") + specs.append(mod.skill) + except Exception as e: + print(f"Skip {p.name}: {e}") + return specs +``` + +📌 不再: + +* 扫 dir +* 猜 class +* 做反射魔法 + +--- + +## 六、Agent Node 的关键优化:职责分离 + +### 1️⃣ 决策 Node:**只负责“选技能 + 参数”** + +```python +class DecideNode(Node): + def prep(self, shared): + return shared["query"], shared["skills"] + + def exec(self, data): + query, skills = data + skill_desc = "\n".join( + f"- {s.name}: {s.description}" for s in skills + ) + + return llm(f""" +可用技能: +{skill_desc} + +用户问题: +{query} + +请返回 JSON: +{{"skill": "...", "args": {{}}}} +""") + + def post(self, shared, _, res): + shared["decision"] = json.loads(res) + return "default" +``` + +--- + +### 2️⃣ Skill Executor Node(关键抽象) + +```python +class ExecuteSkillNode(Node): + def prep(self, shared): + return shared["decision"], shared["skills"] + + def exec(self, data): + decision, skills = data + spec = next(s for s in skills if s.name == decision["skill"]) + return spec.run(decision["args"]) + + def post(self, shared, _, res): + shared["skill_result"] = res + return "default" +``` + +📌 未来你可以把这一行换成: + +* MCP client +* HTTP API +* 子 Agent +* 多 step tool chain + +**Flow 完全不用动** + +--- + +## 七、Flow 进一步简化(可复用) + +### `agent/flow.py` + +```python +def build_agent_flow(): + decide = DecideNode() + exec_skill = ExecuteSkillNode() + answer = AnswerNode() + + flow = Flow(start=decide) + flow.add_edge(decide, exec_skill) + flow.add_edge(exec_skill, answer) + return flow +``` + +👉 这是一个 **通用 Agent Flow** + +* 所有 Skills 通用 +* 所有工具通用 + +--- + +## 八、最终结构(非常干净) + +```text +agent_demo/ +├── agent/ +│ ├── flow.py +│ ├── nodes.py +│ └── skill_loader.py +│ +├── skills/ +│ ├── base.py # SkillSpec +│ ├── add_calc/ +│ │ ├── tool.py +│ │ └── skill.py +│ └── baidu_search/ +│ ├── tool.py +│ └── skill.py +│ +├── llm.py +└── main.py +``` + +--- + +## 九、为什么这套结构“很值钱” + +这套设计天然支持: + +✅ MCP +✅ OpenAI function calling +✅ 多 Agent 协作 +✅ Skill 热插拔 +✅ Tool 多实现(本地 / 远程) +✅ PocketFlow / LangGraph / 自研 Flow 随时切换 + +而且你会发现一句话成立了: + +> **Agent 不知道 Skill 的存在 +> Skill 不知道 Agent 的存在** + +--- + +## 十、你现在可以继续进阶的方向(建议) + +下一步如果你愿意,我可以直接带你做其中一个: + +1️⃣ **给 Skill 自动生成 JSON Schema(强约束参数)** +2️⃣ **支持多 Skill 链式调用(Planner Agent)** +3️⃣ **Skill 结果缓存 / 记忆系统** +4️⃣ **Skill 自动测试(Skill = 可测试单元)** +5️⃣ **对齐 MCP 协议(你这套已经 70% 对齐了)** + +你选一个,我直接给你落代码。 diff --git a/Demo3.md b/Demo3.md new file mode 100644 index 0000000..2832451 --- /dev/null +++ b/Demo3.md @@ -0,0 +1,265 @@ +你这个点**抓得非常准** 👍 +👉 **SKILL.md 不是“可有可无”,而是 Skill 体系里“给 LLM 看、给人读、给系统对齐”的核心文件**。 +前面那版“优化”**确实简化过头了**,把 **Skill 的“描述层”丢了**。 + +下面我给你一版 **“保留 SKILL.md + 仍然保持工程简洁、可扩展”** 的**正确形态**,而且这版是: + +> ✅ 工程友好 +> ✅ LLM 友好 +> ✅ MCP / Function Calling 友好 +> ✅ PocketFlow 原生适配 + +--- + +# 一、先说清楚:SKILL.md 在体系里的“真实角色” + +**SKILL.md ≠ 给人看的 README** + +它的真实职责是三层: + +| 层 | 作用 | +| ------- | ------------------------------- | +| 🧠 LLM | 技能选择、参数理解、意图匹配 | +| 🧩 系统 | Skill 元数据来源(name / desc / args) | +| 👨‍💻 人 | 文档、调试、规范 | + +👉 **skill.py 是“执行体”,SKILL.md 是“能力声明”** + +所以 **两者必须共存,但职责必须分离**。 + +--- + +# 二、推荐的最终 Skill 目录结构(标准答案) + +```text +skills/ +├── add_calc/ +│ ├── SKILL.md # 描述 & schema(给 LLM) +│ ├── skill.py # SkillSpec(给系统) +│ └── tool.py # 纯执行 +``` + +--- + +# 三、SKILL.md 的“正确写法”(不是随便写) + +### `skills/add_calc/SKILL.md` + +```markdown +--- +name: add-calc +description: > + 当用户需要对两个或多个数字进行加法计算时使用。 +parameters: + numbers: + type: array + items: number + description: 需要相加的数字列表 +required: [numbers] +--- + +用于执行基础的加法运算。 +``` + +⚠️ 注意: + +* **YAML frontmatter 是机器可读的** +* Markdown 正文是 **LLM 可读补充说明** + +这已经 **80% 对齐 MCP / OpenAI tools schema** + +--- + +# 四、SkillSpec:同时承载「代码 + 描述」 + +### `skills/base.py` + +```python +from dataclasses import dataclass +from typing import Callable, Any + +@dataclass +class SkillSpec: + name: str + description: str + parameters: dict + run: Callable[[dict], Any] +``` + +👉 **SkillSpec 是系统内部的统一技能模型** + +--- + +# 五、skill.py:不再写描述,描述来自 SKILL.md + +### `skills/add_calc/skill.py` + +```python +from .tool import add + +def run(args: dict): + numbers = args["numbers"] + return { + "numbers": numbers, + "result": add(numbers) + } +``` + +📌 skill.py **只负责执行逻辑** + +* 不再关心 name / description +* 不再关心 prompt +* 更容易测试 + +--- + +# 六、Skill Loader:加载 SKILL.md + skill.py(关键) + +### `agent/skill_loader.py` + +```python +import importlib +import yaml +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 + + skill_md = p / "SKILL.md" + skill_py = f"skills.{p.name}.skill" + + if not skill_md.exists(): + continue + + meta = yaml.safe_load(skill_md.read_text()) + + module = importlib.import_module(skill_py) + + skills.append( + SkillSpec( + name=meta["name"], + description=meta["description"], + parameters=meta.get("parameters", {}), + run=module.run + ) + ) + + return skills +``` + +📌 这一步非常重要: + +* ✅ **SKILL.md → 系统元数据** +* ✅ **skill.py → 可执行能力** +* ✅ **二者彻底解耦** + +--- + +# 七、Decide Node:直接用 SKILL.md 信息 + +### `agent/nodes.py`(关键片段) + +```python +class DecideNode(Node): + def prep(self, shared): + return shared["query"], shared["skills"] + + def exec(self, data): + query, skills = data + + skill_desc = "\n".join( + f""" +- {s.name} + 描述: {s.description} + 参数: {s.parameters} +""" + for s in skills + ) + + return llm(f""" +你可以使用以下技能: +{skill_desc} + +用户问题: +{query} + +请选择一个技能,并返回 JSON: +{{ + "skill": "技能名", + "args": {{ 参数 }} +}} +""") +``` + +👉 **现在 LLM 拿到的是:** + +* 技能名 +* 使用时机 +* 参数结构 + +🎯 幻觉率会明显下降 + +--- + +# 八、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"]) +``` + +--- + +# 九、这套设计为什么是“最优解” + +你现在这套是: + +> **SKILL.md = 能力声明(协议) +> SkillSpec = 运行时模型 +> skill.py = 执行体 +> PocketFlow = 编排引擎** + +它天然支持: + +✅ MCP +✅ OpenAI function calling +✅ 自动生成 tool schema +✅ Skill 市场 / 插件化 +✅ 技能文档即代码 + +而且你以后可以直接加一步: + +```python +SkillSpec → JSON Schema → LLM tools +``` + +**不用改任何 Skill 实现** + +--- + +# 十、一句判断你现在到哪一步了 + +如果一个人能自然意识到: + +> “少了 SKILL.md,这个体系就不完整了” + +那他已经在 **Agent Framework 设计层**,而不是“用 Agent”。 + +--- + +如果你愿意,下一步我可以直接帮你做: + +1️⃣ **SKILL.md → JSON Schema 自动生成** +2️⃣ **参数校验(Pydantic)** +3️⃣ **Skill Planner(多技能组合)** +4️⃣ **MCP Server 适配(你这套非常合适)** + +你点一个,我直接继续。 diff --git a/scripts/run_test.sh b/scripts/run_test.sh index ce8304c..7bbcfdd 100644 --- a/scripts/run_test.sh +++ b/scripts/run_test.sh @@ -1,4 +1,4 @@ #!/usr/bin/env bash export $(cat .env | xargs) -pytest -s -W ignore::DeprecationWarning src/tests/test_utils.py::test_skills +pytest -s -W ignore::DeprecationWarning src/tests/test_nodes.py::test_agent diff --git a/src/pipeline/core/nodes.py b/src/pipeline/core/nodes.py index f48c778..9c6cb06 100644 --- a/src/pipeline/core/nodes.py +++ b/src/pipeline/core/nodes.py @@ -1,8 +1,7 @@ -import json import uuid import re from src.pipeline.core.pocket_flow import AsyncBatchNode, AsyncNode -from src.pipeline.core.utils import fixed_size_chunk, load_document, logger, baidu_search_async +from src.pipeline.core.utils import fixed_size_chunk, load_document, logger, baidu_search_async, parse_llm_json from src.pipeline.core import llm from src.pipeline.core import es from itertools import chain @@ -223,3 +222,59 @@ class RerankNode(AsyncBatchNode): # Agent 相关 # ----------------------------- +class DecideNode(AsyncNode): + """ + 选择技能 + """ + async def prep_async(self, shared): + return shared["query"], shared["skills"] + + 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 + ] + ) + prompt = f""" +你可以使用以下技能: +{skill_desc} + +用户问题: +{query} + +请选择一个技能,并返回 JSON: +{{ + "skill": "技能名", + "args": {{ 参数 }} +}} +""" + res = await llm.client.chat([{"role": "user", "content": prompt}]) + logger.debug(res) + return parse_llm_json(res) + + async def post_async(self, shared, prep_res, exec_res): + shared["selected_skill"] = 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"] + + async def exec_async(self, prep_res): + decision, skills = prep_res + skill = next(s for s in skills if s.name == decision["skill"]) + return await skill.run(**decision["args"]) + + async def post_async(self, shared, prep_res, exec_res): + shared["results"] = exec_res + return "default" diff --git a/src/pipeline/core/skills.py b/src/pipeline/core/skills.py index 8ff4a33..03e0eac 100644 --- a/src/pipeline/core/skills.py +++ b/src/pipeline/core/skills.py @@ -9,7 +9,8 @@ from dataclasses import dataclass @dataclass class Skill: name: str - desc: str + description: str + parameters: dict run: callable @@ -27,7 +28,8 @@ async def _load_single_skill(path: str, folder: str) -> Skill | None: frontmatter = content.split("---")[1] meta = yaml.safe_load(frontmatter) name = meta.get("name", "") - desc = meta.get("description", "") + description = meta.get("description", "") + parameters = meta.get("parameters", "") # 动态加载 run.py(这一步只能 sync) spec = importlib.util.spec_from_file_location(f"{folder}_skill", py) @@ -44,7 +46,7 @@ async def _load_single_skill(path: str, folder: str) -> Skill | None: run = async_run - return Skill(name=name, desc=desc, run=run) + return Skill(name=name, description=description, parameters=parameters, run=run) async def load_skills(path="./skills") -> list[Skill]: diff --git a/src/pipeline/core/utils.py b/src/pipeline/core/utils.py index a1e6014..66c4328 100644 --- a/src/pipeline/core/utils.py +++ b/src/pipeline/core/utils.py @@ -5,6 +5,7 @@ import aiofiles import io import re import sys +import json from pathlib import Path from PIL import Image from loguru import logger @@ -215,3 +216,26 @@ async def baidu_search_async(query: str, max_results: int = 5): } ) return docs + + +def parse_llm_json(text: str) -> dict: + """ + 从 LLM 输出中提取并解析 JSON + """ + # 1. 优先提取 ```json ``` 代码块 + match = re.search(r"```json\s*(\{.*?\})\s*```", text, re.S) + if match: + return json.loads(match.group(1)) + + # 2. 尝试直接解析整个文本 + try: + return json.loads(text) + except json.JSONDecodeError: + pass + + # 3. 兜底:提取第一个 {...} + match = re.search(r"(\{.*\})", text, re.S) + if match: + return json.loads(match.group(1)) + + raise ValueError("无法从 LLM 输出中解析 JSON") diff --git a/src/skills/search-baidu/SKILL.md b/src/skills/search-baidu/SKILL.md index 379c739..f9d302d 100644 --- a/src/skills/search-baidu/SKILL.md +++ b/src/skills/search-baidu/SKILL.md @@ -4,8 +4,13 @@ description: > 当用户问题需要查询互联网公开信息、技术概念解释、 新闻动态、人物或项目信息,且本地知识库无法覆盖时使用。 本技能用于指导如何通过百度搜索获取信息并整理成答案。 -allowed_tools: [baidu_search] -tags: [search, web, baidu] +parameters: + query: + type: str + description: 需要搜索的内容 + max_results: + type: int + description: 返回的搜索结果数 --- # 百度搜索技能 diff --git a/src/tests/test_nodes.py b/src/tests/test_nodes.py index f7fa0f1..3953da4 100644 --- a/src/tests/test_nodes.py +++ b/src/tests/test_nodes.py @@ -1,9 +1,11 @@ import asyncio +from platform import node import pytest import json from src.pipeline.core.pocket_flow import AsyncFlow from src.pipeline.core.utils import logger from src.pipeline.core import llm, es, nodes, utils +from src.pipeline.core.skills import load_skills @pytest.mark.asyncio @@ -108,3 +110,33 @@ async def test_rerank(): await llm.close_client() await es.close_client() + + +@pytest.mark.asyncio +async def test_agent(): + await llm.init_client() + await es.init_client() + + shared = { + "query": "山海经中描述了哪里盛产矿石", + "skills": await load_skills("./src/skills"), + } + + decideNode = nodes.DecideNode() + runNode = nodes.ExecuteSkillNode() + + decideNode >> runNode + + flow = AsyncFlow(decideNode) + await flow.run_async(shared) + + res = await llm.client.chat( + messages=[ + {"role": "system", "content": utils.rag_system_prompt()}, + {"role": "user", "content": utils.rag_user_prompt(shared["query"], shared["results"])}, + ] + ) + + logger.debug(res) + await llm.close_client() + await es.close_client() diff --git a/src/tests/test_utils.py b/src/tests/test_utils.py index 83fb142..c1c40d3 100644 --- a/src/tests/test_utils.py +++ b/src/tests/test_utils.py @@ -16,6 +16,6 @@ async def test_skills(): skills = await load_skills("./src/skills") logger.debug("Loaded skills") for s in skills: - logger.debug(f"\n- {s.name}: {s.desc}") - res = await s.run("今天广州天气如何") + logger.debug(f"\n- {s.name}: {s.description} : {s.parameters}") + res = await s.run(**{"query": "今天星期几", "max_results": 10}) logger.debug(f"\n{json.dumps(res, indent=4, ensure_ascii=False)}")