被动交互示例

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

8 个被动交互示例按输入输出复杂度分为三类。每个 *_example.py 都是可独立运行的 main，仅在 register_* 注册的回调类型与 on_request 签名上有差异，其余流程（创建实例、注册鉴权、基类共享回调、初始化）完全一致。

通用骨架（所有示例共享）

python

import logging
from linksoul_agentsdk import AgentAuthCallback, AgentSdk


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


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())

# 第三步：注册业务回调（按场景选择 register_* 方法，见下文各示例）
sdk.register_xxx(YourCallback(sdk))

# 第四步：初始化（建立 WebSocket，自动重连）
sdk.initialize()

基类共享回调（v1.4.0+）

所有 *Callback 都继承自 PassiveCallback，可选重载以下方法（不重载则使用默认日志实现）：

python

from linksoul_agentsdk import AgentMeta, AgentParam
from linksoul_agentsdk.passive import GreetResponse, PassiveCallback


class MyCallbackBase(PassiveCallback):
    # （以下是基类已经提供的默认实现，仅在需要自定义时重载）

    def on_robot_online(self, agent_id, agent_meta: AgentMeta):
        print(f"on_robot_online => {agent_id}, wakeup={agent_meta.get_wakeup_word()}")

    def on_robot_offline(self, agent_id):
        print(f"on_robot_offline => {agent_id}")

    def on_face_info(self, agent_id, event_id, param: AgentParam):
        # 识别到人脸 / 声纹 UID（v1.4.0 开放）
        ...

    def on_video_frame(self, agent_id, event_id, flag, buf, is_key_frame, local_ts, param):
        # 视频帧透传：flag=2 H264，flag=3 图片
        ...

    def on_greet_signal(self, agent_id, event_id, param, response: GreetResponse):
        # 主动打招呼信令，可通过 response.on_greet_vlm_* / response.on_greet_tts_* 应答
        ...

    def on_state(self, agent_id, event_id, state_name, state_value):
        # 机器人状态推送（v1.4.0 起合并到此处，不再单独注册 StateListenCallback）
        ...

后续每个示例只展示 register_* 调用 + on_request 主体，其余按"通用骨架"组装即可。

一、纯音频类

Audio2LlmExample — 音频输入 → ASR + LLM 文本输出

源码：audio2_llm_example.py
适用场景：语音对话，纯文本回复（不合成 TTS）。

python

from linksoul_agentsdk import AgentParam, IdGenerator
from linksoul_agentsdk.passive import Audio2LlmCallback, Audio2LlmResponse


class _Cb(Audio2LlmCallback):

    def on_request(
        self, agent_id, event_id, flag, buf, param: AgentParam,
        response: Audio2LlmResponse,
    ):
        """flag: 0 VAD 开始 / 1 中间音频帧 / 2 VAD 结束"""

        if flag == 0:
            return  # 初始化 ASR 客户端
        if flag == 1:
            return  # 把 buf 送 ASR
        # flag == 2 —— 拿到 final ASR
        response.on_asr_middle(event_id, "今天")
        response.on_asr_final(event_id, "今天天气咋样")

        if not "今天天气咋样":  # 拒识占位逻辑
            return

        item_id = IdGenerator.generate_item_id()
        # 有效指令必须先打断
        response.on_interrupt(event_id, "chat", None)
        response.on_llm_item_delta(event_id, item_id, "今天天气不错，适合出行。")
        response.on_llm_item_done(event_id, item_id)
        response.on_llm_done(event_id)


sdk.register_audio2_llm(_Cb(sdk))

Audio2TtsExample — 音频输入 → ASR + LLM + TTS 完整管道

源码：audio2_tts_example.py
适用场景：最常见的语音对话→语音回复闭环。on_request 签名比 Audio2Llm 多一个 item_id 形参。

python

from linksoul_agentsdk.passive import Audio2TtsCallback, Audio2TtsResponse


class _Cb(Audio2TtsCallback):
    def on_request(
        self, agent_id, event_id, item_id, flag, buf, param,
        response: Audio2TtsResponse,
    ):
        if flag in (0, 1):
            return

        # flag == 2 VAD 结束
        response.on_asr_final(event_id, "今天天气怎么样")
        response.on_interrupt(event_id, "chat", None)
        response.on_llm_item_delta(event_id, item_id, "今天天气不错。")
        response.on_llm_item_done(event_id, item_id)
        response.on_llm_done(event_id)

        tts_audio = b"\x00" * 1024  # 替换为真实 TTS 合成结果
        response.on_tts_item_delta(event_id, item_id, tts_audio)
        response.on_tts_item_done(event_id, item_id)
        response.on_tts_done(event_id)


sdk.register_audio2_tts(_Cb(sdk))

二、纯文本类

Asr2LlmExample — ASR 文本输入 → LLM 输出

源码：asr2_llm_example.py
适用场景：上游已完成 ASR 识别，只做语义理解/LLM 处理。

python

from linksoul_agentsdk.passive import Asr2LlmCallback, Asr2LlmResponse


class _Cb(Asr2LlmCallback):
    def on_request(self, agent_id, event_id, text, param, response: Asr2LlmResponse):
        if not text:
            return  # 拒识 → 静默
        item_id = IdGenerator.generate_item_id()
        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)

        response.on_llm_item_delta(event_id, item_id, "好的，正在处理。")
        response.on_llm_item_done(event_id, item_id)
        response.on_llm_done(event_id)


sdk.register_asr2_llm(_Cb(sdk))

Asr2TtsExample — ASR 文本输入 → LLM + TTS 输出

源码：asr2_tts_example.py

python

from linksoul_agentsdk.passive import Asr2TtsCallback, Asr2TtsResponse


class _Cb(Asr2TtsCallback):
    def on_request(self, agent_id, event_id, text, param, response: Asr2TtsResponse):
        if not text:
            return
        item_id = IdGenerator.generate_item_id()
        response.on_interrupt(event_id, "chat", None)

        response.on_llm_item_delta(event_id, item_id, "好的，我来回答你的问题。")
        response.on_llm_item_done(event_id, item_id)
        response.on_llm_done(event_id)

        response.on_tts_item_delta(event_id, item_id, b"\x00" * 1024)
        response.on_tts_item_done(event_id, item_id)
        response.on_tts_done(event_id)


sdk.register_asr2_tts(_Cb(sdk))

三、多模态类（文本/音频 + 视频）

AsrVideo2VlmExample — ASR 文本 + 视频 → VLM 输出

源码：asr_video2_vlm_example.py

python

from linksoul_agentsdk.passive import AsrVideo2VlmCallback, AsrVideo2VlmResponse


class _Cb(AsrVideo2VlmCallback):
    def on_request(
        self, agent_id, event_id, flag, text, buf, param,
        response: AsrVideo2VlmResponse,
    ):
        """flag: 3 ASR 文本，4 H264 视频帧，5 图片帧"""
        if flag != 3 or not text:
            return  # 视频帧自己缓存，等 final ASR

        item_id = IdGenerator.generate_item_id()
        response.on_interrupt(event_id, "chat", None)
        response.on_vlm_item_delta(event_id, item_id, "我看到了一个杯子。")
        response.on_vlm_item_done(event_id, item_id)
        response.on_vlm_done(event_id)


sdk.register_asr_video2_vlm(_Cb(sdk))

AsrVideo2TtsExample — ASR 文本 + 视频 → VLM + TTS 输出

源码：asr_video2_tts_example.py

python

from linksoul_agentsdk.passive import AsrVideo2TtsCallback, AsrVideo2TtsResponse


class _Cb(AsrVideo2TtsCallback):
    def on_request(
        self, agent_id, event_id, flag, text, buf, param,
        response: AsrVideo2TtsResponse,
    ):
        if flag != 3 or not text:
            return
        item_id = IdGenerator.generate_item_id()
        response.on_interrupt(event_id, "chat", None)
        response.on_vlm_item_delta(event_id, item_id, "看到杯子，正在描述。")
        response.on_vlm_item_done(event_id, item_id)
        response.on_vlm_done(event_id)
        response.on_tts_item_delta(event_id, item_id, b"\x00" * 1024)
        response.on_tts_item_done(event_id, item_id)
        response.on_tts_done(event_id)


sdk.register_asr_video2_tts(_Cb(sdk))

AudioVideo2VlmExample — 音频 + 视频 → ASR + VLM 输出

源码：audio_video2_vlm_example.py

python

from linksoul_agentsdk.passive import AudioVideo2VlmCallback, AudioVideo2VlmResponse


class _Cb(AudioVideo2VlmCallback):
    def on_request(
        self, agent_id, event_id, flag, buf, param,
        response: AudioVideo2VlmResponse,
    ):
        """flag: 0 VAD 开始 / 1 音频帧 / 2 VAD 结束 / 3 H264 视频帧"""
        if flag in (0, 1, 3):
            return
        # flag == 2 (VAD 结束) → 拿到 ASR 文本，再结合缓存的视频帧调用 VLM
        response.on_asr_final(event_id, "前面是什么？")
        item_id = IdGenerator.generate_item_id()
        response.on_interrupt(event_id, "chat", None)
        response.on_vlm_item_delta(event_id, item_id, "前面是一张桌子。")
        response.on_vlm_item_done(event_id, item_id)
        response.on_vlm_done(event_id)


sdk.register_audio_video2_vlm(_Cb(sdk))

AudioVideo2TtsExample — 音频 + 视频 → ASR + VLM + TTS 输出

源码：audio_video2_tts_example.py

python

from linksoul_agentsdk.passive import AudioVideo2TtsCallback, AudioVideo2TtsResponse


class _Cb(AudioVideo2TtsCallback):
    def on_request(
        self, agent_id, event_id, flag, buf, param,
        response: AudioVideo2TtsResponse,
    ):
        if flag in (0, 1, 3):
            return
        response.on_asr_final(event_id, "前面是什么？")
        item_id = IdGenerator.generate_item_id()
        response.on_interrupt(event_id, "chat", None)
        response.on_vlm_item_delta(event_id, item_id, "前面是一张桌子。")
        response.on_vlm_item_done(event_id, item_id)
        response.on_vlm_done(event_id)
        response.on_tts_item_delta(event_id, item_id, b"\x00" * 1024)
        response.on_tts_item_done(event_id, item_id)
        response.on_tts_done(event_id)


sdk.register_audio_video2_tts(_Cb(sdk))

横切关注点

技能指令返回

python

skill_param = (
    AgentParam.create()
    .set_integer("step", 3)
    .set_double("speed", 0.5)
    .set_string("target", "kitchen")
)

response.on_skill(event_id, item_id, "movement", "walk_forward", skill_param)

错误处理

python

try:
    # 业务处理...
    pass
except AsrServiceError as e:
    response.on_error(event_id, 5001, f"ASR 服务不可用: {e}")
except LlmServiceError as e:
    response.on_error(event_id, 5002, "LLM 服务超时")
except Exception as e:
    response.on_error(event_id, 5999, "未知错误")

选择哪个示例？

你的输入 → 输出	推荐示例
音频 → 文本	`audio2_llm_example.py`
音频 → 语音	`audio2_tts_example.py`
文本 → 文本	`asr2_llm_example.py`
文本 → 语音	`asr2_tts_example.py`
文本+视频 → 文本	`asr_video2_vlm_example.py`
文本+视频 → 语音	`asr_video2_tts_example.py`
音频+视频 → 文本	`audio_video2_vlm_example.py`
音频+视频 → 语音	`audio_video2_tts_example.py`

测试约定

Python SDK 内置 mock 测试支持。在你的工程里编写单测时，推荐：

python

from linksoul_agentsdk import AgentSdk


class FakeLinkskyClient:
    """记录所有 send_text 调用，便于断言出站消息。"""

    def __init__(self, agent_sdk):
        self.agent_sdk = agent_sdk
        self.sent = []
        self._closed = False

    def send_text(self, agent_id, event_id, text, log_text):
        self.sent.append({"agent_id": agent_id, "event_id": event_id, "text": text})

    def is_closed(self):
        return self._closed

    def connect(self):
        pass

    def stop(self):
        self._closed = True


def test_my_callback():
    sdk = AgentSdk.create("wss://test/v1", "app", "test-key", "0123456789abcdef" * 2)
    sdk.set_linksky_client_factory(FakeLinkskyClient)
    sdk.initialize()
    # ... 注入构造好的消息到 MessageRouter，断言 sdk.linksky_client.sent ...

完整可运行参考：agentsdk_for_python/tests/。