AIMA开放平台文档中心Agent SDK (Python)

快速开始

快速开始

← 返回首页

环境要求

依赖版本
Python3.9+
pip23+（或 uv / poetry 等）
websocket-client≥ 1.7.0（自动安装）

安装

pip（推荐）

bash
pip install linksoul-agentsdk
# 或者从源码安装：
git clone <repo> && cd linksoul-agentsdk/agentsdk_for_python
pip install -e .

Poetry / PDM

toml
# pyproject.toml
[project]
dependencies = [
    "linksoul-agentsdk>=1.4.0a0",
]

注意：app_key 与 app_secret 用于 HMAC-SHA256 网关签名，请妥善保存，不要写进可被公开的代码或日志。

完整接入示例

下面以最经典的 Asr2Llm（文本输入 → LLM 文本输出）为例展示完整接入流程，其他被动回调 接入方式完全相同，只是 register_* 与 on_request 签名不同（详见 被动示例）。

python
import logging

from linksoul_agentsdk import (
    AgentAuthCallback,
    AgentParam,
    AgentSdk,
    IdGenerator,
)
from linksoul_agentsdk.passive import Asr2LlmCallback, Asr2LlmResponse


# 鉴权回调
class MyAuth(AgentAuthCallback):
    def on_auth_state(self, app_id: str, code: int, msg: str) -> None:
        if code == 0:
            print(f"连接认证成功: {app_id}")
        else:
            print(f"认证失败: code={code}, msg={msg}")


# 业务回调：Asr2Llm（机器人已完成 ASR，SDK 仅做语义理解/LLM）
class MyCallback(Asr2LlmCallback):
    def on_request(
        self,
        agent_id: str,
        event_id: str,
        text: str,
        param: AgentParam,
        response: Asr2LlmResponse,
    ) -> None:
        # ① 拒识 → 静默不应答
        if not text:
            return

        item_id = IdGenerator.generate_item_id()

        # ② 有效指令——必须先打断，再回填技能 / LLM
        response.on_interrupt(event_id, "chat", None)

        # ③ 如分类为技能则下发技能指令（可选）
        if "向前" in text:
            skill_param = (
                AgentParam.create()
                .set_string("movement", "move_forward")
                .set_integer("meter", 10)
            )
            response.on_skill(event_id, item_id, "movement", "move_forward", skill_param)

        # ④ LLM 流式输出
        response.on_llm_item_delta(event_id, item_id, "好的，正在处理。")
        response.on_llm_item_done(event_id, item_id)
        response.on_llm_done(event_id)


def main() -> None:
    logging.basicConfig(level=logging.INFO)

    # ① 创建 SDK 实例（单例：同一 app_id 多次调用返回同一实例）
    sdk = AgentSdk.create(
        url="wss://open.agibot.com/api/V1/open-portal/app/wss/agent-sdk",
        app_id="您的 appId",
        app_key="您的 appKey",
        app_secret="您的 appSecret",
    )

    # ② 注册鉴权回调（监听连接状态）
    sdk.register_auth(MyAuth())

    # ③ 注册被动回调（按场景选择一种）
    sdk.register_asr2_llm(MyCallback(sdk))

    # ④ 初始化（建立 WebSocket 连接，自动重连）
    sdk.initialize()


if __name__ == "__main__":
    main()

核心概念速查

概念说明
app_id / app_key / app_secret灵心开放平台创建应用后获取；app_key + app_secret 用于 HMAC-SHA256 网关签名
被动交互机器人发起请求 → SDK 处理 → 通过 Response 返回结果
主动交互SDK 主动向机器人发起任务流（register_task_flow）
agent_id灵心平台配置的智能体 ID（如 AGENT_0000001）。机器人在端侧绑定该智能体后上线，SDK 通过 on_robot_online(agent_id, ...) 回传——不是机器人硬件标识
event_id一次完整对话的唯一标识
item_id同一对话中多个意图/响应片段的标识
flag音频/视频回调中的生命周期标志（仅 Audio2* / AudioVideo2* / AsrVideo2* 有该字段）
AgentParam扩展参数（KV 结构，支持链式调用）
AgentMeta机器人元数据（唤醒词、城市、机器人名称等）

接入流程

text
AgentSdk.create() → register_auth() → register_*_callback() → initialize()
                                                                      │
                                                                      ▼
                                                              WebSocket 连接
                                                                      │
                                                          ┌───────────┴───────────┐
                                                          ▼                       ▼
                                                    on_auth_state()         on_robot_online()
                                                    （认证结果）            （机器人上线）
                                                                                  │
                                                                                  ▼
                                                                            on_request()
                                                                          （处理音频/文本）
                                                                                  │
                                                                                  ▼
                                                                        response.on_*()
                                                                        （返回处理结果）

下一步