Live_agent = input_processor + live_processor + play_output

pip install genai-processors

# For Google AI Studio
export GOOGLE_API_KEY="your-api-key"
# Or for Google Cloud
gcloud auth application-default login

import genai_processors
print(genai_processors.__version__)

# Clone for examples or contributions
git clone https://github.com/google-gemini/genai-processors.git
cd genai-processors
pip install -e

↓ ↓ ↓ ↓ ↓
ProcessorPart → ProcessorPart → ProcessorPart → ProcessorPart → ProcessorPart

live_agent/
── main.py
── config.py
└── requirements.txt

config.py
import os
from genai_processors.core import audio_io
# API configuration
GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")
if not GOOGLE_API_KEY:
raise ValueError("Please set GOOGLE_API_KEY environment variable")
# Audio configuration
AUDIO_CONFIG = audio_io.AudioConfig(
sample_rate=16000,
channels=1,
chunk_size=1024,
format="int16"
)
# Video configuration
VIDEO_CONFIG = {
"width": 640,
"height": 480,
"fps": 30
}

main.py
import asyncio
from genai_processors.core import (
 audio_io,
 live_model,
 video,
 streams
)
from config import AUDIO_CONFIG, VIDEO_CONFIG, GOOGLE_API_KEY
class LiveAgent:
 def __init__(self):
 self.setup_processors()
 def setup_processors(self):
 """Initialize all processors for the live agent"""
 # Input processor: combines camera and microphone
 self.input_processor = (
 video.VideoIn(
 device_id=0,
 width=VIDEO_CONFIG["width"],
 height=VIDEO_CONFIG["height"],
 fps=VIDEO_CONFIG["fps"]
 ) + 
 audio_io.PyAudioIn(
 cnotallow=AUDIO_CONFIG,
 device_index=None # Use default microphone
 )
 )
 # Gemini Live API processor
 self.live_processor = live_model.LiveProcessor(
 api_key=GOOGLE_API_KEY,
 model_name="gemini-2.0-flash-exp",
 system_instructinotallow="You are a helpful AI assistant. Respond naturally to user interactions."
 )
 # Output processor: handles audio playback with interruption support
 self.output_processor = audio_io.PyAudioOut(
 cnotallow=AUDIO_CONFIG,
 device_index=None, # Use default speaker
 enable_interruptinotallow=True
 )
 # Complete agent pipeline
 self.agent = (
 self.input_processor + 
 self.live_processor + 
 self.output_processor
 )
 async def run(self):
 """Start the live agent"""
 print("?? Live Agent starting...")
 print("?? Camera and microphone active")
 print("?? Audio output ready")
 print("?? Start speaking to interact!")
 print("Press Ctrl+C to stop")
 try:
 async for part in self.agent(streams.endless_stream()):
 # Process different types of output
 if part.part_type == "text":
 print(f"?? AI: {part.text}")
 elif part.part_type == "audio":
 print(f"?? Audio chunk: {len(part.audio_data)} bytes")
 elif part.part_type == "video":
 print(f"?? Video frame: {part.width}x{part.height}")
 elif part.part_type == "metadata":
 print(f"?? Metadata: {part.metadata}")
 except KeyboardInterrupt:
 print("\n?? Live Agent stopping...")
 except Exception as e:
 print(f"? Error: {e}")
# Advanced agent with custom processing
class CustomLiveAgent(LiveAgent):
 def __init__(self):
 super().__init__()
 self.conversation_history = []
 self.user_emotions = []
 def setup_processors(self):
 """Enhanced setup with custom processors"""
 from genai_processors.core import (
 speech_to_text,
 text_to_speech,
 genai_model,
 realtime
 )
 # Custom input processing with STT
 self.input_processor = (
 audio_io.PyAudioIn(cnotallow=AUDIO_CONFIG) + 
 speech_to_text.SpeechToText(
 language="en-US",
 interim_results=True
 )
 )
 # Custom model with conversation memory
 self.genai_processor = genai_model.GenaiModel(
 api_key=GOOGLE_API_KEY,
 model_name="gemini-pro",
 system_instructinotallow="""You are an empathetic AI assistant. 
 Remember our conversation history and respond with emotional intelligence.
 If the user seems upset, be supportive. If they're excited, share their enthusiasm."""
 )
 # Custom TTS with emotion
 self.tts_processor = text_to_speech.TextToSpeech(
 voice_name="en-US-Neural2-J",
 speaking_rate=1.0,
 pitch=0.0
 )
 # Audio rate limiting for smooth playback
 self.rate_limiter = audio_io.RateLimitAudio(
 sample_rate=AUDIO_CONFIG.sample_rate
 )
 # Complete custom pipeline
 self.agent = (
 self.input_processor +
 realtime.LiveModelProcessor(
 turn_processor=self.genai_processor + self.tts_processor + self.rate_limiter
 ) +
 audio_io.PyAudioOut(cnotallow=AUDIO_CONFIG)
 )
if __name__ == "__main__":
 # Choose your agent type
 agent_type = input("Choose agent type (1: Simple, 2: Custom): ")
 if agent_type == "2":
 agent = CustomLiveAgent()
 else:
 agent = LiveAgent()
 # Run the agent
 asyncio.run(agent.run())

class EmotionAwareLiveAgent(LiveAgent):
 def __init__(self):
 super().__init__()
 self.emotion_history = []
 async def process_with_emotion(self, text_input):
 """Process input with emotion awareness"""
 # Simple emotion detection (in practice, use more sophisticated methods)
 emotions = {
 "happy": ["great", "awesome", "fantastic", "wonderful"],
 "sad": ["sad", "disappointed", "down", "upset"],
 "excited": ["amazing", "incredible", "wow", "fantastic"],
 "confused": ["confused", "don't understand", "what", "how"]
 }
 detected_emotion = "neutral"
 for emotion, keywords in emotions.items():
 if any(keyword in text_input.lower() for keyword in keywords):
 detected_emotion = emotion
 break
 self.emotion_history.append(detected_emotion)
 return detected_emotion
 def get_emotional_response_style(self, emotion):
 """Customize response based on detected emotion"""
 styles = {
 "happy": "Respond with enthusiasm and positivity!",
 "sad": "Respond with empathy and support. Offer help.",
 "excited": "Match their excitement! Use energetic language.",
 "confused": "Be patient and explanatory. Break down complex ideas.",
 "neutral": "Respond naturally and helpfully."
 }
 return styles.get(emotion, styles["neutral"])

requirements.txt
genai-processors>=0.1.0
google-generativeai>=0.3.0
pyaudio>=0.2.11
opencv-python>=4.5.0
asyncio>=3.4.3

pip install -r requirements.txt
python main.py