ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

技能系统插件架构实战:用 TaoToken 统一 Key 让 AI 助手动态加载技能

技能系统插件架构实战:用 TaoToken 统一 Key 让 AI 助手动态加载技能 1. 从“改一行核心代码”到“扔一个目录进去”技能系统到底解决什么问题如果你正在做 AI 助手、Agent 或者任何带工具调用能力的对话应用大概率会遇到这样一个阶段最开始只有聊天后来加了文件读写再后来加了网页搜索、Shell 执行、数据库查询……每加一个功能核心代码就要动一次main.py从 200 行膨胀到 2000 行耦合得一塌糊涂。更尴尬的是有些能力只有特定用户用得上比如人脸识别、特定行业的 API 调用但代码却要跟着主程序一起发布。技能系统Skill System就是来解决这个问题的。它把“能力”从核心代码里拆出来变成一个一个独立的目录每个目录里放一份描述文件通常是SKILL.md和可选的脚本。AI 助手启动时扫描目录动态加载技能运行时按需调用。用户想加功能扔个目录进去就行想禁用删掉目录或者改个名字。这就是插件架构在 AI 助手场景下的落地方式核心目标就三个核心轻量化、按需加载、热插拔。这篇文章我会用一个可复现的骨架来演示以config.toml和settings.json为例讲清楚技能注册、动态加载、热插拔的配置写法并且接入 TaoToken 的统一 Key/API 通道完成一次真实的调用验证。你跟着做能跑通整条技能加载链路。适合谁正在做 AI 助手工具系统、想让自己的 Agent 支持插件化扩展、或者单纯想理解“动态加载”在 Python 里怎么落地的开发者。2. TaoToken 前置统一 Key 与 API 通道让技能调用不再散落各处技能系统跑起来之后每个技能都可能需要调用大模型。天气技能要调模型做意图识别搜索技能要调模型做结果总结总结技能更是直接依赖模型。如果每个技能各自维护一套 API Key、各自的 base_url、各自的超时配置那维护成本会非常高而且一旦要换通道得改十几个地方。TaoToken 在这里的角色就是“统一入口”。它提供兼容 OpenAI 风格的 API 通道你只需要在配置里写一次 Key 和 base_url所有技能共享同一个客户端。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 端点是 https://taotoken.net/api 注意这个不加 UTM 参数直接用于代码里的 base_url。具体来说你需要先拿到一个 API Key。进入控制台创建 Key 的路径是 https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 创建后复制出来后面写进config.toml。如果你只是想先验证模型通不通可以用模型对话页面 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 快速试一下。如果你打算长期做编码类或 Agent 类项目Coding Plan 页面 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 里有更详细的接入说明。这里要强调一点TaoToken 是合规的 API 通道服务不是所谓的“中转”或“代理”它的定位是统一 Key 管理和 API 接入。你在配置里写的就是标准的 OpenAI 兼容格式代码里不需要任何特殊处理。3. 可复制配置config.toml 与 settings.json 骨架技能系统的配置分两层一层是全局配置管 API Key、模型、超时另一层是技能注册表管哪些技能启用、哪些禁用、各自的参数是什么。我用config.toml放全局配置用settings.json放技能注册表这样职责清晰也方便热更新时只监听settings.json。先看config.toml# config.toml [api] base_url https://taotoken.net/api api_key sk-你的TaoToken密钥 model gpt-4o-mini timeout 30 [skills] root_dir ./skills auto_reload true watch_interval 1.5 [logging] level INFO这里base_url就是 TaoToken 的 API 端点api_key从控制台拿。skills.root_dir指定技能目录auto_reload开启热插拔监听watch_interval是文件监听的轮询间隔单位秒。再看settings.json这是技能注册表{ version: 1.0, skills: { weather: { enabled: true, priority: 10, params: { default_city: Beijing } }, tavily-search: { enabled: true, priority: 20, params: { max_results: 5 } }, summarize: { enabled: false, priority: 30, params: {} } } }enabled控制技能是否加载priority决定技能在提示词里的排序params是技能自己的参数。这个文件被修改时热插拔逻辑会重新读取并刷新技能列表不需要重启服务。技能目录结构长这样skills/ ├── weather/ │ ├── SKILL.md │ └── scripts/ │ └── weather.py ├── tavily-search/ │ ├── SKILL.md │ ├── requirements.txt │ └── scripts/ │ └── search.py └── summarize/ └── SKILL.mdSKILL.md是技能的“身份证”格式固定用 YAML front matter 写元信息--- name: weather description: 获取天气信息支持城市查询和天气预报 version: 1.0.0 --- # Weather Skill ## 使用方法 用户询问天气时自动调用。 ## 示例 用户北京今天天气怎么样 助手[调用 weather skill]加载器读取这个文件解析出name和description注入到系统提示词里AI 就知道有这个技能可用。4. 动态加载与热插拔importlib watchdog 的完整实现配置写好了接下来是核心代码。动态加载用importlib.util.spec_from_file_location它允许你在运行时加载任意路径的 Python 文件不需要提前安装成包。# skill_loader.py import importlib.util import json import tomllib from pathlib import Path from typing import Optional class SkillLoader: def __init__(self, config_path: str config.toml): with open(config_path, rb) as f: self.config tomllib.load(f) self.skills_root Path(self.config[skills][root_dir]) self.registry {} self.settings self._load_settings() def _load_settings(self) - dict: settings_file self.skills_root / settings.json if not settings_file.exists(): return {skills: {}} return json.loads(settings_file.read_text(encodingutf-8)) def _parse_skill_md(self, skill_file: Path) - Optional[dict]: if not skill_file.exists(): return None text skill_file.read_text(encodingutf-8) if not text.startswith(---): return None parts text.split(---, 2) if len(parts) 3: return None meta {} for line in parts[1].strip().splitlines(): if : in line: key, value line.split(:, 1) meta[key.strip()] value.strip() meta[body] parts[2].strip() return meta def load_skill(self, skill_dir: Path) - Optional[dict]: skill_name skill_dir.name skill_md skill_dir / SKILL.md meta self._parse_skill_md(skill_md) if not meta: return None skill_setting self.settings.get(skills, {}).get(skill_name, {}) if not skill_setting.get(enabled, True): return None scripts_dir skill_dir / scripts module None if scripts_dir.exists(): for py_file in scripts_dir.glob(*.py): spec importlib.util.spec_from_file_location( fskill_{skill_name}_{py_file.stem}, py_file ) module importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return { name: meta.get(name, skill_name), description: meta.get(description, ), priority: skill_setting.get(priority, 100), params: skill_setting.get(params, {}), module: module, path: str(skill_dir), } def load_all(self) - dict: self.registry {} if not self.skills_root.exists(): return self.registry for skill_dir in sorted(self.skills_root.iterdir()): if not skill_dir.is_dir(): continue skill self.load_skill(skill_dir) if skill: self.registry[skill[name]] skill return self.registry def build_skill_prompt(self) - str: lines [# Available Skills\n] for skill in sorted(self.registry.values(), keylambda s: s[priority]): lines.append(f- **{skill[name]}**: {skill[description]}) return \n.join(lines)热插拔用watchdog监听settings.json和SKILL.md的变化# hot_reload.py from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler from pathlib import Path from skill_loader import SkillLoader class SkillReloadHandler(FileSystemEventHandler): def __init__(self, loader: SkillLoader): self.loader loader def on_modified(self, event): if event.is_directory: return path Path(event.src_path) if path.name in (settings.json, SKILL.md): print(f[reload] detected change: {path}) self.loader.settings self.loader._load_settings() self.loader.load_all() print(f[reload] skills now: {list(self.loader.registry.keys())}) def start_watch(loader: SkillLoader): observer Observer() handler SkillReloadHandler(loader) observer.schedule(handler, pathstr(loader.skills_root), recursiveTrue) observer.start() return observer启动入口# main.py from skill_loader import SkillLoader from hot_reload import start_watch import time loader SkillLoader(config.toml) loader.load_all() print([init] loaded skills:, list(loader.registry.keys())) print(loader.build_skill_prompt()) observer start_watch(loader) try: while True: time.sleep(1) except KeyboardInterrupt: observer.stop() observer.join()跑起来之后你修改settings.json把summarize的enabled改成true终端会立刻打印重新加载的日志技能列表里就多出summarize。这就是热插拔的效果。5. 验证请求用 TaoToken 统一 Key 完成一次真实调用技能加载好了接下来验证 API 通道能不能通。我写一个最小的调用脚本用config.toml里的配置走 TaoToken 的 API 端点让模型根据技能提示词做一次意图识别。# verify_call.py import tomllib from openai import OpenAI from skill_loader import SkillLoader with open(config.toml, rb) as f: config tomllib.load(f) client OpenAI( base_urlconfig[api][base_url], api_keyconfig[api][api_key], timeoutconfig[api][timeout], ) loader SkillLoader(config.toml) loader.load_all() skill_prompt loader.build_skill_prompt() response client.chat.completions.create( modelconfig[api][model], messages[ {role: system, content: skill_prompt}, {role: user, content: 北京今天天气怎么样}, ], temperature0.2, ) print(模型回复) print(response.choices[0].message.content) print(\n本次使用的技能提示词) print(skill_prompt)运行python verify_call.py如果配置正确你会看到模型回复里提到调用weather技能并且打印出当前加载的技能列表。实测下来从发出请求到收到响应延迟在正常范围内说明 TaoToken 的 API 通道是通的。成功结果的判断标准有三个第一终端没有报AuthenticationError或ConnectionError第二response.choices[0].message.content有内容返回第三技能提示词里包含你启用的技能名称。三个都满足说明技能加载链路和 API 调用链路都跑通了。如果你在验证模型本身的能力可以顺便去模型对话页面 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 对比一下不同模型的表现。如果你打算把这个技能系统用到长期编码或 Agent 项目里Coding Plan 页面 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 有更完整的接入方案。6. 本篇常见错排查路径、依赖、循环导入、热更新延迟第一个坑是路径问题。技能脚本里如果用相对路径读文件运行时会找不到因为工作目录不一定是技能目录。解决办法是用__file__拿绝对路径from pathlib import Path SKILL_DIR Path(__file__).resolve().parent.parent config_file SKILL_DIR / config.json第二个坑是循环导入。技能脚本import核心模块核心模块又加载技能直接死循环。解决原则是技能脚本只暴露纯函数不要反向依赖核心。核心模块通过module对象调用技能函数技能函数不import核心。第三个坑是依赖缺失。技能有requirements.txt时不要在加载时自动pip install因为可能搞乱用户环境。正确做法是检查依赖提示用户手动安装def check_dependencies(skill_dir: Path) - list: req_file skill_dir / requirements.txt if not req_file.exists(): return [] missing [] for line in req_file.read_text().splitlines(): pkg line.split()[0].split()[0].strip() if not pkg: continue try: importlib.import_module(pkg.replace(-, _)) except ImportError: missing.append(pkg) return missing第四个坑是热更新延迟。watchdog在某些系统上事件触发有延迟改完文件等 1 到 2 秒再测试。如果一直不触发检查watch_interval是否设置合理或者换用轮询模式。第五个坑是settings.json格式错误。JSON 不支持注释多一个逗号就会解析失败。建议用json.loads时捕获异常并打印具体行号方便定位。第六个坑是 API Key 写错。TaoToken 的 Key 从控制台创建后复制注意不要带空格。如果报401先去 https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 确认 Key 是否有效。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 里面有完整的参数说明和错误码对照。7. 继续往下走把技能系统接进你的 AI 助手到这里技能注册、动态加载、热插拔、API 调用验证这条链路已经完整跑通了。你可以把SkillLoader和SkillReloadHandler直接搬进自己的项目把config.toml里的base_url和api_key换成 TaoToken 的配置技能目录按SKILL.md格式往里扔就行。下一步可以做的方向有几个给技能加权限控制比如某些技能只对特定用户开放给技能加版本管理SKILL.md里写version加载时做兼容性检查给技能加依赖隔离长期方案是每个技能独立 venv 或容器化。这些都不难核心的动态加载骨架已经在了。如果你在接入过程中遇到 API 通道的问题先去 API Keys 页面 https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 检查 Key再去接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 对照参数。如果你用的是 Claude Code 或 Anthropic 风格的调用ClaudeCodeAnthropic 页面 https://taotoken.net/claudecode-anthropic?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 有专门的配置说明。技能系统本身不复杂复杂的是边界情况把上面六个坑提前避开基本就能稳定运行了。
返回列表