从零理解Agent与Skills的关系：新手开发者避坑指南

1次阅读

共计 3172 个字符，预计需要花费 8 分钟才能阅读完成。

在智能体开发中，Agent（智能体）相当于人类的大脑，而 Skills（技能） 则是大脑可以调用的各种能力。举个生活中的例子：Agent 就像一位厨师长，Skills 则是切菜、炒菜、摆盘等具体技能。厨师长根据顾客需求（输入）决定调用哪些技能组合（决策），最终完成菜品（输出）。

Agent 核心职责：
接收外部输入（如用户请求）
决策技能调用链（Routing）
管理技能生命周期
处理异常与监控
Skill 核心特征：
单一职责原则（一个技能只做一件事）
标准化输入输出（如统一 JSON 格式）
无状态设计（优先使用参数传递而非内部状态）

# 技术术语对照表
GLOSSARY = {
    "Agent": "决策中枢",
    "Skill": "能力单元",
    "Routing": "路由决策",
    "Lifecycle": "生命周期"
}

当两个技能都定义了 utils.py 时：

# 错误案例
skill_a/
    utils.py  # 定义了 handle_data()
skill_b/
    utils.py  # 同名函数不同实现

解决方案：强制技能包命名隔离

# 正确做法
skills/
    finance_utils/  # 领域前缀
        __init__.py
    image_utils/
        __init__.py

陷阱 1 ：在 __init__ 中初始化数据库连接（应懒加载）
陷阱 2 ：技能内部维护全局变量（应通过 Agent 传递上下文）
陷阱 3 ：未实现 shutdown() 方法导致资源泄漏

当技能 A 阻塞时，整个 Agent 停止响应：

[Agent 线程] → [SkillA(10 秒)] → [SkillB] ❌ 被延迟

优化方案：异步调度 + 超时控制

async def run_skill(skill, timeout=3):
    try:
        return await asyncio.wait_for(skill.execute(), timeout)
    except TimeoutError:
        logging.warning(f"{skill} timeout")

from typing import Protocol, runtime_checkable

@runtime_checkable
class SkillProtocol(Protocol):
    skill_version: str

    def execute(self, input: dict) -> dict:
        ...

    @classmethod
    def health_check(cls) -> bool:
        ...

class BaseSkill:
    """所有技能必须继承的基类"""
    registry: dict[str, type] = {}

    def __init_subclass__(cls, skill_name: str):
        if skill_name in cls.registry:
            raise KeyError(f"Duplicate skill: {skill_name}")
        cls.registry[skill_name] = cls
        cls.skill_name = skill_name

# 使用示例
@BaseSkill.register("weather_query")
class WeatherSkill(BaseSkill):
    skill_version = "1.2"

    def execute(self, input):
        return {"temperature": 25}

import importlib
from pathlib import Path

class SkillLoader:
    @staticmethod
    def load_from_path(skill_path: Path):
        """从指定路径加载技能包"""
        try:
            module = importlib.import_module(f"skills.{skill_path.stem}"
            )
            # 自动注册所有继承 BaseSkill 的类
            for obj in module.__dict__.values():
                if isinstance(obj, type) and issubclass(obj, BaseSkill):
                    return obj
        except Exception as e:
            logging.error(f"Load {skill_path} failed: {e}")
            raise

flowchart TD
    A[接收请求] --> B{是否需要权限?}
    B -->|Yes| C[调用 Auth Skill]
    B -->|No| D[选择技能组]
    D --> E[优先级排序]
    E --> F[并发执行]
    F --> G[聚合结果]

预热加载：启动时预加载高频技能

@agent.on_startup
async def preload():
    await warmup("weather", "stock")

结果缓存：对幂等技能使用 LRU 缓存

@lru_cache(maxsize=100)
def get_weather(city: str):
    return requests.get(f"https://api.weather/{city}")

连接池管理：数据库 /API 连接复用

import restrictedpython

def safe_execute(code: str, inputs: dict):
    """在受限环境中执行非信任技能"""
    loc = {"inputs": inputs}
    bytecode = restrictedpython.compile_restricted(code)
    exec(bytecode, {}, loc)
    return loc["result"]

禁止在__init__中执行 I / O 操作
错误案例：初始化时连接数据库
正确做法：懒加载 + 连接池

必须实现版本兼容检查

class PaymentSkill(BaseSkill):
    @classmethod
    def health_check(cls):
        return check_api_version() >= "2.0"

推荐使用 Protocol 定义接口

class ChatProtocol(Protocol):
    def respond(self, text: str) -> str:
        ...

技能输入输出必须序列化
避免技能间直接调用
超时设置必须小于 Agent 响应时限
技能日志需包含唯一请求 ID
…（更多见完整版文档）

# 基础方案提示
class HotReloader:
    def watch(self, path: Path):
        """使用 watchdog 监测文件变化"""
        from watchdog.observers import Observer
        observer.schedule(
            handler=self._on_modified,
            path=str(path)
        )

进阶要求：
– 版本灰度发布
– 旧请求继续使用老版本
– 回滚机制