Agent Skill 从入门到实战：如何高效构建智能体技能链

6次阅读

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

在智能体 (Agent) 开发中，开发者常遇到以下典型问题：

技能复用困难：不同场景下重复开发相似功能，缺乏标准化接口
上下文管理混乱：跨技能数据传递依赖全局变量或复杂参数传递
调试成本高 ：错误在技能链(Skill Chain) 中传播时难以定位
性能不稳定：未考虑超时、重试等容错机制

# 典型反模式：所有逻辑耦合在单一函数
def handle_request(input):
    # 技能 A 逻辑
    a_result = _skill_a(input)
    # 技能 B 逻辑 
    b_result = _skill_b(a_result)
    # 技能 C 逻辑
    return _skill_c(b_result)

缺点：
– 修改任一技能需重新测试整个流程
– 难以单独扩展或替换组件

[用户请求]
    ↓
[路由技能] → 根据输入选择技能链
    ↓
[技能 A] → 通过上下文 (Context) 传递结果
    ↓
[技能 B] → 可插拔替换实现
    ↓
[响应组装]

优势：
– 每个技能 (Skill) 独立开发测试
– 支持运行时动态编排
– 上下文隔离确保安全性

from typing import Callable, Dict, Any

class SkillRegistry:
    def __init__(self):
        self._skills: Dict[str, Callable] = {}

    def register(self, name: str) -> Callable:
        def decorator(func: Callable) -> Callable:
            self._skills[name] = func
            return func
        return decorator

    def trigger(self, name: str, context: Dict[str, Any]) -> Any:
        if name not in self._skills:
            raise ValueError(f"Skill {name} not registered")
        return self._skills[name](context)

# 初始化注册中心
registry = SkillRegistry()

@registry.register("weather_query")
def query_weather(ctx: dict) -> dict:
    """查询天气技能"""
    location = ctx.get("location")
    # 实际业务逻辑...
    return {"temperature": 25, "status": "sunny"}

def inject_context(*required_keys):
    def wrapper(func):
        def inner(ctx):
            missing = [k for k in required_keys if k not in ctx]
            if missing:
                raise ValueError(f"Missing context keys: {missing}")
            return func(ctx)
        return inner
    return wrapper

@registry.register("travel_advice")
@inject_context("location", "weather")
def generate_advice(ctx):
    # 可安全使用上下文中的 location 和 weather 字段
    return f"建议携带{' 雨伞 'if ctx['weather']['status'] =='rainy'else' 防晒 '}"

from concurrent.futures import ThreadPoolExecutor, TimeoutError
import functools

def with_timeout(timeout: int):
    def decorator(func):
        @functools.wraps(func)
        def wrapped(ctx):
            with ThreadPoolExecutor() as executor:
                future = executor.submit(func, ctx)
                try:
                    return future.result(timeout=timeout)
                except TimeoutError:
                    # 记录超时日志
                    ctx.setdefault("errors", []).append(f"{func.__name__} timeout")
                    return None
        return wrapped
    return decorator

@registry.register("slow_api_call")
@with_timeout(3)
def call_external_api(ctx):
    # 模拟耗时操作
    import time
    time.sleep(5)  # 实际会触发超时
    return "result"

# 应用启动时预加载资源
@registry.register("nlp_processing")
class NLPSkill:
    def __init__(self):
        self._model = load_ai_model()  # 初始化耗时操作

    def __call__(self, ctx):
        return self._model.predict(ctx["text"])

from functools import lru_cache

@registry.register("expensive_calculation")
@lru_cache(maxsize=100)
def calculate(ctx):
    # 对相同输入只计算一次
    return heavy_computation(ctx["input"])

from threading import Semaphore

class RateLimitedSkill:
    def __init__(self, max_concurrent=5):
        self.sem = Semaphore(max_concurrent)

    def __call__(self, ctx):
        with self.sem:
            return do_work(ctx)

所有写操作技能应支持重复执行
使用唯一请求 ID 防止重复处理

@registry.register("payment")
def make_payment(ctx):
    payment_id = ctx["payment_id"]
    if check_already_processed(payment_id):
        return {"status": "duplicate"}
    # 实际支付逻辑...

import logging

logging.basicConfig(format="%(asctime)s [%(levelname)s] %(skill)s: %(message)s",
    level=logging.INFO
)

def logged_skill(name):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(ctx):
            logger = logging.getLogger(name)
            try:
                logger.info(f"Processing with {ctx}")
                result = func(ctx)
                logger.info(f"Completed successfully")
                return result
            except Exception as e:
                logger.error(f"Failed: {str(e)}", exc_info=True)
                raise
        return wrapper
    return decorator

避免技能中缓存无限增长
使用弱引用 (weakref) 处理交叉引用
定期检查技能内存占用

任务要求：为现有技能添加熔断机制(Circuit Breaker)

# 现有技能代码
@registry.register("unstable_api")
def call_unstable_api(ctx):
    response = requests.get("https://unstable.example.com")
    return response.json()

# 目标实现特性：# 1. 当连续失败次数超过阈值 (如 3 次) 时
# 2. 熔断 5 分钟不再实际调用 API
# 3. 直接返回缓存值或默认值
# 4. 半开状态尝试恢复

提示：
– 使用类属性记录状态
– 考虑线程安全问题
– 记录状态切换日志

通过模块化设计、明确的上下文管理和完善的异常处理，可以构建出高可用的智能体技能链。建议从简单技能开始逐步验证架构，再扩展到复杂业务流程。实际开发中应特别注意技能间的接口约定和性能监控。

正文完