오디오 및 동영상 스트림 전송

이 문서에서는 Gemini 모델과의 실시간 양방향 통신을 위해 오디오 및 동영상 스트림을 Live API로 전송하는 방법을 설명합니다. 동적인 대화형 애플리케이션을 빌드하기 위해 오디오 및 동영상 데이터를 구성하고 전송하는 방법을 알아봅니다.

오디오 스트림 전송

실시간 오디오를 구현하려면 샘플링 레이트 사양을 엄격하게 준수하고 버퍼 관리를 신중하게 수행하여 지연 시간을 줄이고 자연스러운 중단 가능성을 보장해야 합니다.

Live API는 다음 오디오 형식을 지원합니다.

  • 입력 오디오: 16kHz의 원시 16비트 PCM 오디오, little-endian
  • 출력 오디오: 24kHz의 원시 16비트 PCM 오디오, little-endian 형식

다음 코드 샘플은 스트리밍 오디오 데이터를 전송하는 방법을 보여줍니다.

import asyncio
# Assumes session is an active Live API session
# and chunk_data contains bytes of raw 16-bit PCM audio at 16 kHz.
from google.genai import types
# Send audio input data in chunks
await session.send_realtime_input(
    audio=types.Blob(data=chunk_data, mime_type="audio/pcm;rate=16000")
)

클라이언트는 재생 버퍼를 유지해야 합니다. 서버는 server_content 메시지 내에서 오디오를 청크 단위로 스트리밍합니다. 클라이언트는 데이터 디코딩, 버퍼링, 재생을 담당합니다.

다음 코드 샘플은 스트리밍 오디오 데이터를 처리하는 방법을 보여줍니다.

import asyncio
# Assumes session is an active Live API session
# and audio_queue is an asyncio.Queue for buffering audio for playback.
import numpy as np

async for msg in session.receive():
    server_content = msg.server_content
    if server_content:
        # 1. Handle Interruption
        if server_content.interrupted:
            print("\n[Interrupted] Flushing buffer...")
            # Clear the Python queue
            while not audio_queue.empty():
                try: audio_queue.get_nowait()
                except asyncio.QueueEmpty: break
            # Send signal to worker to reset hardware buffers if needed
            await audio_queue.put(None)
            continue

        # 2. Process Audio chunks
        if server_content.model_turn:
            for part in server_content.model_turn.parts:
                if part.inline_data:
                    # Add PCM data to playback queue
                    await audio_queue.put(np.frombuffer(part.inline_data.data, dtype='int16'))

동영상 스트림 전송

동영상 스트리밍은 시각적 컨텍스트를 제공합니다. Live API는 개별 이미지 프레임 시퀀스를 예상하며 1FPS로 동영상 프레임 입력을 지원합니다. 최상의 결과를 얻으려면 1FPS에서 기본 768x768 해상도를 사용하세요.

다음 코드 샘플은 스트리밍 동영상 데이터를 전송하는 방법을 보여줍니다.

import asyncio
# Assumes session is an active Live API session
# and chunk_data contains bytes of a JPEG image.
from google.genai import types
# Send video input data in chunks
await session.send_realtime_input(
    media=types.Blob(data=chunk_data, mime_type="image/jpeg")
)

클라이언트 구현은 동영상 피드에서 프레임을 캡처하고 JPEG blob으로 인코딩한 후 realtime_input 메시지 구조를 사용하여 전송합니다.

import cv2
import asyncio
from google.genai import types

async def send_video_stream(session):
    # Open webcam
    cap = cv2.VideoCapture(0)

    while True:
        ret, frame = cap.read()
        if not ret: break

        # 1. Resize to optimal resolution (768x768 max)
        frame = cv2.resize(frame, (768, 768))

        # 2. Encode as JPEG
        _, buffer = cv2.imencode('.jpg', frame,)

        # 3. Send as realtime input
        await session.send_realtime_input(
            media=types.Blob(data=buffer.tobytes(), mime_type="image/jpeg")
        )

        # 4. Wait 1 second (1 FPS)
        await asyncio.sleep(1.0)

    cap.release()

미디어 해상도 구성

세션 구성에서 media_resolution 필드를 설정하여 입력 미디어의 해상도를 지정할 수 있습니다. 해상도가 낮을수록 토큰 사용량과 지연 시간이 줄어드는 반면, 해상도가 높을수록 세부사항 인식이 향상됩니다. 지원되는 값은 low, medium, high입니다.

config = {
    "response_modalities": ["audio"],
    "media_resolution": "low",
}

다음 단계