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):
        ...

    def on_histories(self, agent_id, event_id, histories):
        # new in v1.5.0: de-duplicate by event_id, sort by timestamp
        ...

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

Function Call dispatch & Arbiter (new in v1.5.0)

v1.5.0 opens Function Call dispatch and Arbiter arbitration on the passive path: the gateway forwards its semantic-classification result (v3 classifier + skill router) as a function call (on_function_call); your app branches on FunctionCallInfo.policy to either delegate to the default cloud pipeline or override and take over. All 8 passive callbacks may override on_function_call.

python

from linksoul_agentsdk import ArbiterDecision, FunctionCallInfo
from linksoul_agentsdk.passive import Audio2TtsCallback, Audio2TtsResponse


class _Cb(Audio2TtsCallback):
    def on_function_call(
        self, agent_id, event_id, function_call: FunctionCallInfo,
        response: Audio2TtsResponse,
    ):
        if function_call is None:
            return
        if function_call.policy == "body":
            # on-device skill → delegate to the default cloud pipeline
            response.on_arbiter(event_id, ArbiterDecision.DELEGATE.value, None)
        elif function_call.policy == "cloud":
            # cloud skill → the app overrides and takes over
            response.on_arbiter(event_id, ArbiterDecision.OVERRIDE.value, None)

FunctionCallInfo (linksoul_agentsdk.FunctionCallInfo) fields: source / policy (e.g. body / cloud) / type / value / param (dict). Decision enum ArbiterDecision: DELEGATE / OVERRIDE. See Passive guide · Function Call exposure & Arbiter.

Conversation history (on_histories, new in v1.5.0)

The gateway forwards the session history to the SDK via on_histories, usable as XLM context:

python

def on_histories(self, agent_id, event_id, histories):
    if not histories:
        return
    for h in sorted(histories, key=lambda x: x.timestamp):
        print(f"history => ts {h.timestamp}, query {h.query}, answer {h.answer}, complete {h.complete}")

HistoryInfo (linksoul_agentsdk.HistoryInfo) fields:

Field	Type	Description
`timestamp`	`int`	timestamp, for sorting
`query`	`str`	user-side input
`answer`	`str`	agent-side reply
`complete`	`bool`	whether this turn finished
`event_id`	`str`	session event ID, for de-duplication

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/.