はじめに
こんにちは、ドイツのコンスタンツ大学大学院の伊藤です！
私は東大発AIベンチャーの株式会社2WINSの元でインターンをしています！
2WINSは隔週で勉強会を開催していて、今回は第2回の内容をもとに発展させた記事です。
FastAPIでLang Chainを使ったので、学んだことをまとめてみました。
具体的には、富士山の画像をアップロードし、クエリに「Mt. Fuji」と入力したときに、富士山を検出できるのかを試しました。
完成イメージ
/detect, /visualizeの二つエンドポイントを作成した。
/detectでは、画像をアップロードしてクエリを入力したら、そのクエリに対応する対象を検出し、JSON形式で結果を返してくれる。
{
"description": "string",
"detected_objects": [
{
"label": "string",
"box_2d": [
null,
null,
null,
null
]
}
]
}
/visualizeでは、上の/detectで得た、バウンディングボックスの座標が正しいのか確かめるために、バウンディングボックス付きの画像を返してくれる。
バウンディングボックスとは以下の画像にあるような検出したことを示してくれる緑色の枠のこと。
フォルダ構成
フォルダ構成は以下の通り。
simple-detection/
├── .env #APIキーを保存しているファイル
├── pyproject.toml #依存パッケージの定義ファイル、これ無くても動く
├── chain.py
├── schema.py
└── main.py
chain.py - 「AIへの支持と処理の流れ」を定義する
# 1
llm = ChatGoogleGenerativeAI(
model="gemini-2.5-flash",
api_key=SecretStr(os.environ["GOOGLE_API_KEY"]),
temperature=0,
).with_structured_output(DetectionResult)
# 2
prompt = ChatPromptTemplate([...])
# 3
chain = (
{"image": lambda x: base64.b64encode(x["image"]).decode(), "query": lambda x: x["query"]}
| prompt
| llm
)
このファイルが今回の処理の中核を担ってくれる。
LangChainを使って、「画像バイナリを受け取ってAIの返答をPythonオブジェクトに変換するまでの処理」を1本のパイプライン(Chain)として定義している。
1. LLMの初期化
ChatGoogleGenerativeAIでGeminiを呼び出すための設定をする。
temperature=0を指定すると、AIの返答の確率的なランダム性を0にするにという意味。
ただし、結果が完全に同一になるというわけではない。GeminiはGoogleのサーバー上で動いており、負荷分散のために複数のハードウェアで並列処理されている。浮動小数演算の丸め誤差がハードウェアごとに少し異なるため、同じ入力でも結果が少し変わる。今回の例でいえば、富士山の裾野がどこまでかという境界には明確な正解がないため、毎回わずかにバウンディングボックスの大きさが変わる。
.with_structured_output(DetectionResult)を末尾に着けることで、AIの返答を直接DetectionResultのPythonオブジェクトに変換してくれる。
補足:以前の書き方(PydantiOutputParser)との比較
LangChainの以前の書き方では、以下のようにPydanticOutputParserを使っていた。
parser = PydanticOutputParser(pydantic_object=DetectionResult)
そもそもパーサーとは、文字列として返ってきたテキストデータを、プログラムで扱えよるようなデータ構造に変換すること。
PydanticOutputParser は、AIが返してきた以下のようなJSON文字列を DetectionResult のPythonオブジェクトに自動変換してくれる。
{
"description": "富士山が写っている画像です",
"detected_objects": [
{
"label": "富士山",
"box_2d": [80, 150, 500, 850]
}
]
}
ただしこの方法では、プロンプトに「このフォーマットで返してください」という指示文(get_format_instructions())を手動で埋め込む必要があった。
with_structured_output()を使えばその指示文の埋め込みが不要になり、Chainの末尾に|parserを書く必要もなくなる。
2. プロンプト
画像はそのままAIに渡せないため、base64という形式に変換してテキストとして埋め込む。プロンプトには画像データと「何を検出してほしいか」というテキスト指示の両方を含んでいる。
3. Chainの組み立て
LangChainでは|でつなぐことで、処理の流れを直観的に記述できる。
今回の流れは以下の通り。
入力 -> base64変換 -> プロンプト生成 -> Gemini呼び出し -> DetectionResultに変換
scheme.py - 「AIの返答の型」を定義する
from pydantic import BaseModel
class DetectedObject(BaseModel):
label: str
box_2d: tuple[int, int, int, int]
class DetectionResult(BaseModel):
description: str
detected_objects: list[DetectedObject]
このファイルでは、AIが返す結果の「型」を定義している。
pydanticはvalidationを自動で行ってくれる。validationとは、そのデータが正しいかチェックすること。
取得したデータの型が、指定した型と異なった場合にエラーを出してくれる。
main.py - 「APIのエンドポイント」を定義する
このファイルはユーザーが画像を送ってきたときの「窓口」にあたる。今回はFastAPIを使ってエンドポイントを2つ定義している。
- 入力バリデーション（門番）
async def get_image_file(file: UploadFile = File(...)) -> UploadFile:
if file.content_type not in ["image/jpeg", "image/jpg", "image/png"]:
raise HTTPException(status_code=400, detail="Unsupported file type.")
return file
送られてきたファイルがJPEGまたはPNGかどうかをチェックする関数。それ以外のファイル（PDFやテキストなど）が送られてきた場合は HTTPException で400エラーを返す。
ValueError ではなく HTTPException を使う理由は、ValueError だと500エラー（サーバーエラー）になってしまうため。「送ってきたデータが不正」なので、クライアント側のエラーである400を返すことができるようにしている。
このFile(...)を書くことで「このパラメータはファイルとして受け取る」とFastAPIに伝わり、Swagger UIが自動で「ファイルを選択」ボタンを表示してくれる。
- Depends() による依存性注入
FastAPIには Depends() という依存性注入の仕組みがある。
# Depends()を使わない場合
async def detect_objects(file: ImageFile, query: QueryStr = "object"):
...
# Depends()を使う場合
async def detect_objects(
file: Annotated[UploadFile, Depends(get_image_file)],
query: str = Form(default="object"),
):
...
Depends(get_image_file) と書くことで、エンドポイントが呼ばれる前にFastAPIが自動で get_image_file を実行してくれる。バリデーション処理をエンドポイントの関数の外に切り出せるため、/detect と /visualize の両方で同じバリデーションを使い回せる。
Annotatedは、型と追加情報(この場合は、Depends(get_image_file))をひとまとめにしてくれて、わかりやすくなる。
UploadFileは、from fastapi import UploadFileでインポートしたもので、自分で定義したものではない。
- エンドポイントについて
FastAPIのエンドポイントは以下のような形で定義する。
@router.post("/エンドポイントのパス")
async def 関数名(パラメータ):
# 処理
return 結果
@ はデコレーターと呼ばれる記号で、「この下の関数に設定を付け加える」という意味。@router.post("/detect") と書くだけで「/detect にPOSTリクエストが来たらこの関数を呼ぶ」という登録ができる。
「/detect」というURLにPOSTリクエストが来たら、この下の関数を呼び出してね」とFastAPIに登録している。
- /detect エンドポイント — 検出結果をJSONで返す
@router.post("/detect", response_model=DetectionResult)
async def detect_objects(
file: Annotated[UploadFile, Depends(get_image_file)],
query: str = Form(default="object"),
) -> Any:
image_bytes = await file.read()
return await chain.ainvoke({"image": image_bytes, "query": query})
response_model=DetectionResult は「返すJSONの形はこの型ですよ」とSwagger UIに伝えるための設定で、/docs のレスポンス例の表示に使われる。
この関数が受け取るパラメータは2つ。
| パラメータ | 内容 |
|---|---|
file |
アップロードされた画像ファイル（バリデーション済み） |
query |
検出対象の文字列（省略時は "object"） |
await chain.ainvoke({"image": image_bytes, "query": query})がGeminiに外注しているコード。chain.pyファイルで定義したパイプラインを非同期で実行し、返ってきたDetectionResultオブジェクトをそのままレスポンスとして返す。awaitは「この処理が終わるまで待つ」という意味で、async defの中でしか使えない。chain.ainvoke(...)は、chainを非同期で実行するコードで、a が「asynchronous（非同期）」の意味。
- /visualize エンドポイント — BBox付き画像を返す
@router.post("/visualize")
async def visualize_objects(
file: Annotated[UploadFile, Depends(get_image_file)],
query: str = Form(default="object"),
) -> StreamingResponse:
image_bytes = await file.read()
result = await chain.ainvoke({"image": image_bytes, "query": query})
img = Image.open(io.BytesIO(image_bytes)).convert("RGB")
w, h = img.size
draw = ImageDraw.Draw(img)
try:
font = ImageFont.truetype("arial.ttf", size=max(16, h // 40))
except OSError:
font = ImageFont.load_default()
for obj in result.detected_objects:
y_min, x_min, y_max, x_max = obj.box_2d
x0, y0 = x_min / 1000 * w, y_min / 1000 * h
x1, y1 = x_max / 1000 * w, y_max / 1000 * h
draw.rectangle([x0, y0, x1, y1], outline="lime", width=3)
draw.text((x0 + 4, y0 + 2), obj.label, fill="lime", font=font)
buf = io.BytesIO()
img.save(buf, format="JPEG")
buf.seek(0)
return StreamingResponse(buf, media_type="image/jpeg")
/detectと違ってresponse_modelがなく、それは返すのがJSONではなく画像ファイルだから。/detectが純粋な窓口なのに対し、/visualizeは窓口兼作業場になっている。
処理の流れは以下の通り。
-
Geminiで検出
/detectと同様にchain.ainvoke(...)でGeminiに外注し、検出結果を受け取る。 -
画像オブジェクトに変換
Image.open(io.BytesIO(image_bytes)).convert("RGB")で受け取った画像バイナリをPIL(画像処理ライブラリ)の画像オブジェクトに変換する。.convert("RGB")はPNGなど透過情報を持つ画像をJPEGで保存できる形式に変換するための処理。PILは、Python Imaging Libraryを意味し、Pythonで画像を扱うためのライブラリで、今回は画像にBBoxを描くために使っている。透過情報(アルファチャンネル)とは、PNGは「透明な部分」を持つことができ、その情報のことである。例えば、ロゴ画像な背景が透明なものなど。しかし、JPEGは透明を表現できないため、透過情報を持つ画像をそのままJPEGに変換しようとするとエラーになるから、.convert("RGB")で透過情報を削除してからJPEGに変換している。 - フォントの読み込み
try:
font = ImageFont.truetype("arial.ttf", size=max(16, h // 40))
except OSError:
font = ImageFont.load_default()
ラベルを書くためのフォントを読み込む。arial.ttfが環境にない場合はOSErrorが発生するため、exceptでデフォルトフォントに切り替えている。フォントサイズはmax(16, h//40)で画像の高さに応じて自動調整しており、最小でも16pxになるようにしている。
4. BBoxとラベルの描画
検出された物体をforループで1つずつ取り出し、座標変換してから描画する。Geminiが返す座標は0〜1000スケールのため、実際のピクセル座標に変換する必要がある。
x0 = x_min / 1000 * w # 例: x_min=300, w=1200px → 360px
draw.rectangle(...) で緑色（lime）の矩形を描き、draw.text(...) でラベル名を描く。+4, +2 は枠線の内側に少しずらすための余白。
5. 画像をレスポンスとして返す
描画済みの画像をメモリ上のバッファ（buf）にJPEGとして保存し、StreamingResponse で返す。ディスクに保存せずメモリ上のデータを直接返せるのが StreamingResponse の利点で、「画像を描画 → ディスクに保存 → 読み込んで返す → 削除」という手間が不要になる。buf.seek(0) はバッファの読み取り位置を先頭に戻す操作で、これがないと空のデータが返ってしまう。media_type="image/jpeg" を指定することでブラウザが「これは画像だ」と認識して表示してくれる。bufはbufferの略で、「一時的にデータを置いておく場所」という意味。今回はPILで描いた画像をいったん bufに保存してから、HTTPレスポンスとして返している。
- APIRouter によるルーティング
router = APIRouter()
...
app = FastAPI()
app.include_router(router)
エンドポイントを app に直接書かず router にまとめてから app に登録している。コンビニ全体（app）の中に担当コーナー（router）を作るイメージ。今回は1ファイルだが、機能が増えたときにエンドポイントをファイル単位で分割しやすくなる。
動作確認
この富士山の画像をアップロードしてみた結果。
- /detect
{
"description": "この画像は、日の出か日没時の富士山を捉えた美しい風景写真です。雪をかぶった山頂は暖かな光に照らされ、その姿は手前の穏やかな湖面に完璧に映し出されています。空は左側が鮮やかなオレンジと黄色から右側に向かって柔らかな青と紫へとグラデーションを描き、湖面にもその色彩が鏡のように反射しています。山と湖の間には、木々や陸地が帯状に広がっています。",
"detected_objects": [
{
"label": "Mt. Fuji",
"box_2d": [
69,
104,
417,
885
]
},
{
"label": "Mt. Fuji",
"box_2d": [
417,
104,
625,
885
]
}
]
}
- /visualize
使用バージョン
fastapi: 0.115.0
langchain-core: 1.3.2
langchain-google-genai: 4.2.2