AnythingLLM中Agent Skills开发实战：从零构建智能对话技能

1次阅读

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

在 AnythingLLM 中扩展 Agent Skills 时，开发者常遇到几个典型问题：

技能注册机制不透明：不知道如何正确注册新技能，导致功能无法被系统识别
上下文处理复杂：难以理解对话状态的保持和传递机制
技能触发混乱：多个技能同时响应时优先级处理不当
调试困难：缺乏有效的日志和错误追踪手段

生命周期管理 ：每个 Skill 经历注册(Register)→初始化(Init)→执行(Execute)→销毁(Destroy) 四个阶段
状态保持：通过 Session Context 实现跨对话轮次的状态维护
会话感知：可以访问完整的对话历史上下文

特性	Agent Skill	普通插件
状态保持	✅ 支持多轮对话状态	❌ 无状态
上下文感知	✅ 完整对话历史	❌ 仅当前输入
触发方式	语义 + 关键词复合触发	单一条件触发

关键词匹配（优先处理）
NLP 意图识别（次级处理）
手动指定技能链（强制触发）

# 在 skills 目录下创建 weather_skill.py
from anythingllm.skill import BaseSkill

class WeatherSkill(BaseSkill):
    def __init__(self):
        super().__init__(
            name="weather",
            description="查询城市天气情况",
            triggers=["天气", "weather"]  # 触发关键词
        )

def execute(self, context):
    # 从对话中提取城市参数
    city = self._extract_entity(context, 'city')

    if not city:
        return "请告诉我您想查询哪个城市的天气"

    # 保留查询状态
    self.session.set('last_city', city)
    return self._get_weather(city)

def _get_weather(self, city):
    # 示例使用 OpenWeatherMap API
    params = {
        'q': city,
        'appid': API_KEY,
        'units': 'metric'
    }

    try:
        resp = requests.get(API_URL, params=params)
        data = resp.json()
        return f"{city}当前天气：{data['weather'][0]['description']}, 温度{data['main']['temp']}℃"
    except Exception as e:
        self.logger.error(f"天气查询失败: {str(e)}")
        return "暂时无法获取天气信息"

使用 session.set()/session.get() 管理短期状态
重要数据建议持久化到数据库

async def execute_async(self, context):
    # 设置超时控制
    try:
        async with asyncio.timeout(10):
            result = await self._async_api_call()
            return result
    except TimeoutError:
        return "请求超时，请稍后再试"