emos-workshop-rosconfr-2026/02_proactive_perception_agent.py at main · automatika-robotics/emos-workshop-rosconfr-2026 · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
"""
Workshop recipe 02 — Proactive perception agent (Events).

Recipe 01 was reactive: you ask, it answers. This one is autonomous: the agent
watches the webcam and greets a person the moment they appear — no prompting.

This is the heart of the EMOS programming model: an **Event** (an asynchronous
condition evaluated over live topic data) triggers an **Action** (here, running
the VLM). Detection runs as a small cheap local detector; the heavier reasoning
(VLM) and speech (TTS) run as separate services — right model in the right place.

    webcam ──▶ Vision (local detector) ──▶ detections
       Event: a person appears  ──▶ triggers ──▶ VLM (Ollama, local) ──▶ greeting
       greeting ──▶ TextToSpeech (RoboML container) ──▶ speaker (web UI)

SETUP
  1. Fill in endpoints.py (same endpoints as recipe 01).
  2. Start the webcam:  ros2 run usb_cam usb_cam_node_exe      (publishes /image_raw)
  3. Run, open http://localhost:5001, then step into frame — it greets you.

NOTE: the local detector (enable_local_classifier) runs on onnxruntime (already
in the EMOS pixi/docker install). Its model weights download from HuggingFace on
FIRST run, so the first launch pauses briefly — run it once to warm the cache.
"""

from agents.components import Vision, VLM, TextToSpeech
from agents.config import VisionConfig, MLLMConfig, TextToSpeechConfig
from agents.clients import OllamaClient, RoboMLHTTPClient
from agents.models import OllamaModel, TransformersTTS
from agents.ros import Topic, Launcher, FixedInput, Event

# from agents.clients import GenericHTTPClient
# from agents.models import GenericMLLM

from endpoints import (
    ROBOML_HOST,
    ROBOML_PORT,
    OLLAMA_HOST,
)
# from endpoints import VLM_BASE_URL, VLM_API_KEY, VLM_CHECKPOINT

# ---- Topics ----------------------------------------------------------------
image_raw  = Topic(name="/image_raw",  msg_type="Image")       # usb_cam feed
detections = Topic(name="/detections", msg_type="Detections")  # Vision output
greeting   = Topic(name="greeting",    msg_type="String")      # VLM output
audio_out  = Topic(name="audio_out",   msg_type="Audio")       # TTS output, played by web UI

# ---- Vision: lightweight object detection, runs LOCALLY ---------------------
# No model_client + enable_local_classifier=True => a small on-device detector,
# so every webcam frame is processed without touching a remote endpoint.
vision = Vision(
    inputs=[image_raw],
    outputs=[detections],
    trigger=image_raw,                                   # run on every frame
    config=VisionConfig(threshold=0.5, enable_local_classifier=True),
    component_name="vision",
)

# ---- The Event: "a person appears" -----------------------------------------
# A condition over the LIVE detections topic. on_change makes it fire on the
# transition into "person present" (not every frame), and keep_event_delay
# suppresses re-firing for a few seconds so it greets once, not continuously.
# (Conditions like this are the boolean building blocks of the Recipe paradigm —
#  you can extend them to richer triggers as the workshop goes on.)
person_appeared = Event(
    detections.msg.labels.contains_any(["person"]),
    on_change=True,
    keep_event_delay=5,
)

# ---- VLM: greet the person (local Ollama, multimodal — same as recipe 01) ----
vlm_client = OllamaClient(
    OllamaModel(name="vlm", checkpoint="qwen3.5:latest"),
    host=OLLAMA_HOST,
)

# Workshop endpoint (offline). The Modal vLLM client this replaced:
# vlm_client = GenericHTTPClient(
#     GenericMLLM(name="vlm", checkpoint=VLM_CHECKPOINT),
#     host=VLM_BASE_URL,
#     api_key=VLM_API_KEY,
# )
# A fixed instruction injected whenever the event fires (no user question here).
greet_prompt = FixedInput(
    name="prompt",
    msg_type="String",
    fixed=(
        "A person just appeared in front of you. Greet them warmly in one short, "
        "friendly sentence, mentioning something you can actually see about them."
    ),
)
greeter = VLM(
    inputs=[greet_prompt, image_raw],
    outputs=[greeting],
    model_client=vlm_client,
    trigger=person_appeared,                             # KEY: runs only on the event
    config=MLLMConfig(stream=False),
    component_name="greeter",
)
greeter.set_system_prompt(
    "You are a friendly robot. Keep your answers short. Do not use emojis or "
    "special symbols; they are read aloud by a text-to-speech system and "
    "cannot be pronounced."
)

# ---- TTS: speak the greeting (remote, shared RoboML pod) -------------------
tts_client = RoboMLHTTPClient(TransformersTTS(name="tts"), host=ROBOML_HOST, port=ROBOML_PORT)
text_to_speech = TextToSpeech(
    inputs=[greeting],
    outputs=[audio_out],
    trigger=greeting,
    model_client=tts_client,
    config=TextToSpeechConfig(play_on_device=False),     # play in the browser
    component_name="text_to_speech",
)
#
# ---- Launch with the web UI (outputs only — the agent drives itself) -------
launcher = Launcher()
launcher.enable_ui(outputs=[image_raw, detections, greeting, audio_out])
launcher.add_pkg(
    components=[vision, greeter, text_to_speech],
    multiprocessing=True,
    package_name="automatika_embodied_agents",
)
launcher.bringup()