複数の問い合わせ要求を受け取り、回答を返す
これだけでは、ただGemini APIとの間にひとつエンドポイントが増えただけで、何もしてくれません。つまり、嬉しくありません。
次はいよいよ、2つ以上の役割のGeminiに、まとめて同じ質問を聞けるようにしましょう。
といっても、先ほどと同じように、Gemini APIに聞く部分はすでに関数ができています。次のsearchapi.pyの以下の部分を見てみましょう。
def grid_query_gemini(
q: str,
roles: tuple[str, ...],
model_names: tuple[AVAILABLE_MODELS, ...],
temperature: float,
max_tokens: int | None = None,
) -> list[tuple[str, QueryArgs]]:
"""
...（中略）...
"""
results = []
for model_name in model_names:
for role in roles:
result, args = query_gemini(
q=q,
role=role,
model_name=model_name,
temperature=temperature,
max_tokens=max_tokens,
)
results.append((result, args))
return results
関数の引数roles、model_namesの部分を見てみると、複数の引数が受け取れるようになっています。そして、受け取った引数のroles、model_namesを、for文で繰り返し問い合わせているわけです。
さて、この関数を使ったエンドポイントも、main.pyに書き込んでいきます。以下の部分に書き込んでみましょう。
@app.post("/multi", response_model=MultiQueryResponse)
def multi(data: MultiRequest):
"""
複数の問い合わせを行うエンドポイント
"""
# ここにコードを書く
return MultiQueryResponse(
data=[
MultiQueryItem(
id=1,
result="Not Implemented",
args=None,
)
],
meta={"duration": 0},
)
といっても、実は、singleの時とそんなに変わりません。/singleのときと大きく違う行をdiffで示してみました。
@app.post("/multi", response_model=MultiQueryResponse)
def multi(data: MultiRequest):
start_time = time.time()
+ model_names = tuple(data.options.models)
+ roles = tuple(data.options.roles)
+ max_tokens = data.options.max_tokens
try:
+ results = grid_query_gemini(
q=data.q,
roles=roles,
model_names=model_names,
temperature=0.7,
max_tokens=max_tokens,
)
except ValueError as e:
raise HTTPException(status_code=500, detail=str(e))
else:
+ multi_query_items = []
+ for idx, (result, args) in enumerate(results, 1):
+ multi_query_items.append(
+ MultiQueryItem(
+ id=idx,
+ result=result,
+ args=args,
)
)
end_time = time.time()
duration = end_time - start_time
# 応答を作成
response = MultiQueryResponse(
+ data=multi_query_items,
meta={"duration": duration},
)
return response
一つ一つ見ていきましょう。
- 複数の引数の取得
/singleでは単一のモデルと役割を指定していたのに対し、/multiではdata.optionsからモデル名と役割のタプルを取得しています。これにより、複数のモデルと役割の組み合わせでクエリを実行できるようになります。
model_names = tuple(data.options.models) roles = tuple(data.options.roles) max_tokens = data.options.max_tokens
- 一括クエリの実行
この関数は、複数のモデル、役割、その他の引数を受け取り、それらの組み合わせに対してクエリを一括で実行する役割を担っています。
/singleでは単一のクエリを実行していたのに対し、ここでは「グリッド」という名前の通り、複数のクエリを連続で実行する処理が行われます。
results = grid_query_gemini(...)
- 応答データの整形
grid_query_geminiから返されたresultsは、クエリ結果のリストになっています。このリストをループ処理し、各結果と使用された引数（args）を組み合わせてMultiQueryItemという独自のデータ構造に格納しています。
multi_query_items = [] for idx, (result, args) in enumerate(results, 1): multi_query_items.append(...)
実際に試してみましょう。
- APIサーバを起動する
利用されるポートは、デフォルトでは8000番です。
$ uv run uvicorn main:app --reload
- curlでアクセスしてみる
$ curl -X POST "http://127.0.0.1:8000/multi" \
-H "accept: application/json" \
-H "Content-Type: application/json" \
-d '{
"q": "FastAPI にはどんな特徴がありますか？",
"options": {
"models": [
"gemini-2.0-flash"
],
"roles": [
"小学校の先生",
"専門学校の先生",
"IT 企業の上司"
],
"max_tokens": 128
}
}'
すると、一つの回答が返ってきます。整形したものがこちらです。うまく動いたでしょうか。
{
"data": [
{
"id": 1,
"result": "はい、FastAPIはPythonでAPIを構築するためのモダンなWebフレームワークです。小学校の先生のようにわかりやすく説明しますね。FastAPIには、主に次のような良いところがあります。1. 速い（高速）:まるで運動会の短距離走みたいに、APIの処理がとても速いです...",
"args": {
"query": "FastAPI にはどんな特徴がありますか？",
"role": "小学校の先生",
"model_name": "gemini-2.0-flash",
"temperature": 0.7,
"max_tokens": 128
}
},
{
"id": 2,
"result": "はい、FastAPI の特徴についてですね。FastAPI は、Python で API を構築するための比較的新しいフレームワークですが、非常に人気があり、多くの開発者に支持されています...",
"args": {
"query": "FastAPI にはどんな特徴がありますか？",
"role": "専門学校の先生",
"model_name": "gemini-2.0-flash",
"temperature": 0.7,
"max_tokens": 128
}
},
{
"id": 3,
"result": "はい、承知いたしました。FastAPI の特徴について説明しますね。FastAPI は、Python で API を構築するためのモダンで高速な (高パフォーマンス) Web フレームワークです。特に、以下のような特徴が挙げられます。主な特徴:高速性:ASGI (Asynchronous Server Gateway Interface) に基づいており、非同期処理をサポートしています。これにより、Node.js や Go と同等のパフォーマンスを実現できます。Pydantic によるデータ検証とシリアライゼーションの最適化により、高速なデータ処理が可能です。",
"args": {
"query": "FastAPI にはどんな特徴がありますか？",
"role": "IT 企業の上司",
"model_name": "gemini-2.0-flash",
"temperature": 0.7,
"max_tokens": 128
}
}
],
"meta": { "duration": 5.0235841274261475 }
}
このように、一度のリクエストで複数の役割から見た回答を得ることができました。しかし、durationを見ると、3つの回答を得るのに5秒ほどかかっていることがわかります。これは長いです。なぜなら、for文で一つ一つの回答を順々に問い合わせているからです。前の回答が到着するまで、次の回答に進みません。ああ、どうにかして複数の回答を同時に問い合わせできないものか。
複数の問い合わせ要求を受け取り、回答を「すばやく」返す（非同期処理）
できます。次の関数を見てみましょう。
async def agrid_query_gemini(
q: str,
roles: tuple[str, ...],
model_names: tuple[AVAILABLE_MODELS, ...],
temperature: float,
max_tokens: int | None = None,
) -> list[tuple[str, QueryArgs]]:
"""
（中略）
"""
tasks = []
for model_name in model_names:
for role in roles:
task = aquery_gemini(
q=q,
role=role,
model_name=model_name,
temperature=temperature,
max_tokens=max_tokens,
)
tasks.append(task)
# 並列に実行して結果を待つ
results = await asyncio.gather(*tasks)
return results
重要なのはtaskに順に代入され、tasksリストに追加されていきます。このtaskは、一番最初の/singleエンドポイントで利用した単一のリクエストを積んでいきます。
そのあと、await asyncio.gather(*tasks)を実行することで、これらのコルーチンを「ほぼ同時に」走らせ、すべての結果が返るまで待ちます（内部的には、I/O待ちの間に別の処理へ切り替わることで高速化します）。
これを実際に Web APIに実装すると、以下のようになります。といっても、変更点は些細なものです。
@app.post("/multi-async", response_model=MultiQueryResponse)
async def multi_async(data: MultiRequest):
"""
複数の問い合わせを非同期で行うエンドポイント
"""
start_time = time.time()
model_names = tuple(data.options.models)
roles = tuple(data.options.roles)
max_tokens = data.options.max_tokens
try:
# agrid_query_geminiを呼び出して、複数の組み合わせで非同期に問い合わせる
results = await agrid_query_gemini(
q=data.q,
roles=roles,
model_names=model_names,
temperature=0.7,
max_tokens=max_tokens,
except ValueError as e:
# Gemini APIの環境変数が設定されていない場合など
raise HTTPException(status_code=500, detail=str(e))
else:
# 回答をMultiQueryItemに変換
multi_query_items = []
for idx, (result, args) in enumerate(results, 1):
multi_query_items.append(
MultiQueryItem(
id=idx,
result=result,
args=args,
)
)
end_time = time.time()
duration = end_time - start_time
# 応答を作成
response = MultiQueryResponse(
data=multi_query_items,
meta={"duration": duration},
)
return response
1つ目は、resultsに結果が代入される関数が、非同期に対応した関数agrid_query_geminiに置き換わっています。これは、当然のことです。
2つ目は、エンドポイントの関数定義def multi_async(data: MultiRequest): ...の前にasyncというワードが付与され、resultsに代入される関数agrid_query_geminiが書かれる前にawaitと書かれている点です。
このasync/awaitは、Pythonにおける非同期処理のための構文です。まず、asyncは、関数定義の前に付与することで、その関数は「非同期関数（coroutine）」になります。通常の関数とは異なり、呼び出すと即時に処理結果を返すのではなく、「実行すべき処理のタスクオブジェクト」を返します。したがって、この関数を呼び出しても、awaitを付けない限り実際には処理が進みません。そして、awaitは、非同期関数内でのみ使用できるキーワードです。先に他の非同期処理の完了を待ち、その結果を受け取ります。もしawaitを付けずに呼び出すと「タスクオブジェクト」が返るだけで、本来の処理は進みません。
この仕組みによって、複数の外部APIリクエストを同時に投げて結果を待つ、というような「待ち時間の多い処理」を高速化できるのです。
実際にこれにアクセスするには、以下の手順を行います。
- APIサーバを起動する
利用されるポートは、デフォルトでは8000番です。
$ uv run uvicorn main:app --reload
- curlでアクセスしてみる
$ curl -X POST "http://127.0.0.1:8000/multi-async" \
-H "accept: application/json" \
-H "Content-Type: application/json" \
-d '{
"q": "FastAPI にはどんな特徴がありますか？",
"options": {
"models": [
"gemini-2.0-flash"
],
"roles": [
"小学校の先生",
"専門学校の先生",
"IT 企業の上司"
],
"max_tokens": 128
}
}'
すると、応答が返ってきます。応答の様子は先ほどと大差がないですが、実行時間durationが大きく短縮されています。先ほどの非同期関数を利用しない場合の実行時間は、筆者の環境では5.0秒ほどかかりましたが、非同期関数を利用した場合、2.0秒ほどになりました。
このように、FastAPIとPythonは非同期処理を簡単な記述で利用できるように設計されており、複数のリクエストを同時に処理することが可能です。これにより、APIの応答時間を大幅に短縮することができます。
