智能体构建：基于SKILL的AI智能体构建：模块化能力编排+实时交互系统全实现.136

一、前言现在不管是做企业应用、内部效率工具还是对外AI产品大家都越来越清楚一件事单纯靠大模型聊天根本解决不了真实业务问题。用户要的不是说得好听而是能真正把事办成比如自动查数据、生成报表、处理工单、对接业务系统甚至完成一整串复杂流程。但直接把一堆 API、工具塞给大模型去调用很容易出现逻辑混乱、执行出错、不可控、不好维护的问题。业务一变代码就要重写模型也跟着乱决策。所以行业里慢慢形成了一种更稳定、更工程化的思路把 、AI的能力拆成一个个标准化、可复用、可管理的技能单元也就是我们常说的SKILL体系。基于SKILL构建 AI 智能体本质上就是让AI从即兴发挥变成按规办事让每一项能力都可注册、可调度、可监控、可迭代。今天我们就从业务落地的实际应用视角完整讲清楚如何从零搭建一套可直接用在生产环境的SKILL驱动AI智能体从架构设计、技能定义到前后端实现、调度执行与部署全部一次性说清道明了解透彻。二、核心概念1. 什么是 SKILLSKILL可以理解未智能体的“能力插件”是封装特定任务逻辑、输入输出、触发条件与执行流程的标准化单元也是AI智能体能够落地业务、解决实际问题的核心基础。区别于普通 Tool工具SKILL不只是简单的调用接口而是包含场景描述、任务目标、触发条件、执行步骤、输出规范、示例案例、依赖关系七大完整要素是可独立开发、独立测试、独立部署、独立迭代的最小能力单元比如“查询天气”这个 SKILL不只是调用天气API还会明确适用场景、输入的城市格式、输出的天气字段、异常处理方式甚至给出测试示例方便开发和维护。区别于子Agent子智能体SKILL更轻量、更聚焦只负责完成某一个单一的具体任务没有独立的上下文管理和记忆能力所有的调度、上下文衔接、记忆存储都由主Agent统一管控比如“生成周报”和“发送邮件”是两个独立SKILL主Agent会根据用户需求调度这两个SKILL协同完成“生成周报并发送”的完整任务而每个SKILL只专注自己的核心功能。2. SKILL 架构的核心价值2.1 模块化将智能体的能力拆分为独立的SKILL单元每个SKILL负责一项具体任务开发时可分工协作不同开发人员负责不同SKILL迭代时只需修改对应 SKILL无需改动整个智能体代码大幅降低系统耦合度提升开发效率。2.2 可扩展支持SKILL的动态注册与卸载新增业务能力时只需按标准开发新的SKILL并放入指定目录系统会自动加载无需修改主 Agent 核心代码淘汰无用能力时直接删除 SKILL 目录即可灵活适配业务迭代需求。2.3 可观测每个SKILL的执行过程都有清晰的日志记录从触发、参数校验、执行到输出全链路可监控、可审计;一旦出现问题能快速定位到具体SKILL无需排查整个系统降低问题排查成本。2.4 高可靠单一SKILL出现故障如 API 调用失败、代码报错时不会影响整个智能体的正常运行系统会自动触发降级机制返回友好提示或调用备用 SKILL保障业务流程不中断提升系统稳定性。2.5 易维护所有SKILL按统一目录结构组织每个SKILL都有标准化的元数据SKILL.md清晰记录其功能、用法和依赖后续维护时只需查看对应 SKILL 的文档和代码无需梳理整个系统逻辑降低维护成本也便于新人快速上手。3. SKILL 的能力分类类别核心功能典型示例实现方式基础交互对话理解、意图识别问候、澄清、结束大模型 Prompt信息获取数据查询、API 调用天气、新闻、数据库查询代码 API内容生成文案、报告、代码写邮件、生成周报大模型 模板工具操作文件、系统、第三方工具读写文件、发送邮件代码执行流程编排多技能组合、任务规划差旅预订、报告生成工作流引擎领域专业垂直行业能力医疗问诊、金融分析领域知识 技能4. SKILL 的标准结构skill-name/├── SKILL.md # 技能元数据核心├── __init__.py # 技能入口├── impl.py # 技能实现逻辑├── examples/ # 输入输出示例├── tests/ # 单元测试└── resources/ # 配置/依赖资源三、SKILL.md 详细说明SKILL.md是SKILL体系最核心的文件相当于每个技能的身份证 说明书 接口文档 运行手册系统、开发者、大模型都靠它工作1. SKILL.md 是什么SKILL.md 是技能的标准化元数据文件采用YAML Markdown 混合格式。它的作用让系统自动识别技能让调度引擎知道何时触发让执行器知道入参和出参让大模型知道如何调用让开发者知道功能、边界、示例让运维知道依赖、权限、版本总结来说没有SKILL.mdSKILL就无法被智能体识别与使用。2. SKILL.md 固定标准格式--- # YAML 格式机器可解析字段核心系统读取这里 name: 技能唯一标识 description: 技能一句话功能描述 version: 版本号 author: 开发者 category: 技能分类 priority: 执行优先级 trigger: keywords: [] intent: [] context: [] confidence: 0.8 input_schema: type: object properties: {} required: [] output_schema: type: object properties: {} examples: - input: output: {} - input: output: {} dependencies: api: [] skills: [] permissions: [] limits: max_retries: 2 timeout: 5 rate_limit: 10/min --- # Markdown 格式注释文档系统不解析 ## 功能说明 ... ## 执行流程 1. ... 2. ... ## 异常处理 ... ## 使用注意事项 ...3. 格式内容详解3.1 基础信息段name: weather_query # 唯一标识英文、下划线不能重复 description: 查询城市天气 # 功能描述给大模型人看 version: 1.0.0 # 版本方便升级/回滚 author: AI-Team # 开发者 category: info_query # 分类info_query / operate / generate / workflow priority: 5 # 优先级1-10数字越高越先匹配用途系统唯一识别技能展示在管理后台做版本控制做技能分类检索3.2 触发规则 triggertrigger: keywords: [天气, 温度, 下雨, 预报] # 关键词匹配 intent: [查询天气, 获取气象] # 意图名称 context: [出行, 旅游, 日常] # 场景上下文 confidence: 0.8 # 触发置信度阈值作用让智能体知道什么时候用这个技能keywords用户输入包含这些词优先匹配intent意图标识大模型可直接选择context场景过滤避免乱触发confidence高于 0.8 才触发避免误调用业务意义精准触发不幻觉、不错误调用、不乱执行。3.3 输入规范 input_schemainput_schema: type: object properties: city: type: string description: 城市名称 days: type: integer default: 1 minimum: 1 maximum: 7 required: [city]作用告诉大模型必须提取什么参数告诉执行器接收什么格式自动做参数校验缺失参数时自动反问用户3.4 输出规范 output_schemaoutput_schema: type: object properties: success: {type: boolean} message: {type: string} data: {type: object}作用统一所有技能输出格式主 Agent 可以稳定解析结果前端可以统一渲染异常统一返回格式3.5 示例 examplesexamples: - input: 杭州明天天气 output: {success:true,data:{city:杭州,...}} - input: 上海未来3天天气 output: {success:true,...} - input: 查询火星天气 output: {success:false,message:不支持该城市}作用给大模型做小样本学习给测试做自动化用例给开发者理解输入输出提升意图识别准确率3.6 依赖 dependenciesdependencies: api: [weather_api] skills: [] permissions: [user:normal]作用系统启动时检查依赖是否就绪权限控制技能之间的依赖编排API 密钥检查3.7 限制 limitslimits: max_retries: 2 timeout: 5 rate_limit: 10/min作用失败自动重试超时强制结束防刷限流保证系统稳定3.8 注释说明文档## 功能说明 用于查询国内城市实时天气与预报。 ## 执行流程 1. 解析用户输入提取城市、天数 2. 校验城市是否支持 3. 调用天气API 4. 返回格式化结果 ## 异常处理 - API 失败返回友好提示 - 城市不支持明确提示 ## 注意事项 仅支持国内一线/新一线城市作用团队协作文档维护手册上线说明业务边界说明4. SKILL.md 必不可少系统自动加载技能大模型知道如何调用调度引擎知道如何触发执行器知道参数格式开发者知道功能边界测试知道用例标准运维知道依赖与权限业务能做到可管控、可审计、可迭代它是SKILL体系的“契约文件”没有它模块化、工程化都无法实现。四、SKILL 智能体整体架构1. 分层架构图用户交互层与前端/API网关作为入口将请求汇聚至 Agent核心层。Agent核心层依次调用 技能注册中心 → 技能调度引擎 → 技能执行层 → 基础设施层完成从意图理解到技能执行的全链路。各层职责清晰便于扩展与维护。2. 核心组件详解2.1 Agent Core智能体核心接收用户请求作为入口统一处理用户输入完成初步清洗与格式归一化。意图理解与任务拆解调用大模型进行语义解析识别用户意图将复杂任务拆解为可执行的子任务。调用 SkillRegistry 匹配技能根据解析结果查询 SkillRegistry 获取可用的技能列表。调度执行引擎编排技能流程将技能组合成工作流DAG并交由调度引擎执行支持串行、并行、条件分支。管理对话记忆与上下文维护当前会话的短期记忆历史轮次、关键实体必要时从长期向量库中召回相关记忆。合成最终响应整合技能执行结果利用大模型生成自然语言回复返回给用户。2.2 SkillRegistry技能注册中心扫描技能目录加载 SKILL.md 元数据自动发现技能模块解析其元数据名称、描述、输入输出、依赖、版本。提供技能查询、匹配、发现能力支持按名称、标签、语义相似度进行技能检索为 Agent Core 提供候选技能列表。管理技能版本、依赖、权限维护技能升级策略确保依赖版本兼容控制技能调用权限如仅管理员可用。支持动态注册/卸载允许运行时添加新技能或停用过时技能无需重启服务实现热插拔。2.3 Skill Scheduler技能调度引擎根据任务规划选择最优技能组合基于成本、耗时、成功率等指标从候选技能中选出最优执行路径。处理技能依赖、并行/串行执行解析技能间的输入输出依赖自动并行无依赖的技能提升执行效率。异常处理、重试、降级对执行失败的技能进行重试如可重试错误若仍失败则执行降级方案如使用备选技能或返回友好提示。执行状态跟踪与日志记录实时记录每个技能的执行状态、耗时、输入输出便于监控和问题排查。2.4 Skill Executor技能执行器加载技能实现代码/Prompt根据技能类型函数式、API、Prompt 模板加载对应的实现逻辑。解析输入校验参数将 Agent Core 传递的上下文参数与技能要求的输入进行匹配校验必填字段与数据类型。执行技能逻辑调用外部工具实际运行技能可能调用本地函数、远程 API、数据库查询或大模型推理。格式化输出返回结果将执行结果按统一格式返回给调度引擎便于后续技能使用或最终响应合成。2.5 Infrastructure基础设施大模型API、第三方服务、数据库提供底层能力支撑如LLM推理、知识库检索、业务系统对接。向量库、文件存储、消息队列存储非结构化数据文档、图片、管理异步任务、实现服务解耦。认证、监控、日志系统保障服务安全实时观测性能指标记录全链路日志用于审计与优化。五、SKILL 智能体完整实现1. 技术选型框架FastAPIAPIWebSocket技能管理自定义 SkillRegistry 目录扫描大模型本地大模型或在线api调用数据校验Pydantic技能发现文件系统监听 动态加载部署Docker 云服务器2. 步骤 1设计 SKILL2.1 SKILL 设计五步法场景定义明确技能适用场景、用户画像、业务目标触发设计关键词、意图、上下文触发条件IO 规范输入输出 SchemaJSON Schema执行逻辑步骤化流程、工具调用、异常处理示例与测试3 输入输出示例覆盖正常、边界、异常2.2 实战天气查询 SKILL设计2.2.1 目录结构skill_agent_project/├── .env # 环境变量API KEY、配置├── main.py# 后端入口FastAPI WebSocket Agent调度├── skill_registry.py # 技能注册中心 动态加载 文件监听├── requirements.txt # 所有依赖清单├── Dockerfile # 生产部署可选├── static/ # 前端静态资源│ └── index.html# 实时对话前端界面└── skills/ # 所有技能统一存放目录核心└── weather_query/ # 示例技能天气查询├── SKILL.md # 技能元数据Schema、触发、描述├── impl.py# 技能真正执行逻辑├── examples/ # 示例输入输出可选└── tests/ # 技能单元测试可选注意加粗的部分完整的一个SKILL所包含的完整文件以及文章种提供的完整示例代码实际目录结构可参考下图2.2.2 完整SKILL.md--- name: weather_query description: 查询指定城市实时天气与未来N天预报支持国内主要城市 version: 1.0.0 author: DevTeam trigger: keywords: [天气, 温度, 下雨, 晴天, 预报, 穿衣] intent: [查询天气, 获取气象信息, 天气建议] context: [出行, 旅游, 日常] input_schema: type: object properties: city: type: string description: 城市名称如杭州、北京 days: type: integer default: 1 minimum: 1 maximum: 7 description: 预报天数1-7天 required: [city] output_schema: type: object properties: success: {type: boolean} message: {type: string} data: type: object properties: current: {type: object} forecast: {type: array} examples: - input: 杭州明天天气怎么样 output: success: true message: 查询成功 data: {current: {...}, forecast: [...]} - input: 查询上海未来3天天气 output: {success: true, ...} - input: 查询火星天气 output: {success: false, message: 不支持该城市} dependencies: api: [open-weather-map] skills: [] --- # 执行流程 1. 解析用户输入提取city和days参数 2. 校验城市有效性仅支持国内城市 3. 调用OpenWeatherMap API获取天气数据 4. 格式化数据为标准输出结构 5. 处理API异常返回友好提示 # 边界限制 - 仅支持中国内地城市 - 预报天数范围1-7天 - API调用频率限制10次/分钟2.2.3 技能实现impl.pyimport requests import os from pydantic import BaseModel, ValidationError from typing import Optional, Dict, Any # 输入输出模型 class WeatherInput(BaseModel): city: str days: int 1 class WeatherOutput(BaseModel): success: bool message: str data: Optional[Dict[str, Any]] None # 技能实现 class WeatherQuerySkill: def __init__(self): self.api_key os.getenv(WEATHER_API_KEY) self.base_url https://api.openweathermap.org/data/2.5 self.supported_cities [杭州, 北京, 上海, 广州, 深圳] # 城市中文名到拼音的映射 self.city_name_map { 杭州: Hangzhou, 北京: Beijing, 上海: Shanghai, 广州: Guangzhou, 深圳: Shenzhen } def execute(self, input_data: dict) - dict: 执行天气查询技能 try: # 1. 校验输入 validated WeatherInput(**input_data) if validated.city not in self.supported_cities: return WeatherOutput( successFalse, messagef不支持城市{validated.city}仅支持{, .join(self.supported_cities)} ).dict() # 2. 调用API将城市名转换为拼音 city_pinyin self.city_name_map.get(validated.city, validated.city) params { q: city_pinyin, appid: self.api_key, units: metric, lang: zh_cn } response requests.get(f{self.base_url}/weather, paramsparams) response.raise_for_status() current response.json() # 3. 获取预报 forecast_params params.copy() forecast_params[cnt] validated.days forecast_resp requests.get(f{self.base_url}/forecast, paramsforecast_params) forecast_resp.raise_for_status() forecast forecast_resp.json() # 4. 格式化结果 result { current: { temp: current[main][temp], desc: current[weather][0][description], humidity: current[main][humidity] }, forecast: [ { date: item[dt_txt], temp: item[main][temp], desc: item[weather][0][description] } for item in forecast[list] ] } return WeatherOutput( successTrue, message查询成功, dataresult ).dict() except ValidationError as e: return WeatherOutput(successFalse, messagef参数错误{str(e)}).dict() except Exception as e: return WeatherOutput(successFalse, messagef服务异常{str(e)}).dict() # 技能入口 def get_skill(): return WeatherQuerySkill()3. 步骤 2实现技能注册中心SkillRegistry核心功能扫描技能目录、加载元数据、提供查询匹配对应文件skill_registry.pyimport os import yaml import importlib from typing import Dict, List, Any from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler class SkillRegistry: def __init__(self, skill_dir: str skills): self.skill_dir skill_dir self.skills: Dict[str, Dict[str, Any]] {} # name: skill_data self._load_all_skills() self._start_watcher() def _load_skill(self, skill_path: str): 加载单个技能 skill_name os.path.basename(skill_path) skill_md os.path.join(skill_path, SKILL.md) if not os.path.exists(skill_md): return # 解析SKILL.md with open(skill_md, r, encodingutf-8) as f: content f.read() if --- in content: yaml_part content.split(---)[1] meta yaml.safe_load(yaml_part) meta[path] skill_path meta[module] fskills.{skill_name}.impl self.skills[meta[name]] meta def _load_all_skills(self): 加载所有技能 self.skills.clear() for item in os.listdir(self.skill_dir): skill_path os.path.join(self.skill_dir, item) if os.path.isdir(skill_path): self._load_skill(skill_path) print(f已加载 {len(self.skills)} 个技能: {, .join(self.skills.keys())}) def match_skills(self, query: str) - List[Dict[str, Any]]: 根据用户查询匹配技能 matched [] query_lower query.lower() for skill in self.skills.values(): # 关键词匹配 for kw in skill[trigger][keywords]: if kw.lower() in query_lower: matched.append(skill) break return matched def get_skill_instance(self, skill_name: str): 获取技能实例 if skill_name not in self.skills: raise ValueError(f技能不存在{skill_name}) skill_meta self.skills[skill_name] module importlib.import_module(skill_meta[module]) return module.get_skill() # 文件监听技能变更自动重载 class SkillChangeHandler(FileSystemEventHandler): def __init__(self, registry): self.registry registry def on_modified(self, event): if event.is_directory: return if SKILL.md in event.src_path: print(技能文件变更重新加载...) self.registry._load_all_skills() def _start_watcher(self): 启动文件监听 event_handler self.SkillChangeHandler(self) observer Observer() observer.schedule(event_handler, self.skill_dir, recursiveTrue) observer.start()4. 步骤 3实现Agent Core 调度执行引擎这里我们构建了一个基于SKILL的 AI 智能体后端服务主要用于演示如何通过意图识别来调用特定功能。核心用途接收用户的自然语言输入分析意图匹配并执行相应的技能如计算、翻译、查询天气等并通过 WebSocket 实现实时对话交互。核心逻辑接收请求 → 意图理解 → 匹配技能 → 执行 → 返回结果文件名称main.pyfrom fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.staticfiles import StaticFiles from fastapi.responses import FileResponse import os from dotenv import load_dotenv from skill_registry import SkillRegistry load_dotenv() app FastAPI(titleSKILL-Based AI Agent) # 确保使用绝对路径 import os BASE_DIR os.path.dirname(os.path.abspath(__file__)) STATIC_DIR os.path.join(BASE_DIR, static) # 挂载静态文件 app.mount(/static, StaticFiles(directorySTATIC_DIR), namestatic) registry SkillRegistry() # 模拟LLM类 class MockLLM: def __init__(self, **kwargs): pass def run(self, **kwargs): # 模拟意图识别逻辑 query kwargs.get(query, ) skills kwargs.get(skills, ) # 简单的基于规则的匹配 - 返回实际的技能名称 if 天气 in query: return weather_query elif 旅游 in query or 景点 in query: return travel_guide elif 翻译 in query: return translate elif 计算 in query or 加 in query or 乘 in query: return calculator else: # 返回第一个技能 skill_list skills.split(, ) return skill_list[0] if skill_list else unknown llm MockLLM() # 模拟意图识别链 class MockIntentChain: def __init__(self, llm, prompt): self.llm llm def run(self, **kwargs): return self.llm.run(**kwargs) intent_chain MockIntentChain(llmllm, promptNone) # 对话记忆模拟版本 class MockMemory: def __init__(self, **kwargs): self.chat_history [] memory MockMemory() # 连接管理 class ConnectionManager: def __init__(self): self.active_connections [] async def connect(self, ws): await ws.accept() self.active_connections.append(ws) def disconnect(self, ws): self.active_connections.remove(ws) async def send(self, msg, ws): await ws.send_text(msg) manager ConnectionManager() # 技能执行逻辑 async def execute_skill_flow(query: str) - str: 执行技能流程 # 1. 匹配技能 matched registry.match_skills(query) if not matched: return 暂不支持该功能请尝试其他问题 # 2. 意图识别选择最优技能 skill_names [s[name] for s in matched] selected_skill_name intent_chain.run(queryquery, skills, .join(skill_names)).strip() # 3. 获取技能实例 try: skill registry.get_skill_instance(selected_skill_name) except: return f技能 {selected_skill_name} 加载失败 # 4. 解析输入简化版实际需用大模型提取参数 input_data _extract_params_from_query(selected_skill_name, query) # 5. 执行技能 result skill.execute(input_data) return f【{selected_skill_name}】结果{result} def _extract_params_from_query(skill_name: str, query: str) - dict: 从用户查询中提取技能参数 query query.strip() # 计算器技能参数提取 if skill_name calculator: # 提取数字和运算符 import re # 匹配表达式数字、运算符、括号 expr_match re.search(r[\d\\-\*\/\^\%\(\)\. ], query) if expr_match: expression expr_match.group(0).strip() return {expression: expression} else: # 查找 计算xxx 或 xxx等于多少 模式 if 计算 in query: expr query.replace(计算, ).replace(等于多少, ).replace(等于, ).strip() return {expression: expr} return {expression: 0} # 翻译技能参数提取 elif skill_name translate: import re result {target_lang: en, text: 你好} # 提取目标语言 if 中文 in query: result[target_lang] zh elif 英文 in query or 英语 in query: result[target_lang] en elif 日语 in query: result[target_lang] ja elif 韩语 in query: result[target_lang] ko elif 德语 in query: result[target_lang] de elif 法语 in query: result[target_lang] fr # 提取待翻译文本引号内的内容 quote_match re.search(r[\«「『](.?)[\」』], query) if quote_match: result[text] quote_match.group(1) else: # 简单提取把xxx翻译成... if 把 in query and 翻译成 in query: text_part query.split(翻译成)[0].replace(把, ).strip() result[text] text_part elif 将 in query and 翻译成 in query: text_part query.split(翻译成)[0].replace(将, ).strip() result[text] text_part return result # 旅游指南技能参数提取 elif skill_name travel_guide: input_data {city: 杭州, days: 1} if 上海 in query: input_data[city] 上海 elif 北京 in query: input_data[city] 北京 if 3天 in query: input_data[days] 3 return input_data # 天气查询技能参数提取 elif skill_name weather_query: input_data {city: 杭州, days: 1} if 上海 in query: input_data[city] 上海 elif 北京 in query: input_data[city] 北京 elif 广州 in query: input_data[city] 广州 elif 深圳 in query: input_data[city] 深圳 if 3天 in query: input_data[days] 3 return input_data # 默认返回空字典 return {} # WebSocket实时对话 app.websocket(/ws) async def websocket_endpoint(websocket: WebSocket): await manager.connect(websocket) try: while True: user_msg await websocket.receive_text() await manager.send(f[你] {user_msg}, websocket) await manager.send([AI处理中...], websocket) # 执行技能流程 response await execute_skill_flow(user_msg) await manager.send(f[AI] {response}, websocket) except WebSocketDisconnect: manager.disconnect(websocket) app.get(/) async def root(): return FileResponse(os.path.join(BASE_DIR, static, index.html)) app.get(/api/skills) async def get_skills(): return {message: SKILL-Based AI Agent, skills: list(registry.skills.keys())} if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8002)重点功能技能注册与匹配利用 SkillRegistry 管理可用技能根据用户输入匹配候选技能列表。意图识别通过 MockLLM模拟大语言模型从候选技能中确定用户的具体意图例如区分“计算”和“翻译”。参数提取内置 _extract_params_from_query 函数使用正则表达式从自然语言中提取关键参数如提取算式、目标语言、城市名。实时通信基于 FastAPI 的 WebSocket 实现全双工通信支持流式或即时响应的聊天界面。静态服务挂载静态文件目录直接提供前端HTML页面index.html以便用户访问交互。5. 步骤 4前端实时交互界面前端提供了一个可以满足交互的页面样式可以根据我们的要求让AI替我们美化即可!DOCTYPE html html langzh-CN head meta charsetUTF-8 titleSKILL AI Agent 实时对话/title style *{margin:0;padding:0;box-sizing:border-box;font-family:Arial} .container{max-width:900px;margin:20px auto;height:90vh;display:flex;flex-direction:column} .chat{flex:1;border:1px solid #ddd;border-radius:10px;padding:20px;overflow-y:auto;background:#f9f9f9} .input{display:flex;margin-top:15px} #msg{flex:1;padding:15px;border:1px solid #ddd;border-radius:8px;font-size:16px} button{padding:15px 25px;background:#007bff;color:white;border:none;border-radius:8px;margin-left:10px;cursor:pointer} button:hover{background:#0056b3} .msg{margin:10px 0;padding:12px 16px;border-radius:8px;max-width:80%} .user{background:#d3eafd;align-self:flex-end;margin-left:auto} .ai{background:#fff;align-self:flex-start;border:1px solid #eee} .log{color:#666;font-size:14px;padding:5px} /style /head body div classcontainer h2 styletext-align:centerSKILL AI 智能体/h2 div classchat idchat/div div classinput input idmsg placeholder输入问题... button onclicksend()发送/button /div /div script const chat document.getElementById(chat); const msgInput document.getElementById(msg); // 获取WebSocket地址处理不同端口 const wsProtocol window.location.protocol https: ? wss: : ws:; const wsHost window.location.host || localhost:8002; const wsUrl ${wsProtocol}//${wsHost}/ws; console.log(Connecting to WebSocket:, wsUrl); let ws null; let isConnected false; function connect() { ws new WebSocket(wsUrl); ws.onopen () { console.log(WebSocket connected); isConnected true; addLog(已连接到服务器); }; ws.onclose () { console.log(WebSocket closed); isConnected false; addLog(连接已断开); // 3秒后尝试重连 setTimeout(connect, 3000); }; ws.onerror (error) { console.error(WebSocket error:, error); addLog(连接错误); }; ws.onmessage (e) { console.log(Received message:, e.data); const div document.createElement(div); div.className e.data.includes([你]) ? msg user : e.data.includes([AI]) ? msg ai : log; div.textContent e.data; chat.appendChild(div); chat.scrollTop chat.scrollHeight; }; } // 初始连接 connect(); function addLog(msg) { const div document.createElement(div); div.className log; div.textContent msg; chat.appendChild(div); chat.scrollTop chat.scrollHeight; } function send() { const msg msgInput.value.trim(); if(!msg) return; if (!isConnected) { addLog(等待连接中...); return; } console.log(Sending message:, msg); try { ws.send(msg); msgInput.value ; } catch (error) { console.error(Send error:, error); addLog(发送失败); } } msgInput.addEventListener(keypress, e { if (e.key Enter) { send(); } }); /script /body /html六、运行与测试1. 启动服务# 1. 创建.env文件echo WEATHER_API_KEY你的天气API密钥 .env# 2. 启动后端python main.py# 3. 访问前端http://127.0.0.1:8002注意app.mount(/static, StaticFiles(directorySTATIC_DIR), namestatic) 静态文件挂载由于我们已经挂载静态文件启动后直接运行http://localhost:8002/即可运行文件2. 测试用例输入查询北京的天气 → 调用 weather_query返回{temp: 12.94, desc: 阴多云, humidity: 23}输入推荐杭州的旅游景点 → 调用travel_guide返回杭州1天旅游推荐西湖, 雷峰塔输入将你好翻译成英文 → 调用translate返回{original_text: 你好, translated_text: Hello, source_lang: auto, target_lang: en}输入计算 123 456 → 调用calculator返回{expression: 123 456, result: 579, steps: [123456 579]}输入自定义无主题内容 → 无匹配技能返回提示七、总结整体来看基于 SKILL 构建 AI 智能体本质上是把一套杂乱、松散的 AI 能力整理成标准化、可管理、可落地的工程化体系。SKILL 作为智能体的最小能力单元不只是简单的工具调用而是自带完整描述、触发规则、输入输出规范、示例与边界约束的标准化插件既比普通 Tool 更严谨又比子 Agent 更轻量、更易调度。而 SKILL.md 作为整个体系的核心契约文件承担了机器可读、人类可懂的双重角色从基础信息、触发条件、参数规范到依赖限制一应俱全让系统能自动识别、调度引擎能精准匹配、大模型能规范调用、开发者能快速维护真正实现技能的可插拔、可扩展、可管控。对于想要把大模型真正落地到业务、从对话交互走向流程执行的场景来说SKILL 体系是一条非常成熟、可直接量产的技术路线也是 AI 智能体从 Demo 走向生产环境的关键一步。