5.6 KiB
5.6 KiB
这个 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 里抽出来,变成“能力注册表 + 统一调用协议”
最终目标是:
PocketFlow 只管流程
Agent 只管决策
Skill 只管能力
Tool 只管执行
三、优化后的核心设计(精简但更强)
✅ 关键变化一:引入 SkillSpec(统一技能协议)
skills/base.py
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 化(非常重要)
✨ 优化前
class AddCalcSkill:
name = "add-calc"
def call(self, args): ...
✨ 优化后(推荐)
skills/add_calc/skill.py
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
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:只负责“选技能 + 参数”
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(关键抽象)
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
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 通用
- 所有工具通用
八、最终结构(非常干净)
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% 对齐了)
你选一个,我直接给你落代码。