feat: agent

This commit is contained in:
李如威 2025-12-30 15:43:23 +08:00
parent 61f348327d
commit 5562641431
7 changed files with 408 additions and 53 deletions

View File

@ -254,12 +254,3 @@ SkillSpec → JSON Schema → LLM tools
那他已经在 **Agent Framework 设计层**,而不是“用 Agent”。
---
如果你愿意,下一步我可以直接帮你做:
1⃣ **SKILL.md → JSON Schema 自动生成**
2⃣ **参数校验Pydantic**
3⃣ **Skill Planner多技能组合**
4⃣ **MCP Server 适配(你这套非常合适)**
你点一个,我直接继续。

284
Demo4.md Normal file
View File

@ -0,0 +1,284 @@
很好,这一步你已经**想清楚“为什么拆”了**,那接下来就该**最小侵入式地改 PocketFlow 项目结构**,而不是推倒重来。
我按 **“不破坏 PocketFlow 使用习惯 + 增量升级”** 的原则,给你一套**可直接落地的调整方案**。
---
# 总体目标(你现在要做的事)
**不改变 PocketFlow Node / Flow 编排方式** 的前提下,引入:
* `brief` → **超轻量 Router**
* `description + parameters` → **工具级调用**
* `body` → **按需加载语义**
👉 PocketFlow 继续负责 **流程**
👉 你新增的是 **Skill Semantic Layer**
---
# 一、调整 SkillSpec核心数据结构
## 原来(简化版)
```python
@dataclass
class SkillSpec:
name: str
description: str
parameters: dict
body: str
run: Callable
```
## 调整后(渐进式友好)
```python
from dataclasses import dataclass
from typing import Callable, Any, Optional
@dataclass
class SkillSpec:
name: str
# Stage 1Router 用(极短)
brief: str
# Stage 2工具说明
description: str
parameters: dict
# Stage 3按需加载
body: Optional[str]
run: Callable[[dict], Any]
```
📌 **这是全套调整的“锚点”**
---
# 二、调整 SKILL.md 规范(一次到位)
你现在的 SKILL.md 要**明确支持分层**。
```markdown
---
name: add-calc
brief: 对数字列表执行加法计算
description: 对两个或多个数字进行加法运算
parameters:
numbers:
type: array
items: number
required: [numbers]
---
## 使用场景
- 求和
- 相加
- 计算总数
## 不适用场景
- 需要乘法、减法
- 非数值输入
```
📌 约定:
* `brief`**10~20 token**
* `description`**工具级说明**
* body**可长、但不默认进 prompt**
---
# 三、Skill Loader 调整PocketFlow 友好版)
## 1⃣ 解析 SKILL.mdfrontmatter + body
```python
import re
import yaml
def parse_skill_md(text: str):
pattern = r"^---\s*\n(.*?)\n---\s*\n(.*)$"
m = re.match(pattern, text, re.S)
if not m:
raise ValueError("SKILL.md 缺少 YAML frontmatter")
meta = yaml.safe_load(m.group(1))
body = m.group(2).strip()
return meta, body
```
---
## 2⃣ load_skills最小侵入
```python
import importlib
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
md = p / "SKILL.md"
if not md.exists():
continue
meta, body = parse_skill_md(md.read_text())
module = importlib.import_module(f"skills.{p.name}.skill")
skills.append(
SkillSpec(
name=meta["name"],
brief=meta.get("brief", meta["description"]),
description=meta["description"],
parameters=meta.get("parameters", {}),
body=body,
run=module.run,
)
)
return skills
```
👉 **兼容旧 skill没有 brief 也能跑**
---
# 四、调整 DecideNode真正体现“拆分价值”
## Stage 1只用 brief强烈推荐
```python
class DecideNode(Node):
def prep(self, shared):
return shared["query"], shared["skills"]
def exec(self, data):
query, skills = data
index = "\n".join(
f"- {s.name}: {s.brief}"
for s in skills
)
return llm(f"""
你可以使用以下技能(简要):
{index}
用户问题:
{query}
请选择最合适的技能名称,返回 JSON
{{ "skill": "skill_name" }}
""")
```
📌 **这一轮 token 极低skill 再多也扛得住**
---
## Stage 2可选参数校验 / 二次确认
如果你发现 **参数经常乱填**,再加这一层:
```python
def refine_skill(skill: SkillSpec, query: str):
return llm(f"""
技能:{skill.name}
说明:{skill.description}
参数结构:{skill.parameters}
用户问题:
{query}
请生成该技能需要的参数 JSON。
""")
```
---
## Stage 3仅在歧义时拉 body
```python
def disambiguate(skills, query):
context = "\n".join(
f"### {s.name}\n{s.body}"
for s in skills
)
return llm(f"""
以下是技能详细说明:
{context}
用户问题:
{query}
请选择最合适的技能。
""")
```
👉 **90% 情况不会触发**
---
# 五、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"])
```
这正是你这套设计**最优雅的地方**。
---
# 六、这套改造对 PocketFlow 的影响
| 维度 | 影响 |
| -------------- | ------ |
| Node / Flow | ❌ 不变 |
| 执行模型 | ❌ 不变 |
| Skill 扩展 | ✅ 大幅增强 |
| Token 成本 | ✅ 可控 |
| MCP / tools 对齐 | ✅ 顺滑 |
👉 **PocketFlow 被你“语义升级”,但没被你“污染”**
---
# 七、你现在这个阶段的“最佳状态判断”
你现在已经做到:
> **Skill = 语义声明 + 运行实体 + 渐进式暴露**
这已经是:
* LangGraph
* OpenAI tools
* MCP
背后用的**同一套思想**。
---
## 如果你愿意下一步继续进阶
我可以直接帮你做三选一:
1. 🔌 **Skill → OpenAI tools schema 自动生成**
2. 🧠 **Skill body 向量化SkillRAG**
3. 🧩 **Planner + Router + Executor 三节点拆分**
你现在这个系统,已经非常值得继续往上走了。

View File

@ -158,6 +158,9 @@ class AsyncLLm:
logger.exception("Chat request failed")
return ""
async def chat_prompt(self, prompt:str):
return await self.chat([{"role": "user", "content": prompt}])
async def stream_chat(
self,
messages: list[dict],

View File

@ -222,7 +222,7 @@ class RerankNode(AsyncBatchNode):
# Agent 相关
# -----------------------------
class DecideNode(AsyncNode):
class DecideSkillNode(AsyncNode):
"""
选择技能
"""
@ -231,16 +231,7 @@ class DecideNode(AsyncNode):
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
]
)
skill_desc = "\n".join(f"- {s.name}: {s.brief}" for s in skills)
prompt = f"""
你可以使用以下技能
{skill_desc}
@ -248,27 +239,106 @@ class DecideNode(AsyncNode):
用户问题
{query}
请选择一个技能并返回 JSON
{{
"skill": "技能名",
"args": {{ 参数 }}
}}
请选择最可能相关的技能最多 2 返回 JSON
{{ "candidates": ["skill_name"] }}
"""
res = await llm.client.chat([{"role": "user", "content": prompt}])
res = await llm.client.chat_prompt(prompt)
logger.debug(res)
return parse_llm_json(res)
async def post_async(self, shared, prep_res, exec_res):
shared["selected_skill"] = exec_res
shared["candidates"] = exec_res["candidates"] or []
return "default"
class RefineSkillNode(AsyncNode):
"""
- 不读 body 的情况下
- 尝试生成可用的 skill + args
- 如果失败 返回 disambiguate
"""
async def prep_async(self, shared):
candidates = shared["candidates"]
skills = [s for s in shared["skills"] if s.name in candidates]
return shared["query"], skills
async def exec_async(self, prep_res):
query, skills = prep_res
skill_desc = "\n".join(f"""
### {s.name}
说明: {s.description}
参数结构: {s.parameters}
""" for s in skills)
prompt = f"""
以下是候选技能
{skill_desc}
用户问题
{query}
请选择最合适的技能并生成参数
如果无法确定请返回
{{ "need_disambiguation": true }}
否则返回
{{ "skill": "name", "args": {{...}} }}
"""
res = await llm.client.chat_prompt(prompt)
logger.debug(res)
return parse_llm_json(res)
async def post_async(self, shared, prep_res, exec_res):
if "need_disambiguation" in exec_res and exec_res["need_disambiguation"]:
return "disambiguate"
if "skill" in exec_res and "args" in exec_res:
shared["decision"] = exec_res
return "default"
return "disambiguate"
class DisambiguateSkillNode(AsyncNode):
async def prep_async(self, shared):
candidates = shared["candidates"]
skills = [s for s in shared["skills"] if s.name in candidates]
return shared["query"], skills
async def exec_async(self, prep_res):
query, skills = prep_res
skill_desc = "\n".join(
f"""
### {s.name}
说明: {s.description}
参数结构: {s.parameters}
{s.body}
""" for s in skills)
prompt = f"""
以下是候选技能
{skill_desc}
用户问题
{query}
请选择最合适的技能并生成参数返回 JSON
{{ "skill": "name", "args": {{...}} }}
"""
res = await llm.client.chat_prompt(prompt)
logger.debug(res)
return parse_llm_json(res)
async def post_async(self, shared, prep_res, exec_res):
shared["decision"] = 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"]
return shared["decision"], shared["skills"]
async def exec_async(self, prep_res):
decision, skills = prep_res

View File

@ -6,11 +6,14 @@ import asyncio
import yaml
from dataclasses import dataclass
@dataclass
class Skill:
name: str
brief: str
description: str
parameters: dict
body: str
run: callable
@ -25,11 +28,13 @@ async def _load_single_skill(path: str, folder: str) -> Skill | None:
content = ""
async with aiofiles.open(md, "r", encoding="utf-8") as f:
content = await f.read()
frontmatter = content.split("---")[1]
meta = yaml.safe_load(frontmatter)
frontmatter = content.split("---")
meta = yaml.safe_load(frontmatter[1])
name = meta.get("name", "")
description = meta.get("description", "")
parameters = meta.get("parameters", "")
brief = meta.get("parameters", "")
body = frontmatter[-1]
# 动态加载 run.py这一步只能 sync
spec = importlib.util.spec_from_file_location(f"{folder}_skill", py)
@ -46,7 +51,14 @@ async def _load_single_skill(path: str, folder: str) -> Skill | None:
run = async_run
return Skill(name=name, description=description, parameters=parameters, run=run)
return Skill(
name=name,
brief=brief,
description=description,
parameters=parameters,
body=body,
run=run,
)
async def load_skills(path="./skills") -> list[Skill]:

View File

@ -1,5 +1,6 @@
---
name: baidu-search
brief: 网页搜索
description: >
当用户问题需要查询互联网公开信息、技术概念解释、
新闻动态、人物或项目信息,且本地知识库无法覆盖时使用。
@ -20,8 +21,6 @@ parameters:
通过互联网搜索获取公开信息,
并将多个搜索结果整理为可信、简洁的自然语言回答。
---
## 适用场景
- 查询某个概念 / 技术 / 框架是什么
@ -29,8 +28,6 @@ parameters:
- 需要获取较新的公开资料
- 本地知识库未命中或信息不足
---
## 工作流程SOP
### 1. 分析用户意图
@ -40,35 +37,25 @@ parameters:
- 现状 / 进展
- 对比或事实查询
---
### 2. 构造搜索关键词
- 使用简洁、明确的中文关键词
- 避免过长句子
- 必要时添加限定词(如:官网、介绍、教程)
---
### 3. 获取搜索结果
- 优先关注权威来源
- 多条结果进行交叉参考
---
### 4. 信息整理
- 去除广告和无关内容
- 合并重复信息
- 提炼核心要点
---
### 5. 生成最终回答
- 使用自然语言总结
- 不暴露搜索过程
- 直接回答用户问题
---
## 注意事项
- 搜索结果可能存在噪声,应谨慎判断

View File

@ -122,10 +122,18 @@ async def test_agent():
"skills": await load_skills("./src/skills"),
}
decideNode = nodes.DecideNode()
decideNode = nodes.DecideSkillNode()
refineNode = nodes.RefineSkillNode()
disambiguateNode = nodes.DisambiguateSkillNode()
runNode = nodes.ExecuteSkillNode()
decideNode >> runNode
decideNode >> refineNode
refineNode >> runNode
disambiguateNode >> runNode
refineNode - "disambiguate" >> disambiguateNode
flow = AsyncFlow(decideNode)
await flow.run_async(shared)