Passive Interaction Examples

← Home
Source: agentsdk_for_python/examples/

The 8 passive examples are grouped by input/output complexity. Every *_example.py is a runnable main; only the register_* call and on_request signature differ. Everything else (SDK creation, auth registration, base callbacks, initialisation) is identical.

Shared skeleton

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 = 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(MyAuth())
sdk.register_xxx(YourCallback(sdk))   # pick one
sdk.initialize()

Shared base callbacks (v1.4.0+)

python

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


class MyCallbackBase(PassiveCallback):
    # These are the base-class defaults — override only what you need.

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

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

    def on_face_info(self, agent_id, event_id, param: AgentParam):
        ...

    def on_video_frame(self, agent_id, event_id, flag, buf, is_key_frame, local_ts, param):
        ...

    def on_greet_signal(self, agent_id, event_id, param, response: GreetResponse):
        ...

    def on_state(self, agent_id, event_id, state_name, state_value):
        ...

1. Pure audio

Audio2LlmExample

Source: audio2_llm_example.py — voice in, text reply only (no 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, response: Audio2LlmResponse):
        # flag: 0 VAD start / 1 audio frame / 2 VAD end
        if flag in (0, 1):
            return
        response.on_asr_final(event_id, "what's the weather today")
        if not "what's the weather today":   # rejected sentinel
            return
        item_id = IdGenerator.generate_item_id()
        response.on_interrupt(event_id, "chat", None)
        response.on_llm_item_delta(event_id, item_id, "The weather is nice.")
        response.on_llm_item_done(event_id, item_id)
        response.on_llm_done(event_id)


sdk.register_audio2_llm(_Cb(sdk))

Audio2TtsExample

Source: audio2_tts_example.py — most common voice-in voice-out flow. The signature carries an extra item_id compared to Audio2Llm.

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
        response.on_asr_final(event_id, "what's the weather today")
        response.on_interrupt(event_id, "chat", None)
        response.on_llm_item_delta(event_id, item_id, "The weather is nice.")
        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_audio2_tts(_Cb(sdk))

2. Pure text

Asr2LlmExample

Source: asr2_llm_example.py — text in, LLM text out.

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 "forward" in text.lower():
            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, "Sure.")
        response.on_llm_item_done(event_id, item_id)
        response.on_llm_done(event_id)


sdk.register_asr2_llm(_Cb(sdk))

Asr2TtsExample

Source: 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, "Sure, here is the answer.")
        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))

3. Multimodal (text/audio + video)

AsrVideo2VlmExample

Source: 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 text, 4 H264, 5 image
        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, "I see a cup.")
        response.on_vlm_item_done(event_id, item_id)
        response.on_vlm_done(event_id)


sdk.register_asr_video2_vlm(_Cb(sdk))

AsrVideo2TtsExample

Source: 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, "I see a cup. Describing now.")
        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

Source: 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 start / 1 audio / 2 VAD end / 3 H264
        if flag in (0, 1, 3):
            return
        response.on_asr_final(event_id, "what's in front?")
        item_id = IdGenerator.generate_item_id()
        response.on_interrupt(event_id, "chat", None)
        response.on_vlm_item_delta(event_id, item_id, "There's a table in front.")
        response.on_vlm_item_done(event_id, item_id)
        response.on_vlm_done(event_id)


sdk.register_audio_video2_vlm(_Cb(sdk))

AudioVideo2TtsExample

Source: 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, "what's in front?")
        item_id = IdGenerator.generate_item_id()
        response.on_interrupt(event_id, "chat", None)
        response.on_vlm_item_delta(event_id, item_id, "There's a table in front.")
        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))

Cross-cutting concerns

Skill dispatch

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)

Error handling

python

try:
    ...
except AsrServiceError as e:
    response.on_error(event_id, 5001, f"ASR service unavailable: {e}")
except LlmServiceError as e:
    response.on_error(event_id, 5002, "LLM service timeout")
except Exception as e:
    response.on_error(event_id, 5999, "unknown error")

Which example to pick?

Your input → output	Recommended example
audio → text	`audio2_llm_example.py`
audio → speech	`audio2_tts_example.py`
text → text	`asr2_llm_example.py`
text → speech	`asr2_tts_example.py`
text + video → text	`asr_video2_vlm_example.py`
text + video → speech	`asr_video2_tts_example.py`
audio + video → text	`audio_video2_vlm_example.py`
audio + video → speech	`audio_video2_tts_example.py`

Test conventions

python

from linksoul_agentsdk import AgentSdk


class FakeLinkskyClient:
    """Records all send_text calls so you can assert outbound messages."""

    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()
    # ... feed messages into MessageRouter, then assert on sdk.linksky_client.sent ...

Full reference: agentsdk_for_python/tests/.