扩展技能(ExtSkill)指南

扩展技能(ExtSkill)指南

⚠️ 尚未上线:本能力为 v1.5.0-SNAPSHOT 开发中特性,接口与协议可能调整,正式发布前请勿用于生产环境。

v1.5.0 新增能力。ExtSkill = 扩展技能:二开应用把自定义技能挂到智能体上,运行期先 query 拉取该智能体当前可用的扩展技能清单,再 invoke 触发某个技能执行。它与「被动交互 / 主动交互 / 超视距遥操」并列,走独立的事件族与独立的策略枚举 AgentPolicy.EXT_SKILL(wire 值 "extskill")。Python 端接口与 Java 版同版本 API 完全等价,仅遵循 PEP8 的 snake_case 命名。

使用场景

  • 二开应用在灵心平台为智能体配置了若干扩展技能(如天气查询、工单创建、设备控制),需要在运行期动态获取清单并按需触发;
  • 需要旁路对话链路直接调用一个确定性的能力(带结构化入参、拿结构化出参),而不是走 LLM 自由生成;
  • 一次 invoke 分三段收结果:先 on_invoke_ack 确认网关已受理,再 on_state 收执行期间的状态上报(可选),最后 on_invoke_result 拿到技能真正的执行输出。

调用流程

register_ext_skill 注册后,可任意次 query 拉取清单、invoke 触发执行;不再需要时 unregister_ext_skill 释放。

text
   ┌─────────────────────────────┐
   │  ExtSkillRequest.query       │  拉取可用扩展技能清单
   │    (on_query_result)         │
   └─────────────────────────────┘

   ┌─────────────────────────────┐
   │  ExtSkillRequest.invoke      │  触发某个扩展技能执行
   │    (on_invoke_ack)           │  ← 第一段:网关已受理
   │    (on_state)                │  ← 第二段:执行期间状态上报(0 到多次)
   │    (on_invoke_result)        │  ← 第三段:技能执行输出
   └─────────────────────────────┘
  • query 通过 on_query_result 一次性回结果;code == 0 时 result 承载技能清单等返回内容;
  • invoke 分三段:先 on_invoke_ack(下发确认),再 on_state(执行期间技能侧通过 agentsdk.ext_skill.report_state 上报进度,0 到多次),最后 on_invoke_result(执行输出,code == 0 时 param 承载出参);
  • 若发送瞬间 SDK 连接已关闭,query / invoke 回调会在原线程同步以 code == 1000 + msg == "Agentsdk connection is closed" 触发一次失败回执。

事件族与协议

Python 端事件类型枚举位于 linksoul_agentsdk.enums.AgentEventType:

枚举成员事件类型字符串说明
AGENTSDK_EXT_SKILL_QUERYagentsdk.ext_skill.query查询可用扩展技能清单(SDK → 网关)
AGENTSDK_EXT_SKILL_QUERY_RESULTagentsdk.ext_skill.query_result查询结果(网关 → SDK)
AGENTSDK_EXT_SKILL_INVOKEagentsdk.ext_skill.invoke触发扩展技能执行(SDK → 网关)
AGENTSDK_EXT_SKILL_INVOKE_ACKagentsdk.ext_skill.invoke_ack下发受理回执(网关 → SDK)
AGENTSDK_EXT_SKILL_REPORT_STATEagentsdk.ext_skill.report_state执行期间状态上报(网关 → SDK)
AGENTSDK_EXT_SKILL_INVOKE_RESULTagentsdk.ext_skill.invoke_result执行结果(网关 → SDK)

所有 ExtSkill 消息 wire 结构:

json
{
  "type": "<event type>",
  "agentId": "<目标智能体 agentId>",
  "eventId": "event_xxxxxxxxxxxxxxxxxxxxx",
  "agentMode": "extskill",
  "requestId": "extskill_xxxxxxxxxxxxxxxxxxxxx"
}
  • 与 Teleop 用 sn 不同,ExtSkill 的机器人 / 智能体标识字段用 agentId;
  • requestId 由 IdGenerator.generate_ext_skill_request_id() 生成,是一次扩展技能会话的身份,query / invoke 复用同一个;
  • invoke 额外携带 extSkillId(要触发的技能 ID)与 input(扁平的入参键值,对齐 input_param.get_ext_param());
  • agentMode 恒为 "extskill"(对应 AgentPolicy.EXT_SKILL)。

invoke 入参

invoke(ext_skill_id, input_param, invoke_callback, timeout) 的 input_param 是一个 AgentParam,承载该技能的结构化入参,具体键由技能定义决定。构造示例:

python
input_param = (
    AgentParam.create()
    .set_string("city", "shanghai")
    .set_integer("days", 3)
)
event_id = ext_skill_request.invoke("skill_weather", input_param, invoke_callback, 2000)

wire 上以扁平键值下发({"input": {"city": "shanghai", "days": 3}}),与 dict(input_param.get_ext_param()) 输出一致。

结果错误码

query / invoke 回调的 code 语义统一:

code含义
0成功(result / param 承载返回内容)
-1服务端返回失败,msg 携带 errorMsg 详情
1000本地失败:发送时 SDK 侧 LinkskyClient 已关闭,SDK 同步在原线程回调该 code,业务侧应视为发送失败

code != 0 时,on_query_result 的 result、on_invoke_result 的 param 均为 None。

API 速览

  • 请求类:linksoul_agentsdk.ExtSkillRequest
  • 回调基类:ExtSkillQueryCallback(on_query_result)/ ExtSkillInvokeCallback(on_invoke_ack + on_state + on_invoke_result)
  • 注册:AgentSdk.register_ext_skill(ExtSkillRequest) / AgentSdk.unregister_ext_skill(request_id)
  • ID:IdGenerator.generate_ext_skill_request_id()

完整签名见 ExtSkill API。可运行示例见 ExtSkill 示例。

生命周期与清理

  • AgentSdkExtSkillMgr 按 request_id 维护活跃 request;调用 query / invoke 前必须先 register_ext_skill(...);
  • 每次 request 方法调用会刷新内部 update_ts;超过 7200s 未活动的 request 会被后台清理线程回收;
  • 用完或不再需要时,建议调用 agent_sdk.unregister_ext_skill(request_id) 显式回收;
  • AgentSdk.release() 会连带清理该实例名下的所有 ExtSkill request。

与 Teleop 的区别

维度Teleop(超视距遥操)ExtSkill(扩展技能)
策略AgentPolicy.TELEOP("teleop")AgentPolicy.EXT_SKILL("extskill")
标识字段sn + teleop_idagentId + request_id
核心动作enter / audio / keepalive / exitquery / invoke
结果模型单次 ACKquery 单次结果;invoke 三段(ack + state + result)

与 Java SDK 对齐

Python 版接口命名遵循 PEP8(snake_case),语义完全对齐 Java:

JavaPython
ExtSkillRequest.query(cb, timeout)ExtSkillRequest.query(query_callback, timeout)
ExtSkillRequest.invoke(extSkillId, input, cb, timeout)ExtSkillRequest.invoke(ext_skill_id, input_param, invoke_callback, timeout)
ExtSkillQueryCallback.onQueryResult(...)ExtSkillQueryCallback.on_query_result(...)
ExtSkillInvokeCallback.onInvokeAck(...)ExtSkillInvokeCallback.on_invoke_ack(...)
ExtSkillInvokeCallback.onState(...)ExtSkillInvokeCallback.on_state(...)
ExtSkillInvokeCallback.onInvokeResult(...)ExtSkillInvokeCallback.on_invoke_result(...)
AgentSdk.registerExtSkill(...)AgentSdk.register_ext_skill(...)
AgentSdk.unregisterExtSkill(...)AgentSdk.unregister_ext_skill(...)
IdGenerator.generateExtSkillRequestId()IdGenerator.generate_ext_skill_request_id()