GenAI Processors v2.0: Google's Unified Framework for Modular, Streaming AI Pipelines

Executive Summary

Picture this: You’re building an AI pipeline that needs to handle real‑time streaming of text, images, and audio, orchestrate multiple LLM calls, and process results concurrently — all while maintaining a clean, composable codebase. Until now, that meant stitching together half a dozen libraries and writing hundreds of lines of asyncio boilerplate.

Enter GenAI Processors, Google’s open‑source Python library that provides a unified, streaming‑first abstraction for building modular AI pipelines. In the 11 months since its release, it has attracted 2,108 GitHub stars and 212 forks, becoming the go‑to foundation for teams building production‑grade generative AI systems on Google’s Gemini platform.

Here’s what the data tells us:

• 2,108 stars (GitHub API, retrieved 2026‑04‑20) – steady adoption since launch in April 2025
• 212 forks – active community extending the library
• Latest release: v2.0.0 (2026‑03‑10) – major API stabilization
• Current version: 2.0.3 (per __init__.py) – incremental improvements post‑release
• Dependencies: google‑genai SDK, asyncio, PIL – lightweight, focused stack
• License: Apache 2.0 – production‑friendly, corporate‑safe

This isn’t just another wrapper. GenAI Processors is a paradigm shift in how we think about AI pipelines: every component is a Processor, every piece of content is a ProcessorPart, and streaming is the default — not an afterthought.

If you’re building with Gemini (or any LLM) and want to move beyond simple model.generate() calls, you need to understand how GenAI Processors works. It’s the missing middleware that turns a collection of API calls into a scalable, maintainable AI pipeline.

1. The Core Problem: Fragmented AI Pipelines

Before GenAI Processors, building a non‑trivial AI workflow meant:

• Different content types – text, images, audio, JSON – each with their own serialization
• Mixed synchronous/asynchronous – asyncio for streaming, blocking calls for turn‑based
• Ad‑hoc error handling – retry logic scattered across the codebase
• Manual streaming plumbing – concatenating chunks, managing buffers, handling backpressure

Google’s solution: three core pillars that unify the entire stack.

Pillar 1: Unified Content Model

The library introduces ProcessorPart – a wrapper around Gemini’s Part type that adds metadata, MIME‑type detection, role tracking, and substream labeling.

from genai_processors import ProcessorPart
 
# Text part with role and metadata
text_part = ProcessorPart("Hello, world!", role="user", metadata={"source": "cli"})
 
# Image part with automatic MIME‑type detection
image_part = ProcessorPart(open("diagram.png", "rb").read(), mimetype="image/png")
 
# Custom JSON part
json_part = ProcessorPart({"action": "summarize", "priority": "high"})

Every piece of content in your pipeline is a ProcessorPart. This consistency eliminates the “what format is this?” guessing game that plagues multi‑modal AI code.

Pillar 2: Processors – The Dual‑Interface Pattern

The Processor class is the library’s central abstraction. It follows a dual‑interface pattern that separates the PRODUCER (author) and CONSUMER (caller) concerns.

PRODUCER interface (for library/component authors):

from typing import AsyncIterable
from genai_processors import Processor, ProcessorStream
 
class EchoProcessor(Processor):
    async def call(self, content: ProcessorStream) -> AsyncIterable[ProcessorPartTypes]:
        async for part in content:
            yield part  # Stream‑in, stream‑out

CONSUMER interface (for pipeline builders):

# Input automatically converted to ProcessorParts
input_content = ["Hello ", ProcessorPart("World")]
 
# 1. Get all output as a single object
result = await echo_processor(input_content).gather()
 
# 2. Get just the text (for text‑only processors)
text = await echo_processor(input_content).text()
 
# 3. Stream results as they arrive
async for part in echo_processor(input_content):
    print(part.text, end="")

This duality means authors write streaming‑aware code (the call method), while callers use simple, intuitive methods (gather(), text(), iteration). The library handles all the asyncio complexity behind the scenes.

There are two base classes: Processor (stream‑to‑stream) and PartProcessor (part‑to‑part). PartProcessor is optimized for per‑part transformations and supports the // parallel operator, while Processor is more general and can manage entire streams.

Pillar 3: Built‑in Streaming

Streaming isn’t an add‑on — it’s the default. Every Processor consumes and produces AsyncIterable[ProcessorPart]. This means:

• Zero‑copy streaming between processors
• Automatic backpressure via asyncio queues
• Parallel processing with // operator
• Chaining with + operator

2. The Architecture: How Processors Compose

GenAI Processors provides four composition operators that let you build complex pipelines from simple parts.

Chaining (+) – Sequential Execution

pipeline = text_cleaner + sentiment_analyzer + formatter
result = await pipeline("Some text").gather()

Each processor receives the stream from the previous one. The entire chain streams end‑to‑end: text_cleaner yields cleaned chunks, which sentiment_analyzer processes as they arrive, and so on.

Parallel (//) – Concurrent Execution

parallel = translator // summarizer // keyword_extractor
results = await parallel("Long document").gather()

All three processors receive the same input stream and execute concurrently. Their outputs are interleaved as they complete — perfect for CPU‑bound operations or multiple LLM calls that don’t depend on each other.

Parallel‑Concat (parallel_concat) – Ordered Merge

from genai_processors import parallel_concat
 
pipeline = parallel_concat([translator, summarizer, keyword_extractor])
result = await pipeline("Document").gather()

Like parallel execution, but outputs are concatenated in the order of the processor list (not completion order). Useful when you need deterministic output ordering.

Filtering (create_filter) – Conditional Processing

from genai_processors import create_filter
 
image_filter = create_filter(lambda part: part.mimetype.startswith("image/"))
image_only_stream = image_filter(mixed_stream)

Filters are processors too — they just pass through matching parts and drop others.

3. The Gemini Integration: GenaiModel Processor

The library includes a ready‑to‑use GenaiModel processor that wraps Gemini API calls, turning them into first‑class pipeline components.

from genai_processors.core.genai_model import GenaiModel
from google.genai import Client
 
client = Client(api_key="...")
model = GenaiModel(client, model="gemini-2.0-flash")
 
# Use it like any other processor
response = await model("Explain quantum computing").gather()

Because GenaiModel is a Processor, you can:

• Chain it with pre‑/post‑processing steps
• Run it in parallel with other models or tools
• Stream results token‑by‑token
• Handle errors with processor‑level retry logic

The library also provides LiveProcessor for real‑time streaming interactions where the model can respond while the user is still typing — a feature that’s notoriously difficult to implement correctly.

4. Real‑World Example: Critic‑Reviser Loop

Let’s examine a practical example from the repository: CriticReviser (in examples/smart_model.py). This processor implements a self‑improvement loop where a draft response is critiqued and refined multiple times.

class CriticReviser(Processor):
    def __init__(self, model: Processor, max_iterations: int = 5):
        self._model = model
        self._max_iterations = max_iterations
 
    async def call(self, content: ProcessorStream) -> AsyncIterable[ProcessorPartTypes]:
        input_content = await content.gather()
        current_response = await self._model(input_content).gather()
 
        for _ in range(self._max_iterations):
            critic_response = await self._model([
                input_content,
                '\n\nDraft response:\n\n',
                current_response,
                ('\n\nYou are a harsh critic. Review the draft response...')
            ]).gather()
 
            critic_text = await critic_response.text(strict=False)
            if critic_text.strip().upper() == 'OK':
                break
 
            # Revise based on criticism
            current_response = await self._model([
                input_content,
                '\n\nCriticism:\n\n',
                critic_text,
                '\n\nImprove the draft response based on the criticism.'
            ]).gather()
 
        yield current_response

What’s notable here:

1. Processor composition – CriticReviser wraps another Processor (the base model)
2. Stream‑aware – Uses gather() when it needs full content, but still implements call() for streaming compatibility
3. Natural error handling – The loop can break early when the critic says “OK”

This pattern — wrapping processors with additional logic — is how you build agentic behaviors without rewriting your entire pipeline.

5. Performance & Production Considerations

Memory Efficiency

Because processors stream by default, they can handle multi‑gigabyte inputs without loading everything into memory. The ProcessorStream object provides methods like gather() for when you need the full content, but you can also process chunk‑by‑chunk.

Error Handling & Retries

The library integrates with Gemini’s retry configuration:

from google.genai import types as genai_types
 
retry_config = genai_types.HttpOptions(
    retry_options=genai_types.HttpRetryOptions(attempts=1000)
)
 
model = GenaiModel(client, model="gemini-2.0-flash", http_options=retry_config)

Processor‑level errors propagate cleanly through the pipeline, and you can wrap processors in custom error‑handling logic.

Testing & Debugging

The trace module (genai_processors.dev.trace) provides instrumentation for debugging complex pipelines. You can see exactly how parts flow through each processor, identify bottlenecks, and log intermediate results.

6. When to Use GenAI Processors (and When Not To)

Use GenAI Processors when:

• Building multi‑step AI pipelines – Summarization → translation → sentiment analysis chains
• Processing real‑time streams – Live transcription, video analysis, chat applications
• Orchestrating multiple LLM calls – Agent‑of‑agents, self‑critique loops, parallel tool execution
• Working with mixed media – Documents with text, images, and structured data
• You want clean, testable AI code – Processors are easy to unit test in isolation

Avoid GenAI Processors when:

• Simple one‑off API calls – If you just need model.generate(), use the SDK directly
• Synchronous‑only codebases – The library is async‑first; retrofitting sync code requires asyncio.run()
• Non‑Gemini models – While you could wrap other APIs, the integration is optimized for Gemini

Actionable Takeaways

If you’re ready to upgrade your AI pipeline architecture, here’s your playbook:

1. Install – pip install genai‑processors
2. Start with the examples – Run examples/smart_model.py to see the critic‑reviser loop in action
3. Wrap your first model – Create a GenaiModel processor and experiment with gather() vs. streaming
4. Build a simple chain – Connect a pre‑processor, model, and post‑processor with +
5. Add parallelism – Use // to run multiple analyses concurrently
6. Instrument – Add tracing to see how your pipeline performs

For library authors: Study the dual‑interface pattern in processor.py. It’s a masterclass in abstracting async complexity while keeping the caller’s API simple.

For production teams: The Apache 2.0 license means you can use this in commercial products. Combine it with Google’s Gemini for a fully supported, enterprise‑ready AI stack.

Conclusion

GenAI Processors represents Google’s answer to a critical problem in modern AI development: how to build maintainable, scalable pipelines as we move beyond simple prompt‑response patterns.

With 2,108 stars and 212 forks, it’s clear the community recognizes this need. The library’s elegant abstractions — Processor, ProcessorPart, streaming‑by‑default, and compositional operators — provide a foundation that will only grow more valuable as AI systems become more complex.

One question to leave you with: If your current AI pipeline is a tangle of asyncio calls, manual chunk concatenation, and inconsistent error handling, how much engineering time are you wasting that GenAI Processors could save?

Sources & Acknowledgments

This analysis synthesizes:

• GitHub Repository: google‑gemini/genai‑processors – source code, README, examples
• GitHub API: Repository statistics (stars, forks, issues) retrieved 2026‑04‑20
• Release Data: Latest release v2.0.0 published 2026‑03‑10
• Code Analysis: Direct examination of genai_processors/, examples/, notebooks/ directories
• Documentation: Official documentation – API reference and guides

Data Points Cited:

1. Stars: 2,108 – GitHub API /repos/google‑gemini/genai‑processors
2. Forks: 212 – same endpoint
3. Open issues: 9 – same endpoint
4. Created: 2025‑04‑29 – same endpoint
5. Latest release: v2.0.0 – GitHub API /repos/google‑gemini/genai‑processors/releases/latest
6. Current version: 2.0.3 – genai_processors/__init__.py line 17
7. License: Apache 2.0 – LICENSE file
8. Dependencies: pyproject.toml

All data accurate as of 2026‑04‑21. GenAI Processors is actively developed; check the repository for the latest.