ai_pipeline/Demo2.md

279 lines
5.6 KiB
Markdown
Raw Blame History

This file contains invisible Unicode characters

This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

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