轻量级 Agent 开发方案：基于 Skill 插件的工具调用循环

引言：为什么需要 Skill 插件系统？

很多 Agent 实现停留在”一次性问答 + 硬编码提示词”阶段：

• 能力固化：所有技能写死在系统提示里，无法扩展
• Token 浪费：每次对话都注入全部技能说明，即使大部分用不上
• 无法定制：用户不能添加自己的技能，只能等待官方更新
• 单轮对话：没有工具调用循环，模型无法”边思考边行动”

本文介绍一个轻量级 Agent 架构，核心特性：

1. Skill = 带 SKILL.md 的目录：YAML frontmatter 声明元数据，正文是技能说明
2. 渐进式披露：系统提示只给 name + description，模型需要时通过 load_skill 工具拉取正文
3. 工具调用循环：流式响应 + 多轮工具调用，模型可以”边思考边行动”
4. 用户自定义：支持前端新增/编辑/删除技能，持久化存储
5. 定向技能：/skill_name 语法预注入技能正文，无需模型再调用 load_skill

---

核心设计

1. Skill 文件格式

每个技能是一个目录，包含 SKILL.md：

skills/
  ├── file-ops/
  │   └── SKILL.md
  ├── api-client/
  │   └── SKILL.md
  └── data-analysis/
      └── SKILL.md

SKILL.md 格式：

---
name: file-ops
title: 文件操作
description: 读取、写入、搜索、统计文件内容
---

# 文件操作技能

## 能力范围
- 读取文件内容（支持分页）
- 写入/覆盖文件
- 搜索文件内容（正则）
- 统计代码行数

## 工具调用
- `read_file(path, offset=1, limit=500)`
- `write_file(path, content)`
- `search_files(pattern, target="content")`

## 使用示例
用户：查看 main.py 的前 50 行
模型：调用 read_file(path="main.py", limit=50)

2. 渐进式披露

问题：如果系统提示包含所有技能的完整说明，Token 消耗巨大。

解法：系统提示只包含技能目录（name + description），模型判断相关后，通过 load_skill 工具拉取正文。

系统提示示例：

可用技能（调用 load_skill 获取详情）：
- file-ops: 读取、写入、搜索、统计文件内容
- api-client: HTTP 请求、JSON 解析、认证管理
- data-analysis: 数据清洗、统计、可视化

当前任务：分析日志文件中的错误模式

模型思考：

需要文件操作能力 → 调用 load_skill(name="file-ops")
获取技能详情 → 看到 read_file + search_files 工具
执行任务 → 调用 search_files(pattern="ERROR", path="logs/")

优势：

• Token 节省：系统提示从 2000 Token → 200 Token
• 灵活扩展：新增技能不影响现有对话
• 按需加载：只加载当前任务相关的技能

3. 工具调用循环

核心引擎：一个同步生成器形式的流式工具调用循环。

伪代码：

async def agent_chat_stream(user_message, history, max_steps=128):
    messages = [system_prompt] + history + [user_message]
    
    for step in range(max_steps):
        # 1. 调用 LLM，流式返回
        tool_calls = []
        async for chunk in llm_stream(messages, tools=ALL_TOOLS):
            if chunk.type == "text":
                yield {"type": "text", "content": chunk.text}
            elif chunk.type == "tool":
                tool_calls.append(chunk.tool)
        
        # 2. 没有工具调用 → 对话结束
        if not tool_calls:
            yield {"type": "done"}
            return
        
        # 3. 执行工具调用
        for call in tool_calls:
            result = await dispatch_tool(call.name, call.arguments)
            messages.append({"role": "assistant", "tool_calls": [call]})
            messages.append({"role": "tool", "content": result})
            
            yield {"type": "tool", "name": call.name, "result": result}
    
    # 4. 超过最大步数 → 强制结束
    yield {"type": "error", "message": "达到最大工具调用次数"}

关键点：

• 流式响应：边生成边推送，用户可以看到”思考过程”
• 多轮循环：工具结果回喂给模型，模型可以继续调用工具
• 异常兜底：任何工具错误都归一化为文本，模型可以解释/换路

4. 技能来源与权限

两个技能来源：

来源	路径	权限	用途
内置	skills/	只读	官方技能，随代码发布
用户	data/skills/	可写	用户自定义，持久化存储

加载逻辑：

def load_all_skills():
    builtin_skills = scan_directory(SKILLS_DIR)
    user_skills = scan_directory(USER_SKILLS_DIR)
    
    # 同名以内置为准（内置只读，不可被覆盖）
    all_skills = {**user_skills, **builtin_skills}
    
    return all_skills

CRUD 接口：

# 新增/更新用户技能
def save_user_skill(name: str, content: str):
    if name in BUILTIN_SKILLS:
        raise PermissionError("内置技能不可覆盖")
    if not re.match(r'^[a-z0-9][a-z0-9-]{0,48}$', name):
        raise ValueError("技能名格式错误")
    
    path = USER_SKILLS_DIR / name / "SKILL.md"
    path.parent.mkdir(exist_ok=True)
    path.write_text(content)

# 删除用户技能
def delete_user_skill(name: str):
    if name in BUILTIN_SKILLS:
        raise PermissionError("内置技能不可删除")
    
    path = USER_SKILLS_DIR / name
    shutil.rmtree(path)

5. 定向技能

语法：/skill_name 任务描述

实现：

def parse_user_message(message: str) -> tuple[str, str | None]:
    """解析 /skill_name 语法"""
    if message.startswith("/"):
        parts = message.split(None, 1)
        if len(parts) == 2:
            skill_name = parts[0][1:]  # 去掉 /
            task = parts[1]
            return task, skill_name
    return message, None

def build_system_prompt(forced_skill: str | None):
    prompt = "可用技能：\n" + skill_catalog()
    
    if forced_skill:
        skill_content = read_skill(forced_skill)
        prompt += f"\n\n【已激活】{skill_content}"
    
    return prompt

优势：

• 用户可以直接指定技能，无需模型判断
• 减少 load_skill 调用，节省 Token 和延迟
• 适合”我知道要用哪个技能”的场景

---

实现示例

1. 后端：技能加载器

# skills.py
from pathlib import Path
import yaml
from typing import Dict, List

SKILLS_DIR = Path(__file__).parent / "skills"
USER_SKILLS_DIR = Path("~/data/skills").expanduser()

def parse_skill_file(path: Path) -> dict:
    """解析 SKILL.md，返回 {name, title, description, content}"""
    content = path.read_text()
    
    # 提取 frontmatter
    if content.startswith("---"):
        end = content.find("---", 3)
        if end > 0:
            frontmatter = yaml.safe_load(content[3:end])
            body = content[end+3:].strip()
            return {
                **frontmatter,
                "content": body,
                "path": str(path)
            }
    
    return {"name": path.parent.name, "content": content}

def load_skills() -> Dict[str, dict]:
    """加载所有技能（内置 + 用户）"""
    skills = {}
    
    # 加载用户技能
    for skill_dir in USER_SKILLS_DIR.glob("*"):
        if skill_dir.is_dir():
            skill_file = skill_dir / "SKILL.md"
            if skill_file.exists():
                skills[skill_dir.name] = parse_skill_file(skill_file)
    
    # 加载内置技能（覆盖同名）
    for skill_dir in SKILLS_DIR.glob("*"):
        if skill_dir.is_dir():
            skill_file = skill_dir / "SKILL.md"
            if skill_file.exists():
                skill = parse_skill_file(skill_file)
                skill["builtin"] = True
                skills[skill_dir.name] = skill
    
    return skills

def skill_catalog() -> str:
    """生成技能目录（name + description，用于系统提示）"""
    skills = load_skills()
    lines = []
    for name, skill in sorted(skills.items()):
        marker = "（内置）" if skill.get("builtin") else ""
        lines.append(f"- {name}: {skill.get('description', '无说明')} {marker}")
    return "\n".join(lines)

def read_skill(name: str) -> str | None:
    """读取技能正文"""
    skills = load_skills()
    if name in skills:
        return skills[name]["content"]
    return None

2. 后端：工具调用循环

# agent_loop.py
import json
import requests
from typing import Iterator, Dict, Any

CORE_TOOLS = [
    {
        "name": "load_skill",
        "description": "加载技能详情",
        "parameters": {
            "type": "object",
            "properties": {
                "name": {"type": "string", "description": "技能名"}
            }
        }
    },
    {
        "name": "read_file",
        "description": "读取文件内容",
        "parameters": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "offset": {"type": "integer", "default": 1},
                "limit": {"type": "integer", "default": 500}
            }
        }
    }
]

TOOL_HANDLERS = {
    "load_skill": lambda args: read_skill(args["name"]) or "技能不存在",
    "read_file": lambda args: Path(args["path"]).read_text()
}

def agent_chat_stream(user_message: str, history: list, max_steps: int = 128) -> Iterator[str]:
    """流式工具调用循环"""
    messages = [
        {"role": "system", "content": build_system_prompt()},
        *history,
        {"role": "user", "content": user_message}
    ]
    
    for step in range(max_steps):
        # 调用 LLM
        tool_calls = []
        response_text = ""
        
        response = requests.post(
            "http://localhost:8000/v1/chat/completions",
            json={
                "model": "claude-3",
                "messages": messages,
                "tools": CORE_TOOLS,
                "tool_choice": "auto",
                "stream": True
            }
        )
        
        for line in response.iter_lines():
            if line.startswith(b"data: "):
                data = json.loads(line[6:])
                if "choices" in data and data["choices"]:
                    delta = data["choices"][0]["delta"]
                    
                    if delta.get("content"):
                        yield json.dumps({"type": "text", "content": delta["content"]})
                        response_text += delta["content"]
                    
                    if delta.get("tool_calls"):
                        tool_calls.extend(delta["tool_calls"])
        
        if not tool_calls:
            yield json.dumps({"type": "done"})
            return
        
        # 执行工具
        messages.append({"role": "assistant", "tool_calls": tool_calls})
        
        for call in tool_calls:
            result = TOOL_HANDLERS[call["function"]["name"]](
                json.loads(call["function"]["arguments"])
            )
            
            yield json.dumps({
                "type": "tool",
                "name": call["function"]["name"],
                "result": result
            })
            
            messages.append({
                "role": "tool",
                "tool_call_id": call["id"],
                "content": result
            })

3. 前端：流式对话组件

// AgentChatPanel.tsx
function AgentChatPanel() {
  const [messages, setMessages] = useState<Message[]>([]);
  const [input, setInput] = useState("");
  const [isLoading, setIsLoading] = useState(false);

  const sendMessage = async () => {
    if (!input.trim() || isLoading) return;
    
    const userMsg = { role: "user", content: input };
    setMessages(prev => [...prev, userMsg]);
    setInput("");
    setIsLoading(true);
    
    const response = fetch("/api/agent/chat", {
      method: "POST",
      body: JSON.stringify({ message: input, messages: messages })
    });
    
    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    
    let assistantMsg = { role: "assistant", content: "", tools: [] };
    
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      
      const line = decoder.decode(value);
      const event = JSON.parse(line);
      
      if (event.type === "text") {
        assistantMsg.content += event.content;
      } else if (event.type === "tool") {
        assistantMsg.tools.push(event);
      } else if (event.type === "done") {
        break;
      }
      
      // 实时更新消息
      setMessages(prev => {
        const newMessages = [...prev, userMsg, assistantMsg];
        return newMessages;
      });
    }
    
    setIsLoading(false);
  };

  return (
    <div>
      <div className="messages">
        {messages.map((msg, i) => (
          <MessageBubble key={i} message={msg} />
        ))}
      </div>
      <div className="input-area">
        <input value={input} onChange={e => setInput(e.target.value)} />
        <button onClick={sendMessage} disabled={isLoading}>
          {isLoading ? "思考中..." : "发送"}
        </button>
      </div>
    </div>
  );
}

---

最佳实践

1. 技能设计原则

• 单一职责：每个技能只解决一类问题
• 自包含：技能正文包含完整的使用说明和工具定义
• 可组合：技能之间可以协作，但不强依赖

2. 工具设计原则

• 幂等性：工具可以安全重试
• 原子性：一次工具调用完成一个完整操作
• 可观察：工具调用轨迹对用户可见

3. 安全边界

• 技能名白名单：拒绝路径穿越（如 ../../etc/passwd）
• 内置技能只读：用户技能不能覆盖内置技能
• 工具输出截断：防止单条工具结果撑爆上下文（如限制 4000 字符）
• 权限控制：敏感操作（如删除技能）需要管理员权限

---

总结

这个轻量级 Agent 架构的核心思想：

1. Skill = 文件：技能是可版本化、可分享、可自定义的 Markdown 文件
2. 渐进式披露：系统提示只给目录，模型按需加载，节省 Token
3. 工具调用循环：流式响应 + 多轮工具调用，模型可以”边思考边行动”
4. 用户自定义：支持前端 CRUD，技能持久化存储

适用场景：

• 内部工具平台（如运维助手、数据分析助手）
• 开发者工具（如代码审查、文档生成）
• 企业知识库（如客服问答、故障排查）

不适用场景：

• 需要复杂状态管理的长期任务（考虑 LangGraph 等框架）
• 需要人类审核的敏感操作（考虑 Human-in-the-loop 模式）

---

延伸阅读

• LangGraph：复杂状态机设计模式
• Karpathy LLM Wiki 模式
• OpenAI Function Calling 文档

---

相关文章：

• [[LangGraph 复杂状态机设计模式]]
• [[Karpathy LLM Wiki 模式：Obsidian 知识库重构指南]]
• [[LLM 在垂直领域的微调策略]]