はじめに
AI開発って難しそう？GoogleのGenAI Processorsなら、まるでレゴブロックのようにAI機能を組み合わせて、驚くほど簡単に面白いアプリが作れます！
この記事では、GenAI Processorsの概要から実際に動く4つのアプリケーションの作り方まで、コピペですぐに試せるコード付きでご紹介します。
この記事で学べること
- GenAI Processorsの基本的な使い方
- 段階的なAIアプリ開発（超入門→発展まで）
- 実際に動くコード例とその解説
- APIキーの安全な管理方法
GenAI Processorsとは？
GenAI Processorsは、AIアプリを簡単に作成可能とするライブラリで、Googleが2025年7月に発表したPythonライブラリです。その魅力は以下の3つの特徴にあります：
1. 🧩 パイプライン処理
複数のAI処理を+演算子で直感的に連結できます。
# 音声入力 → 文字起こし → 翻訳 → 音声出力
audio_input + speech_to_text + translator + audio_output
2. ⚡ 非同期・ストリーミング
リアルタイム処理が簡単に実装できます。
async for result in processor(input_stream):
print(result.text) # リアルタイムで結果が流れる
3. 🔧 拡張性
カスタムプロセッサーを作成して、独自の処理を組み込めます。
準備編：3ステップで完了！開発環境のセットアップ
Step 1: ライブラリのインストール
pip install genai-processors
たったこれだけ！Pythonがあれば即座に使い始められます。
Step 2: Gemini APIキーの取得
- Google AI Studioにアクセス
- 「Create API Key」をクリック
- 無料でAPIキーを取得（無料枠あり）
Step 3: APIキーの安全な設定
絶対にコードに直書きしないでください！
# .envファイルを作成
echo "GOOGLE_API_KEY=your_api_key_here" > .env
# python-dotenvをインストール
pip install python-dotenv
# 安全な読み込み方法
from dotenv import load_dotenv
import os
load_dotenv()
api_key = os.getenv('GOOGLE_API_KEY')
実践編：コピペで動く！AIアプリケーション作例4選
作例1：【超入門】絶対肯定！AIほめほめ執事 🎩
目的: どんなネガティブな言葉も、執事風の丁寧な口調でポジティブに変換
学習ポイント: GenaiModelの基本的な使い方、カスタムプロンプト設定
#!/usr/bin/env python3
import asyncio
import os
from dotenv import load_dotenv
from genai_processors import content_api, streams
from genai_processors.core.genai_model import GenaiModel
from google.genai import types
# 環境変数の読み込み
load_dotenv()
class PositiveButler:
def __init__(self):
"""執事AIの初期化"""
api_key = os.getenv('GOOGLE_API_KEY')
if not api_key:
raise ValueError("GOOGLE_API_KEYが設定されていません。.envファイルを確認してください。")
# カスタムプロンプトで執事キャラクターを設定
system_prompt = """
あなたは最高級ホテルの執事です。以下の特徴を持っています：
1. 常に丁寧で品格のある言葉遣い
2. どんなネガティブな内容も、前向きで建設的な視点に変換
3. 「〜でございます」「〜いたします」などの敬語を使用
4. 相手を励まし、希望を与える言葉選び
5. 具体的で実用的なアドバイスを含める
ユーザーの発言を受け取り、執事として適切に応答してください。
"""
# GenerateContentConfigでシステムプロンプトを設定
config = types.GenerateContentConfig(
system_instruction=system_prompt,
temperature=0.7
)
self.model = GenaiModel(
model_name="gemini-1.5-flash",
api_key=api_key,
generate_content_config=config
)
async def process_message(self, message: str) -> str:
"""メッセージを処理して執事風の返答を生成"""
try:
# 入力ストリームを作成
input_stream = streams.stream_content([message])
# モデルで処理
response_parts = []
async for part in self.model(input_stream):
if hasattr(part, 'text') and part.text:
response_parts.append(part.text)
return ''.join(response_parts)
except Exception as e:
return f"申し訳ございません。エラーが発生いたしました: {str(e)}"
async def main():
"""メイン実行関数"""
print("🎩 絶対肯定！AIほめほめ執事 🎩")
print("=" * 50)
print("どんなネガティブな言葉も、執事がポジティブに変換いたします！")
print("終了するには 'quit' と入力してください。")
print("=" * 50)
try:
butler = PositiveButler()
while True:
# ユーザー入力を受け取り
user_input = input("\n💬 あなた: ")
if user_input.lower() in ['quit', 'exit', '終了']:
print("\n🎩 執事: お疲れ様でございました。またお会いできる日を楽しみにしております。")
break
if not user_input.strip():
continue
# 処理中の表示
print("🤔 執事が考えています...")
# 執事の応答を生成
response = await butler.process_message(user_input)
print(f"\n🎩 執事: {response}")
except KeyboardInterrupt:
print("\n\n🎩 執事: 失礼いたします。お疲れ様でございました。")
except Exception as e:
print(f"\n❌ エラーが発生しました: {e}")
# 実行例の関数
async def demo_examples():
"""デモ用の実行例"""
print("\n🎯 実行例デモ")
print("=" * 30)
butler = PositiveButler()
test_messages = [
"今日は最悪な一日だった...",
"仕事でミスばかりしてしまう",
"もう何もやる気が起きない",
"みんなに嫌われているような気がする"
]
for i, message in enumerate(test_messages, 1):
print(f"\n📝 例{i}: {message}")
response = await butler.process_message(message)
print(f"🎩 執事: {response}")
await asyncio.sleep(1) # 見やすくするための小休止
if __name__ == "__main__":
# 実行モードの選択
import sys
if len(sys.argv) > 1 and sys.argv[1] == "--demo":
# デモモード
asyncio.run(demo_examples())
else:
# 対話モード
asyncio.run(main())
実行結果例：
🎩 絶対肯定！AIほめほめ執事 🎩
==================================================
どんなネガティブな言葉も、執事がポジティブに変換いたします！
終了するには 'quit' と入力してください。
==================================================
💬 あなた: 今日発表のセリフを全部飛ばして、1分くらい無言の時間が続いた。頭が真っ白になっていた。もう終わりだ
🤔 執事が考えています...
🎩 執事: ああ、それはさぞかしご心労の極みであったかと存じます。しかしながら、ご安心くださいませ。発表の場で、一瞬の沈黙が生じたことは、決して「終わり」などではございません。むしろ、これから素晴らしい挽回の機会が訪れる、前向きな転換点と捉えるべきでございます。
1分間の無言…それは、聴衆の皆様に、ご発表の内容をより深く心に刻み込ませる、意外な効果を生み出した可能性もございます。 緊張からくる一時的な空白は、誰にでも起こりうることであり、むしろ、その後のご挽回によって、皆様の真摯な姿勢、そしてプレゼンテーションへの情熱がより一層際立つことでしょう。
今回の経験を活かし、次回に向けては、以下のような対策を立ててみてはいかがでしょうか。
* **事前にリハーサルを徹底的に行うこと**: 鏡の前で練習し、声のトーンやジェスチャー、間の取り方などを確認することで、本番での自信へと繋がります。 さらに、想定問答集を作成し、様々な質問への回答を準備しておけば、万が一の事態にも対応できます。
* **深呼吸とマインドフルネスの練習**: 発表直前に深呼吸を行うことで、緊張を和らげ、落ち着きを取り戻すことができます。マインドフルネス瞑想も効果的です。
* **スライドに視覚的な要素を多く取り入れる**: 視覚的な情報が豊富であれば、仮に言葉が詰まったとしても、聴衆の皆様はスライドの内容から情報を理解できます。
* **万が一の事態に備えた、アドリブ対応の練習**: 完全に想定外の事態が起こった場合に備え、状況に応じて柔軟に対応できるよう、アドリブの練習も有効です。
今回の経験は、決して無駄にはなりません。この学びを糧に、より洗練された、そして記憶に残る素晴らしい発表を、必ずや成し遂げられると信じております。 どうぞご自分を責めずに、前向きに進んでいきましょう。 何かお手伝いできることがございましたら、いつでもお申し付けくださいませ。
作例2：【中級編】積み上げた論文PDFも怖くない！AIサマリージェネレーター 📄
目的: PDFファイルを読み込んで要約・キーワード・質問回答を行う
学習ポイント: ファイルI/O連携、外部ライブラリとの組み合わせ、RAGの基礎
#!/usr/bin/env python3
import asyncio
import os
from dotenv import load_dotenv
from genai_processors import content_api, streams
from genai_processors.models.genai_model import GenaiModel
import PyPDF2
load_dotenv()
class PDFSummaryGenerator:
def __init__(self):
api_key = os.getenv('GOOGLE_API_KEY')
# 要約用のプロンプト設定
self.summary_prompt = \"\"\"
あなたは優秀な研究者です。提供されたPDFの内容を分析し、
以下の形式で構造化された要約を作成してください：
## 📄 文書情報
- タイトル: [文書のタイトル]
- 文書の種類: [論文/報告書/技術文書/その他]
## 📝 要約
[3-5文程度の簡潔な要約]
## 🔍 キーワード
[重要なキーワード5-10個をカンマ区切りで]
## 🎯 主要なポイント
1. [重要なポイント1]
2. [重要なポイント2]
3. [重要なポイント3]
## 💡 技術的な特徴
[技術的な内容がある場合の特徴や新規性]
\"\"\"
self.model = GenaiModel(
model_name="gemini-1.5-flash",
api_key=api_key,
system_instruction=self.summary_prompt
)
def extract_text_from_pdf(self, pdf_path: str) -> str:
\"\"\"PDFファイルからテキストを抽出\"\"\"
try:
with open(pdf_path, 'rb') as file:
pdf_reader = PyPDF2.PdfReader(file)
text = ""
print(f"📖 PDFを読み込み中... ({len(pdf_reader.pages)}ページ)")
for page_num, page in enumerate(pdf_reader.pages):
try:
page_text = page.extract_text()
text += f"\\n--- ページ {page_num + 1} ---\\n{page_text}\\n"
if (page_num + 1) % 10 == 0:
print(f" 処理済み: {page_num + 1}/{len(pdf_reader.pages)} ページ")
except Exception as e:
print(f"⚠️ ページ{page_num + 1}の処理でエラー: {e}")
continue
return text
except Exception as e:
raise ValueError(f"PDFの読み込みに失敗しました: {e}")
async def generate_summary(self, pdf_path: str) -> str:
\"\"\"PDFの要約を生成\"\"\"
try:
print("📄 PDFからテキストを抽出中...")
text = self.extract_text_from_pdf(pdf_path)
# テキストが長すぎる場合は制限
max_length = 100000
if len(text) > max_length:
text = text[:max_length] + "\\n\\n[文書が長いため、一部のみ表示]"
print(f"⚠️ テキストが長いため、最初の{max_length}文字に制限しました。")
print("🤖 AIが要約を生成中...")
input_stream = streams.stream_content([text])
response_parts = []
async for part in self.model(input_stream):
if hasattr(part, 'text') and part.text:
response_parts.append(part.text)
return ''.join(response_parts)
except Exception as e:
return f"❌ 要約の生成に失敗しました: {str(e)}"
async def main():
print("📚 PDFサマリージェネレーター 📚")
generator = PDFSummaryGenerator()
pdf_path = input("\\n📄 PDFファイルのパスを入力してください: ").strip()
if not os.path.exists(pdf_path):
print("❌ ファイルが見つかりません。")
return
summary = await generator.generate_summary(pdf_path)
print("\\n" + "="*50)
print(summary)
print("="*50)
if __name__ == "__main__":
asyncio.run(main())
実行結果例：
📄 PDFファイルのパスを入力してください: research_paper.pdf
📖 PDFを読み込み中... (12ページ)
🤖 AIが要約を生成中...
## 📄 文書情報
- タイトル: 機械学習による画像認識の新手法
- 文書の種類: 学術論文
## 📝 要約
本論文では、深層学習とトランスフォーマーアーキテクチャを組み合わせた
新しい画像認識手法を提案している。従来手法と比較して15%の精度向上を
達成し、計算コストも30%削減することに成功した。
## 🔍 キーワード
機械学習, 画像認識, トランスフォーマー, 深層学習, CNN, 精度向上
## 🎯 主要なポイント
1. 新しいアーキテクチャによる精度向上
2. 計算効率の大幅な改善
3. 実用的なアプリケーションへの応用可能性
作例3：【応用編】マイク入力でOK！リアルタイム関西弁翻訳ちゃん 🎌
目的: 音声入力をリアルタイムで関西弁に翻訳
学習ポイント: 音声処理、非同期ストリーミング、継続的な処理ループ
#!/usr/bin/env python3
import asyncio
import os
from dotenv import load_dotenv
from genai_processors import content_api, streams
from genai_processors.models.genai_model import GenaiModel
import speech_recognition as sr
load_dotenv()
class KansaiTranslator:
def __init__(self):
api_key = os.getenv('GOOGLE_API_KEY')
# 関西弁変換用のプロンプト
self.kansai_prompt = \"\"\"
あなたは関西弁の達人です。標準語で話された内容を、
自然で親しみやすい関西弁に変換してください。
関西弁の特徴:
- 「〜や」「〜やん」「〜やねん」などの語尾
- 「おおきに」「めっちゃ」「ほんま」などの表現
- 「何してはるん？」「行きはった」などの敬語
- 自然で親しみやすい雰囲気
\"\"\"
self.model = GenaiModel(
model_name="gemini-1.5-flash",
api_key=api_key,
system_instruction=self.kansai_prompt
)
# 音声認識の設定
self.recognizer = sr.Recognizer()
self.microphone = sr.Microphone()
print("🎤 マイクの調整中...")
with self.microphone as source:
self.recognizer.adjust_for_ambient_noise(source)
print("✅ マイクの調整が完了しました")
async def translate_to_kansai(self, text: str) -> str:
\"\"\"標準語を関西弁に翻訳\"\"\"
try:
input_stream = streams.stream_content([text])
response_parts = []
async for part in self.model(input_stream):
if hasattr(part, 'text') and part.text:
response_parts.append(part.text)
return ''.join(response_parts)
except Exception as e:
return f"翻訳エラー: {str(e)}"
async def listen_and_translate(self):
\"\"\"音声を聞いて関西弁に翻訳\"\"\"
print("📢 話しかけてください！（Ctrl+Cで終了）")
print("=" * 50)
while True:
try:
print("\\n🎤 聞いています...")
# 音声入力を取得
with self.microphone as source:
audio = self.recognizer.listen(source, timeout=5, phrase_time_limit=10)
# 音声をテキストに変換
try:
recognized_text = self.recognizer.recognize_google(audio, language='ja-JP')
print(f"🗣️ 標準語: {recognized_text}")
# 関西弁に翻訳
print("🤔 関西弁に変換中...")
kansai_text = await self.translate_to_kansai(recognized_text)
print(f"🎌 関西弁: {kansai_text}")
print("-" * 30)
except sr.UnknownValueError:
print("⚠️ 音声が認識できませんでした")
except sr.RequestError as e:
print(f"⚠️ 音声認識エラー: {e}")
except sr.WaitTimeoutError:
print("⏰ タイムアウト（再度お話しください）")
except KeyboardInterrupt:
print("\\n🎌 おおきに〜！")
break
except Exception as e:
print(f"❌ エラー: {e}")
async def main():
print("🎌 リアルタイム関西弁翻訳ちゃん 🎌")
translator = KansaiTranslator()
await translator.listen_and_translate()
if __name__ == "__main__":
asyncio.run(main())
実行結果例：
🎤 マイクの調整中...
✅ マイクの調整が完了しました
📢 話しかけてください！（Ctrl+Cで終了）
🎤 聞いています...
🗣️ 標準語: おはようございます
🤔 関西弁に変換中...
🎌 関西弁: おはようさん！今日もええ天気やね〜
🎤 聞いています...
🗣️ 標準語: 今日はいい天気ですね
🤔 関西弁に変換中...
🎌 関西弁: ほんまええ天気やで〜！気持ちええわ〜
作例4：【発展編】アイデアの化学反応！"概念ミキサー"画像生成AI 🎨
目的: 2つの概念を融合した独創的な画像生成プロンプトを作成
学習ポイント: 複数AI連携、パイプライン処理、創造的なプロンプトエンジニアリング
#!/usr/bin/env python3
import asyncio
import os
import json
from dotenv import load_dotenv
from genai_processors import content_api, streams
from genai_processors.models.genai_model import GenaiModel
load_dotenv()
class ConceptMixerImageGenerator:
def __init__(self):
api_key = os.getenv('GOOGLE_API_KEY')
# プロンプト生成用のAI
self.prompt_generator = GenaiModel(
model_name="gemini-1.5-flash",
api_key=api_key,
system_instruction=self._get_prompt_generation_instruction()
)
# 画像説明用のAI
self.image_describer = GenaiModel(
model_name="gemini-1.5-flash",
api_key=api_key,
system_instruction=self._get_image_description_instruction()
)
def _get_prompt_generation_instruction(self) -> str:
return \"\"\"
あなたは創造的なプロンプトエンジニアです。2つの異なる概念を
融合させた、独創的で魅力的な画像生成プロンプトを作成してください。
プロンプト生成の原則:
1. 2つの概念を自然かつ創造的に融合させる
2. 視覚的に魅力的で印象的な要素を含める
3. 具体的な描写を含める（色、質感、雰囲気など）
4. 芸術的なスタイルや技法を指定する
5. 英語で40-60単語程度にまとめる
出力形式:
{
"fusion_concept": "融合概念の日本語説明",
"english_prompt": "英語の画像生成プロンプト",
"style_description": "スタイルの説明",
"expected_result": "期待される画像の説明"
}
JSON形式で回答してください。
\"\"\"
def _get_image_description_instruction(self) -> str:
return \"\"\"
あなたは画像生成AIの専門家です。与えられたプロンプトから、
どのような画像が生成されるかを詳細に説明してください。
魅力的で分かりやすい説明を心がけてください。
\"\"\"
async def generate_fusion_prompt(self, concept1: str, concept2: str) -> dict:
\"\"\"2つの概念を融合したプロンプトを生成\"\"\"
try:
input_text = f\"\"\"
概念1: {concept1}
概念2: {concept2}
これら2つの概念を創造的に融合させた画像生成プロンプトを作成してください。
\"\"\"
input_stream = streams.stream_content([input_text])
response_parts = []
async for part in self.prompt_generator(input_stream):
if hasattr(part, 'text') and part.text:
response_parts.append(part.text)
response_text = ''.join(response_parts)
try:
result = json.loads(response_text)
return result
except json.JSONDecodeError:
return {
"fusion_concept": f"{concept1}と{concept2}の融合",
"english_prompt": f"A creative fusion of {concept1} and {concept2}",
"style_description": "創造的な融合スタイル",
"expected_result": response_text
}
except Exception as e:
return {
"fusion_concept": f"エラーが発生しました: {str(e)}",
"english_prompt": f"A fusion of {concept1} and {concept2}",
"style_description": "基本スタイル",
"expected_result": "エラーが発生しました"
}
async def describe_generated_image(self, prompt: str) -> str:
\"\"\"生成される画像の説明を作成\"\"\"
try:
input_text = f\"\"\"
以下の画像生成プロンプトから、どのような画像が生成されるかを詳細に説明してください：
プロンプト: {prompt}
\"\"\"
input_stream = streams.stream_content([input_text])
response_parts = []
async for part in self.image_describer(input_stream):
if hasattr(part, 'text') and part.text:
response_parts.append(part.text)
return ''.join(response_parts)
except Exception as e:
return f"説明の生成に失敗しました: {str(e)}"
async def generate_concept_fusion(self, concept1: str, concept2: str) -> dict:
\"\"\"概念融合の完全なプロセス\"\"\"
print(f"🎨 概念融合を開始: '{concept1}' × '{concept2}'")
print("-" * 50)
# ステップ1: 融合プロンプトの生成
print("⚙️ ステップ1: 創造的なプロンプトを生成中...")
fusion_data = await self.generate_fusion_prompt(concept1, concept2)
# ステップ2: 画像の説明生成
print("📝 ステップ2: 画像の詳細説明を生成中...")
image_description = await self.describe_generated_image(fusion_data["english_prompt"])
result = {
**fusion_data,
"detailed_description": image_description
}
return result
async def main():
print("🎨 概念ミキサー画像生成AI 🎨")
generator = ConceptMixerImageGenerator()
concept1 = input("\\n💭 概念1を入力してください: ").strip()
concept2 = input("💭 概念2を入力してください: ").strip()
result = await generator.generate_concept_fusion(concept1, concept2)
print("\\n" + "=" * 60)
print("🎉 概念融合結果")
print("=" * 60)
print(f"✨ 融合概念: {result.get('fusion_concept', 'N/A')}")
print(f"🎨 画像生成プロンプト: {result.get('english_prompt', 'N/A')}")
print(f"📖 詳細な説明: {result.get('detailed_description', 'N/A')}")
if __name__ == "__main__":
asyncio.run(main())
実行結果：
💭 概念1を入力してください: 海
💭 概念2を入力してください: ラーメン
🎨 概念融合を開始: '海' × 'ラーメン'
⚙️ ステップ1: 創造的なプロンプトを生成中...
📝 ステップ2: 画像の詳細説明を生成中...
🎉 概念融合結果
✨ 融合概念: 海洋風味のラーメン世界
🎨 画像生成プロンプト: A mystical ramen bowl floating in deep ocean waters,
with noodles made of golden seaweed flowing like underwater currents,
surrounded by luminescent jellyfish and coral formations, artistic digital painting style
📖 詳細な説明: 深海の神秘的な世界に浮かぶラーメン丼。麺は金色の海藻で作られ、
海流のように美しく流れています。周囲には光る クラゲやサンゴ礁が配置され、
幻想的で美しい海中レストランのような雰囲気を演出しています。
考察編：GenAI Processorsの「ココが凄い！」と「ココは注意！」
使ってみて感じたメリット（Pros）
1. 直感的なパイプライン構文
+演算子での連結は本当に直感的で、コードがスッキリします。
# 従来の書き方
result1 = processor1(input)
result2 = processor2(result1)
result3 = processor3(result2)
# GenAI Processorsの書き方
result = processor1 + processor2 + processor3
2. 非同期処理が簡単
async forを使うだけでリアルタイム処理が実現できます。
async for part in model(input_stream):
print(part.text) # リアルタイムで出力
3. Gemini APIとの親和性
設定が簡単で、すぐに高性能なAI機能を使えます。
ハマりどころとTips
1. 非同期処理のデバッグの難しさ
問題: エラーの原因を特定するのが困難
解決策:
try:
async for part in processor(input_stream):
print(part.text)
except Exception as e:
print(f"詳細エラー: {e}")
import traceback
traceback.print_exc()
2. APIの無料枠管理
問題: RPM（リクエスト/分）、TPM（トークン/分）の制限
解決策:
import time
import asyncio
async def rate_limited_request(processor, input_stream, delay=1.0):
\"\"\"レート制限を考慮したリクエスト\"\"\"
await asyncio.sleep(delay)
return await processor(input_stream)
3. エラーハンドリングの考え方
問題: ストリーミング処理中のエラー処理
解決策:
async def safe_stream_processing(processor, input_stream):
try:
async for part in processor(input_stream):
yield part
except Exception as e:
yield f"エラー: {str(e)}"
まとめ
GenAI Processorsは、AI開発の敷居を大幅に下げる素晴らしいライブラリです。
この記事のポイント:
- 🚀 手軽さ: 3ステップで環境構築完了
- 📚 段階的学習: 基本から応用まで体系的に学習
- 💡 実用性: すぐに使える具体的なコード例
- 🔒 セキュリティ: APIキーの安全な管理方法
次のステップ:
- 今回の作例を実際に動かしてみる
- 自分なりにカスタマイズしてみる
- オリジナルのAIアプリケーションを作成する
GenAI Processorsの世界で、あなたも素晴らしいAIアプリケーションを作ってみませんか？
「このアイデアなら自分でも作れそう！」
そう感じたら、ぜひ今すぐ手を動かしてみてください。きっと新しい発見があるはずです！