このノートブックでは、Google Colab AIをOpenAI互換APIとして利用する方法を学びます。各コードブロックを順番に実行して、完全に動作するAPIサーバーを構築しましょう！
🎯 このノートブックで学べること
- Google Colab AIをOpenAI互換APIとして公開する方法
- FastAPIとuvicornを使ったWebサーバーの構築
- ストリーミング対応のチャット補完API
- OpenAIクライアントライブラリでの接続テスト
Google Colabで定額で Gemini / Gemma を使えるライブラリ google-colab-ai をOpenAIフォーマットで呼び出せるラッパを作ってみた！！！
＊📒ノートブック付 https://t.co/bOAC3WxNpt pic.twitter.com/cmOeQV2dNf
— Maki@Sunwood AI Labs. (@hAru_mAki_ch) July 16, 2025
📦 必要なライブラリのインストール
# 必要なライブラリのインストール
!pip install fastapi uvicorn nest-asyncio
print("✅ ライブラリのインストールが完了しました！")📋 基本ライブラリのインポート
import os
import time
import json
import uuid
import subprocess
import threading
import requests
from typing import List, Dict, Optional, Union, AsyncGenerator
from datetime import datetime
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
import uvicorn
import asyncio
import nest_asyncio
nest_asyncio.apply()
# Google Colab AI インポート
from google.colab import ai
print("✅ ライブラリのインポートが完了しました！")🏗️ OpenAI互換データモデルの定義
# OpenAI互換データモデル定義
class ChatMessage(BaseModel):
role: str = Field(..., description="メッセージの役割 (system, user, assistant)")
content: str = Field(..., description="メッセージの内容")
class ChatCompletionRequest(BaseModel):
model: str = Field(default="google/gemini-2.5-flash", description="使用するモデル")
messages: List[ChatMessage] = Field(..., description="会話メッセージのリスト")
temperature: Optional[float] = Field(default=0.7, ge=0, le=2, description="応答のランダム性")
max_tokens: Optional[int] = Field(default=None, gt=0, description="最大トークン数")
stream: Optional[bool] = Field(default=False, description="ストリーミングレスポンス")
top_p: Optional[float] = Field(default=1.0, ge=0, le=1, description="核サンプリング")
frequency_penalty: Optional[float] = Field(default=0, ge=-2, le=2)
presence_penalty: Optional[float] = Field(default=0, ge=-2, le=2)
class ChatCompletionUsage(BaseModel):
prompt_tokens: int
completion_tokens: int
total_tokens: int
class ChatCompletionChoice(BaseModel):
index: int
message: ChatMessage
finish_reason: str
class ChatCompletionResponse(BaseModel):
id: str
object: str = "chat.completion"
created: int
model: str
choices: List[ChatCompletionChoice]
usage: ChatCompletionUsage
class ChatCompletionStreamChoice(BaseModel):
index: int
delta: Dict[str, Optional[str]]
finish_reason: Optional[str] = None
class ChatCompletionStreamResponse(BaseModel):
id: str
object: str = "chat.completion.chunk"
created: int
model: str
choices: List[ChatCompletionStreamChoice]
class ModelInfo(BaseModel):
id: str
object: str = "model"
created: int = int(time.time())
owned_by: str = "google"
class ModelsResponse(BaseModel):
object: str = "list"
data: List[ModelInfo]
print("✅ データモデルの定義が完了しました！")🤖 Google Colab AI ラッパークラス
class ColabAIWrapper:
"""Google Colab AIをOpenAI形式でラップするクラス"""
def __init__(self):
self.available_models = self._get_available_models()
print(f"✅ 利用可能なモデル: {len(self.available_models)}個")
for model in self.available_models:
print(f" • {model}")
def _get_available_models(self) -> List[str]:
"""利用可能なモデル一覧を取得"""
try:
return ai.list_models()
except Exception as e:
print(f"⚠️ モデル一覧取得エラー: {e}")
return ["google/gemini-2.5-flash"]
def _convert_messages_to_prompt(self, messages: List[ChatMessage]) -> str:
"""OpenAI形式のメッセージを単一プロンプトに変換"""
prompt_parts = []
for message in messages:
role = message.role
content = message.content
if role == "system":
prompt_parts.append(f"[システム指示] {content}")
elif role == "user":
prompt_parts.append(f"ユーザー: {content}")
elif role == "assistant":
prompt_parts.append(f"アシスタント: {content}")
if not prompt_parts[-1].startswith("アシスタント:"):
prompt_parts.append("アシスタント:")
return "\n".join(prompt_parts)
def _estimate_tokens(self, text: str) -> int:
"""トークン数の概算"""
japanese_chars = sum(1 for c in text if ord(c) > 127)
other_chars = len(text) - japanese_chars
english_words = len(text.split())
estimated_tokens = int(japanese_chars * 1.5 + english_words * 1.3)
return max(estimated_tokens, len(text) // 4)
async def generate_completion(
self,
request: ChatCompletionRequest
) -> Union[ChatCompletionResponse, AsyncGenerator]:
"""チャット補完を生成"""
prompt = self._convert_messages_to_prompt(request.messages)
model = request.model if request.model in self.available_models else self.available_models[0]
completion_id = f"chatcmpl-{uuid.uuid4().hex[:10]}"
created = int(time.time())
if request.stream:
return self._generate_stream(completion_id, model, prompt, created)
else:
return await self._generate_single(completion_id, model, prompt, created, request)
async def _generate_single(
self,
completion_id: str,
model: str,
prompt: str,
created: int,
request: ChatCompletionRequest
) -> ChatCompletionResponse:
"""非ストリーミング補完を生成"""
try:
# Google Colab AI呼び出し
if model in self.available_models:
response_text = ai.generate_text(prompt, model_name=model)
else:
response_text = ai.generate_text(prompt)
# トークン数計算
prompt_tokens = self._estimate_tokens(prompt)
completion_tokens = self._estimate_tokens(response_text)
return ChatCompletionResponse(
id=completion_id,
created=created,
model=model,
choices=[
ChatCompletionChoice(
index=0,
message=ChatMessage(role="assistant", content=response_text.strip()),
finish_reason="stop"
)
],
usage=ChatCompletionUsage(
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens
)
)
except Exception as e:
print(f"❌ 生成エラー: {e}")
raise HTTPException(status_code=500, detail=f"生成エラー: {str(e)}")
async def _generate_stream(
self,
completion_id: str,
model: str,
prompt: str,
created: int
) -> AsyncGenerator[str, None]:
"""ストリーミング補完を生成"""
try:
# ストリーミング生成
if model in self.available_models:
stream = ai.generate_text(prompt, model_name=model, stream=True)
else:
stream = ai.generate_text(prompt, stream=True)
# 最初のチャンクでroleを送信
first_chunk = ChatCompletionStreamResponse(
id=completion_id,
created=created,
model=model,
choices=[
ChatCompletionStreamChoice(
index=0,
delta={"role": "assistant", "content": ""},
finish_reason=None
)
]
)
yield f"data: {first_chunk.model_dump_json()}\n\n"
# ストリーミングデータを送信
for chunk_text in stream:
chunk = ChatCompletionStreamResponse(
id=completion_id,
created=created,
model=model,
choices=[
ChatCompletionStreamChoice(
index=0,
delta={"content": chunk_text},
finish_reason=None
)
]
)
yield f"data: {chunk.model_dump_json()}\n\n"
await asyncio.sleep(0.01)
# 終了チャンク
final_chunk = ChatCompletionStreamResponse(
id=completion_id,
created=created,
model=model,
choices=[
ChatCompletionStreamChoice(
index=0,
delta={},
finish_reason="stop"
)
]
)
yield f"data: {final_chunk.model_dump_json()}\n\n"
yield "data: [DONE]\n\n"
except Exception as e:
error_chunk = {
"error": {
"message": f"ストリーミングエラー: {str(e)}",
"type": "server_error"
}
}
yield f"data: {json.dumps(error_chunk)}\n\n"
print("✅ ColabAIWrapperクラスの定義が完了しました！")🚀 FastAPIアプリケーションの設定
# FastAPI アプリケーション
app = FastAPI(
title="Google Colab AI OpenAI互換API",
description="Google Colab AIをOpenAI形式で呼び出すためのAPIサーバー",
version="1.0.0"
)
# CORS設定
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Colab AIラッパーのインスタンス化
colab_ai = ColabAIWrapper()
print("✅ FastAPIアプリケーションの設定が完了しました！")🔗 APIエンドポイントの定義
# APIエンドポイント
@app.get("/")
async def root():
"""APIの基本情報"""
return {
"message": "Google Colab AI OpenAI互換API",
"version": "1.0.0",
"available_models": colab_ai.available_models,
"endpoints": {
"chat_completions": "/v1/chat/completions",
"models": "/v1/models",
"health": "/health"
}
}
@app.get("/health")
async def health_check():
"""ヘルスチェック"""
return {"status": "healthy", "timestamp": datetime.now().isoformat()}
@app.get("/v1/models", response_model=ModelsResponse)
async def list_models():
"""利用可能なモデル一覧を取得"""
models = [
ModelInfo(
id=model_id,
owned_by="google"
)
for model_id in colab_ai.available_models
]
return ModelsResponse(data=models)
@app.post("/v1/chat/completions")
async def create_chat_completion(request: ChatCompletionRequest):
"""チャット補完を作成（OpenAI互換）"""
if not request.messages:
raise HTTPException(status_code=400, detail="メッセージが空です")
print(f"📨 リクエスト受信: model={request.model}, stream={request.stream}")
print(f" メッセージ数: {len(request.messages)}")
try:
result = await colab_ai.generate_completion(request)
if request.stream:
return StreamingResponse(
result,
media_type="text/plain",
headers={
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache",
"Connection": "keep-alive",
}
)
else:
return result
except Exception as e:
print(f"❌ エラー: {e}")
raise HTTPException(status_code=500, detail=str(e))
print("✅ APIエンドポイントの定義が完了しました！")🏃 サーバー起動関数の定義
def run_server():
"""subprocessでサーバーを起動"""
print("🚀 Google Colab AI OpenAI互換APIサーバーを起動中...")
def start_uvicorn():
"""uvicornサーバーを別スレッドで起動"""
config = uvicorn.Config(app, host="0.0.0.0", port=8000, log_level="warning")
server = uvicorn.Server(config)
# 新しいイベントループで実行
asyncio.set_event_loop(asyncio.new_event_loop())
asyncio.get_event_loop().run_until_complete(server.serve())
# バックグラウンドでサーバー起動
server_thread = threading.Thread(target=start_uvicorn)
server_thread.daemon = True
server_thread.start()
# サーバー起動待機
time.sleep(3)
# 接続テスト
try:
response = requests.get("http://localhost:8000/health", timeout=5)
if response.status_code == 200:
print("✅ サーバーが正常に起動しました！")
print("🔗 ローカルアクセス: http://localhost:8000")
print("📡 チャット補完: http://localhost:8000/v1/chat/completions")
print("📋 モデル一覧: http://localhost:8000/v1/models")
print("🏥 ヘルスチェック: http://localhost:8000/health")
# Google Colabの場合、外部からのアクセス情報を表示
print("\n💡 Google Colabからの外部アクセス:")
print(" ポート8000が0.0.0.0でリッスンしています")
print(" ローカルネットワークからアクセス可能です")
return True
else:
print(f"❌ サーバー起動エラー: {response.status_code}")
return False
except Exception as e:
print(f"❌ サーバー接続エラー: {e}")
return False
print("✅ サーバー起動関数の定義が完了しました！")🚀 APIサーバーの起動
# APIサーバーを起動
run_server()🧪 基本APIテスト関数の定義
def test_api():
"""APIの簡単なテスト"""
base_url = "http://localhost:8000"
print("🧪 APIテスト開始...")
# ヘルスチェック
try:
response = requests.get(f"{base_url}/health", timeout=5)
if response.status_code == 200:
print("✅ ヘルスチェック: OK")
else:
print(f"❌ ヘルスチェック失敗: {response.status_code}")
return False
except Exception as e:
print(f"❌ ヘルスチェックエラー: {e}")
return False
# モデル一覧取得
try:
response = requests.get(f"{base_url}/v1/models", timeout=5)
if response.status_code == 200:
models = response.json()
print(f"✅ モデル一覧取得: {len(models['data'])}個のモデル")
else:
print(f"❌ モデル一覧取得失敗: {response.status_code}")
except Exception as e:
print(f"❌ モデル一覧取得エラー: {e}")
# チャット補完テスト
try:
data = {
"model": "google/gemini-2.5-flash",
"messages": [
{"role": "user", "content": "こんにちは！元気ですか？"}
],
"stream": False
}
response = requests.post(f"{base_url}/v1/chat/completions", json=data, timeout=30)
if response.status_code == 200:
result = response.json()
content = result["choices"][0]["message"]["content"]
print(f"✅ チャット補完テスト: {content[:50]}...")
print(f"💰 使用トークン: {result['usage']['total_tokens']}")
return True
else:
print(f"❌ チャット補完失敗: {response.status_code}")
print(response.text)
return False
except Exception as e:
print(f"❌ チャット補完エラー: {e}")
return False
print("✅ APIテスト関数の定義が完了しました！")🧪 基本APIテストの実行
# 基本APIテストを実行
test_api()🔧 OpenAIクライアントテスト関数の定義
def test_openai_client():
"""OpenAIクライアントでのテスト"""
try:
import openai
except ImportError:
print("📦 OpenAIライブラリをインストール中...")
!pip install openai
import openai
print("🧪 OpenAIクライアントテスト...")
client = openai.OpenAI(
api_key="dummy-key",
base_url="http://localhost:8000/v1"
)
try:
# 非ストリーミング
response = client.chat.completions.create(
model="google/gemini-2.5-flash",
messages=[
{"role": "system", "content": "あなたは親切なAIアシスタントです。"},
{"role": "user", "content": "Pythonの基本的な使い方を教えてください。"}
],
temperature=0.7
)
print("✅ OpenAIクライアント (非ストリーミング):")
print(f" {response.choices[0].message.content[:100]}...")
# ストリーミング
print("\n🌊 ストリーミングテスト:")
stream = client.chat.completions.create(
model="google/gemini-2.5-flash",
messages=[
{"role": "user", "content": "短い俳句を作ってください。"}
],
stream=True
)
for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
print("\n✅ ストリーミング完了")
return True
except Exception as e:
print(f"❌ OpenAIクライアントエラー: {e}")
return False
print("✅ OpenAIクライアントテスト関数の定義が完了しました！")🧪 OpenAIクライアントテストの実行
# OpenAIクライアントテストを実行
test_openai_client()🎯 カスタムテスト（自由に試してください！）
import openai
# OpenAIクライアントの設定
client = openai.OpenAI(
api_key="dummy-key",
base_url="http://localhost:8000/v1"
)
# 好きな質問をしてみてください！
response = client.chat.completions.create(
model="google/gemini-2.5-flash",
messages=[
{"role": "system", "content": "あなたは博識なAI助手です。"},
{"role": "user", "content": "量子コンピューターについて簡単に説明してください。"}
],
temperature=0.8
)
print("🎯 カスタムテスト結果:")
print(response.choices[0].message.content)📚 まとめ
🎉 おめでとうございます！ Google Colab AI OpenAI互換APIサーバーが正常に動作しています。
✅ 完了したこと
- ライブラリインストール - FastAPI、uvicorn等の必要ライブラリ
- データモデル定義 - OpenAI完全互換のPydanticモデル
- AIラッパー作成 - Google Colab AIをOpenAI形式で使用
- APIサーバー構築 - FastAPIベースのWebサーバー
- エンドポイント実装 - チャット補完、モデル一覧等のAPI
- サーバー起動 - バックグラウンドでのサーバー実行
- 動作テスト - 基本API & OpenAIクライアントテスト
🔗 利用可能なエンドポイント
- ルートエンドポイント:
http://localhost:8000/ - ヘルスチェック:
http://localhost:8000/health - モデル一覧:
http://localhost:8000/v1/models - チャット補完:
http://localhost:8000/v1/chat/completions
💡 活用方法
このAPIサーバーを使って：
- OpenAI互換のクライアントアプリケーション開発
- 既存のOpenAIを使ったプロジェクトでの代替利用
- チャットボットやAIアシスタントの構築
- ストリーミング対応のリアルタイムチャット
📒ノートブック
Google Colab
colab.research.google.com
🚀 次のステップ
- ngrokを使った外部公開
- 認証機能の追加
- ログ機能の実装
- パフォーマンス最適化
楽しいAI開発をお楽しみください！ 🤖✨
コメント