智元灵犀X2机器人二开05:端云联合，语音意图控制动作与导航

本系列：从 0 到 1 入门智元灵犀 X2 二次开发
本篇产出：一个修复后的端云联合工程，以及一套可复用的缺陷定位方法。
前置：第 03 篇（端侧接口）+ 第 04 篇（云端 SDK）。本篇需要真机。

一、职责边界

端云接缝处的代码该写在哪一层，直接决定系统是否可维护。判据如下：

维度	端侧（AIMDK / ROS 2）	云端（AgentSDK + LLM）
职责	实时控制、安全急停、感知、音频采集	语义理解、规划、知识、长期记忆
时延量级	毫秒级闭环	百毫秒 ~ 秒级
离线能力	必须可降级运行	可降级为规则兜底
典型能力	走跑、避障、关节 / 末端、灵创动作、表情灯带	意图仲裁、RAG、任务编排、多轮对话
迭代成本	高（需真机发版）	低（服务端灰度）

判据：需要毫秒级反应的放端侧，需要”理解”和”决策”的放云端。迭代频率是第二判据——需要频繁调优的逻辑放云端，否则每次改动都要过一遍真机发版流程。

一次完整对话的时序

一次对话的整个生命周期由 event_id 关联——跨多次 on_request 与多次 response.on_* 调用共用同一个 event_id。

本系列的选型

采用 asr2llm：ASR 在端侧完成，云端只处理文本。理由是链路最短、延迟最低、调试最直观。代价是云端拿不到音频原始特征（情绪、语速、停顿）——若业务需要这些信息做情绪判断，应改用 audio2llm，并接受链路变长。

二、`on_state`：让 Agent 看得见

官方 demo 完全没实现 on_state，这是它离”智能”最远的一步。机器人通过 agentsdk.state_request.meta 事件上报自身状态：

1	PassiveCallback.on_state(agent_id, event_id, state_name, state_value)

⚠️ state_value 的类型不统一：可能是简单字符串、数字、对象或对象数组。结构化字段需自行 json.loads(state_value) 解析——但 robot_form 是裸字符串，对它调用 json.loads 会抛异常。

22 个状态模块

`state_name`	主题	频率
`map_info`	地图基础信息与导航点位	低频 / 事件触发
`map_2d_grid`	二维栅格地图快照	极低频
`current_location`	当前语义位置	事件触发
`robot_pose`	几何位姿	高频（≤10Hz）
`planned_path`	当前算路路径	事件触发
`locomotion_status`	导航执行状态	事件触发
`pnc_status`	规控避障状态摘要	事件触发
`task_status`	任务状态列表	事件触发
`exhibit_tasks`	展厅讲解任务列表	事件触发
`silisoon`	业务模式与状态扩展	事件触发
`robot_form`	机器人形态	一次性 / 偶尔
`manipulation`	机械臂 / 抓取状态	事件触发
`supported_calligraphy_list`	书法可写文字列表	一次性 / 偶尔
`tts_status`	TTS 运行态	事件触发
`vad_status`	VAD 运行态	事件触发
`dialog_status`	对话状态机	事件触发
`action_status`	动作 / 技能任务状态	事件触发
`listening_status`	端侧聆听 / 唤醒状态	事件触发
`wakeup_behavior_status`	唤醒表现覆盖配置状态	事件触发
`vision_perception`	视觉感知摘要	高频（≤10Hz）
`speaker`	当前发言人信息	事件触发
`settings`	端侧开关状态面板	事件触发

三条使用纪律：

频率敏感。robot_pose 与 vision_perception 默认按模块 ≤10Hz 限频，不要在回调里做重 I/O（写数据库、发 HTTP）。
整体替换语义。planned_path / exhibit_tasks / supported_calligraphy_list 等是整数组替换，不要做点级 merge。
空态显式上发。无任务时约定上发 task_status = [{"state":"idle","task_name":""}]。

World Model

把所有状态聚合成一个按 agent_id 分桶的世界模型：

import json

WORLD = {}          # agent_id -> 世界模型


def get_world(agent_id: str) -> dict:
    return WORLD.setdefault(agent_id, {
        "pose": None, "nav": None, "poi": {}, "tasks": None,
        "tts": None, "vision": None, "speaker": None, "form": None,
    })


class MyCallback(Asr2LlmCallback):
    def on_state(self, agent_id, event_id, state_name, state_value):
        wm = get_world(agent_id)
        try:
            if state_name == "robot_pose":
                wm["pose"] = json.loads(state_value)
            elif state_name == "locomotion_status":
                wm["nav"] = json.loads(state_value)
            elif state_name == "map_info":
                info = json.loads(state_value)
                wm["poi"] = {p["name"]: p for p in info.get("navi_points", [])}
            elif state_name == "task_status":
                wm["tasks"] = json.loads(state_value)
            elif state_name == "tts_status":
                wm["tts"] = json.loads(state_value)
            elif state_name == "vision_perception":
                wm["vision"] = json.loads(state_value)
            elif state_name == "speaker":
                wm["speaker"] = json.loads(state_value)
            elif state_name == "robot_form":
                wm["form"] = state_value          # 裸字符串，不要 json.loads
        except (json.JSONDecodeError, TypeError):
            pass                                   # 状态解析失败不应影响主链路

    def on_robot_offline(self, agent_id):
        WORLD.pop(agent_id, None)

有了 World Model，一批”看起来需要额外开发”的能力变成查表：

需求	依赖字段
避免重复下发导航	`locomotion_status.state`
回答”还有多久到”	`planned_path.eta_sec` / `remaining_length_m`
判断能否接新动作	`action_status.state`
判断是否正在播报、能否打断	`tts_status.state`
区分服务对象	`speaker.speaker_id` + `vision_perception`
“去前台”解析为坐标	`map_info.navi_points`

三、语义点位：把地名变成坐标

map_info 携带 navi_points：

{
  "map_id": "MAP_001",
  "map_name": "1F 大厅",
  "navi_points": [
    { "name": "前台", "id": 1, "point_x": 1.2, "point_y": 3.4 },
    { "name": "茶水间", "id": 2, "point_x": 5.6, "point_y": 7.8 }
  ]
}

def resolve_poi(agent_id: str, name: str):
    """把语义地点名解析为坐标"""
    poi = get_world(agent_id)["poi"]
    for k, v in poi.items():
        if name in k or k in name:      # 容忍 "前台" / "前台区" 这类差异
            return v
    return None

取舍：把坐标硬编码在代码里的成本，在地图变更时会一次性暴露——地图重采后所有点位失效，而代码里没有任何地方能提示这一点。走 navi_points 后，地图变、代码不变。唯一需要写死的只有地图 ID 与初始位姿，这两项也应外置到配置文件。

另一种导航方式是云端任务流提供的 pilot_task（定点导航 / 动线讲解），支持 pilot_start_task / pilot_resume_task / pilot_stop_task，配合 target_name 直接使用点位名。第 06 篇展开。

五、Demo

整体架构如下：

#!/usr/bin/env python3
"""端云联合：语音控制灵创动作 + 语义导航（修复版）"""

import json
import logging
import os
import threading
import time

import rclpy
from linksoul_agentsdk import AgentAuthCallback, AgentParam, AgentSdk, IdGenerator
from linksoul_agentsdk.passive import Asr2LlmCallback

PLAYER = None                    # 端侧句柄；多机器人场景见第 06 篇
WORLD = {}


def get_world(agent_id):
    return WORLD.setdefault(agent_id, {"poi": {}, "nav": None, "pose": None})


def run_async(fn, *args):
    threading.Thread(target=fn, args=args, daemon=True).start()


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


class MyCallback(Asr2LlmCallback):
    # ---------- 状态 ----------
    def on_state(self, agent_id, event_id, state_name, state_value):
        wm = get_world(agent_id)
        try:
            if state_name == "map_info":
                info = json.loads(state_value)
                wm["poi"] = {p["name"]: p for p in info.get("navi_points", [])}
            elif state_name == "locomotion_status":
                wm["nav"] = json.loads(state_value)
            elif state_name == "robot_pose":
                wm["pose"] = json.loads(state_value)
        except (json.JSONDecodeError, TypeError):
            pass

    def on_robot_offline(self, agent_id):
        WORLD.pop(agent_id, None)

    # ---------- 主回调 ----------
    def on_request(self, agent_id, event_id, text, param, response):
        raw = (text or "").strip()
        if not raw:
            return                                  # 拒识

        item_id = IdGenerator.generate_item_id()
        nlu = self._classify(agent_id, 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)

        response.on_llm_item_delta(event_id, item_id, nlu["reply"])
        response.on_llm_item_done(event_id, item_id)
        response.on_llm_done(event_id)

    # ---------- 意图分类（规则占位；第 06 篇替换为 LLM） ----------
    def _classify(self, agent_id, text):
        wm = get_world(agent_id)

        # 1) 停止类必须最先判断
        if any(w in text for w in ("停止导航", "取消导航", "别走了")):
            if wm["nav"] and wm["nav"].get("state") == "running":
                run_async(self._do_stop_nav)
                return {"type": "skill", "skill_type": "status",
                        "skill_name": "freeze",
                        "params": {"status": "freeze"},
                        "reply": "好的，已停止导航。"}
            return {"type": "chat", "reply": "当前没有正在进行的导航。"}

        # 2) 导航
        if any(w in text for w in ("导航", "带我去", "去一下")):
            if wm["nav"] and wm["nav"].get("state") == "running":
                return {"type": "chat", "reply": "我正在导航中，请稍等。"}
            dest = self._resolve_poi(wm, text)
            if dest is None:
                return {"type": "chat",
                        "reply": "抱歉，我没找到这个地点，您可以换个说法吗？"}
            run_async(self._do_navigate, dest)
            return {"type": "chat",
                    "reply": f"好的，带您去{dest['name']}，请跟我来。"}

        # 3) 灵创动作
        act = self._match_action(text)
        if act:
            return {"type": "skill", "skill_type": "action",
                    "skill_name": act, "params": {"action": act},
                    "reply": f"好的，为您表演{act}。"}

        # 4) 闲聊
        if any(w in text for w in ("你好", "嗨", "谢谢", "名字")):
            return {"type": "chat",
                    "reply": "你好。你可以让我导航去某个地点，或者播放一个动作。"}

        return {"type": "reject"}

    @staticmethod
    def _resolve_poi(wm, text):
        for name, p in wm["poi"].items():
            if name in text:
                return {"name": name,
                        "x": p.get("point_x", 0.0),
                        "y": p.get("point_y", 0.0)}
        return None

    @staticmethod
    def _match_action(text):
        for k in ("新年快乐", "太极", "孤勇者", "特技连踢"):
            if k in text:
                return k
        return None

    # ---------- 端侧执行（独立线程） ----------
    def _do_navigate(self, dest):
        if PLAYER is None:
            return
        ok = PLAYER.send_start(
            map_id=PLAYER.map_id,
            x=dest["x"], y=dest["y"], z=0.0,
            ox=0.0, oy=0.0, oz=0.0, ow=1.0,
        )
        print(f"[nav] -> {dest['name']} : {'OK' if ok else 'FAIL'}")

    def _do_stop_nav(self):
        if PLAYER is not None:
            PLAYER.send_stop()
            print("[nav] stopped")


def main():
    global PLAYER
    logging.basicConfig(level=logging.INFO)

    rclpy.init()
    # PLAYER = LinkCraftPlayer()          # 见第 03 篇
    # PLAYER.map_id = map_id

    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("端云链路就绪，等待语音指令...")
    try:
        while True:
            time.sleep(600)
    except KeyboardInterrupt:
        sdk.release()
        rclpy.shutdown()


if __name__ == "__main__":
    main()

六、调试与排障

环节	排查手段
云端连不上	看 `on_auth_state` 的 code；确认网关与凭证是 v1.4.0
收不到语音	确认机器人绑定的智能体 ID 与 SDK 注册的回调类型匹配
`on_state` 无数据	确认端侧已开启状态上报；先打印 `state_name` 看实际收到哪些模块
端侧调用失败	看 service 返回的 `header.header.code` 与 `header.message`
机器人不响应后续指令	检查 `on_request` 内是否有同步阻塞
状态解析抛异常	`state_value` 类型不统一；`robot_form` 是裸字符串

经验：接 on_state 的第一步不是写业务逻辑，而是在回调里加一行开关式日志，把 22 个模块实际收到的原始载荷打出来。文档给的是字段定义，真机上报的内容与频率往往有出入，先看真实数据再决定用哪些字段。

结论

端云判据：毫秒级反应放端侧，理解和决策放云端；迭代频率是第二判据。本系列选 asr2llm，链路最短。
on_state 是让 Agent 看得见的关键，最该先接的是 map_info / locomotion_status / robot_pose / tts_status。
用 map_info.navi_points 把地名解析为坐标，地图变更时无需改代码。
官方 demo 的 5 个缺陷：navHandler 作用域、_parse_nav 分支顺序、缺 on_interrupt、on_request 阻塞、凭证硬编码。
规则式意图解析的原则是先排除后命中——否定类规则必须排在正向规则之前。
端侧调用异步化是安全的：response.on_* 可在任意线程调用，但受 120 秒 Response TTL 约束。

下一篇是完结篇：从”语音遥控器”升级为真正的 Agent——主动交互任务流、四层闭环架构、任务状态机与任务栈、记忆层，以及两个完整的复杂场景案例。

声明

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