编程笔记

lifelong learning & practice makes perfect

智元灵犀X2机器人二开06:从Demo到产品，复杂场景下的智能Agent

发表于 更新于 分类于
本文字数： 11k 阅读时长 ≈ 10 分钟
完结篇：把语音遥控器升级为真正的 Agent——主动交互任务流、四层闭环架构、任务状态机与任务栈、记忆层，以及两个完整的复杂场景案例。

06 · 从 Demo 到产品：复杂场景下的智能 Agent

本系列：从 0 到 1 入门智元灵犀 X2 二次开发
本篇产出：一套可落地的 Agent 架构（感知 / 理解 / 决策 / 执行 + 记忆），两个完整案例，以及 P0–P5 分阶段路线。
前置：第 05 篇（端云已打通）。

一、”关键词遥控器”的边界

第 05 篇版本能跑通，但骨架仍是”关键字 if-else → 固定动作”。把它放进真实场景，会稳定地在三个地方失效：

失效场景 现场例子 现有实现为什么撑不住
多轮指代 “带我去前台” → “那边人有点多，换个地方” → “去茶水间吧” 每轮独立解析，无上下文，”换个地方”无法关联到”改目的地”
任务中断 导航途中用户问”还有多久到？” 没有任务概念，只能在”丢弃导航”和”无法应答”之间二选一
组合指令 “先跳个舞，然后带我去前台” 一次文本只能命中一个分支，无法拆成两个动作

三者的根因是同一个：Agent 既没有世界状态，也没有任务状态。它只有一次输入和一次输出。

补齐它需要三件事，外加一件”做得好”：

  1. 看得见 —— 感知闭环（第 05 篇的 World Model）
  2. 想得清 —— LLM 意图仲裁 + 工具调用
  3. 记得住 —— 任务状态机 + 记忆
  4. 做得好 —— 主动交互编排

二、主动交互：任务流

第 04 篇的 on_skill 是”被动回调中对单条文本的一次性技能下发”。任务流不同：它是主动交互，支持顺序 / 并行 / 随机编排多个动作，并支持流式追加动作组。

2.1 两种模式

任务流的两种模式 SIMPLE 模式在 start_request 时一次性携带全部 payload 后执行；COMPLEX 模式以空 payload 开流，通过多次 task_request 流式追加动作组，最后 end_request 结束。 任务流 · 两种模式 SIMPLE · 单步任务 start_request payload 一次带全 执行 动作集合在发起时就已知 COMPLEX · 多步任务 start_request · payload = [] task_request ① task_request ② … 流式追加 end_request 需按运行时状态逐段下发

判据：任务的动作集合在发起时就已知，用 SIMPLE；需要根据运行时状态逐段下发（先唤醒、再启动讲解、中途追加继续/取消），用 COMPLEX。

2.2 payload 协议

⚠️ payload 必须是 JSON 数组。即使只下发一个 action_group，也必须写成 [{...}]，不能直接传单个对象。这是最常见的首个报错点。

1
2
3
4
5
6
7
8
9
10
[
  {
    "type": "action_group",
    "execution_mode": "parallel",
    "continue_on_failure": false,
    "actions": [
      { "action_type": "tts", "config": { "tts": "hello" } }
    ]
  }
]

action_group 字段：

字段 必填 默认 说明
type 否 action_group 传了就必须是 action_group
actions 是 — action 数组
execution_mode 否 parallel parallel / sequential / random
continue_on_failure 否 false 真正生效的是 action 自身的同名字段

action 字段：

字段 必填 说明
action_type 是 见下表
config 是 各类型的配置
continue_on_failure 否 sequential 模式下，失败是否继续

2.3 支持的 action_type

场景 action_type 可否等待完成
动作 motion 支持
TTS tts 支持
表情 emoticon 不支持
定点导航任务开启 / 继续 / 取消 pilot_task（旧名 fixed_line_intro 仍兼容） 不支持
主动唤醒收音 / 退出全双工 setting 不支持
转弯 move 当前不建议使用（未实现）

经验：execution_scenarios.wait_status 只对 motion / tts 有效。指望用它等导航结束，是设计阶段的常见误判。

2.4 两个容易误解的点

误解 1：多个 group 会并行。

不会。同一个 request 里带多个 action_group 时，当前实现会把它们排队串行执行。group 内是否并行由 execution_mode 决定，但 request 内多个 group 之间不并行。要真并行，就合并进同一个 group。

误解 2：pilot_task 返回即导航完成。

不会。pilot_task 返回的是”控制命令已下发”，不是“导航任务真正完成”，且当前不支持通过 execution_scenarios 等待其执行结束。判断导航完成只能看 on_state 的 locomotion_status。

2.5 编排示例：边讲解边移动边比心

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
[
  {
    "type": "action_group",
    "execution_mode": "parallel",
    "actions": [
      { "action_type": "tts",
        "config": { "tts": "好的，带您去前台，请跟我来" } },
      { "action_type": "emoticon", "config": { "id": 5 } },
      { "action_type": "motion",
        "config": { "id": 59,
                    "execution_scenarios": { "time": 5000,
                                             "wait_status": "kCompleted" } } }
    ]
  },
  {
    "type": "action_group",
    "execution_mode": "sequential",
    "actions": [
      { "action_type": "pilot_task",
        "config": { "control_type": "pilot_start_task",
                    "cmd": "target_poi",
                    "target_name": "前台",
                    "play_tts": true } }
    ]
  }
]

注意两段的语义：第一段三个动作同时起（说话 + 表情 + 动作），第二段在它们之后串行执行导航。导航放第二个 group 而非塞进第一个，正是因为 group 之间是串行的——这里串行恰好是想要的效果。

2.6 COMPLEX 模式：流式追加

适合”先唤醒 → 启动讲解任务 → 中途追加继续/取消”这类多阶段编排。

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
import json

from linksoul_agentsdk import AgentSdk, IdGenerator
from linksoul_agentsdk.enums import FlowMode
from linksoul_agentsdk.active import (
    FlowEndCallback, FlowStartCallback, FlowTaskCallback, TaskFlowRequest,
)

flow_id = IdGenerator.generate_flow_id()
flow = TaskFlowRequest(agent_sdk, "AGENT_001", flow_id, FlowMode.COMPLEX)
agent_sdk.register_task_flow(flow)

class _Start(FlowStartCallback):
    def on_start_request_ack(self, flow_id, event_id, code, msg):
        print(f"启动确认 => code={code}, msg={msg}")
    def on_start_execute_ack(self, flow_id, event_id, code, msg):
        print(f"开始执行 => code={code}, msg={msg}")

# ① 开流（payload 为空数组，仅 mode 生效）
flow.start_request('[]', _Start(), timeout=10000)

class _Task(FlowTaskCallback):
    def on_task_request_ack(self, flow_id, event_id, code, msg):
        print(f"任务确认 => code={code}, msg={msg}")
    def on_task_execute_ack(self, flow_id, event_id, code, msg):
        print(f"任务执行完成 => code={code}, msg={msg}")

# ② 追加动作组（可多次）
flow.task_request(json.dumps([{
    "type": "action_group",
    "execution_mode": "sequential",
    "actions": [
        {"action_type": "setting", "config": {"active_wakeup": True}}
    ],
}], ensure_ascii=False), _Task(), timeout=30000)

# ③ 结束
class _End(FlowEndCallback):
    def on_end_request_ack(self, flow_id, event_id, code, msg):
        print(f"流程结束 => code={code}, msg={msg}")

flow.end_request(_End(), timeout=10000)
agent_sdk.unregister_task_flow(flow_id)

timeout 单位是毫秒，传小于 50 会被钳制到 50ms。任务流整体 2 小时后自动清理——长流程必须在应用层自己持有超时逻辑，不能依赖 SDK 兜底。

三、四层闭环架构

把前面所有碎片整合成一条完整链路：

四层闭环架构 感知层由 on_state 汇聚为按 agent_id 分桶的 World Model；理解层做 LLM 意图仲裁；决策层用任务状态机与任务栈；执行层分 on_skill 与任务流两路。执行结果回写世界状态构成闭环，记忆层向理解层与决策层提供上下文。 执行结果回写 ① 感知闭环 on_state · 22 个状态模块 World Model · 位姿/导航/POI/人群/播报 ② 理解层 LLM 意图仲裁 · 结构化 JSON + 槽位抽取 ③ 决策层 任务状态机 任务栈 · continue/interrupt/switch ④ 执行层 on_skill · 单步技能 任务流 action_group · 多步编排 记忆层 · Session / 长期记忆 / RAG

自下而上的读法更清楚：执行层产生的效果会改变世界状态，所以 E2 → S2 那条虚线不是装饰——它是让闭环成立的关键边。缺了它，系统就退化成一条单向流水线。

3.1 感知闭环

见第 05 篇第二节。两个要点：高频模块（robot_pose / vision_perception，≤10Hz）不要在回调里做重 I/O；planned_path 等数组型状态是整体替换，不是增量合并。

3.2 理解层：LLM + 工具调用

用一次结构化输出替代正则——同时拿到意图、与当前任务的关系、槽位、完成判定、置信度：

1
2
3
4
5
6
7
{
  "intent": "guide",
  "relation": "interrupt",
  "slots": { "品类": "跑鞋", "预算": "800" },
  "task_completed": false,
  "confidence": 0.92
}

用 JSON Schema 严格约束输出，路由决策完全可控、可打日志。这比”在一次对话里顺便判断意图”稳定得多——后者的失败模式不可预测，且无法回归测试。

把能力封装成工具，交给模型调度：

工具 作用 落到哪个接口
navigate_to(poi_name) 语义导航 on_skill("movement", ...) 或任务流 pilot_task
play_action(name) 播放动作 on_skill("action", ...) 或灵创资源
speak(text) 播报 on_llm_* / 任务流 tts
query_robot_state() 查世界模型 读 WORLD[agent_id]
ask_confirm(question) 二次确认 on_skill("confirmation", ...)

🔑 核心原则：状态推进只能通过工具调用发生，闲聊不改变状态。这是整套架构里最值钱的一条约束——它保证状态永远可控、可回溯，模型不会”聊飞”。

延迟优化：门店/展厅场景要求首响 < 1.5 秒。做法是高频意图用规则或小模型前置，只有复杂输入才走大模型。仲裁这一步可以整体换成小模型或本地分类器，接口不用改——这正是把仲裁独立成一层而不是内联在 prompt 里的收益。

3.3 决策层：任务状态机 + 任务栈

长任务（导航讲解、交付流程、导览）建模为显式状态机，用任务栈处理插入：

1
2
3
4
class TaskStack:
    def push(self, task): ...            # 挂起当前，压入新任务（短任务插入）
    def replace(self, task, keep_slots=None): ...   # 同类切换，复用槽位
    def complete_current(self): ...      # 完成并恢复被挂起的任务

任务在栈上的生命周期：

任务在栈上的生命周期 任务由 push 进入 Active；interrupt 使其压栈变为 Suspended，complete_current 恢复为 Active；任务达成进入 Done，switch 换目标则进入 Abandoned。 任务在栈上的生命周期 Active 当前执行中 Suspended 挂起（被插入） Done Abandoned interrupt（压栈） complete_current 恢复 任务达成 switch（同类换目标）

三种关系：

relation 含义 处理
continue 当前任务延续（”有红色的吗”在导购语境下是延续） 不新建任务
interrupt 插入短任务（导航途中问”还有多久到”） 压栈 → 完成后显式复述上下文再恢复
switch 同类换目标（去前台 → 去茶水间） 旧任务 abandoned，保留可复用槽位

三条配套纪律：

  1. 槽位与任务绑定，不跨任务污染——用户在导购中途问”卫生间在哪”，导航任务不应污染导购任务的槽位。
  2. suspended 任务必须过期（门店/展厅建议 3–5 分钟），否则半小时前的话题会被错误恢复。
  3. 恢复任务时显式复述：”刚才您还在选购（品类是跑鞋、预算 800），需要继续吗？”现场用户注意力分散，静默恢复只会让人困惑。

置信度低时追问，不要猜：

1
2
if decision["confidence"] < 0.6:
    return "不好意思，没太听清您的意思，您是想看商品、问活动，还是找位置呢？"

取舍：追问会让单轮交互变长，但比”猜错后执行一个错误动作”代价低得多——尤其在移动类技能上，猜错意味着机器人真的走错了地方。

3.4 执行层：平台原生优先

场景 用什么 理由
单步技能（动作 / 表情 / 移动 / 确认） response.on_skill(...) 平台按型号过滤，含安全语义
多步编排（边讲边动、定点讲解） 任务流 action_group 支持顺序 / 并行 / 随机，可流式追加
平台未覆盖（自研感知、抓取、精细控制） 端侧 AIMDK 直连 只在必要时使用

💡 不要绕过平台。第 05 篇的官方 demo 就是自己直连 ROS，结果是换型号即失效、绕过安全仲裁、重复造轮子——三个代价一次付清。

四、记忆层

层 做法
短期 Session 按 session_id 隔离（多客户各一份状态 + 消息历史）
长期 用 on_face_info 的 face_uid 绑定用户身份，做个性化与长期记忆
知识 RAG 知识库——商品库 / 楼层导览 / 售后 FAQ 分开建索引，意图路由后再检索，准确率远高于混在一起
持久化 Session 落 Redis / DB，对话中断后可恢复

多人支持的隔离设计：

1
2
3
4
5
6
7
8
@dataclass
class Session:
    customer: str
    stage: Stage = Stage.WELCOME
    checked: dict = field(default_factory=dict)
    messages: list = field(default_factory=list)

SESSIONS: dict[str, Session] = {}   # session_id -> Session

⚠️ 但 SDK 的回调是单例，所有智能体共享同一个回调实例。所以 SESSIONS / WORLD 这类状态必须按 agent_id 再分一层，否则多机器人上线即串台：

1
SESSIONS: dict[str, dict[str, Session]] = {}   # agent_id -> session_id -> Session

这一条在单机联调时完全暴露不出来，是多机部署最常见的翻车点。

五、端云分工再平衡

第 05 篇给过一张分工表，这里补充”什么时候该把能力下沉到端侧”：

判断 放哪
需要毫秒级反应（急停、避障、平衡） 端侧
断网也必须能用（基础动作、安全） 端侧
涉及大模型、知识库、长期记忆 云端
需要频繁迭代、灰度发布 云端（端侧发版成本高）

关于自研语音系统：官方提供过渡方案——通过 SetAgentPropertiesRequest 把运行模式切到 only_voice，关闭内置交互，释放音频流给自己的系统。但注意三点：

  • 这是临时过渡手段，后续版本会支持”一键切换交互二开模式”
  • 关闭后端侧的基础交互能力也没了，需要自己补齐 VAD、唤醒、播报排队
  • 因此更稳的做法是：只在确实需要定制语音链路时才关，否则优先复用端侧内置能力

取舍：自研语音链路的真实成本不在模型选型，而在 VAD 端点检测、打断时序、播报排队这些”看不见的工程量”。评估时把这三项算进去，往往结论就变成”先复用内置”。

六、两个完整案例

6.1 案例: 门店导购 Agent（任务栈 + 意图仲裁）

三个 Skill：导购（商品推荐）、问答（RAG 知识库）、导航（店内 POI），统一接口：

1
2
3
4
class Skill:
    """统一接口：handle(task, user_input) -> (回复文本, 任务是否完成)"""
    def retrieve(self, task, user_input) -> str: ...
    def handle(self, task, user_input) -> tuple[str, bool]: ...

仲裁与执行是两次 LLM 调用。看似多花一次，但仲裁用严格 JSON Schema，路由决策完全可控、可打日志。延迟敏感时把仲裁换成小模型或本地分类器，接口不用改。

一次典型的多轮交互，注意任务栈在其中的作用：

门店导购一次多轮交互的时序 顾客提出选购需求后压入导购任务；中途插入导航问答任务使导购任务挂起，完成后恢复并显式复述上下文；随后顾客的追问被判定为 continue，不新建任务。 门店导购 · 一次多轮交互的时序 顾客 Agent 任务栈 导购 Skill relation = continue，不新建任务 想买双跑鞋，预算 800 以内 卫生间在哪 有红色的吗 push(guide, 品类=跑鞋, 预算=800) push(nav_qa) ← 挂起 guide complete_current() ← 恢复 guide handle(task, input) handle(task, input) 推荐飞速跑鞋 Pro，799 元 2 楼扶梯右转直走 20 米 刚才您还在选购（跑鞋 / 800）， 需要继续吗？ 飞速跑鞋 Pro 有红色…

映射到机器人：导航 Skill 直接对接 pilot_task；导购/问答 Skill 的回复走 on_llm_*；配合 vision_perception 与 speaker 做多人区分。

七、复杂场景的三个专项

7.1 多人场景（最容易翻车）

多人场景的三层分工 输入层负责说话人区分（声纹、麦克风阵列声源定位、人脸追踪、内容连贯性兜底），状态层做会话隔离，输出层决定交互策略。 多人场景 · 三层分工 输入层 · 说话人区分 失败模式：认错人 声纹识别 麦克风阵列声源定位 speaker_id / doa 人脸追踪 face_uid 内容连贯性兜底 状态层 · 会话隔离 失败模式：串台 每个 speaker_id 独立会话 + 任务栈 群组会话 结伴顾客例外 输出层 · 交互策略 失败模式：答错人 明确应答对象 插话排队，不丢弃当前会话 无法区分 → 降级单会话 + 主动确认

三层各有各的失败模式：输入层错在认错人，状态层错在串台，输出层错在答错人。三者中状态层最容易修（纯代码问题），输入层最难（依赖硬件与算法）。

📌 实施建议：先把”单人 + 任务栈”跑通再做多人。多人场景 80% 的复杂度在感知层（声纹、视觉追踪），不在对话层，因此建议分期实施，不要让感知层的进度阻塞对话层上线。

7.2 延迟优化

手段 说明
意图前置 高频意图用规则 / 分类器，复杂输入才走大模型
流式输出 ASR / LLM / TTS 都是流式（*_delta + *_done），边产出边消费
端侧兜底 网络抖动时用端侧预置应答，避免”没反应”
动作与播报并行 on_skill 与 on_llm_* 互不阻塞，可同时下发

7.3 兜底与风控

场景 做法
敏感话题 on_interrupt(event_id, "risk_control", "<话术>")，interrupt_tips 必填
移动类技能 先走 confirmation 二次确认（文档已标注哪些技能”需二次确认”）
识别失败 / 任务冲突 统一转”人工呼叫”路径——现场用户耐心远低于线上
端侧调用失败 调 response.on_error(event_id, code, msg) 上报

八、工程化

项目 做法
异步化 on_request 内禁止同步阻塞；耗时任务丢线程池，回调立即返回
状态分桶 World Model / Session / 任务栈全部按 agent_id 分桶，避免单例串台
可观测 每轮打印工具调用与状态迁移；生产环境换落库 + 埋点
异常上报 统一走 on_error，让用户端有反馈
凭证管理 环境变量或密钥管理服务，绝不硬编码
配置外置 地图 ID、点位、速度参数从配置读，不写死在代码里
降级预案 云端不可用时，端侧保留基础动作与安全能力

前两项是架构级约束，后五项是运维级约束。架构级的坑一旦踩了，后期改造要动全部业务代码；运维级的坑可以逐个补。

九、落地路线

P0 到 P5 的落地路线 从 P0 修复让现有功能可用，到 P1 感知让 Agent 看得见，再到 P2 理解、P3 决策、P4 编排与 P5 多人，逐级推进。 落地路线 · P0 → P5 P0 修复 让现有功能可用 P1 感知 Agent 看得见 P2 理解 规则 → LLM P3 决策 单轮 → 任务 P4 编排 复杂表现 P5 多人 多人场景 止损 能力起点 · ROI 最高 规模化
阶段 目标 关键动作 验证标准
P0 修复 让现有功能可用 修作用域 bug、分支顺序、补 on_interrupt；凭证外置 导航有语音反馈；”停止导航”能停
P1 感知 Agent 看得见 实现 on_state，建 World Model + 语义点位表 “去前台”命中真实点位
P2 理解 从规则到 LLM 意图仲裁（结构化输出）+ function calling 多轮指代、槽位抽取可用
P3 决策 从单轮到任务 任务状态机 + 任务栈 + Session 导航中途插话后能恢复
P4 编排 复杂表现 任务流 action_group；用 on_skill 替代直连 ROS “边讲解边移动边比心”
P5 多人 多人场景 vision_perception + speaker + 声纹/人脸绑定 区分服务对象不串台

投入产出比最高的是 P1——只加一个 on_state，就能让”去前台”从写死坐标变成语义导航。P0 是止损，P1 才是真正的能力起点。

结论

六篇走下来，能力曲线是这样的：

篇 获得的
01 知道二开有两条路，以及各自的能力边界
02 能把机器人连起来，跑通第一个 AIMDK 节点
03 掌握端侧控制、交互、感知接口
04 能独立写一个合规的云端应答 Agent
05 能打通端云，并识别 / 修复常见陷阱
06 具备设计复杂场景 Agent 的架构能力

如果只记住三件事：

  1. 让 Agent 看得见——on_state 是最被低估的接口，先接它。
  2. 让状态可控——状态迁移只通过工具调用发生，模型就不会”聊飞”。
  3. 别绕过平台——on_skill 和任务流能做的事，不要自己直连 ROS 重造。

这三条分别对应本文档的三个层次：感知（世界状态）→ 决策（任务状态）→ 执行（平台能力）。缺任何一条，系统都能跑，但都撑不住复杂场景。

声明

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

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