主动请求 API 参考

← 返回首页

v1.4.0：当前 AgentSdk 仅公开 任务流 一个主动入口。技能查询、状态查询、拉流、推流监听暂未开放。状态推送已合并到 PassiveCallback.on_state(...)，详见被动回调指南。

TaskFlowRequest

来自 linksoul_agentsdk.active。

构造

python

from linksoul_agentsdk.active import TaskFlowRequest
from linksoul_agentsdk.enums import FlowMode

# COMPLEX 模式（默认）
flow = TaskFlowRequest(agent_sdk, agent_id, flow_id)

# 指定模式
flow = TaskFlowRequest(agent_sdk, agent_id, flow_id, FlowMode.SIMPLE)

方法

方法	返回值	说明
`start_request(payload, callback, timeout)`	`str`（event_id）	启动流程
`task_request(payload, callback, timeout)`	`str`（event_id）	发送任务步骤
`end_request(callback, timeout)`	`str`（event_id）	结束流程
`interrupt_request(callback, timeout)`	`str`（event_id）	中断流程
`flow_id` (property)	`str`	流程 ID
`agent_id` (property)	`str`	目标智能体 ID（灵心平台配置的智能体 ID）
`flow_mode` (property)	`FlowMode`	当前模式

payload: JSON 字符串（任务流协议要求为 JSON 数组，如 '[{...}]'，详见任务流 payload 协议）
timeout: 超时（毫秒），小于 50ms 会自动钳制到 50ms

流程回调

python

from linksoul_agentsdk.active import (
    FlowEndCallback,
    FlowInterruptCallback,
    FlowStartCallback,
    FlowTaskCallback,
)


class _Start(FlowStartCallback):
    def on_start_request_ack(self, flow_id, event_id, code, msg): ...
    def on_start_execute_ack(self, flow_id, event_id, code, msg): ...


class _Task(FlowTaskCallback):
    def on_task_request_ack(self, flow_id, event_id, code, msg): ...
    def on_task_execute_ack(self, flow_id, event_id, code, msg): ...


class _End(FlowEndCallback):
    def on_end_request_ack(self, flow_id, event_id, code, msg): ...


class _Interrupt(FlowInterruptCallback):
    def on_interrupt_request_ack(self, flow_id, event_id, code, msg): ...

两阶段确认：RequestAck = 机器人收到请求，ExecuteAck = 机器人执行完成。code=0 表示成功。同一个 task_request 的 request_ack 与 execute_ack 共享同一个 event_id。

工具类

AgentParam（扩展参数）

来自 linksoul_agentsdk.AgentParam。支持链式调用：

python

from linksoul_agentsdk import AgentParam

# 创建并设置参数
param = (
    AgentParam.create()
    .set_string("action", "navigate")
    .set_integer("step", 3)
    .set_double("speed", 1.5)
    .set_boolean("safe", True)
    .set_long("timestamp", int(time.time() * 1000))
)

# 读取参数
action = param.get_string("action")    # → "navigate"
step = param.get_integer("step")       # → 3

完整方法列表：

方法	类型
`get_string` / `set_string`	`str`
`get_integer` / `set_integer`	`int`（兼容 `get_long` / `set_long` / `get_short` / `set_short`）
`get_double` / `set_double`	`float`
`get_boolean` / `set_boolean`	`bool`（兼容 `"true"` / `"false"` 字符串）
`get_object` / `set_object`	`Any`
`get_list` / `set_list`	`list`
`get_ext_param()`	底层 `dict`

Python 特别注意：bool 在 Python 里是 int 的子类，get_integer 已做防护—— 对 set_boolean 设入的值返回 None，避免误读为 0/1。

AgentMeta（机器人元数据）

机器人上线时通过 on_robot_online(agent_id, agent_meta) 获取：

python

def on_robot_online(self, agent_id: str, agent_meta: AgentMeta) -> None:
    wakeup = agent_meta.get_wakeup_word()    # 唤醒词
    city = agent_meta.get_city()              # 所在城市
    name = agent_meta.get_robot_name()        # 机器人名字
    buzz_word = agent_meta.get_buzz_word()    # 热词
    custom = agent_meta.get_string("your_key")  # 自定义字段

IdGenerator（ID 生成器）

python

from linksoul_agentsdk import IdGenerator

flow_id = IdGenerator.generate_flow_id()       # "flow_xxxxx..."
event_id = IdGenerator.generate_event_id()     # "event_xxxxx..."
item_id = IdGenerator.generate_item_id()       # "item_xxxxx..."
signal_id = IdGenerator.generate_signal_id()   # "signal_xxxxx..."

所有 ID 为 前缀 + 21 位随机字符串（大小写字母+数字），使用 secrets.choice(...) 保证安全随机。