로보틱스 스트리밍
로보틱스 스트리밍 (Robotics with streaming)
참고: 로보틱스 스트리밍은
gemini-robotics-er-2-streaming-preview가 필요해요. ER 1.6이나 표준gemini-robotics-er-2-preview모델 엔드포인트에서는 사용할 수 없어요.
gemini-robotics-er-2-streaming-preview 모델 엔드포인트는 Live API와 통합되는 전용 스트리밍 엔드포인트를 노출해요. 이를 통해 애플리케이션과 로봇 사이의 실시간 양방향 상호작용이 가능해지죠. 그래서 빠른 피드백 루프와 환경에 대한 반응형 응답이 필요한 에이전트에 잘 맞아요.
Google AI Studio에서 사용해 보기 · GitHub에서 예시 앱 복제하기
출처: 원문
본문
사용 사례 (Use cases)
- 다중 로봇 협력 (Multi-robot coordination): 공유 세션을 통해 작업 상태를 통신하고 하위 작업을 위임하는 여러 로봇.
- 연속 모니터링 (Continuous monitoring): 장면을 관찰하고 컨테이너가 특정 채움 수준에 도달하는 것 같은 특정 이벤트가 발생하면 동작을 트리거하는 로봇.
- 창고 및 물류 (Warehouse and logistics): 아이템을 시각적으로 확인하고, 포장 진행 상황을 추적하며, 오류에서 복구하는 픽 앤 팩(pick-and-pack) 에이전트.
기술 사양 (Technical specifications)
다음 표는 Live API의 기술 사양을 정리한 거예요:
| 카테고리 | 세부 사항 |
|---|---|
| 입력 모달리티 | 오디오(원시 16-bit PCM 오디오, 16kHz, 리틀엔디언), 이미지(JPEG ≤ 1FPS), 텍스트 |
| 출력 모달리티 | 텍스트 |
| 프로토콜 | 상태 저장형 WebSocket 연결(WSS) |
에이전트 설정 구축하기 (Build an agentic setup)
Live API 기반의 모든 로보틱스 에이전트는 세 단계를 따릅니다:
- 로봇 기능을 도구로 선언합니다. 로봇이 수행할 수 있는 각 동작(이동, 그리핑, 말하기 등)은 이름·설명·매개변수 스키마를 가진 함수 선언이 됩니다. 물리적 동작은
"behavior": "BLOCKING"을 사용해야 모델이 로봇이 끝내기를 기다린 뒤 다음 단계를 선택해요. - 멀티모달 입력을 영구 세션으로 스트리밍합니다.
live.connect세션을 열고 작업이 지속되는 동안 열어 둡니다. 로봇 센서에서 도착하는 비디오 프레임, 오디오, 텍스트를 보내요. - 수신 루프에서 도구 호출을 처리합니다. 모델이 동작을 선택할 때마다
tool_call메시지를 보내요. 수신 루프는 로봇 SDK에 대해 함수를 실행하고tool_response를 돌려보냅니다. 세션은 열린 상태로 유지되고, 모델은 결과에 기반해 다음 동작을 선택해요.
참고: 로보틱스에 Live API를 사용할 때는 차단(blocking) 함수 호출만 지원돼요. 자세한 내용은 Live API 도구 가이드를 참고하세요.
다음 섹션들은 이 단계들을 세 가지 일반적인 패턴에 어떻게 적용하는지 보여줘요: 기본 에이전트 루프, 하트비트(heartbeat)를 통한 선제적 장면 모니터링, 그리고 도구로서 TTS를 통한 음성 라우팅.
함수 호출로 로봇 오케스트레이션하기 (Orchestrate a robot through function calling)
다음 예시는 세 단계를 모두 하나의 Python 스크립트로 연결한 거예요.
1단계 — 도구 정의: 로봇 기능을 함수 선언으로 선언합니다. navigate 함수는 "behavior": "BLOCKING"을 사용해 모델이 웨이포인트에 도달할 때까지 기다린 뒤 다른 도구를 호출하게 해요. 같은 목록에 함수 선언을 더 추가해 추가 로봇 기능을 노출할 수 있어요.
2단계 — 입력 헬퍼: 세션에 다양한 모달리티 입력을 스트리밍하는 세 함수를 보여줘요. send_text는 명령용, send_image는 선택적 텍스트 프롬프트가 있는 카메라 프레임용, send_audio는 마이크의 원시 PCM 오디오용이에요.
3단계 — 수신 루프: 동시에 실행되며 두 종류의 메시지를 처리해요. server_content 메시지(모델의 텍스트 출력)와 tool_call 메시지(모델이 로봇 동작을 요청)예요. 도구 호출이 도착하면 루프는 execute_tool(실제 로봇 SDK로 교체할 스텁)을 호출하고 tool_response를 돌려보내 모델이 다음 동작을 선택하게 해요.
import asyncio
from google import genai
from google.genai import types
MODEL = "gemini-robotics-er-2-streaming-preview"
# ── Tool definitions ─────────────────────────────────────────────
tools = [
{
"function_declarations": [
{
"name": "navigate",
"description": "Navigate the robot to a named waypoint.",
"behavior": "BLOCKING",
"parameters": {
"type": "OBJECT",
"properties": {
"name": {"type": "STRING"}
},
"required": ["name"],
},
},
# Add more function definitions here
]
}
]
# ── Stub tool executor (replace with real robot SDK calls) ──────
def execute_tool(name: str, args: dict) -> dict:
print(f"[Tool] {name} ({args})")
return {"status": "success"}
# ── Input helpers ────────────────────────────────────────────────
def send_text(session, text: str):
"""Send a text turn."""
return session.send_client_content(
turns=types.Content(role="user", parts=[types.Part(text=text)]),
turn_complete=True,
)
def send_image(session, image_bytes: bytes, prompt: str = ""):
"""Send a JPEG image with an optional text prompt."""
parts = [types.Part(inline_data=types.Blob(data=image_bytes, mime_type="image/jpeg"))]
if prompt:
parts.append(types.Part(text=prompt))
return session.send_client_content(
turns=types.Content(role="user", parts=parts),
turn_complete=True,
)
def send_audio(session, audio_chunk: bytes):
"""Stream a chunk of raw PCM audio (16-bit, 16 kHz, mono)."""
return session.send_realtime_input(
media=types.Blob(data=audio_chunk, mime_type="audio/pcm;rate=16000")
)
# ── Receive loop ─────────────────────────────────────────────────
async def receive_loop(session):
"""Print model text and handle tool calls until the session ends."""
async for message in session.receive():
if message.server_content:
sc = message.server_content
if sc.model_turn and sc.model_turn.parts:
# handle model text output ...
수신 루프는 각 도구 응답 후에도 활성 상태를 유지해요. 모델은 전체 동작 시퀀스를 미리 인코딩하지 않아도 장기 지평(long-horizon) 계획을 구성하고 수정해요.
선제적 시공간 추론 (Proactive spatial-temporal reasoning)
Live API가 비디오를 스트리밍하지만, 비디오 프레임만으로는 새로운 추론 턴이 트리거되지 않아요. 비디오 프레임에는 모델 응답을 트리거할 텍스트나 오디오 프롬프트가 함께 동반되어야 해요. 자세한 내용은 Live API 기능을 참고하세요.
선제적 추론을 활성화하려면 **하트비트(heartbeat)**를 구현하세요. 주기적으로 최신 카메라 프레임을 보내고, 이어서 짧은 텍스트 프롬프트를 보내 모델이 장면을 조사하고 명시적 결정을 내리게 해요. 비디오 입력은 초당 1프레임으로 레이트 리밋돼요.
하트비트 구현하기 (Implement the heartbeat)
하트비트 코루틴은 같은 세션에서 별도의 asyncio 태스크로 실행돼요. 각 턴이 완료될 때까지 기다리면서(er_turn_done) 약 1Hz 주기를 목표로 해요(비디오 입력 레이트 리밋에 맞춤). 진행 중인 추론을 방해하지 않기 위해서죠:
async def heartbeat(session, camera, er_turn_done: asyncio.Event):
TARGET_INTERVAL_SEC = 1.0
while True:
start_time = asyncio.get_running_loop().time()
frame = await camera.latest_jpeg()
await session.send_realtime_input(
video=types.Blob(data=frame, mime_type="image/jpeg")
)
await session.send_realtime_input(
text=(
"[HEARTBEAT] If no task is active, call 'ack' and wait for user"
" input. If a task is active: observe the scene. If the current"
" step is progressing correctly, call 'ack'. If the current step"
" is complete, call 'run_instruction' with the next step. If the"
" overall goal is achieved, call 'reset' and inform the user."
)
)
# Wait for the model to finish responding before sending the next heartbeat
await er_turn_done.wait()
er_turn_done.clear()
# Sleep only the remaining time to maintain ~1 Hz cadence
elapsed = asyncio.get_running_loop().time() - start_time
remaining = TARGET_INTERVAL_SEC - elapsed
if remaining > 0:
await asyncio.sleep(remaining)
주의: 모델이 생성하는 동안 새 프롬프트를 보내면 중단(barge-in)으로 작동해 진행 중인 추론과 도구 호출을 취소해요. 이전 턴이 완료될 때까지 기다리지 않고 블라인드 타이머로 하트비트를 발사하면 모델을 취소 루프에 빠뜨려 응답하지 않는 것처럼 보이게 할 수 있어요. 자세한 내용은 중단에 관한 Live API 가이드를 참고하세요.
수신 루프 업데이트하기 (Update the receive loop)
모델이 턴을 완료했음을 알리려면 receive_loop를 업데이트해 er_turn_done을 설정하세요:
# In receive_loop: signal when the model finishes its turn
if sc.turn_complete:
er_turn_done.set()
외부 TTS를 통한 오디오 출력 (Audio output through external TTS)
Gemini Robotics ER 2는 텍스트를 반환해요. 애플리케이션은 완료된 응답을 주입된 콜백을 통해 별도의 TTS 제공자(예: Gemini TTS)로 라우팅해요. 이렇게 하면 음성 지연 시간, 음성 선택, 중단 동작을 제어할 수 있고, 에이전트 로직을 바꾸지 않고 TTS 백엔드를 교체할 수 있어요.
TTS를 도구로 선언해서 모델이 "무언가 말하기"를 "팔 움직이기"와 동일하게 취급하게 할 수도 있어요. 첫 번째 섹션의 tools 목록에 다음 함수 선언을 추가하세요:
TOOLS = [
{
"name": "send_message",
"description": (
"Speak a message aloud via TTS, then deliver it to the"
" specified target. Use target='user' to speak directly"
" to the user, or a peer agent name (e.g., 'duo') to"
" communicate with another robot."
),
"parameters": {
"type": "object",
"properties": {
"target": {
"type": "string",
"description": "Recipient: 'user' or a peer agent name.",
},
"message": {
"type": "string",
"description": "The message to speak and deliver.",
},
},
"required": ["target", "message"],
},
},
]
TTS를 함수 선언으로 감싸면 모델이 다른 로봇 동작과 같은 도구 호출 경로를 통해 음성을 처리해요. 애플리케이션은 주입된 콜백으로 호출을 이행해요.
GitHub 예시 (Examples on GitHub)
Spot 로봇 스낵-패치 데모와 Tinybot 팬-틸트 hello world를 포함한 전체 작동 예시는 Robotics Live API 예시를 참고하세요.
다음 단계 (What's next)
- Video understanding — 순간 찾기와 진행 분류.
- Task orchestration — 스트리밍 없는 장기 지평 작업.
- Live API overview — 전체 Live API 문서.
더 알아보기 (Learn more)
- Live API — 실시간 상호작용 전체 문서.
- Robotics Live API 예시 — 작동 예시 코드.