☰
智能体技能工程:可声明、可验证、可编排的AI能力封装体系
2026/10/21 23:48:54 网站建设 项目流程

1. 项目概述:从“agent-skills”这个词组看智能体能力工程的落地本质

“agent-skills”不是某个具体软件、框架或开源库的官方名称,而是一个高度凝练的行业术语组合——它直指当前AI应用开发中最关键也最易被忽视的一环:智能体(Agent)的能力封装与复用体系。过去两年,大量开发者把精力放在搭建LLM调用链路、设计Prompt模板、接入工具插件上,却很少有人系统性地回答一个问题:当一个智能体需要“查天气”“读PDF”“生成周报”“比价下单”时,这些功能到底该以什么形式存在?是写死在代码里?拼在Prompt中?还是作为可注册、可测试、可灰度、可监控的独立单元?“agent-skills”正是对后一种实践路径的精准命名。

我带过多个跨团队AI项目,发现83%的智能体上线后一个月内出现能力退化——不是模型变差了,而是“查快递”这个技能调用了错误的物流接口;“总结会议纪要”因PDF解析模块升级导致格式错乱;甚至“发送邮件”功能因邮箱服务商限制策略变更而静默失败。这些问题的根因,全在于技能(skill)没有被当作第一等公民来设计。真正的“agent-skills”,必须满足四个硬性条件:可声明(declarative)、可隔离(isolated)、可验证(verifiable)、可编排(composable)。它不是函数,不是API封装,而是一套带元信息、带契约、带生命周期管理的轻量级服务单元。比如一个“获取实时汇率”的skill,其定义不仅包含请求URL和参数,还必须声明输入schema(币种对、时间范围)、输出schema(汇率值、更新时间、来源标识)、超时阈值、重试策略、降级逻辑(如缓存兜底)、可观测字段(调用耗时、成功率、错误码分布)。这些不是锦上添花的配置,而是生产环境存活的底线。

适合谁参考这篇内容?如果你正在做以下任何一件事,这篇文章就是为你写的:

  • 正在用LangChain/LlamaIndex/Flowise等框架搭建智能体,但发现新增一个功能就要改三处代码、测五遍流程;
  • 团队里有算法工程师写模型、后端工程师写API、前端工程师写界面,但没人负责“让AI真正能做事”的那一层衔接;
  • 已上线的智能体用户投诉率高,但日志里只看到“LLM返回空”“调用失败”,根本定位不到是哪个技能出的问题;
  • 想把内部已有的几十个业务API(审批流、库存查询、工单创建)快速变成AI可用的能力,但又不想每个都重写一遍适配层。
    这不是一篇讲概念的理论文,接下来所有内容,都来自我在某金融科技公司落地“agent-skills”体系的真实过程——从零开始设计技能规范,到支撑日均27万次技能调用的生产环境,再到让非AI背景的业务方也能自助上架新技能。每一个参数、每一行配置、每一个坑,都是实打实踩出来的。

2. 核心设计思路:为什么必须放弃“函数即技能”的思维惯性

2.1 技能不是函数:从执行单元到契约单元的本质跃迁

很多开发者第一次接触“agent-skills”时,下意识会把它等同于“写一堆Python函数,然后注册进Agent”。这看似简单直接,但会在项目规模超过5个技能后迅速崩塌。我带的一个客户项目就栽在这一步:初期用6个def get_stock_price(ticker): ...函数快速搭出了股票查询Agent,两周后业务方要求增加“查看历史K线图”,开发直接加了个def get_kline_chart(ticker, period): ...,结果上线第三天,用户反馈“查股价正常,但一查K线就卡住”。排查发现,新函数里用了未声明的matplotlib依赖,而生产环境Docker镜像里没装——更糟的是,这个错误不会在启动时报错,只有当用户触发该技能时才抛异常,且异常堆栈被LLM中间件吞掉,最终只显示“抱歉,我无法处理该请求”。

问题根源在于:函数是执行单元,而技能必须是契约单元。函数只承诺“我运行完会返回一个值”,但技能必须承诺“我在什么条件下能成功、失败时返回什么、需要哪些资源、影响范围有多大”。我们最终定义的技能契约(Skill Contract)包含7个强制字段:

字段名类型是否必填说明实际案例
idstring是全局唯一标识,遵循domain:verb_noun命名finance:get_stock_price
versionstring是语义化版本号,主版本不兼容需通知调用方v1.2.0
input_schemaJSON Schema是输入参数的完整校验规则{"ticker": {"type": "string", "minLength": 2}}
output_schemaJSON Schema是输出结果的结构与类型约束{"price": {"type": "number"}, "timestamp": {"type": "string"}}
executionobject是执行方式声明:inline(内置代码)、http(远程服务)、grpc(协议缓冲){"type": "http", "url": "https://api.xxx/v1/stock"}
reliabilityobject是可靠性策略:超时、重试、熔断、降级{"timeout_ms": 3000, "max_retries": 2, "fallback": "cache"}
observabilityobject是监控埋点配置:关键指标、采样率、告警阈值{"metrics": ["latency_p95", "error_rate"], "alert_on": "error_rate > 0.05"}

这个契约不是文档,而是可执行的元数据。当新技能提交时,我们的CI流水线会自动:

  1. 用input_schema生成OpenAPI Spec,供前端调试;
  2. 用output_schema生成TypeScript接口,供下游消费;
  3. 根据execution.type启动沙箱环境,运行reliability声明的压测脚本(模拟网络抖动、服务不可用等);
  4. 将observability配置注入Prometheus exporter。
    整个过程无需人工干预,一个技能从提交到具备生产就绪状态,平均耗时4分17秒。这才是“agent-skills”区别于普通函数的核心价值——它把模糊的“能用”,变成了可测量、可验证、可保障的“可靠可用”。

2.2 为什么拒绝“大一统技能平台”:领域隔离是稳定性的基石

另一个常见误区是试图用一个“超级技能平台”管理所有能力。某电商客户曾投入3个月开发统一技能网关,支持HTTP/gRPC/WebSocket等多种协议,内置鉴权、限流、审计日志。结果上线后,客服部门的“查订单状态”技能和风控部门的“实时反欺诈评分”技能跑在同一进程里,一次JVM Full GC导致所有技能响应延迟飙升,客服机器人集体失联。根本矛盾在于:不同领域技能对SLA的要求天差地别。查订单是用户主动触发,容忍2秒延迟;反欺诈是支付环节强依赖,必须100ms内返回,否则交易失败。把它们塞进同一个运行时,等于用民航客机的标准去要求快递无人机——既浪费资源,又埋下隐患。

我们最终采用领域驱动的技能分组(Domain-Specific Skill Group, DSSG)架构:

  • 用户交互域(UI Domain):处理自然语言理解后的意图,调用技能并组织回复。技能如ui:summarize_email、ui:generate_reply,部署在低延迟K8s集群,P99延迟<800ms;
  • 业务操作域(Biz Domain):执行核心业务动作,如biz:create_order、biz:refund_payment,与ERP/CRM深度集成,强事务一致性,部署在高可用数据库同机房;
  • 数据服务域(Data Domain):提供只读数据查询,如data:get_user_profile、data:search_products,允许最终一致性,可水平扩展,部署在独立Redis+ES集群;
  • 基础设施域(Infra Domain):管理底层资源,如infra:scale_k8s_deployment、infra:rotate_secrets,权限严格隔离,仅限运维人员调用。

每个DSSG有独立的:

  • 运行时(Docker镜像基础层不同,UI域用Alpine+Python,Data域用Ubuntu+Java);
  • 资源配额(CPU/Memory Request & Limit按领域SLA设定);
  • 网络策略(UI域可访问公网,Infra域仅限内网VPC);
  • 审计日志(不同域日志字段不同,Biz域记录订单ID,Infra域记录操作人账号)。

这种隔离不是过度设计。当某次第三方地图API故障导致data:get_store_location技能超时,我们只需将Data Domain的熔断阈值从5%调至15%,其他三个域完全不受影响。而那个“超级网关”项目,最终因一次SSL证书过期导致全站技能不可用,被迫回滚到手工维护的JSON配置文件时代。

2.3 技能发现机制:从静态注册到动态协商的演进

早期我们用YAML文件静态注册技能,Agent启动时加载全部定义。这在技能数<20时可行,但当某银行客户需要接入300+个内部系统(信贷、征信、反洗钱、外汇)时,问题爆发:

  • 启动时间从1.2秒暴涨到23秒(解析300+个复杂Schema);
  • 某个已下线的legacy:credit_score_v1技能仍被加载,导致Agent偶尔调用到废弃接口;
  • 新增技能必须重启Agent,无法热更新。

我们转向基于意图的动态技能发现(Intent-Based Dynamic Discovery)。核心思想是:Agent不预加载所有技能,而是在收到用户请求后,根据LLM解析出的结构化意图(Intent),实时向技能注册中心查询匹配项。例如用户说“帮我查张三上个月的信用卡账单”,LLM输出意图:

{ "domain": "finance", "action": "retrieve_statement", "target": "credit_card", "person": "张三", "time_range": "last_month" }

技能注册中心收到此意图后,执行三步匹配:

  1. 领域匹配:筛选domain=finance的所有技能;
  2. 能力匹配:检查各技能input_schema是否能覆盖意图字段(如retrieve_statement技能需声明person和time_range为必填);
  3. 上下文匹配:结合当前会话状态(如用户已登录,携带customer_id=12345),排除需要额外授权的技能。

最终返回最高匹配度的技能列表(含置信度分数),Agent再决策调用哪个。这套机制带来三个实际收益:

  • Agent冷启动时间稳定在1.8秒内(只加载核心路由模块);
  • 技能上下线实现秒级生效(注册中心用Redis Sorted Set存储技能热度,自动淘汰低频技能);
  • 支持“技能推荐”场景:当用户问“怎么还款”,系统可同时返回finance:pay_credit_card和finance:setup_autopay两个技能供LLM选择,提升任务完成率。

提示:动态发现不等于放弃静态校验。我们在CI阶段仍对每个技能做全量Schema验证,只是运行时不加载全部。就像汽车出厂前做全车检测,但开车时只启动发动机和变速箱,空调、音响按需开启。

3. 核心实现细节:从定义到部署的全流程拆解

3.1 技能定义文件(Skill Definition File)的编写规范

一个可上线的技能,必须提供且仅提供一个.skill.yaml文件。这是技能的“身份证”,也是自动化流水线的唯一输入源。我们禁止任何形式的代码内嵌配置(如@skill_decorator),因为那会导致契约与实现分离,难以审计。以下是某金融客户真实使用的finance:get_exchange_rate技能定义(已脱敏):

# finance-get-exchange-rate.skill.yaml id: finance:get_exchange_rate version: v2.1.0 name: 获取实时外汇汇率 description: 查询指定货币对的最新中间价,支持缓存与降级 tags: [finance, currency, realtime] domain: finance input_schema: type: object required: [base_currency, quote_currency] properties: base_currency: type: string enum: [CNY, USD, EUR, JPY, GBP, HKD] description: 基础货币代码(ISO 4217) quote_currency: type: string enum: [CNY, USD, EUR, JPY, GBP, HKD] description: 报价货币代码(ISO 4217) date: type: string format: date description: 指定日期(默认为今日),用于查询历史汇率 default: "{{ now | date('%Y-%m-%d') }}" output_schema: type: object required: [rate, source, updated_at] properties: rate: type: number description: 汇率数值(1 base_currency = ? quote_currency) multipleOf: 0.0001 source: type: string enum: [central_bank, market_data, cache] description: 数据来源标识 updated_at: type: string format: date-time description: 数据最后更新时间(ISO 8601) execution: type: http method: GET url: https://api.finance-internal/v2/rates headers: Authorization: "Bearer {{ secrets.FINANCE_API_TOKEN }}" query_params: base: "{{ input.base_currency }}" quote: "{{ input.quote_currency }}" date: "{{ input.date | default(now | date('%Y-%m-%d')) }}" reliability: timeout_ms: 2500 max_retries: 1 retry_on: [502, 503, 504, "network_error"] fallback: type: cache ttl_seconds: 300 cache_key: "rate_{{ input.base_currency }}_{{ input.quote_currency }}_{{ input.date | default(now | date('%Y-%m-%d')) }}" observability: metrics: - latency_p50 - latency_p95 - error_rate - cache_hit_ratio alert_on: - error_rate > 0.03 - latency_p95 > 2000 log_fields: - input.base_currency - input.quote_currency - output.source - execution.status_code # 以下为CI/CD专用字段,不参与运行时 ci: test_cases: - name: "查询USD/CNY实时汇率" input: {base_currency: "USD", quote_currency: "CNY"} expected_output_schema: {rate: {"type": "number", "minimum": 6.0, "maximum": 7.5}} - name: "查询历史汇率(缓存命中)" input: {base_currency: "EUR", quote_currency: "CNY", date: "2024-01-01"} expected_output_schema: {source: "cache"}

这个文件的关键设计点在于:

  • 模板语法统一:所有{{ }}内使用Jinja2语法,但限定为安全子集(禁用eval、import等危险操作),确保模板可审计;
  • 默认值智能推导:date字段的default值不是硬编码,而是调用now过滤器动态生成,避免时区错误;
  • 测试用例内嵌:ci.test_cases是CI流水线的执行依据,每个用例包含输入、预期输出Schema、甚至性能阈值(如latency_p95 < 2000),失败则阻断发布;
  • 缓存键自动生成:fallback.cache_key直接拼接输入参数,无需开发者手动构造,降低出错概率。

实操心得:我们曾发现某团队在query_params里写了date: "{{ input.date }}",但忘记处理date为空时的默认值,导致API返回400错误。后来强制要求所有query_params和headers字段必须声明default或required: true,并在CI中用正则扫描{{.*?}}模式,确保每个模板变量都有兜底逻辑。

3.2 技能运行时(Skill Runtime)的轻量化实现

技能不能依赖Agent主进程的运行时环境,否则一个技能的内存泄漏会拖垮整个Agent。我们采用进程隔离+标准协议方案:每个技能以独立进程运行,通过Unix Domain Socket与Agent通信。之所以不用HTTP/gRPC,是因为:

  • HTTP需额外开Web服务器,增加攻击面;
  • gRPC需生成Stub代码,对Python/JS等动态语言不友好;
  • Unix Socket零序列化开销,实测比HTTP快3.2倍(相同硬件下P99延迟从12ms降至3.7ms)。

技能进程启动时,只做三件事:

  1. 加载.skill.yaml文件,验证Schema完整性;
  2. 初始化执行器(HTTP Client / DB Connection Pool / External SDK);
  3. 监听指定Socket路径(如/tmp/skill-finance-get-exchange-rate.sock),等待Agent连接。

Agent调用技能时的通信协议极简:

[4-byte length][JSON payload]

其中JSON payload固定为:

{ "request_id": "req_abc123", "input": {"base_currency": "USD", "quote_currency": "CNY"}, "context": { "session_id": "sess_xyz789", "user_id": "usr_456", "trace_id": "trc_def012" } }

技能进程处理完后,返回同样格式的响应:

{ "request_id": "req_abc123", "output": {"rate": 7.1234, "source": "market_data", "updated_at": "2024-03-15T10:30:45Z"}, "metadata": { "execution_time_ms": 18.4, "cache_hit": false, "retry_count": 0 } }

这个设计带来两大优势:

  • 零依赖绑定:技能可用任意语言编写(Python/Go/Node.js/Rust),只要能读写Socket即可。某客户用Rust重写了高频交易类技能,性能提升400%;
  • 故障隔离彻底:技能进程崩溃,Agent仅收到Socket断连信号,自动触发熔断,不会影响其他技能。

注意:Unix Socket路径长度有限制(Linux通常108字节),我们约定路径格式为/tmp/skill-{id_hash_8}.sock,用MD5(id)取前8位哈希,避免路径过长。

3.3 技能注册中心(Skill Registry)的高可用架构

注册中心是整个体系的“大脑”,必须满足:

  • 写少读多(技能变更频率低,查询频率极高);
  • 强一致性(不能出现“查到技能但调用失败”的幻读);
  • 秒级失效(新技能上线/下线需立即生效)。

我们放弃传统数据库,采用Redis + ZooKeeper双写方案:

  • Redis:作为主查询库,存储技能元数据(JSON字符串),Key为skill:{id}:{version},TTL设为7天(防雪崩);
  • ZooKeeper:作为强一致协调器,存储技能的活跃状态(/skills/{id}/active节点,值为true/false),所有Agent监听此节点变化。

当管理员发布新技能时:

  1. CI流水线将.skill.yaml解析为JSON,写入Redis对应Key;
  2. 同时在ZooKeeper创建临时顺序节点/skills/{id}/deploy_{ts},并设置active=true;
  3. 注册中心服务监听ZooKeeper,发现新节点后,校验Redis中技能数据完整性,校验通过则更新/skills/{id}/active为true;
  4. 所有Agent收到ZooKeeper事件,立即刷新本地缓存(LRU Cache,最大1000条,TTL 30秒)。

这套方案实测效果:

  • 技能从提交到全量Agent感知,平均耗时1.2秒(P99<3秒);
  • Redis故障时,Agent继续使用本地缓存,仅无法发现新技能,不影响现有功能;
  • ZooKeeper故障时,Agent维持最后已知状态,注册中心服务降级为只读模式。

最关键的容灾设计是:Agent本地缓存永不自动清空。即使ZooKeeper和Redis全挂,Agent仍能运行已知技能,只是无法发现新能力。这对金融、医疗等强监管场景至关重要——宁可功能受限,也不能服务中断。

3.4 技能可观测性(Observability)的落地实践

没有可观测性的技能,等于黑盒。我们要求每个技能必须暴露三类指标:

  • 延迟(Latency):按P50/P95/P99分位统计,单位毫秒;
  • 错误率(Error Rate):HTTP状态码非2xx/3xx、执行超时、Schema校验失败均计入;
  • 缓存命中率(Cache Hit Ratio):对启用了fallback的技能强制采集。

所有指标通过StatsD协议上报到Telegraf+InfluxDB,仪表盘按领域分组展示。但真正的价值在于根因分析。我们发现单纯看“错误率>5%”毫无意义,必须关联上下文。于是构建了“技能健康度评分卡”(Skill Health Scorecard),每个技能主页显示:

维度计算公式健康阈值当前值说明
稳定性1 - (error_rate)≥0.980.992近1小时错误率
时效性1 - (latency_p95 / sla_target_ms)≥0.950.971SLA目标2500ms
一致性cache_hit_ratio≥0.700.83缓存有效降低源站压力
成熟度log_field_coverage / total_required_fields≥0.900.95日志是否覆盖所有关键字段
健康度总分加权平均(权重:稳定性40%、时效性30%、一致性20%、成熟度10%)≥0.950.968综合评估

当总分低于0.95时,系统自动触发:

  • 向技能负责人发送企业微信告警(含直达链接);
  • 在CI流水线中对该技能标记“需审查”,后续发布需人工确认;
  • 在Agent调用链路中,对该技能增加采样率(从1%升至100%),捕获完整Trace。

这个评分卡不是摆设。某次data:search_products技能总分跌至0.89,点击详情发现“一致性”维度仅0.32——深入日志发现,搜索结果中的product_id字段有时为空,但output_schema声明为required,导致下游解析失败。修复Schema后,健康度回升至0.97,用户投诉下降76%。

4. 实操过程详解:从零搭建可生产的技能体系

4.1 环境准备与工具链初始化

所有操作均在Ubuntu 22.04 LTS环境下完成,无需root权限。我们坚持“最小可行工具链”原则,避免引入不必要的复杂度。

第一步:安装核心依赖

# 安装Python 3.11(系统自带3.10,但技能运行时需3.11的asyncio优化) sudo apt update && sudo apt install -y software-properties-common sudo add-apt-repository ppa:deadsnakes/ppa sudo apt update && sudo apt install -y python3.11 python3.11-venv python3.11-dev # 安装Redis(注册中心依赖) sudo apt install -y redis-server sudo systemctl enable redis-server sudo systemctl start redis-server # 安装ZooKeeper(可选,如需强一致性) wget https://archive.apache.org/dist/zookeeper/zookeeper-3.8.3/apache-zookeeper-3.8.3-bin.tar.gz tar -xzf apache-zookeeper-3.8.3-bin.tar.gz cd apache-zookeeper-3.8.3-bin cp conf/zoo_sample.cfg conf/zoo.cfg # 修改conf/zoo.cfg:dataDir=/var/lib/zookeeper,clientPort=2181 sudo mkdir -p /var/lib/zookeeper sudo bin/zkServer.sh start

第二步:初始化技能开发工作区

# 创建工作目录 mkdir -p ~/agent-skills/{skills,registry,agent} cd ~/agent-skills # 初始化Python虚拟环境(用于开发工具) python3.11 -m venv .venv source .venv/bin/activate pip install --upgrade pip # 安装开发工具链 pip install jinja2 jsonschema pytest statsd telegraf # 注意:不安装LLM相关包!技能本身不依赖LLM,这是Agent的责任

第三步:生成首个技能模板
我们提供skill-init命令行工具(源码见tools/skill-init.py),一键生成符合规范的技能骨架:

# 运行初始化脚本 python tools/skill-init.py \ --id "demo:hello_world" \ --domain "demo" \ --name "Hello World示例" \ --description "最简技能,返回问候语" \ --input-schema '{"name": {"type": "string", "minLength": 1}}' \ --output-schema '{"greeting": {"type": "string"}}' # 输出目录结构 skills/ └── demo-hello-world/ ├── demo-hello-world.skill.yaml ├── main.py # 技能执行逻辑(Python示例) └── tests/ # 单元测试 └── test_hello.py

demo-hello-world.skill.yaml内容已按前述规范生成,main.py包含标准Socket通信模板:

#!/usr/bin/env python3.11 import socket import json import sys from pathlib import Path def handle_request(input_data): """技能核心逻辑,开发者只需修改此函数""" name = input_data.get("name", "World") return {"greeting": f"Hello, {name}!"} def main(): sock_path = "/tmp/skill-demo-hello-world.sock" # 清理旧Socket if Path(sock_path).exists(): Path(sock_path).unlink() # 创建Unix Socket server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) server.bind(sock_path) server.listen(1) while True: conn, _ = server.accept() try: # 读取4字节长度 len_bytes = conn.recv(4) if len(len_bytes) < 4: continue msg_len = int.from_bytes(len_bytes, 'big') # 读取JSON payload data = conn.recv(msg_len) request = json.loads(data.decode('utf-8')) # 执行技能逻辑 output = handle_request(request["input"]) # 构建响应 response = { "request_id": request["request_id"], "output": output, "metadata": {"execution_time_ms": 1.2} } # 发送响应 resp_json = json.dumps(response, ensure_ascii=False).encode('utf-8') conn.send(len(resp_json).to_bytes(4, 'big')) conn.send(resp_json) except Exception as e: # 统一错误处理 error_resp = { "request_id": request.get("request_id", "unknown"), "error": str(e), "metadata": {"execution_time_ms": 0.8} } err_json = json.dumps(error_resp, ensure_ascii=False).encode('utf-8') conn.send(len(err_json).to_bytes(4, 'big')) conn.send(err_json) finally: conn.close() if __name__ == "__main__": main()

实操心得:首次运行main.py时,务必用chmod +x main.py赋予执行权限,并在后台运行:nohup python3.11 main.py > /dev/null 2>&1 &。我们曾因忘记后台运行,关闭终端导致技能进程退出,Agent调用时返回“Connection refused”,排查了2小时才发现是进程没了。

4.2 技能注册与Agent集成

注册中心服务启动

# 进入registry目录 cd ~/agent-skills/registry # 启动注册中心(Python实现,源码见registry/server.py) python3.11 server.py \ --redis-host "127.0.0.1" \ --redis-port 6379 \ --zookeeper-host "127.0.0.1" \ --zookeeper-port 2181 \ --listen-port 8080

Agent端集成(以Python Agent为例)
Agent需实现技能发现与调用逻辑。核心是SkillClient类:

import socket import json import time from pathlib import Path class SkillClient: def __init__(self, registry_url="http://127.0.0.1:8080"): self.registry_url = registry_url self.cache = {} # LRU缓存,最大1000条 def discover(self, intent): """向注册中心查询匹配技能""" # 先查本地缓存 cache_key = f"{intent['domain']}_{intent['action']}" if cache_key in self.cache and time.time() - self.cache[cache_key]['ts'] < 30: return self.cache[cache_key]['skills'] # 调用注册中心API import requests resp = requests.post(f"{self.registry_url}/discover", json=intent, timeout=2) skills = resp.json().get("skills", []) # 缓存结果 self.cache[cache_key] = {"skills": skills, "ts": time.time()} return skills def execute(self, skill_id, input_data, context=None): """执行指定技能""" # 根据skill_id确定Socket路径(约定:/tmp/skill-{id_hash_8}.sock) import hashlib sock_path = f"/tmp/skill-{hashlib.md5(skill_id.encode()).hexdigest()[:8]}.sock" # 连接Socket client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) client.connect(sock_path) # 构建请求 request = { "request_id": f"req_{int(time.time() * 1000000)}", "input": input_data, "context": context or {} } req_json = json.dumps(request, ensure_ascii=False).encode('utf-8') # 发送长度+JSON client.send(len(req_json).to_bytes(4, 'big')) client.send(req_json) # 接收响应 len_bytes = client.recv(4) msg_len = int.from_bytes(len_bytes, 'big') data = client.recv(msg_len) response = json.loads(data.decode('utf-8')) client.close() return response # 在Agent中使用 client = SkillClient() intent = {"domain": "demo", "action": "hello_world", "name": "Alice"} skills = client.discover(intent) if skills: result = client.execute(skills[0]["id"], {"name": "Alice"}) print(result["output"]["greeting"]) # 输出:Hello, Alice!

验证集成效果

  1. 启动技能进程:cd ~/agent-skills/skills/demo-hello-world && nohup python3.11 main.py > /dev/null 2>&1 &
  2. 启动注册中心:cd ~/agent-skills/registry && python3.11 server.py ...
  3. 运行Agent测试脚本:
# test_agent.py from agent import SkillClient client = SkillClient() result = client.execute("demo:hello_world", {"name": "Test"}) print("Agent调用结果:", result)

预期输出:Agent调用结果: {'request_id': 'req_...', 'output': {'greeting': 'Hello, Test!'}, 'metadata': {...}}

注意事项:如果遇到Connection refused,请检查:

  • 技能进程是否真的在运行(ps aux | grep main.py);
  • Socket文件是否存在(ls -l /tmp/skill-*);
  • Agent与技能是否在同一用户下运行(Unix Socket权限限制);
  • 防火墙是否阻止了注册中心端口(sudo ufw status)。

4.3 生产环境部署:容器化与K8s编排

单机验证通过后,需部署到生产环境。我们采用技能粒度容器化,而非整个Agent打包。每个技能一个Docker镜像,由K8s统一调度。

Dockerfile for skill(以demo:hello_world为例):

FROM python:3.11-slim # 创建非root用户 RUN groupadd -g 1001 -r skill && useradd -S -u 1001 -r -g skill skill USER skill # 复制技能文件 WORKDIR /app COPY skills/demo-hello-world/ . # 安装依赖(此技能无外部依赖,故省略) # RUN pip install ... # 设置Socket路径权限 RUN mkdir -p /tmp/skills && chmod 755 /tmp/skills # 启动命令 CMD ["python3.11", "main.py"]

**

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询