编程笔记

lifelong learning & practice makes perfect

智元灵犀X2机器人二开04:云端 AgentSDK，被动交互与语义技能

发表于 更新于 分类于
本文字数： 11k 阅读时长 ≈ 10 分钟
本篇讲透灵犀X2云端灵心 AgentSDK v1.4.0：HMAC-SHA256 验签、8 种被动回调选型、打断与拒识规范，并给出一个最小云端应答 Agent。

前置：第 01 篇。本篇不需要真机，用文本即可跑通全流程。

一、凭证与验签

云端开发的第一步是拿到凭证：登录灵心开放平台 → 进入「应用」页面 → 创建应用。

凭证 用途
app_id 应用标识
app_key 参与 HMAC-SHA256 签名
app_secret 签名密钥 
1
2
3
export AGIBOT_APP_ID="app_xxxxxxxx"
export AGIBOT_APP_KEY="ak_xxxxxxxx"
export AGIBOT_APP_SECRET="sk_xxxxxxxx"

⚠️ app_key 与 app_secret 用于网关签名，不得写进可公开的代码或日志。生产环境走环境变量或密钥管理服务。

二、v1.4.0 的不兼容升级

这是上手前必须确认的一件事：v1.4.0 与 v1.3.0 及更早版本不兼容。

项目 v1.3.0 及更早 v1.4.0
鉴权机制 JWT（AGIBOT_AUTHORIZATION 头） HMAC-SHA256 五元组请求头
请求头 1 个 X-App-Id / X-App-Key / X-Timestamp / X-Nonce / X-Signature
凭证来源 二开智能体模板 灵心开放平台「应用」
create() 签名 create(url, app_id, app_secret) create(url, app_id, app_key, app_secret)
网关地址 wss://agentsdk.agibot.com/v1/agentsdk wss://open.agibot.com/api/V1/open-portal/app/wss/agent-sdk
回调类型 JWT payload 字段 独立请求头 X-Callback-Types（JSON 数组字符串） 
1
2
3
4
5
6
7
8
from linksoul_agentsdk import AgentSdk

agent_sdk = AgentSdk.create(
    url="wss://open.agibot.com/api/V1/open-portal/app/wss/agent-sdk",
    app_id=app_id,
    app_key=app_key,
    app_secret=app_secret,
)

安装：linksoul_agentsdk 目前未发布到公共 PyPI，需从官方获取离线分发包后本地安装：

1
pip install ./linksoul_agentsdk-1.4.0a0-py3-none-any.whl

依赖 websocket-client>=1.7.0 会自动拉取。版本号 1.4.0a0 是 PEP 440 的 alpha 预发布标识，与 Java SDK 的 1.4.0-SNAPSHOT 语义对齐。

三、生命周期与运行时约束

3.1 四步初始化，顺序不可颠倒

AgentSDK 四步初始化时序 先注册鉴权与被动回调，再调用 initialize 建立 WebSocket 长连接并完成 HMAC-SHA256 验签，随后依次收到认证结果、机器人上线与语音请求回调。 四步初始化 · 顺序不可颠倒 你的程序 AgentSdk LinkskyGateway AgentSdk.create( url, app_id, app_key, app_secret) register_auth(MyAuth()) register_asr2_llm(MyCallback()) initialize() WebSocket 连接 + HMAC-SHA256 验签 response.on_interrupt / on_skill / on_llm_* on_auth_state(code=0) on_robot_online(agent_id, agent_meta) on_request(agent_id, event_id, text, param, response)

必须先 register_*() 再 initialize()。顺序反了回调不会生效，且不会报错——这类静默失败排查成本很高。

3.2 线程模型决定了代码怎么写

线程 数量 职责
linksky-recv-<app_id> 1 WebSocket 接收线程
agent-worker-N 1..50 每个 agent_id 绑定一个串行 worker
threading.Timer daemon 0..1 重连定时器
AgentSDK 线程模型 WebSocket 接收线程把消息交给 MessageRouter，再按 agent_id 分发到各自串行的 agent-worker，所有 worker 共享同一个 Callback 单例。 线程模型 · 每个 agent_id 一个串行 worker linksky-recv-<app_id> WebSocket 接收线程 MessageRouter 按 event_type 分发 agent-worker-1 agent_id = AGENT_001 agent-worker-2 agent_id = AGENT_002 agent-worker-N agent_id = AGENT_00N 同一 Callback 实例 单例，按 agent_id 区分

三条直接约束：

  1. 同一 agent_id 的请求是串行的。on_request 阻塞多久，该智能体的后续消息就积压多久。
  2. response.on_* 可在任意线程安全调用（内部用 threading.Lock 串行化）。
  3. 回调是单例，所有上线的智能体共享同一实例。不能用全局变量存会话状态——必须按 agent_id 分桶。

⚠️ 结论：on_request 内禁止同步阻塞（长时间 sleep、同步 HTTP 调用）。需要等待结果时用线程池，回调立即返回。第 05 篇会看到官方 demo 恰好在这里踩了坑。

3.3 重连

  • 连接失败 / 断开后自动重连，默认延迟 3 秒（构造时可覆盖）
  • 默认无限重连（max_reconnect_attempts=-1），直到调用 agent_sdk.release()
  • 心跳：每 10 秒 Ping（websocket-client 内置）

四、8 种被动回调的选型

被动交互是 SDK 的核心能力：机器人发起（音频 / 视频 / 文本）→ 回调处理 → 通过 Response 返回结果。

按”输入 × 输出”组合共 8 种：

类别 回调类 输入 输出
纯音频 Audio2LlmCallback 音频流 ASR + LLM 文本
Audio2TtsCallback 音频流 ASR + LLM + TTS
纯文本 Asr2LlmCallback ASR 文本 LLM 文本
Asr2TtsCallback ASR 文本 LLM + TTS
多模态 AsrVideo2VlmCallback ASR 文本 + 视频 VLM 文本
AsrVideo2TtsCallback ASR 文本 + 视频 VLM + TTS
AudioVideo2VlmCallback 音频 + 视频 ASR + VLM 文本
AudioVideo2TtsCallback 音频 + 视频 ASR + VLM + TTS

选型是一条三步判断链：

被动回调选型的三步判断链 先看端侧是否已完成 ASR 决定前缀 Asr 或 Audio，再看是否需要输入画面决定是否加 Video，最后看是否需要输出音频决定后缀 Tts 或 Llm、Vlm。 回调选型 · 三步判断链 ① 端侧已完成 ASR？ 是 → 前缀 Asr 否 → 前缀 Audio ② 需要输入画面？ 是 → 加 Video，后缀 Vlm 或 Tts 否 → 不加画面 ③ 需要输出音频？ 是 → 后缀 Tts 否 → 后缀 Llm（无视频）/ Vlm（有视频） 三步组合即得 8 种被动回调；音频类与多模态类额外用 flag 区分帧类型。

取舍：本系列选 Asr2LlmCallback。ASR 在端侧完成，云端只处理文本——链路最短、延迟最低、调试最直观。代价是云端拿不到音频原始特征（情绪、语速、停顿），需要这些时应改用 Audio2Llm。

v1.4.0 变更：Audio2Asr / Audio2Audio / AudioVideo2Audio / Text2Tts 四种回调及其 Response 类已从 SDK 删除。

五、flag 与标准响应流程

5.1 flag：区分同一回调中的不同数据

音频类回调（Audio2*、AudioVideo2*）：

flag 含义 buf 触发时机
0 VAD 开始 None 用户开始说话
1 音频帧 PCM bytes 持续传输中
2 VAD 结束 None 用户停止说话
3 H264 视频帧 H264 字节 视频流传输

AsrVideo2* 类：

flag 含义 有效参数
3 ASR 文本 text 有效，buf=None
4 H264 视频帧 buf 有效，text=None
5 图片帧 buf 有效，text=None

AsrVideo2* 的 on_request 会被多次调用（分别传文本与视频），必须按 flag 分派。

5.2 标准响应流程

on_request 的标准响应流程 语义理解分类后分四种去向：拒识不调用任何 response.on_；风控走 risk_control 打断；技能先 action 打断再下发技能；有效闲聊先 chat 打断再返回文本或语音。 标准响应流程 · on_request(text) on_request(text) 语义理解分类 分类结果 直接 return 不调用任何 response.on_* on_interrupt (risk_control, 话术) on_interrupt (action, None) on_interrupt (chat, None) on_skill(...) 可并行 on_llm_* on_llm_* / on_vlm_* on_tts_* 拒识 风控 技能 有效闲聊

5.3 Response 方法矩阵

Response 类 ASR LLM VLM TTS Skill Interrupt
Audio2LlmResponse ✓ ✓ ✓ ✓
Audio2TtsResponse ✓ ✓ ✓ ✓ ✓
Asr2LlmResponse ✓ ✓ ✓
Asr2TtsResponse ✓ ✓ ✓ ✓
AsrVideo2VlmResponse ✓ ✓ ✓
AsrVideo2TtsResponse ✓ ✓ ✓ ✓
AudioVideo2VlmResponse ✓ ✓ ✓ ✓
AudioVideo2TtsResponse ✓ ✓ ✓ ✓ ✓

常用方法：

方法 说明
on_llm_item_delta(event_id, item_id, text) LLM 流式输出（增量）
on_llm_item_done(event_id, item_id) LLM 单段完成
on_llm_done(event_id) LLM 全部完成
on_skill(event_id, item_id, skill_type, skill_name, param) 下发技能
on_interrupt(event_id, type, tips) 打断当前对话
on_error(event_id, code, msg) 错误上报（所有 Response 均支持）

六、打断与拒识

这两条是最容易写错、且错误后果最严重的部分。

6.1 打断顺序是硬约束

⚠️ 一旦判定为有效指令，必须先调用 on_interrupt，之后才能调用 on_skill / on_llm_* / on_vlm_* / on_tts_*。顺序颠倒会导致机器人无法正确进入新一轮应答状态。

interrupt_type 四种取值：

取值 语义 场景 interrupt_tips
risk_control 风控打断 内容违规、敏感话题 必填，作为给用户的话术
action 技能打断 命中技能 / 动作 传 None
chat 有效闲聊打断 命中普通对话 传 None
realtime 实时对话打断 边听边说场景 传 None

6.2 拒识：一个都不调用

拒识（误唤醒、噪音、无效查询）的正确处理是完全不调用任何 response.on_*：不返回技能、不调用 on_interrupt、不返回 LLM / VLM 结果，让机器人保持当前状态。

常见错误：对任意输入都回一句”我收到你的话：xxx”。这等于把噪音当成有效对话，会让机器人对着环境噪音自言自语。

6.3 分类决策表

语义理解结果 第一步 之后允许的应答
命中风控 on_interrupt(..., "risk_control", "<话术>") 一般不再返回 LLM / VLM / 技能
命中技能 on_interrupt(..., "action", None) on_skill(...)，可并行 on_llm_*
命中有效闲聊 on_interrupt(..., "chat", None) on_llm_* / on_vlm_* / on_tts_*
realtime 对话 on_interrupt(..., "realtime", None) realtime 应答内容
拒识 不调用 一个都不调用

七、语义技能

技能是 ASR 文本经语义理解分类后的一种结果分支，通过 on_skill(event_id, item_id, skill_type, skill_name, skill_param) 下发。skill_param 的键值对即为”语义”。

类别 示例语义
动作（action） {"action":"wave_hands"}、{"action":"cheer","hand":"left"}、{"action":"blow_kiss"}、{"action":"hug"}、{"action":"dance"}、{"action":"taichi"}
移动（movement） {"movement":"move_forward"}、{"movement":"move_forward","meter":10}、{"movement":"turn_left","angle":30}、{"movement":"run"}
表情（expression） {"expression":"smile"}、{"expression":"cry"}、{"expression":"heart_eyes"}
状态控制（status） {"status":"freeze"}（停止行走）、{"status":"hush"}（退出）
二次确认（confirmation） {"confirmation":"yes"}、{"confirmation":"no"}
技能问答 {"action":"sing","genre":"children"}、{"action":"play_music","song":"小星星"}

⚠️ 平台过滤机制：上表是平台设计能力的全集。端侧不同型号 / 系统版本支持的子集可能不同，但二开侧无需关心——按语义结果直接下发即可，平台会按目标机型与技能清单过滤：

  • 命中且端侧支持 → 正常执行
  • 命中但端侧不支持 → 平台向端侧返回 unknown，端侧按”不支持”处理

多意图：同一 event_id 下可多次调用 on_skill，每个意图使用独立 item_id。”先跳个舞再比个心”就是两次 on_skill。

⚠️ 部分移动类技能（move_backward、move_left、move_right、continuous 持续行走）需要二次确认。文档在技能表中已标注，实现时必须先走 confirmation 流程。

八、最小云端 Agent

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
#!/usr/bin/env python3
"""最小云端 Agent：文本进 → 技能 + LLM 回复出"""

import logging
import os
import re
import time

from linksoul_agentsdk import (
    AgentAuthCallback,
    AgentParam,
    AgentSdk,
    IdGenerator,
)
from linksoul_agentsdk.passive import Asr2LlmCallback, Asr2LlmResponse


class MyAuth(AgentAuthCallback):
    def on_auth_state(self, app_id: str, code: int, msg: str) -> None:
        if code == 0:
            print(f"认证成功: {app_id}")
        else:
            print(f"认证失败: code={code}, msg={msg}")


class MyCallback(Asr2LlmCallback):
    def on_request(
        self,
        agent_id: str,
        event_id: str,
        text: str,
        param: AgentParam,
        response: Asr2LlmResponse,
    ) -> None:
        raw = (text or "").strip()
        if not raw:
            return                                    # 拒识：静默

        item_id = IdGenerator.generate_item_id()
        nlu = self._classify(raw)

        if nlu["type"] == "reject":
            return                                    # 拒识：一个 on_* 都不调

        # 有效指令：先打断，再回填
        response.on_interrupt(
            event_id,
            "action" if nlu["type"] == "skill" else "chat",
            None,
        )

        if nlu["type"] == "skill":
            p = AgentParam.create()
            for k, v in nlu["params"].items():
                if isinstance(v, str):
                    p.set_string(k, v)
                elif isinstance(v, bool):
                    p.set_bool(k, v)
                elif isinstance(v, int):
                    p.set_integer(k, v)
                elif isinstance(v, float):
                    p.set_float(k, v)
            response.on_skill(
                event_id, item_id, nlu["skill_type"], nlu["skill_name"], p)

        # 技能与文本可并行下发，互不阻塞
        for chunk in self._reply_chunks(raw, nlu):
            response.on_llm_item_delta(event_id, item_id, chunk)
        response.on_llm_item_done(event_id, item_id)
        response.on_llm_done(event_id)

    # 规则分类器占位；第 06 篇替换为 LLM 结构化输出
    @staticmethod
    def _classify(text: str) -> dict:
        if "挥手" in text or "打招呼" in text:
            return {"type": "skill", "skill_type": "action",
                    "skill_name": "wave_hands",
                    "params": {"action": "wave_hands"}}
        if "向前" in text and "米" in text:
            m = re.search(r"(\d+)\s*米", text)
            meter = int(m.group(1)) if m else 1
            return {"type": "skill", "skill_type": "movement",
                    "skill_name": "move_forward",
                    "params": {"movement": "move_forward", "meter": meter}}
        if any(w in text for w in ("你好", "嗨", "谢谢", "名字")):
            return {"type": "chat"}
        return {"type": "reject"}

    @staticmethod
    def _reply_chunks(text: str, nlu: dict):
        if nlu["type"] == "skill":
            yield "好的，正在执行。"
        elif "名字" in text:
            yield "我是灵心语音助手，支持语音控制机器人动作和移动。"
        elif "谢谢" in text:
            yield "不客气，随时吩咐。"
        else:
            yield "你好。你可以让我挥手，或者说「向前走 2 米」。"


def main() -> None:
    logging.basicConfig(level=logging.INFO)

    sdk = AgentSdk.create(
        url="wss://open.agibot.com/api/V1/open-portal/app/wss/agent-sdk",
        app_id=os.environ["AGIBOT_APP_ID"],
        app_key=os.environ["AGIBOT_APP_KEY"],
        app_secret=os.environ["AGIBOT_APP_SECRET"],
    )
    sdk.register_auth(MyAuth())
    sdk.register_asr2_llm(MyCallback())
    sdk.initialize()

    print("WebSocket 连接就绪，等待语音指令...")
    try:
        while True:
            time.sleep(600)
    except KeyboardInterrupt:
        sdk.release()
        print("SDK 已释放")


if __name__ == "__main__":
    main()

这份代码固化了四条规范：

  1. 拒识分支不调用任何 response.on_*
  2. 有效指令先 on_interrupt，再回填技能 / 文本
  3. 技能与文本并行下发，互不阻塞
  4. 凭证从环境变量读取

九、故障对照表

现象 根因 处理
连接建立瞬间被服务端关闭 使用旧网关，或 create() 缺少 app_key 换新网关 + 四参调用
握手 401 / 403 凭证不同源；本机时钟偏差过大；代理未透传 5 个 X-* 头 核对凭证、date -u 对齐时钟、修代理
握手成功但回调不触发 X-Callback-Types 缺失或格式错误（需 JSON 数组字符串） 抓包确认该头存在且合法
AttributeError: register_state_listen v1.4.0 已移除该 API 状态改用 on_state；主动交互只用 register_task_flow
pip install linksoul_agentsdk 找不到包 未发布到公共 PyPI 用离线 wheel 本地安装
机器人卡住、不响应后续消息 on_request 内同步阻塞，卡死 per-agent worker 改为线程池，回调立即返回

结论

  • v1.4.0 是不兼容升级：HMAC-SHA256 五元组、新网关、四参 create()、X-Callback-Types 头。
  • 生命周期四步 create → register_auth → register_* → initialize，顺序不可颠倒，颠倒后静默失效。
  • 同一 agent_id 串行处理，on_request 内禁止同步阻塞；回调是单例，状态必须按 agent_id 分桶。
  • 8 种回调按”是否已 ASR / 是否要视频 / 是否要音频输出”三步选型。
  • 有效指令必须先 on_interrupt；拒识一个 response.on_* 都不能调。
  • on_skill 按语义全集下发即可，平台按机型过滤；多意图用独立 item_id。

下一篇把端云接起来：拆解官方端云联合 demo，定位并修复其中 5 个致命缺陷，最终实现”一句话控制机器人做动作与导航”。

声明

本文首发智元AIMA开放平台，并在博客同步

欢迎关注我的其它发布渠道