ai_pipeline/Demo2.md

5.6 KiB
Raw Permalink Blame History

这个 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% 对齐了)

你选一个,我直接给你落代码。