279 lines
5.6 KiB
Markdown
279 lines
5.6 KiB
Markdown
这个 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% 对齐了)**
|
||
|
||
你选一个,我直接给你落代码。
|