主动交互示例

← 返回首页
源码位置：agentsdk_for_python/examples/task_flow_example.py

主动交互在 v1.3.0 起就开放了接口（一并暴露过技能查询 / 状态查询 / 拉流 / 推流监听，但这些接口实际未启用，只有任务流真正可用）。v1.4.0 把这些未启用接口从公开 API 移除，对外公开入口收敛为 任务流（register_task_flow / unregister_task_flow），对应的示例文件就一个：task_flow_example.py。

主动交互的通用骨架

python

import time
from linksoul_agentsdk import AgentAuthCallback, AgentSdk


class _Auth(AgentAuthCallback):
    def on_auth_state(self, app_id, code, msg):
        print(f"on_auth_state => app_id={app_id} code={code} msg={msg}")


sdk = AgentSdk.create(
    url="wss://open.agibot.com/api/V1/open-portal/app/wss/agent-sdk",
    app_id="<your appId>",
    app_key="<your appKey>",
    app_secret="<your appSecret>",
)
sdk.register_auth(_Auth())

# 主动交互在 initialize() 之后才能发起
sdk.initialize()

time.sleep(2)  # 等鉴权 + 上线

注意：与被动交互不同，主动交互必须在 initialize() 之后发起。

TaskFlowExample — 任务流（启动 / 任务 / 结束 / 打断）

源码：task_flow_example.py
TaskFlowRequest 把"机器人执行多步任务"抽象成一条流水线： start_request（必须） → task_request（可选，可多次） → end_request（可选） → interrupt_request（可选）。每个方法都有两阶段确认：RequestAck 表示请求被机器人收到，ExecuteAck 表示该步已执行完成。
本示例的 payload 用占位 "[{}]" 演示流程；真实业务请按任务流 payload 协议填写 action_group / action_type。

1) 注册任务流

python

from linksoul_agentsdk import IdGenerator
from linksoul_agentsdk.active import TaskFlowRequest

agent_id = "<the target agentId>"
flow_id = IdGenerator.generate_flow_id()
flow = TaskFlowRequest(sdk, agent_id, flow_id)
sdk.register_task_flow(flow)

2) 打断请求（可选，可在任意阶段调用）

python

from linksoul_agentsdk.active import FlowInterruptCallback


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


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

3) 启动请求（必须）

python

from linksoul_agentsdk.active import FlowStartCallback


class _Start(FlowStartCallback):
    def on_start_request_ack(self, flow_id, event_id, code, msg):
        # 机器人已收到 start 请求
        print(f"start.requestAck => {flow_id} {event_id} code={code} msg={msg}")

    def on_start_execute_ack(self, flow_id, event_id, code, msg):
        # 机器人 start 阶段执行完成
        print(f"start.executeAck => {flow_id} {event_id} code={code} msg={msg}")


flow.start_request("[{}]", _Start(), timeout=2000)

4) 任务请求（可选，可多次调用）

python

from linksoul_agentsdk.active import FlowTaskCallback


class _Task(FlowTaskCallback):
    def on_task_request_ack(self, flow_id, event_id, code, msg):
        print(f"task.requestAck => {flow_id} {event_id} code={code} msg={msg}")

    def on_task_execute_ack(self, flow_id, event_id, code, msg):
        print(f"task.executeAck => {flow_id} {event_id} code={code} msg={msg}")


flow.task_request("[{}]", _Task(), timeout=2000)

5) 结束请求（可选）

python

from linksoul_agentsdk.active import FlowEndCallback


class _End(FlowEndCallback):
    def on_end_request_ack(self, flow_id, event_id, code, msg):
        print(f"end.requestAck => {flow_id} {event_id} code={code} msg={msg}")


flow.end_request(_End(), timeout=2000)

任务流回调一览

回调	触发时机	含义
`on_start_request_ack`	启动请求到达机器人	`code=0` 表示请求被接受
`on_start_execute_ack`	启动阶段执行完成	多步任务可以开始下发
`on_task_request_ack`	任务步骤到达机器人	单步请求被接受
`on_task_execute_ack`	任务步骤执行完成	可以继续下一步
`on_end_request_ack`	结束请求被处理	整条流水线已收口
`on_interrupt_request_ack`	打断请求被处理	已通知机器人中断当前流程

TaskFlow 选择 SIMPLE 还是 COMPLEX？

TaskFlowRequest 的构造函数支持两种模式（不传则默认 COMPLEX）：

python

from linksoul_agentsdk.enums import FlowMode

# COMPLEX：start → task × N → end（适合多步任务）
flow = TaskFlowRequest(sdk, agent_id, flow_id)
flow = TaskFlowRequest(sdk, agent_id, flow_id, FlowMode.COMPLEX)

# SIMPLE：start_request 自带完整 payload，不需要 task_request
flow = TaskFlowRequest(sdk, agent_id, flow_id, FlowMode.SIMPLE)

状态监听（通过被动回调接收）

状态变化推送已合并到 PassiveCallback.on_state(...)，直接在已注册的被动回调里重载即可，详见被动交互指南 - 基类共享回调。

python

from linksoul_agentsdk.passive import Audio2TtsCallback


class MyCb(Audio2TtsCallback):
    def on_state(self, agent_id, event_id, state_name, state_value):
        print(f"状态变化: {state_name} = {state_value}")

    def on_request(self, *a, **k):
        ...