从零构建智能Agent系统：Skill设计与实战避坑指南

5次阅读

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

刚接触 Agent 系统开发时，最容易遇到两个经典问题：

状态泄漏(State Leakage)：前一个 Skill 的变量污染了全局上下文(Context)
意图冲突(Intent Collision)：多个 Skill 同时响应同一个用户请求

举个真实案例：天气查询 Skill 未清理 location 缓存，导致后续的航班查询 Skill 错误使用了旧城市数据。流程图展示典型错误路径：

flowchart TD
    A[用户: 查询北京天气] --> B{WeatherSkill}
    B -->| 缓存 location= 北京 | C[用户: 查航班]
    C --> D{FlightSkill}
    D -->| 错误读取北京 | E[返回错误航班]

优点：
响应时间稳定在 10-50ms
代码可解释性强
缺点：
意图识别准确率约 82%(实测数据)
需要手动维护状态转移表

优点：
意图识别准确率可达 93%
自动学习对话模式
缺点：
响应延迟 200-800ms
需要标注大量训练数据

建议：对实时性要求高的客服场景用 FSM，复杂对话场景用 NLP。

from typing import Protocol, runtime_checkable

@runtime_checkable
class SkillProtocol(Protocol):
    """Skill 接口协议"""
    def execute(self, ctx: 'Context') -> bool:
        """
        前置条件: ctx 已通过验证
        后置条件: 返回 bool 表示是否成功处理
        """
        ...

class BaseSkill:
    """所有 Skill 的基类"""
    def __init__(self, skill_id: str):
        self._id = skill_id
        self._dependencies: set[str] = set()

    @property
    def version(self) -> str:
        return "1.0"

import threading
from contextlib import contextmanager

class Context:
    def __init__(self):
        self._data = {}
        self._lock = threading.RLock()

    @contextmanager
    def safe_access(self, key: str):
        """上下文管理器确保线程安全"""
        with self._lock:
            yield self._data.get(key)
            # 自动释放锁

错误做法：直接reload(skill_module)
正确方案：

# 使用 importlib 的 invalidate_caches
import importlib

def hot_reload(skill: BaseSkill):
    importlib.invalidate_caches()
    module = importlib.reload(skill.__module__)
    return getattr(module, skill.__class__.__name__)()

当 Skill 包含 async 代码时：

避免在同步方法中调用asyncio.run()
使用 asyncio.Lock 代替threading.Lock

import time
from functools import wraps

def timeit(metric_name: str):
    """记录 Skill 执行时间"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            start = time.perf_counter()
            result = func(*args, **kwargs)
            elapsed = (time.perf_counter() - start) * 1000
            print(f"[{metric_name}]耗时: {elapsed:.2f}ms")
            return result
        return wrapper
    return decorator

测试数据（4 核 CPU）：