免费获取学习方案
ARTICLE DETAIL

资讯详情

深耕编程基础知识与建站技术分享的一线实战洞察。

AI Agent开发实战:Skills设计核心与腾讯云部署指南

AI Agent开发实战:Skills设计核心与腾讯云部署指南 1. 为什么我建议每个 Agent 项目都认真对待 Skills 这一层先说一个我自己踩过的坑。去年我做了一个所谓的全能助手底层接了大模型也做了联网搜索、天气查询、待办管理这几个工具。一开始觉得自己挺了不起等真正用起来才发现一个问题Agent 只会调用工具根本不会干活。它能帮你查天气但不会结合你的日程告诉你明天要下雨建议带伞而且你早上 9 点有会最好提前 20 分钟出门。问题出在哪出在我在设计阶段就把技能想得太简单了。后来我把这些能力按照 AI Skills 的思路重新整理了一轮效果天差地别。再配合腾讯云把整套服务部署上线整个项目才算真正养成了一个能独立处理复杂任务的 Agent。这也是我写这篇文章的原因——把你从能用做到好用中间那些容易被忽略的环节以及腾讯云上的部署实践一次性讲清楚。这篇文章适合谁两种人第一种是已经在做 Agent 开发但总觉得自己的 Agent差点意思的开发者第二种是想把 Agent 部署上线但被环境配置、外网访问、服务稳定性这些事折磨过的朋友。本文会围绕 AI Skills 的设计与实现、腾讯云服务器上的部署流程、以及生产环境下的稳定性优化展开全程不绕弯子直接讲操作和思路。需要先明确一点AI Skills 不是一个神秘的新技术它本质上是一种结构化的技能封装方式——让大模型以更可控、更可靠的方式使用外部能力。它解决的问题很具体Agent 怎么知道自己有哪些能力、什么情况下该用哪个能力、调用时参数怎么填、返回结果怎么解析。没有这层封装Agent 就像一个大厨进了没有菜谱、没有备菜的厨房再有本事也做不出一桌像样的菜。2. Skills 设计的核心原则不是把 API 包一层而是把能力教给模型2.1 从一次失败的 Function Calling 说起在讲 Skill 设计之前我先还原一个真实场景。我之前在一个项目里给 Agent 加了个查询数据库的功能用的是最粗暴的方式——直接让模型生成 SQL 然后执行。听起来很直接对吧结果惨不忍睹模型不知道数据库里有哪些表、哪些字段生成了一堆不存在的列名模型把查询最近一周的数据翻译成了WHERE date 2024-01-01但系统里存的是时间戳更离谱的是有一次模型生成了一条DELETE FROM users WHERE ...执行前被我手动拦下了不然后果没法收拾。这说明一个核心问题让模型自由发挥地调用工具本质上是在赌模型的运气。大模型确实很聪明但它对系统边缘情况的了解几乎为零。而 AI Skills 要解决的就是这个——把模糊的意图翻译成确定的动作。2.2 Skill 描述词决定模型判断质量的胜负手设计一个 Skill最重要的不是代码逻辑而是描述Description。模型是通过描述来判断当前用户的这句话该不该用这个技能的。描述写得烂再好的实现也白搭。以我部署在腾讯云上的一个定时巡检技能为例skill_description 当用户要求对服务进行定期健康检查、巡检、或定时执行某段逻辑时使用此技能。 适用场景示例 - 用户说每天凌晨2点检查一下服务器状态 - 用户说每5分钟拉取一次最新的监控数据 - 用户说如果磁盘使用率超过80%就告警 不适用场景 - 用户只是问服务器现在的状态如何一次性查询请使用 query_server_status - 用户要求立即执行某个动作 执行此技能后必须向用户输出下一次任务的预计执行时间。 我特意加了不适用场景这一块。实践下来发现加了这部分的技能描述触发准确率能提升很多。原因很简单模型很多时候是没得选才乱选——它不知道其他技能的存在也不知道什么时候不该用这个技能你把这些边界画清楚它自然就乖了。实操技巧描述词不要只写该技能用于执行巡检任务这是初学者最常见的错误。要写清楚触发条件场景示例不适用场景执行后需要向用户汇报什么。你写给模型看的描述应该像写给一个新同事看的操作手册而不是写给搜索引擎看的关键词堆砌。2.3 输入输出 Schema 的设计细节Skills 的输入输出设计直接决定 Agent 的稳定性。我给你几个我总结出来的硬性规则设计点错误做法正确做法参数命名用a、b、data这种模糊名用server_id、check_interval、alert_threshold这种含义明确的名称参数约束不写取值范围写明minimum、maximum、enum等约束输出格式直接返回 JSON 原文先做标准化封装再返回给模型错误处理异常时返回空字符串返回结构化的error信息告诉模型发生了什么关于参数约束我吃过一次大亏。当时写了一个发送告警的技能参数里有channel渠道我没限定枚举值。结果模型自己造了个叫dingtalk_webhook的渠道——我们的告警系统压根没有这个渠道于是静默失败了。后来我把enum限定为[wechat, sms, email, webhook]再没出过这种问题。关于输出格式我的做法是统一封装成下面这种结构{ success: true, data: { cpu_usage: 67.5, memory_usage: 82.1, disk_usage: 43.2 }, suggestions: [内存使用率超过80%建议检查是否有内存泄漏], timestamp: 2024-06-15T10:30:0008:00 }为什么要加suggestions这个字段因为大部分情况下模型拿到原始数据后依然不知道该怎么理解、该怎么回复用户。你直接给它加工好的建议既降低了模型的理解成本也保证了回复质量。提示Skills 的输出不是给程序看的是给模型看的。你的输出设计得越友好模型越不需要猜测你的意图。3. 腾讯云部署实战从零到一跑通你的第一个 Agent3.1 服务器选型和初始化配置如果你是从零开始部署 Agent我的建议是不要一上来就上大集群。一个 Agent 服务在初期对资源的需求并不高甚至可以跑在一台轻量服务器上。我自己用的是腾讯云的轻量应用服务器2核4G的配置跑一个基于 FastAPI 的 Agent 服务绰绰有余。选型时注意几点地域选择优先选离你的用户近的地域。如果你面向国内用户选上海、广州这些节点就对了如果面向海外考虑新加坡等节点。操作系统我推荐 Ubuntu 22.04 LTS。稳定、社区资料全、兼容性好遇到问题随便一搜就有答案。带宽Agent 服务主要是接口调用流量不算大但如果你要做流式输出SSE需要确保带宽不是瓶颈。5M 起步基本够用。初始化服务器的几个基础操作# 更新系统包 sudo apt update sudo apt upgrade -y # 安装 Python 3.10 和虚拟环境 sudo apt install python3-pip python3-venv -y # 创建项目目录 mkdir -p /opt/agent-project cd /opt/agent-project # 创建虚拟环境 python3 -m venv venv source venv/bin/activate这些操作你肯定很熟了我不过多展开。重点想说一下强烈建议所有服务都用 systemd 托管不要用nohup直接跑。原因后面会讲。3.2 部署前的 Skill 注册与加载机制设计在部署之前你要考虑清楚一个问题Skills 如何注册到 Agent 服务里我见过很多人的做法是在代码里硬编码技能列表。这种做法在技能少的时候没毛病但技能一旦多了维护成本就上来了。而且每次新增技能都要重新部署代码很痛苦。我的做法是做一个Skill Registry技能注册中心。每个 Skill 是一个独立的 Python 模块放在skills/目录下通过统一的接口暴露给 Agent/opt/agent-project/ ├── main.py # FastAPI 入口 ├── agent/ │ ├── core.py # Agent 核心逻辑 │ ├── registry.py # 技能注册与发现 │ └── llm_client.py # 模型调用客户端 ├── skills/ │ ├── scheduled_check/ # 定时巡检技能 │ │ ├── __init__.py │ │ ├── skill.py # 技能定义与逻辑 │ │ └── schema.py # 输入输出 Schema │ ├── wechat_notify/ # 微信通知技能 │ └── serve_restart/ # 服务重启技能 └── requirements.txt每个技能的skill.py都要实现一个统一接口from typing import Any, Dict class BaseSkill: name: str skill_name description: str 描述词 async def execute(self, params: Dict[str, Any]) - Dict[str, Any]: 执行技能逻辑返回标准化输出 raise NotImplementedError注册中心在启动时扫描skills/目录自动加载所有技能然后把它们的name和description拼进系统提示词System Prompt里。这样每次新增技能只需要把目录丢进去重启服务即可不需要改代码逻辑。# registry.py 核心代码 import importlib import pkgutil from typing import List import skills class SkillRegistry: def __init__(self): self._skills: Dict[str, BaseSkill] {} def discover(self): 扫描 skills 包下的所有模块 for module_info in pkgutil.iter_modules(skills.__path__): if module_info.name.startswith(_): continue module importlib.import_module(fskills.{module_info.name}) skill_class getattr(module, Skill, None) if skill_class: skill skill_class() self._skills[skill.name] skill print(f[Registry] 已注册技能: {skill.name}) def get_skill(self, name: str) - BaseSkill: return self._skills.get(name) def get_skill_descriptions(self) - str: 生成给模型看的技能列表描述 lines [] for name, skill in self._skills.items(): lines.append(f- {name}: {skill.description}) return \n.join(lines)这套机制帮了我大忙。有一次我需要同时管理 6 个技能每个技能的参数模型还都不一样硬编码绝对疯了。用注册制之后加技能半个小时搞定而且互不影响。3.3 FastAPI 服务封装与模型接入服务的入口我用的 FastAPI原因很直接它对异步的支持好天然适合做 SSE 流式输出而且文档全、踩坑少。# main.py 核心代码 import json from fastapi import FastAPI, HTTPException from pydantic import BaseModel from agent.core import AgentEngine from agent.registry import SkillRegistry app FastAPI(title全能 Agent 服务) registry SkillRegistry() registry.discover() class ChatRequest(BaseModel): message: str session_id: str default app.post(/agent/chat) async def chat(req: ChatRequest): engine AgentEngine(registry) reply await engine.process(req.session_id, req.message) return {reply: reply} app.post(/agent/skills/{skill_name}/execute) async def execute_skill(skill_name: str, params: dict): skill registry.get_skill(skill_name) if not skill: raise HTTPException(status_code404, detailf技能 {skill_name} 不存在) result await skill.execute(params) return result你注意到我没有直接把技能暴露给用户而是通过 Agent 核心引擎来调用。这一点非常重要不要让用户直接调用技能接口而是让用户先与大模型对话由模型判断该调用哪个技能。如果你让用户直接调技能等于把决策权从模型手里夺走了Agent 就退化成普通 API 服务了。Agent 核心引擎的调用逻辑大概是接收用户消息从 System Prompt 里携带技能列表描述调用大模型如果模型返回的结果包含call_skill动作解析出技能名和参数根据技能名从注册中心取出技能实例执行把技能执行结果返回给模型让模型生成面向用户的最终回复返回最终回复给用户。大模型接入我用的协议是 OpenAI 兼容格式这样换模型厂商只需要改base_url和api_key两个配置不用改业务代码。腾讯云自身也有模型服务但考虑到很多团队已经用着别家的模型保留 OpenAI 兼容的抽象层会灵活很多。你可以理解为这个抽象层就是一层翻译官让角色和工具之间解耦。4. 部署踩坑记录那些文档上不说、但一定会遇到的事4.1 systemd 服务托管与异常退出恢复前面我卖了个关子——为什么一定要用 systemd 托管。因为我第一次部署 Agent 服务的时候直接用nohup python main.py 结果一个周末回来服务挂了都没人发现。后来一查日志原来是内存溢出被内核给杀了。用 systemd 托管最大的好处有两个自动重启和开机自启。不需要写复杂的守护脚本系统原生帮你管好。# /etc/systemd/system/agent.service [Unit] DescriptionAI Agent Service Afternetwork.target [Service] Typesimple Userubuntu WorkingDirectory/opt/agent-project EnvironmentPATH/opt/agent-project/venv/bin ExecStart/opt/agent-project/venv/bin/python main.py Restartalways RestartSec5 # 内存保护 MemoryMax2G [Install] WantedBymulti-user.target启用服务sudo systemctl daemon-reload sudo systemctl enable agent sudo systemctl start agent特别提一下MemoryMax2G这个参数。我这台服务器是 4G 内存给服务限定 2G 上限配合Restartalways即使出现内存泄漏导致进程被杀systemd 也会在 5 秒后自动拉起。实测下来服务可用性从随缘提升到了99.9%以上。4.2 模型调用超时与熔断处理第二个坑是关于模型调用的。最开始的时候我直接在大模型调用外层包了个try...except以为这样就能处理所有异常。结果某次模型服务响应特别慢超过了 60 秒用户的消息直接卡住前面的请求排队的排队、超时的超时整个服务变得不可用。后来我做了三层防护第一层超时控制。所有外部调用包括大模型和技能内部的外部请求都要设超时时间。根据模型服务商提供的响应时间基线我把超时设置为 30 秒。# llm_client.py 片段 from openai import AsyncOpenAI import asyncio client AsyncOpenAI( base_urlhttps://your-model-endpoint.com/v1, api_keysk-xxx, timeout30.0, # 30秒超时 ) async def chat_completion(messages, **kwargs): try: resp await client.chat.completions.create( modelyour-model, messagesmessages, temperature0.3, **kwargs, ) return resp.choices[0].message.content except asyncio.TimeoutError: return 模型响应超时请稍后重试 except Exception as e: return f模型调用发生异常: {str(e)}第二层并发控制。用信号量限制同时对模型服务的调用量。如果并发太高宁愿让一部分请求排队也不要一股脑全打过去把模型服务打垮。import asyncio # 同时最多允许 8 个并发请求 semaphore asyncio.Semaphore(8) async def chat_completion_with_concurrency(messages, **kwargs): async with semaphore: return await chat_completion(messages, **kwargs)第三层熔断开关。如果连续失败超过 N 次直接打开熔断开关后续请求不再调用模型而是返回降级提示。每隔 30 秒尝试半开一次看看模型服务是否恢复。这三层防护下来我最后一次收到服务不可用的告警还是因为腾讯云那台机器的网络配置出问题——那是另一段自己给自己挖的坑下面讲。4.3 安全组、端口与外网访问的配置细节这是所有腾讯云新手都会遇到的一个问题服务在服务器上跑起来了curl localhost:8000也通但外部就是访问不了。问题出在腾讯云的两层防火墙上腾讯云控制台的安全组在服务器实例详情页 - 安全组 - 添加规则放行你需要的端口。比如你的 Agent 服务跑在 8000 端口就要放行TCP:8000。服务器系统防火墙如果有启用 ufwUbuntu 上如果开了ufw也要放行端口。我给客户的部署检查单是这样的# 1. 在腾讯云控制台安全组放行端口 # 入站规则TCP:8000Agent API TCP:80/443如果用 Nginx # 2. 检查服务器内部防火墙 sudo ufw status # 如果启用执行 sudo ufw allow 8000/tcp # 3. 确认服务监听地址不是 127.0.0.1 # main.py 中启动参数 # uvicorn.run(app, host0.0.0.0, port8000)很多人会挂在最后一步uvicorn.run(app)默认 host 是127.0.0.1只有本机能访问。必须改成0.0.0.0才能从外部访问。还有一个进阶细节如果你用nginx做反向代理注意在 nginx 配置里设置proxy_read_timeout否则默认 60 秒超时而你的 Agent 处理一次完整对话可能要 60 秒以上。location /agent { proxy_pass http://127.0.0.1:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_read_timeout 120s; proxy_send_timeout 120s; }这个坑我踩得很冤当时用 nginx 转发所有请求模型偶尔慢一点用户就收到 504 网关超时。排查了半天才发现是proxy_read_timeout默认值太短了。5. 真实业务场景的完整实现从需求到可用5.1 场景设定与 Skill 切分逻辑光讲概念不够我拿一个最近在腾讯云上做的实际业务来复盘——一个业务汇报助手Agent。公司内部运维和开发团队每天需要在群里同步项目进展但数据分散在多个系统里整理起来非常浪费时间。我的任务是做一个 Agent让负责人只需要说一句汇报一下今天的项目进展就自动拉数据、做汇总、发到群里。这个需求看起来简单但细拆下来有多个子任务从 DevOps 平台拉取今天的构建和部署记录调用devops_platform接口从监控系统拉取服务健康状态和告警调用monitoring_system接口从 Git 仓库拉取最近一天的代码提交记录调用git_log接口把以上数据整理成一份结构化汇报文案大模型生成不需要单独技能发送到企业微信群调用wechat_webhook技能。一开始我犯了个典型错误把整理汇报当成一个技能把所有逻辑都塞进一个generate_report技能里。这样做的问题很快就暴露了——技能内部条理混乱一个接口出错整个技能就废了而且大模型无法判断用户只是想部署了没这种单点问题时该不该调用整个大技能。正确的切法应该是一个技能只做一件有明确边界的事。划分后得到 4 个技能fetch_devops_recordsfetch_monitoring_statusfetch_git_commitssend_wechat_message大模型的职责是根据用户的指令决定调用哪几个技能、调用顺序是什么、最后怎么组织回复。5.2 编排流程与上下文传递技能拆细之后的另一个问题技能之间的数据怎么串联拿我们的场景来说send_wechat_message需要的内容是前面三个技能返回数据的汇总结果。这要求模型能记住前几个技能的返回并把它们组装成新的消息内容。具体实现上我是这样做的把每个技能的返回结果都追加到对话上下文中作为一组tool角色的消息传回给模型让模型在多次调用之间保持上下文连贯最后让模型基于全部返回数据生成汇报文本然后调用发送技能。核心代码如下# agent/core.py 片段 async def process(self, session_id: str, user_message: str): messages self.session_manager.get_messages(session_id) # 用户消息 messages.append({role: user, content: user_message}) self.session_manager.save(session_id, messages) # 最多循环 5 次防止模型陷入无限工具调用 for _ in range(5): response await self.llm_client.chat_completion(messages) if response.action call_skill: skill self.registry.get_skill(response.skill_name) if not skill: messages.append({ role: tool, content: json.dumps({ error: f技能 {response.skill_name} 不存在请确认技能名称 }) }) continue # 执行技能 result await skill.execute(response.parameters) # 技能结果返回给模型 messages.append({ role: tool, name: response.skill_name, content: json.dumps(result, ensure_asciiFalse) }) self.session_manager.save(session_id, messages) elif response.action reply: # 面向用户的最终回复 messages.append({role: assistant, content: response.reply}) self.session_manager.save(session_id, messages) return response.reply return 抱歉处理步骤较多请稍后再试或用更简洁的方式描述需求。这套流程最考验人的一点在于模型如何判断该结束调用了。我的经验是在 System Prompt 里明确提醒模型只有当用户的需求已经被完全满足时才生成最终回复。如果还有需要获取的数据请继续调用技能。每次调用技能后根据返回结果判断是否还需要进一步操作。这个提示词让模型的停火判断准确了不少。5.3 沙箱验证上线前必须过的关卡技能上线前我会做一个简单的沙箱验证把所有的技能和执行链路跑一遍确保每个技能都能在真实环境里正确返回。这一步看着简单但特别有用。我的验证方法是写一个自动化脚本模拟用户请求并断言结果# tests/test_report_flow.py import pytest from agent.core import AgentEngine from agent.registry import SkillRegistry pytest.mark.integration async def test_report_flow(): registry SkillRegistry() registry.discover() engine AgentEngine(registry) result await engine.process(test_session, 汇报一下今天的项目进展) assert 构建 in result, f汇报内容应包含构建信息实际为: {result} assert 成功 in result, f汇报内容应包含状态信息实际为: {result} print(f验证通过汇报内容长度: {len(result)} 字符)跑通这条链路的意义在于你验证的不只是单个技能没有问题而是整条意图理解 - 技能编排 - 数据获取 - 汇报生成的链路是否完整。不过我也得提醒一句集成测试能帮你抓住大部分问题但无法覆盖所有边界情况。比如模型偶尔会抽风给出完全不符合预期的输出。别问我怎么知道的都是泪。6. 稳定性、安全与成本生产环境绕不开的三座大山6.1 日志与监控Agent 服务上线后第一件事就是做监控。如果不做监控你根本不知道 Agent 是在正确地干活还是在错误地勤劳。我推荐的最小监控方案请求日志记录每次请求的 session_id、用户消息、模型响应、技能调用情况和耗时。用 JSON 格式输出方便后续检索分析。技能调用统计统计每个技能的调用次数、成功率、平均耗时。这能帮你找出最薄弱的技能。模型成本追踪统计每天的模型调用次数和输入输出 token 量折算成费用。Agent 服务的成本大头就是模型调用这项统计不做月底对账的时候会吓你一跳。在腾讯云上我直接用腾讯云的日志服务CLS采集系统日志和业务日志通过检索语法把耗时过长的接口、报错信息筛出来。配合腾讯云的云监控告警可以在 CPU 超过 80%、内存超过 85% 时第一时间收到短信提醒。6.2 密钥管理与 API 安全这是整个项目里我最想强调的一点。上一个项目里我把 OpenAI 的 API Key 直接写在.env文件里然后一段代码忘了加.gitignore把.env传到了 Git 仓库。虽然那是私有仓库但还是心里发毛——后来我花了一个下午把所有密钥全部重置冷汗直流。正确的做法有几种使用腾讯云的密钥管理服务KMS把 API Key 存进密钥管理运行时代码从 KMS 拉取。使用环境变量 systemd 配置把密钥放到 systemd 服务的 Environment 里利用 systemd 的权限隔离来保护文件。使用 .env 文件 严格的文件权限控制chmod 600 .env确保只有当前用户能读。我现在的标准做法是 systemd Environment 文件权限控制两件套# agent.service 追加 EnvironmentOPENAI_API_KEYsk-xxx EnvironmentWECHAT_WEBHOOK_URLhttps://qyapi.weixin.qq.com/.../xxx然后用sudo chmod 600 /etc/systemd/system/agent.service sudo systemctl daemon-reload对于 Webhook 地址这类半敏感信息我还额外在代码里做了 URL 校验防止模型拼接出合法的恶意 URL。6.3 并发扩容与服务降级如果 Agent 服务要服务比较多的人单机部署很快会碰到瓶颈。我在腾讯云上的扩容思路分三步走:单个服务实例内部优化用gunicorn uvicorn的工作模式同时跑多进程充分利用多核 CPU。多实例部署 负载均衡把 Agent 服务部署到多个实例前端加一层负载均衡器。腾讯云的 CLB 可以直接做到这一点。服务降级策略当系统过载时优先保证核心链路——比如对话功能可用非核心功能比如发送告警可以排队执行。降级策略上有个小技巧把耗时长的技能和基础对话拆成两个独立的服务进程。对话功能保持轻快技能执行任务扔到后台队列里异步返回结果。这样即使某个技能卡住也不会影响用户和 Agent 的基本聊天体验。7. 从能用到好用我的几点复盘与新坑预告项目上线跑了一段时间结合团队的使用反馈我复盘了三条经验可能对你有帮助。第一Skill 的设计迭代要跟着真实使用数据走。上线初期我的“定时巡检”技能描述词里写了一大堆适用场景结果用户根本不这么用他们更常说的是看看服务器有没有事。我根据日志里的真实用户提问把描述词里的表达改成了用户习惯的口语化说法触发准确率一下就上来了。第二不要迷信大模型的一步到位能力。哪怕是一流的模型在处理多步骤任务时也容易出错。与其指望模型完美规划一条最优路径不如把每一步的路标给足——比如在技能描述里写清楚每次执行完必须输出进度信息。这样即使模型走得绕至少不会迷失。第三成本控制要从第一天就做。我提到过把技能拆细的过程拆细后模型经常为了一个简单问题调用两三次模型接口。后来我加了一条限制——某些轻量任务直接走规则匹配不走模型调用。比如查询服务器状态这种固定指令我直接写了个关键词匹配命中后直接调 API省掉一大笔模型调用费用。最后预告一个我正在折腾的方向给 Agent 加上记忆层。目前 Skill 的输入输出都是一次性的Agent 并不知道上次巡检时磁盘是多少、这次的变化趋势如何。我准备引入简单的向量存储来记录每次技能执行的关键状态让 Agent 能够记得上次发生了什么它的决策质量应该会再上一个台阶。等这套玩法在腾讯云上跑通了我再来写一篇实战复盘。
返回列表