【Gradio】関数の入出力と非同期処理のベストプラクティス

こんにちは、JS2IIUです。
Gradioは、Pythonで手軽にWebアプリケーションを構築できるライブラリとして注目されています。特に機械学習モデルのインターフェースを素早く構築できる点で人気がありますが、Gradioに渡す関数の「入出力形式」や「非同期処理」への理解が不十分なまま開発を進めると、パフォーマンスや拡張性に課題が出ることがあります。
本記事では、Gradioで利用する関数の定義方法、入出力の型指定、さらに async def を用いた非同期処理の活用方法まで、ベストプラクティスとして丁寧に確認していきます。今回もよろしくお願いします。

1. Gradioの基本：関数とインターフェースの紐付け

Gradioでは、関数をWebインターフェースに接続することで、簡単にインタラクティブなUIを構築できます。基本形は次の通りです。

import gradio as gr

def greet(name):
    return f"Hello, {name}!"

gr.Interface(fn=greet, inputs="text", outputs="text").launch()

import gradio as gr

def greet(name):
    return f"Hello, {name}!"

gr.Interface(fn=greet, inputs="text", outputs="text").launch()

この例では、greet という関数に "text" 入力と "text" 出力を設定して、シンプルな挨拶アプリを作成しています。

2. 入力・出力の型の詳細

Gradioの inputs と outputs は、UIのコンポーネントとPython関数のデータ型を対応させるための設定です。

複数の引数

複数の引数をとる関数を使いたい場合、inputs にはリストで複数のコンポーネントを指定します。

def add(x, y):
    return x + y

gr.Interface(fn=add, inputs=["number", "number"], outputs="number").launch()

def add(x, y):
    return x + y

gr.Interface(fn=add, inputs=["number", "number"], outputs="number").launch()

辞書形式の出力（JSON）

関数の戻り値として辞書を返すと、JSON形式で表示されます。

def person_info(name, age):
    return {"name": name, "age": age}

gr.Interface(fn=person_info, inputs=["text", "number"], outputs="json").launch()

def person_info(name, age):
    return {"name": name, "age": age}

gr.Interface(fn=person_info, inputs=["text", "number"], outputs="json").launch()

この形式は、APIレスポンスのようなデータを扱いたいときに便利です。

3. 非同期関数 (async def) の基本

Gradioは、Pythonの非同期関数（async def）を関数として直接受け取ることができます。これは、非同期I/Oを用いたアプリケーション、たとえば外部APIの呼び出しや長時間かかる処理において特に有効です。

3.1 なぜ非同期関数が必要なのか

通常の関数（def）では、時間のかかる処理（例えば2秒待つなど）があると、その間Gradioのインターフェースはブロックされます。つまり、UIが固まり、他の入力も受け付けられません。これはWebアプリとしての操作性を著しく損ないます。

一方で非同期関数（async def）は、await を用いることで処理の一時中断と再開を制御できます。これにより他の処理を並行して進められるため、ユーザーの体験を損ねることなく遅延処理を組み込むことが可能になります。

3.2 Pythonにおける async def と await

Pythonでは、非同期関数を async def で定義し、その中で非同期な処理（例：I/O）を行う際に await を使います。代表的な非同期処理対象は以下の通りです：

• asyncio.sleep()：指定秒数だけ非同期で待つ
• 非同期HTTPクライアント（例：httpx.AsyncClient）
• 非同期データベースドライバ（例：asyncpg）

以下は基本的な非同期関数の例です：

import asyncio

async def slow_echo(text):
    await asyncio.sleep(2)  # 非同期に2秒待機
    return f"Echo: {text}"

import asyncio

async def slow_echo(text):
    await asyncio.sleep(2)  # 非同期に2秒待機
    return f"Echo: {text}"

この関数は呼び出されると、2秒待機してから入力テキストを加工して返します。

3.3 Gradioにおける非同期関数の扱い

Gradioでは、このような async def 関数をそのまま Interface や Blocks に渡すことが可能です。

import gradio as gr
import asyncio

async def slow_echo(text):
    await asyncio.sleep(2)
    return f"Echo: {text}"

gr.Interface(fn=slow_echo, inputs="text", outputs="text").launch()

import gradio as gr
import asyncio

async def slow_echo(text):
    await asyncio.sleep(2)
    return f"Echo: {text}"

gr.Interface(fn=slow_echo, inputs="text", outputs="text").launch()

この例を実行すると、入力されたテキストに対し、2秒待ってから結果が返ってきます。しかし重要なのは、その2秒間にもGradioはUI操作や他の処理をブロックしないという点です。

Gradio内部では、async 関数を検知すると、Pythonのイベントループ（asyncio）を適切に扱って呼び出す仕組みが整っており、特別な設定は不要です。

3.4 Gradioで async 関数を使用する際の注意点

1. await 可能な処理しか非同期に意味がない

以下のようなコードには await がありますが、もし await する対象がなければ、関数を非同期にするメリットは基本的にありません。

# 無意味な非同期関数（非推奨）
async def useless_async(text):
    return text  # await がない

# 無意味な非同期関数（非推奨）
async def useless_async(text):
    return text  # await がない

この場合、普通の def を使った方がシンプルです。

2. 関数の呼び出し側は非同期でなければならない

Gradioが内部的に非同期関数を処理できるのは、イベントループを制御してくれているからです。自分で書くコードで async def を呼ぶには、await が必要であることに注意してください。

# 非同期関数を自分で呼ぶ場合は await が必須
result = await slow_echo("hello")  # 呼び出し側も async である必要がある

# 非同期関数を自分で呼ぶ場合は await が必須
result = await slow_echo("hello")  # 呼び出し側も async である必要がある

Gradioはこの仕組みを内部で処理してくれているため、ユーザーは async def を定義するだけで問題ありません。

3. Pythonバージョンに注意

Python 3.7以降で async / await の構文は安定して使えるようになっています。Gradio自体もPython 3.7以降のサポートを前提としているため、古い環境では動作しない可能性があります。

4. 非同期API呼び出しの実用例

Gradioで非同期関数を活用する大きなメリットの一つが、外部APIとの連携処理をUIを固めずに実行できることです。API呼び出しは典型的なI/O操作の一種であり、応答に数秒かかる場合も少なくありません。

ここでは、Pythonの非同期HTTPクライアントライブラリである httpx を用いて、外部APIからデータを取得する実例を通して、非同期処理の構造を詳しく説明します。

4.1 httpx とは何か

httpx は、Python標準ライブラリの requests に代わる、非同期対応のHTTPクライアントライブラリです。以下の特徴があります：

• 非同期 (async) モードと同期 (sync) モードの両方に対応
• HTTP/1.1 および HTTP/2 の両方をサポート
• リクエストのタイムアウト、リトライ、接続管理などが細かく制御可能

httpx.AsyncClient を使用することで、非同期I/Oによる効率的なリクエストが可能になります。

4.2 実装例：ランダムジョーク取得アプリ

以下のコードは、Official Joke API を呼び出して、ランダムなジョークを取得し表示する非同期Gradioアプリです。

import gradio as gr
import httpx

async def get_joke(_):
    async with httpx.AsyncClient() as client:
        response = await client.get("https://official-joke-api.appspot.com/random_joke")
        data = response.json()
        return f"{data['setup']} - {data['punchline']}"

gr.Interface(fn=get_joke, inputs="button", outputs="text").launch()

import gradio as gr
import httpx

async def get_joke(_):
    async with httpx.AsyncClient() as client:
        response = await client.get("https://official-joke-api.appspot.com/random_joke")
        data = response.json()
        return f"{data['setup']} - {data['punchline']}"

gr.Interface(fn=get_joke, inputs="button", outputs="text").launch()

4.3 このコードの構造と解説

非同期関数の定義

async def get_joke(_):

async def get_joke(_):

ここでは、Gradioの button コンポーネントに対応するため、引数 _ を取るようにしています（ボタンは引数なしでも動作しますが、Gradioの内部的には「何かが渡される」仕様です）。

非同期HTTPクライアントの生成とリクエスト

async with httpx.AsyncClient() as client:
    response = await client.get("https://official-joke-api.appspot.com/random_joke")

async with httpx.AsyncClient() as client:
    response = await client.get("https://official-joke-api.appspot.com/random_joke")

• httpx.AsyncClient() によって非同期クライアントを生成し、async with を使って適切にクローズ処理を行います。
• .get(...) メソッドで外部APIにリクエストを送信しますが、これは非同期関数であるため await が必要です。
• 応答 (response) にはHTTPステータスコード、本文、ヘッダなどが含まれます。

レスポンスからJSONを取得

data = response.json()

data = response.json()

この処理は同期的に見えますが、httpx の .json() メソッドはレスポンス本文のパースだけを行うため、非同期である必要はありません。必要に応じて await response.aread() や await response.json() のように await を使うバージョンも利用可能です。

結果の整形と返却

return f"{data['setup']} - {data['punchline']}"

return f"{data['setup']} - {data['punchline']}"

取得したジョークデータから、ジョークの前振り（setup）とオチ（punchline）を文字列として結合し、Gradioに返しています。

4.4 タイムアウト・エラー処理を追加した例

外部APIを呼び出す際には、失敗時や応答遅延時への備えが重要です。以下は、タイムアウト設定と**例外処理（エラーハンドリング）**を追加した改良版です。

async def get_joke_with_timeout(_):
    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            response = await client.get("https://official-joke-api.appspot.com/random_joke")
            data = response.json()
            return f"{data['setup']} - {data['punchline']}"
    except httpx.RequestError as e:
        return f"API呼び出し中にエラーが発生しました: {str(e)}"

async def get_joke_with_timeout(_):
    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            response = await client.get("https://official-joke-api.appspot.com/random_joke")
            data = response.json()
            return f"{data['setup']} - {data['punchline']}"
    except httpx.RequestError as e:
        return f"API呼び出し中にエラーが発生しました: {str(e)}"

• timeout=5.0：5秒以内に応答がなければ httpx.TimeoutException を発生
• RequestError：DNS失敗や接続拒否など、HTTPリクエストに関する例外全般をキャッチ
• ユーザーに適切なエラーメッセージを表示できるようにしています

4.5 同期版（参考）との比較

同期的に同様の処理を行うと以下のようになります：

import requests

def get_joke_sync(_):
    response = requests.get("https://official-joke-api.appspot.com/random_joke")
    data = response.json()
    return f"{data['setup']} - {data['punchline']}"

import requests

def get_joke_sync(_):
    response = requests.get("https://official-joke-api.appspot.com/random_joke")
    data = response.json()
    return f"{data['setup']} - {data['punchline']}"

この同期版をGradioで使うと、API応答の間、他のUI操作がブロックされるというデメリットがあります。特に複数ユーザーが同時にアクセスするような状況では非効率です。

4.6 複数APIを並列呼び出ししたい場合

非同期処理の強みは「並列化」にあります。たとえば複数のAPIを同時に呼びたいときは asyncio.gather() を活用できます。

async def get_multiple_jokes(_):
    urls = [
        "https://official-joke-api.appspot.com/random_joke",
        "https://official-joke-api.appspot.com/random_joke"
    ]
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        jokes = [r.json() for r in responses]
        return "\n".join([f"{j['setup']} - {j['punchline']}" for j in jokes])

async def get_multiple_jokes(_):
    urls = [
        "https://official-joke-api.appspot.com/random_joke",
        "https://official-joke-api.appspot.com/random_joke"
    ]
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        jokes = [r.json() for r in responses]
        return "\n".join([f"{j['setup']} - {j['punchline']}" for j in jokes])

このようにすれば、2つのAPI呼び出しを同時に実行して、短時間で結果を得ることができます。

4.7 まとめ：非同期API呼び出しのポイント

• 外部API呼び出しは遅延が生じやすく、非同期処理が有効
• httpx.AsyncClient を使えば、await による非同期I/Oが簡単に実現可能
• タイムアウトや例外処理を適切に設計することで、実用性が大きく向上
• asyncio.gather() で複数のリクエストを並列化可能

GradioでAPIを呼び出すアプリを構築する場合、非同期関数の活用はパフォーマンスとユーザー体験の両立に欠かせない技術です。

5. Gradioで非同期関数を使う際の注意点と制限

Gradioは非同期関数（async def）を自然に扱える便利なインターフェースを提供していますが、非同期プログラミングそのものが持つ制約と、Gradioの内部実装による挙動の違いを理解しておくことで、より安定したアプリケーションを作ることができます。

この章では、Gradioにおける非同期関数の挙動・制限・落とし穴・回避策を具体的に解説します。

5.1 Gradioが非同期関数をどのように扱うか

Gradioは、関数が async def で定義されている場合、Pythonの asyncio イベントループを使って実行します。以下のような関数を渡した場合：

async def my_func(input_text):
    await asyncio.sleep(1)
    return input_text.upper()

async def my_func(input_text):
    await asyncio.sleep(1)
    return input_text.upper()

Gradioは内部的に次のような処理を行います：

• 呼び出し対象の関数が coroutine（async）かどうかをチェック
• 非同期関数であれば、既存の asyncio イベントループで await 実行
• 必要に応じて nest_asyncio（再帰的ループ許可）を使ってイベントループを多重に扱う（これはJupyter Notebook互換のため）

この仕組みによって、Gradioは非同期処理を「同期関数と同様に」簡単に扱えるように見せています。

5.2 Gradioで async def を使うときの制限と落とし穴

1. 並列処理には非対応（並行 ≠ 並列）

Gradioでは、非同期関数が複数同時に呼ばれても、その都度新しいタスクがイベントループ上でスケジューリングされるだけで、並列にバックグラウンドスレッドやプロセスで動作するわけではありません。

# ユーザーAとBが同時に入力した場合、同じイベントループ内で await を共有する
async def process(x):
    await asyncio.sleep(5)
    return x

# ユーザーAとBが同時に入力した場合、同じイベントループ内で await を共有する
async def process(x):
    await asyncio.sleep(5)
    return x

これは並行処理であり、CPUバウンドの並列処理とは異なります。CPUを占有する重い処理を非同期で書いても、パフォーマンス改善にはなりません。

対策：CPUバウンド処理には concurrent.futures.ThreadPoolExecutor を使って別スレッドで実行します。

import asyncio
import concurrent.futures

executor = concurrent.futures.ThreadPoolExecutor()

def cpu_heavy(x):
    # 例えば画像変換などの重い処理
    import time; time.sleep(3)
    return x.upper()

async def wrapper(x):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(executor, cpu_heavy, x)

import asyncio
import concurrent.futures

executor = concurrent.futures.ThreadPoolExecutor()

def cpu_heavy(x):
    # 例えば画像変換などの重い処理
    import time; time.sleep(3)
    return x.upper()

async def wrapper(x):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(executor, cpu_heavy, x)

2. イベントループが既に存在する環境（Jupyter）では挙動に注意

GradioはJupyter Notebookでも動作しますが、Jupyterはすでに独自のイベントループを起動しているため、通常の asyncio.run() が使えません。

この場合、Gradioは内部的に nest_asyncio を使って既存ループに非同期関数を「ネスト」しています。これにより、一部の非同期コードがうまく動かない、あるいは2回実行されるように見える現象が起こることがあります。

対策：

• Jupyter上では gr.Interface(...).launch(inline=False) を使うと、ブラウザでの起動になり問題が起きにくい
• CLI環境で動作確認するのがベスト

3. 非同期コードの例外処理を明示的に行う必要がある

非同期関数内部で例外が発生しても、明示的に try / except を書かないと、Gradio上では「出力が返らない」「無応答に見える」ことがあります。

async def error_prone(x):
    try:
        if x == "fail":
            raise ValueError("意図的エラー")
        return x
    except Exception as e:
        return f"エラーが発生しました: {str(e)}"

async def error_prone(x):
    try:
        if x == "fail":
            raise ValueError("意図的エラー")
        return x
    except Exception as e:
        return f"エラーが発生しました: {str(e)}"

Gradioは関数がエラーを返すとUIに「エラー発生」の表示をする場合もありますが、ユーザーに明確なメッセージを出したいときは、自分で try / except を書いた方が親切です。

4. 非同期関数から同期関数を呼び出すのはOK（ただし逆は注意）

Pythonでは、非同期関数から同期関数を呼び出すのは問題ありません：

def sync_func(x): return x.upper()

async def async_func(x): return sync_func(x)

def sync_func(x): return x.upper()

async def async_func(x): return sync_func(x)

しかし、同期関数から非同期関数を await せずに呼ぶことはできません。

# これはエラー
def sync_func(x):
    result = async_func(x)  # ← await できない
    return result

# これはエラー
def sync_func(x):
    result = async_func(x)  # ← await できない
    return result

Gradioは fn 引数に渡された関数が async def かどうかを判別して自動的に処理するため、明示的に await を書く必要はありません。しかし、自分で関数をラップしたりチェーンしたりする際は注意が必要です。

5.3 状態管理（gr.State）との併用時の注意

Gradioでは gr.State() を使って関数間で状態（セッション変数）を共有できますが、非同期関数内で状態を変更する際も状態の整合性を保つことが重要です。

例：

state = gr.State("")

async def update_state(x, s):
    new_val = s + x
    return new_val, new_val

state = gr.State("")

async def update_state(x, s):
    new_val = s + x
    return new_val, new_val

この場合、gr.State は引数と戻り値の両方で渡す必要があります。非同期で複数のユーザーが同時に状態を書き換えると、データ競合が発生する可能性があります。

対策：

• gr.State は「セッションごと」にスコープされており、ユーザー間では分離されている
• ただし、外部変数やグローバル変数に非同期関数からアクセスするのは危険

5.4 実用上のまとめ：開発時に意識すべきこと

5.5 補足：非同期デバッグのヒント

Pythonの非同期コードのデバッグは難しいですが、以下の方法が有効です：

• 非同期関数にも print() や logging は使える
• asyncio.get_event_loop().create_task() で手動スケジューリング可能
• asyncio.run() を使うなら一度だけ（トップレベルで）

Gradioと組み合わせたときは、「呼び出し前後で出力が出るか」「await前後で止まっていないか」などを丁寧に追いましょう。

まとめ：Gradioの非同期処理における正しい理解が安定動作の鍵

非同期関数をGradioで安全かつ効率的に使うためには、asyncioの基本的な動作の理解と、Gradio内部での扱われ方の把握が欠かせません。

特に以下の点を抑えておくと、実践でもつまずきにくくなります：

• 非同期はあくまで「I/O向け」だと認識する
• 並列処理が必要な場合はスレッドやプロセスを併用する
• イベントループの多重起動（Jupyter）に注意する
• エラーハンドリングは必ず明示的に行う

6. Gradioにおける非同期処理のベストプラクティス

Gradioで非同期関数を活用する際には、単に async def を使うだけではなく、処理の分離、スケーラビリティ、安全性、ユーザー体験まで考慮した設計が求められます。この章では、非同期処理をGradioに正しく、効率的に組み込むためのベストプラクティスを、具体例とともに紹介します。

6.1 入出力処理の粒度は小さく保つ

原則：

非同期関数の中で複数の役割（データ処理、表示、保存など）を1つにまとめないこと。

悪い例（モノリシックな非同期関数）：

async def handle_all(text):
    await save_to_db(text)
    processed = await call_api(text)
    with open("output.txt", "w") as f:
        f.write(processed)
    return processed

async def handle_all(text):
    await save_to_db(text)
    processed = await call_api(text)
    with open("output.txt", "w") as f:
        f.write(processed)
    return processed

このような関数では、API通信に失敗した場合にファイル保存も巻き込まれて失敗するなど、責任の切り分けが難しくなります。

改善例（関心の分離と明確なエラーハンドリング）：

async def save_input(text):
    await save_to_db(text)
    return text

async def process(text):
    return await call_api(text)

def write_file(text):
    with open("output.txt", "w") as f:
        f.write(text)
    return text

async def save_input(text):
    await save_to_db(text)
    return text

async def process(text):
    return await call_api(text)

def write_file(text):
    with open("output.txt", "w") as f:
        f.write(text)
    return text

Gradioでは、これらを gr.Blocks で連結して順次呼び出す設計が可能です。

6.2 非同期処理に対して明確なユーザー通知を設ける

非同期関数では結果が返るまでに時間がかかることが多いため、ユーザーには「いま処理中である」ことを明示すべきです。

Gradioでは status="in_progress" 表示や gr.Textbox(..., interactive=False) を一時的に使ってUIを制御できます。

例：ボタンの非活性化とメッセージ表示

import gradio as gr
import asyncio

async def slow_func(x):
    await asyncio.sleep(3)
    return f"完了: {x}"

with gr.Blocks() as demo:
    input_text = gr.Textbox()
    output = gr.Textbox()
    btn = gr.Button("送信")

    def disable_ui():
        return gr.update(interactive=False), gr.update(value="処理中...")

    def enable_ui(result):
        return gr.update(interactive=True), gr.update(value=result)

    btn.click(disable_ui, [], [input_text, output])  # UI無効化
    btn.click(slow_func, input_text, output)         # 非同期処理
    btn.click(enable_ui, output, [input_text, output])  # UI復元

demo.launch()

import gradio as gr
import asyncio

async def slow_func(x):
    await asyncio.sleep(3)
    return f"完了: {x}"

with gr.Blocks() as demo:
    input_text = gr.Textbox()
    output = gr.Textbox()
    btn = gr.Button("送信")

    def disable_ui():
        return gr.update(interactive=False), gr.update(value="処理中...")

    def enable_ui(result):
        return gr.update(interactive=True), gr.update(value=result)

    btn.click(disable_ui, [], [input_text, output])  # UI無効化
    btn.click(slow_func, input_text, output)         # 非同期処理
    btn.click(enable_ui, output, [input_text, output])  # UI復元

demo.launch()

この設計では、非同期関数が呼ばれる前にUI制御関数で状態を変更し、処理後に戻すことで、操作感の良いUIが実現できます。

6.3 外部API呼び出しは常にタイムアウトを設定する

非同期I/Oで最も多い失敗は「応答が返ってこないケース」。Gradio上では待ち続けてしまうため、APIクライアントには必ずタイムアウト設定を追加すべきです。

httpx の例：

import httpx

async def fetch_data():
    async with httpx.AsyncClient(timeout=5.0) as client:
        r = await client.get("https://api.example.com/data")
        r.raise_for_status()
        return r.text

import httpx

async def fetch_data():
    async with httpx.AsyncClient(timeout=5.0) as client:
        r = await client.get("https://api.example.com/data")
        r.raise_for_status()
        return r.text

また、タイムアウトやネットワーク例外に対しては、明示的に try / except で対処し、UIに返すエラー文言も整えることでユーザー体験を向上できます。

6.4 CPUバウンド処理は run_in_executor() に逃がす

画像処理、自然言語モデル、暗号化などCPUを多く使う処理は非同期にしても意味がなく、むしろイベントループをブロックしてしまいます。

その場合は、同期関数を別スレッドで実行して、非同期関数で await する形にするのがGradioとの相性も良いです。

例：画像変換を別スレッドに移譲する

from concurrent.futures import ThreadPoolExecutor
import asyncio
from PIL import Image, ImageFilter

executor = ThreadPoolExecutor()

def blur_image(path):
    image = Image.open(path)
    return image.filter(ImageFilter.BLUR)

async def process_image(path):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(executor, blur_image, path)

from concurrent.futures import ThreadPoolExecutor
import asyncio
from PIL import Image, ImageFilter

executor = ThreadPoolExecutor()

def blur_image(path):
    image = Image.open(path)
    return image.filter(ImageFilter.BLUR)

async def process_image(path):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(executor, blur_image, path)

Gradioに渡すのは async def process_image で問題ありません。

6.5 例外処理とログは必ず設ける

非同期関数では例外が握りつぶされて UI に何も表示されないことがあるため、例外ログとユーザー向けの通知は分けて処理すべきです。

例：ログとUIメッセージの分離

import logging

logging.basicConfig(level=logging.INFO)

async def safe_func(x):
    try:
        if x == "error":
            raise ValueError("テストエラー")
        return f"成功: {x}"
    except Exception as e:
        logging.exception("非同期処理中にエラー発生")
        return "エラーが発生しました。再試行してください。"

import logging

logging.basicConfig(level=logging.INFO)

async def safe_func(x):
    try:
        if x == "error":
            raise ValueError("テストエラー")
        return f"成功: {x}"
    except Exception as e:
        logging.exception("非同期処理中にエラー発生")
        return "エラーが発生しました。再試行してください。"

6.6 依存する外部リソースをテストモードに切り替える

本番のAPIやDBに毎回非同期リクエストを送るのは効率的でなく、テスト中のエラー原因にもなります。

開発中はテストモード（モック）に切り替える仕組みを導入しておきましょう。

例：APIのフェイク版

USE_FAKE = True

async def fetch_data():
    if USE_FAKE:
        await asyncio.sleep(1)
        return "フェイクデータ"
    else:
        async with httpx.AsyncClient() as client:
            r = await client.get("https://api.example.com/data")
            return r.text

USE_FAKE = True

async def fetch_data():
    if USE_FAKE:
        await asyncio.sleep(1)
        return "フェイクデータ"
    else:
        async with httpx.AsyncClient() as client:
            r = await client.get("https://api.example.com/data")
            return r.text

まとめ：非同期処理の設計における7つのチェックポイント

7. Blocksとの組み合わせによる柔軟な非同期処理

Gradioの gr.Interface は手軽で便利ですが、UIや処理の流れを細かく制御したい場合には gr.Blocks の使用が必須です。
特に非同期処理では、「UI状態の制御」「並列タスクの扱い」「処理順序の明示化」が重要となります。

この章では、gr.Blocks によって非同期処理がどのように柔軟に管理できるのかを、以下の観点から詳細に解説します。

7.1 Blocks と非同期関数の基本的な接続方法

まずは、シンプルな非同期関数の例：

import gradio as gr
import asyncio

with gr.Blocks() as demo:
    input_box = gr.Textbox(label="入力")
    output_box = gr.Textbox(label="出力")
    button = gr.Button("送信")

    async def async_reply(msg):
        await asyncio.sleep(1)
        return f"Processed: {msg}"

    button.click(fn=async_reply, inputs=input_box, outputs=output_box)

demo.launch()

import gradio as gr
import asyncio

with gr.Blocks() as demo:
    input_box = gr.Textbox(label="入力")
    output_box = gr.Textbox(label="出力")
    button = gr.Button("送信")

    async def async_reply(msg):
        await asyncio.sleep(1)
        return f"Processed: {msg}"

    button.click(fn=async_reply, inputs=input_box, outputs=output_box)

demo.launch()

この例では、次のような処理が非同期で行われます：

• ボタンがクリックされる
• async_reply() が非同期で呼ばれる（await asyncio.sleep(1)）
• 完了後、出力が output_box に表示される

Gradio は内部で asyncio イベントループを管理しているため、 async def 関数をそのまま渡しても問題ありません。

7.2 複数ステップの非同期チェーン処理

複数のステップで非同期処理を順番に行いたい場合も、Blocks はとても便利です。

例：API呼び出し → 要約 → 表示

import gradio as gr
import asyncio

async def step1(text):
    await asyncio.sleep(1)
    return f"[API] {text}"

async def step2(text):
    await asyncio.sleep(1)
    return f"[要約] {text}"

with gr.Blocks() as demo:
    inp = gr.Textbox(label="入力")
    mid = gr.Textbox(label="中間結果")
    out = gr.Textbox(label="最終出力")
    btn = gr.Button("実行")

    btn.click(step1, inp, mid)  # 1段階目
    btn.click(step2, mid, out)  # 2段階目（midを使って続ける）

demo.launch()

import gradio as gr
import asyncio

async def step1(text):
    await asyncio.sleep(1)
    return f"[API] {text}"

async def step2(text):
    await asyncio.sleep(1)
    return f"[要約] {text}"

with gr.Blocks() as demo:
    inp = gr.Textbox(label="入力")
    mid = gr.Textbox(label="中間結果")
    out = gr.Textbox(label="最終出力")
    btn = gr.Button("実行")

    btn.click(step1, inp, mid)  # 1段階目
    btn.click(step2, mid, out)  # 2段階目（midを使って続ける）

demo.launch()

このように、1つのボタンに対して複数の click() を定義し、それぞれの入力・出力を制御できます。

• 非同期関数同士の出力・入力を中間コンポーネントで橋渡し
• ステップごとのUIが可視化され、ユーザーにも分かりやすい
• 並列ではなく「順序制御」される

7.3 UIのインタラクティブな状態制御（enable/disable）

非同期処理の開始・完了に応じて、UIコンポーネントの状態（インタラクティブ・表示内容）を動的に制御するのも、Blocks なら容易です。

import gradio as gr
import asyncio

async def process(msg):
    await asyncio.sleep(2)
    return f"完了: {msg}"

with gr.Blocks() as demo:
    textbox = gr.Textbox()
    button = gr.Button("送信")
    result = gr.Textbox()

    # 実行中: 入力無効＋一時メッセージ
    def disable_ui():
        return (
            gr.update(interactive=False),
            gr.update(interactive=False),
            gr.update(value="処理中です...")
        )

    # 実行完了: 入力有効化
    def enable_ui(output):
        return (
            gr.update(interactive=True),
            gr.update(interactive=True),
            gr.update(value=output)
        )

    button.click(disable_ui, [], [textbox, button, result])
    button.click(process, textbox, result)
    button.click(enable_ui, result, [textbox, button, result])

demo.launch()

import gradio as gr
import asyncio

async def process(msg):
    await asyncio.sleep(2)
    return f"完了: {msg}"

with gr.Blocks() as demo:
    textbox = gr.Textbox()
    button = gr.Button("送信")
    result = gr.Textbox()

    # 実行中: 入力無効＋一時メッセージ
    def disable_ui():
        return (
            gr.update(interactive=False),
            gr.update(interactive=False),
            gr.update(value="処理中です...")
        )

    # 実行完了: 入力有効化
    def enable_ui(output):
        return (
            gr.update(interactive=True),
            gr.update(interactive=True),
            gr.update(value=output)
        )

    button.click(disable_ui, [], [textbox, button, result])
    button.click(process, textbox, result)
    button.click(enable_ui, result, [textbox, button, result])

demo.launch()

このように update() を使うことで、UI状態を動的に切り替える非同期ワークフローが実現できます。

7.4 並列非同期処理の実行（asyncio.gather()）

複数の独立した非同期タスクを並列に実行し、同時に結果を取得することも可能です。

import gradio as gr
import asyncio

async def fetch_1(msg):
    await asyncio.sleep(1)
    return f"1: {msg}"

async def fetch_2(msg):
    await asyncio.sleep(2)
    return f"2: {msg}"

async def process(msg):
    res1, res2 = await asyncio.gather(
        fetch_1(msg), fetch_2(msg)
    )
    return f"{res1} + {res2}"

with gr.Blocks() as demo:
    input_box = gr.Textbox()
    output_box = gr.Textbox()
    btn = gr.Button("並列処理")

    btn.click(process, input_box, output_box)

demo.launch()

import gradio as gr
import asyncio

async def fetch_1(msg):
    await asyncio.sleep(1)
    return f"1: {msg}"

async def fetch_2(msg):
    await asyncio.sleep(2)
    return f"2: {msg}"

async def process(msg):
    res1, res2 = await asyncio.gather(
        fetch_1(msg), fetch_2(msg)
    )
    return f"{res1} + {res2}"

with gr.Blocks() as demo:
    input_box = gr.Textbox()
    output_box = gr.Textbox()
    btn = gr.Button("並列処理")

    btn.click(process, input_box, output_box)

demo.launch()

• asyncio.gather() により並列実行
• Gradioの click() からも問題なく呼び出せる
• これにより待機時間の合計が最長タスクの時間になる

7.5 State を使ったデータの保持と受け渡し

gr.State() を使うと、ユーザーの入力や処理中のデータを一時的にセッション内で保持できます。

with gr.Blocks() as demo:
    state = gr.State()
    inp = gr.Textbox()
    out = gr.Textbox()
    btn1 = gr.Button("一時保存")
    btn2 = gr.Button("表示")

    def save(text):
        return text  # stateに保存

    def show(stored):
        return f"保存された: {stored}"

    btn1.click(save, inp, state)      # 入力をstateに保存
    btn2.click(show, state, out)      # stateから出力へ

demo.launch()

with gr.Blocks() as demo:
    state = gr.State()
    inp = gr.Textbox()
    out = gr.Textbox()
    btn1 = gr.Button("一時保存")
    btn2 = gr.Button("表示")

    def save(text):
        return text  # stateに保存

    def show(stored):
        return f"保存された: {stored}"

    btn1.click(save, inp, state)      # 入力をstateに保存
    btn2.click(show, state, out)      # stateから出力へ

demo.launch()

これにより、一連の非同期処理の中で状態を共有し、段階的にUIへ反映できます。

7.6 イベントの柔軟なバインディング

Blocks では Textbox.change(), Dropdown.select(), Button.click() など、イベント単位で関数をバインドできます。

例：リアルタイム更新と非同期の組み合わせ

async def auto_reply(msg):
    await asyncio.sleep(0.5)
    return f"Auto: {msg}"

inp.change(auto_reply, inp, out)  # 入力変更時に即非同期処理

async def auto_reply(msg):
    await asyncio.sleep(0.5)
    return f"Auto: {msg}"

inp.change(auto_reply, inp, out)  # 入力変更時に即非同期処理

まとめ：Blocks + 非同期処理の設計戦略

gr.Blocks を用いた非同期処理設計は、UIとロジックの結合度を下げつつ、柔軟かつ拡張性の高いアプリ構築に非常に有効です。
この構造をテンプレート化すれば、Gradioを使った本格的な非同期Webアプリのベースとしても活用できます。

8. まとめ

Gradioを活用してWebインターフェースを構築する際、関数の入出力の型と非同期処理の扱いを正しく理解することは非常に重要です。

• inputs と outputs で関数の引数と戻り値をUIに紐づけられる
• async def を使うことで、外部APIや時間のかかる処理にもスムーズに対応可能
• gr.Blocks を使えば、より複雑なUIと非同期処理の連携も実現できる

これらの知識を活かすことで、Gradioによるアプリ開発が一段と効率的かつ高品質になります。

参考リンク

• Gradio公式ドキュメント
  https://www.gradio.app/
• Gradio GitHub Discussions（Async対応）
  https://github.com/gradio-app/gradio/discussions
• httpxライブラリ
  https://www.python-httpx.org/

最後まで読んでいただきありがとうございます。
ご意見、ご感想はコメント欄までお願いします。