GenAI Processors Library

The genai_processors.core module provides a variety of pre-built processors designed for different stages of a GenAI pipeline, including model interaction, media I/O, and stream manipulation.

Key categories of processors include:

• Model Interaction: GenaiModel for turn-based interactions and LiveProcessor for real-time streaming via the Google GenAI Live API.
• Media Input/Output: PyAudioIn/PyAudioOut for microphone/speaker I/O, VideoIn for camera/screen capture, and RateLimitAudio for managing audio playback speed.
• Stream Manipulation: Preamble and Suffix for adding fixed content, Timestamp for adding temporal metadata, and MatchProcessor for regex extraction.
• Event & Data Processing: EventDetection for detecting events in image streams and Docs/Sheets/Slides for downloading Google Workspace files for model grounding.

GenAI Processors is designed for building modular and asynchronous AI pipelines with the following capabilities:

• Modular Design: Tasks can be broken into Processor and PartProcessor units. These can be chained using the + operator or parallelized using the // operator.
• Integrated GenAI Support: Includes built-in processors like GenaiModel (for turn-based calls) and LiveProcessor (for real-time streaming).
• Rich Content Handling: Uses ProcessorPart, a wrapper around genai.types.Part that includes metadata like MIME type, role, and custom attributes. It supports text, images, audio, and custom JSON.
• Asynchronous Orchestration: Built on asyncio to handle concurrent network I/O and compute-heavy tasks.
• Stream Management: Provides utilities for splitting, concatenating, and merging asynchronous streams of ProcessorParts.

A Processor is the fundamental unit of work in the GenAI Processors library. It acts as a transformation pipeline that takes an input stream of ProcessorPart data (representing modalities like text, images, or files) and produces an output stream of parts.

Key characteristics:

• Input/Output: It consumes an AsyncIterable of parts and yields an AsyncIterable of parts.
• Normalization: The library automatically handles data type unification. You can yield raw strings or parts, and the library will normalize them into ProcessorPart objects.
• Stream Handling: When receiving data, you work with a ProcessorStream (which inherits from ContentStream). This class provides convenient accessors like .text() for quick extraction.

The LiveCommentator manages the gap between utterances by predicting the latency of the next generation request. It measures latency as the time from when a request is sent to the model until the first audio part is received (referred to in code as "TTFT" or "time to first token").

To minimize gaps while still allowing for interruptions, the commentator uses a history of recent latency values to predict the next TTFT. It then schedules the next comment to trigger just before the current one finishes.

Key methods for latency management:

• predict_next_ttft(): Calculates the predicted TTFT using a mean-std approach based on recent history.
• tentative_trigger_time(): Calculates the specific timestamp at which the next comment should be triggered, factoring in the predicted TTFT and the duration of the current audio.

Processors can be chained together using the + operator. This composition allows you to build complex pipelines, such as adding a Preamble (system prompt) to a model. Chaining with + is recommended as it leverages framework optimizations to minimize Time To First Token (TTFT) and maximize concurrency.

from genai_processors import content_api
from genai_processors.core import preamble

# Create a processor that adds a prefix to any input.
system_prompt = preamble.Preamble(
    "You are a pirate styling assistant. Answer everything in pirate speak."
)

# Chain it with the model
pirate_bot = system_prompt + model

# Now the model will always act like a pirate
response = pirate_bot(
    content_api.ContentStream(content=["What color matches blue?"])
    )
print(await response.text())

The Research Agent is composed of several specialized modules:

• ResearchAgent (agent.py): The orchestrator that chains all sub-processors into a single pipeline.
• Topic (interfaces.py): A dataclass embedded within ProcessorParts that carries the topic string, its relationship to the query, and the gathered research_text.
• Config (interfaces.py): A dataclass for managing agent settings like model names and tool configurations.
• TopicGenerator (processors/topic_generator.py): A processor that identifies research sub-topics.
• TopicResearcher (processors/topic_researcher.py): A processor that gathers information on topics using tools.
• prompts.py: Contains the string preambles (e.g., TOPIC_GENERATION_PREAMBLE, TOPIC_RESEARCH_PREAMBLE, SYNTHESIS_PREAMBLE) used to instruct models at each stage.

Tracing is built into the Processor and PartProcessor base classes using Python's asyncio and contextvars.

Trace Context

Tracing is activated when a processor is executed within an active trace.Trace asynchronous context.

• If a trace is active, the processor creates a sub-trace and attaches it to the parent.
• If no trace is active, the processor runs normally without tracing.

Trace Events

Each input part consumed and output part produced is logged as a TraceEvent with a timestamp. Nested processor calls (e.g., inside a chain or parallel operation) are captured as sub-traces within the event log.

Exclusions

Certain modules (like genai_processors.debug) are excluded by default to reduce noise. You can manage this list via trace.EXCLUDED_TRACE_MODULES.

A PartProcessor is a specialized class designed to handle a single ProcessorPart rather than a whole stream. This is ideal for tasks like image preprocessing or formatting where parts can be processed independently.

Benefits:

• Simplicity: You write logic for one part instead of managing a stream loop.
• Performance: The library automatically parallelizes PartProcessor logic across the incoming stream, avoiding head-of-line blocking.

Use the @processor.part_processor_function decorator to define them.

@processor.part_processor_function
async def shouter(
    part: content_api.ProcessorPart
    ) -> AsyncIterable[content_api.ProcessorPartTypes]:
    if content_api.is_text(part.mimetype):
        yield part.text.upper()
    else:
        yield part

An AI Studio Applet is a web bundle used for prototyping interactive AI agents with specialized UI or hardware access (microphone/camera). A functional Applet requires three core files:

1. index.html: The UI skeleton.
2. index.js or index.tsx: Client-side logic, including WebSocket handling, audio processing, and UI state management.
3. metadata.json: Defines the app's identity and required browser permissions (e.g., microphone, camera).

{
  "name": "Live Voice Assistant",
  "description": "Real-time voice agent using WebSockets",
  "requestFramePermissions": [
    "microphone",
    "camera"
  ]
}