主动交互开发指南

← 返回首页

主动交互是 SDK 向机器人发起请求的能力。v1.3.0 起在接口层面开放（彼时一并暴露过技能查询 / 状态查询 / 拉流 / 推流监听等接口，但这些接口实际未启用，只有任务流真正可用）。v1.4.0 顺势把这些未启用接口从公开 API 中移除，公开入口收敛到 任务流——底层类仍保留在源码里，但不再通过 AgentSdk.register* 公开。

状态推送、打招呼信令、视频帧透传、人脸/声纹 UID 等场景已统一并入被动交互的 PassiveCallback 基类方法。

任务流

任务流用于向机器人下发多步任务序列。支持两种模式：

SIMPLE：单步任务（start_request 中直接包含 payload，无需 task_request）
COMPLEX：多步任务（start → task × N → end）

完整生命周期示例（COMPLEX 模式）

python

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


# ① 创建任务流（指定目标机器人和流程 ID）
flow_id = IdGenerator.generate_flow_id()
flow = TaskFlowRequest(agent_sdk, "AGENT_001", flow_id, FlowMode.COMPLEX)

# ② 注册到 SDK
agent_sdk.register_task_flow(flow)


# ③ 启动流程（payload 是 JSON 数组，详细协议见 taskflow_payload.md）
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}")


flow.start_request('[{"goal":"navigate_to_kitchen"}]', _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('[{"action":"turn_left","angle":90}]', _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)

中断流程

python

from linksoul_agentsdk.active import FlowInterruptCallback


class _Interrupt(FlowInterruptCallback):
    def on_interrupt_request_ack(self, flow_id, event_id, code, msg):
        print(f"中断 => code={code}, msg={msg}")


flow.interrupt_request(_Interrupt(), timeout=10000)

回调参数说明

参数	说明
`flow_id`	流程唯一标识
`event_id`	本次请求的唯一标识
`code`	0=成功，非 0=失败
`msg`	成功时为 `"success"`，失败时为错误描述

start_request / task_request 的 payload 是 JSON 数组字符串，由若干 action_group 组成，每个 group 包含一组 action（TTS、动作、表情、导航、设置等）。即使只下发一个 action_group，也必须放进数组里（'[{...}]'），不能直接传单个对象。完整协议、各 action_type 字段以及 SIMPLE / COMPLEX 两种模式的端到端示例请参见：

👉 任务流 payload 协议

状态监听

v1.4.0 变更：状态监听已合并到被动交互的 PassiveCallback.on_state(...)，无需再单独注册。

直接在已注册的被动回调里重载即可：

python

class MyCallback(Audio2TtsCallback):
    def on_state(self, agent_id, event_id, state_name, state_value):
        print(f"状态变化 => {state_name} = {state_value}")
        # state_value 可能是 JSON 字符串，需要自行解析

    def on_request(self, *a, **k):
        ...  # 主路径

超时机制

TaskFlowRequest 的每个请求方法都支持 timeout 参数（单位：毫秒），超时未收到响应后回调不会被触发，资源自动清理：

操作	建议超时	自动清理时间
`start_request` / `task_request` / `end_request` / `interrupt_request`	业务自定	任务流整体 2 小时

当传入的 timeout < 50 时会自动钳制到 50ms。

不再开放的接口

技能查询（query_skill）、状态查询（query_state）、拉流（register_pull_stream）、推流监听（register_push_listen）的注册方法已在 v1.4.0 从 AgentSdk 移除，本指南不再展示其用法。对应底层类（SkillQueryRequest / StateQueryRequest / PullStreamRequest / PushListenCallback）也不再公开。