OllamaのAPIをPython asyncioで並列呼び出しする方法｜httpxとasyncio.gatherで複数プロンプトを同時処理してスループットを向上させる手順

この記事のポイント

・OLLAMA_NUM_PARALLELを設定し、httpx + asyncioで複数プロンプトを同時処理するPythonコードを解説
・asyncio.gatherで複数タスクを一斉実行し、完了順にレスポンスを回収する実装パターンを紹介
・asyncio.Semaphoreで同時接続数を制限し、Ollamaのリソースを超えない安全な呼び出しを実現
・社内ドキュメントの一括要約スクリプトを実装例として提示し、すぐに実務へ応用できる構成にする

# override.confを開くコマンド
sudo systemctl edit ollama

[Service]
Environment="OLLAMA_NUM_PARALLEL=4"

# 設定を反映する
sudo systemctl daemon-reload
sudo systemctl restart ollama

# 設定確認（Environmentに反映されているか）
sudo systemctl show ollama | grep -i parallel
# Environment=OLLAMA_NUM_PARALLEL=4

# 仮想環境の作成
python3 -m venv ~/ollama-async-env
source ~/ollama-async-env/bin/activate

# httpxのインストール（asyncioは標準ライブラリのため不要）
pip install httpx

# APIが応答するか確認
curl -s http://localhost:11434/api/tags | python3 -m json.tool | head -5
# {"models": [{"name": "mistral:7b-instruct-q4_0", ...}]} が返れば正常

import asyncio
import httpx
import json

OLLAMA_URL = "http://localhost:11434/api/generate"
MODEL = "mistral:7b-instruct-q4_0"

async def generate(prompt: str, client: httpx.AsyncClient) -> str:
    payload = {
        "model": MODEL,
        "prompt": prompt,
        "stream": False
    }
    response = await client.post(OLLAMA_URL, json=payload, timeout=120.0)
    return json.loads(response.text)["response"]

async def main():
    async with httpx.AsyncClient() as client:
        result = await generate("Linuxのcronとは何ですか？一文で説明してください。", client)
        print(result)

asyncio.run(main())

python3 ollama_async_basic.py
# cronはLinuxのジョブスケジューラーで、指定した日時や周期でコマンドを自動実行する仕組みです。

import asyncio
import httpx
import json
import time

OLLAMA_URL = "http://localhost:11434/api/generate"
MODEL = "mistral:7b-instruct-q4_0"

async def generate(prompt: str, idx: int, client: httpx.AsyncClient) -> tuple[int, str]:
    payload = {"model": MODEL, "prompt": prompt, "stream": False}
    response = await client.post(OLLAMA_URL, json=payload, timeout=120.0)
    return idx, json.loads(response.text)["response"]

async def main():
    prompts = [
        "Linuxのsystemdを一文で説明してください。",
        "Linuxのiptablesを一文で説明してください。",
        "Linuxのcronを一文で説明してください。",
        "Linuxのsshを一文で説明してください。",
    ]
    async with httpx.AsyncClient() as client:
        start = time.time()
        tasks = [generate(p, i, client) for i, p in enumerate(prompts)]
        results = await asyncio.gather(*tasks)
        elapsed = time.time() - start

    for idx, answer in sorted(results):
        print(f"[{idx}] {answer[:60]}")
    print(f"\n並列処理時間: {elapsed:.1f}秒")

asyncio.run(main())

python3 ollama_gather.py
# [0] Linuxのsystemdは、サービス・プロセス・デバイスを一元管理するinitシステムで
# [1] Linuxのiptablesは、パケットフィルタリングを行うファイアウォールツールです。
# [2] Linuxのcronは、指定した日時・周期でコマンドを自動実行するスケジューラです。
# [3] Linuxのsshは、暗号化されたチャネルでリモートマシンに安全に接続するプロトコル
#
# 並列処理時間: 11.3秒
# ※逐次処理（forループで順番に実行）の場合は約38秒

import asyncio
import httpx
import json

OLLAMA_URL = "http://localhost:11434/api/generate"
MODEL = "mistral:7b-instruct-q4_0"
MAX_CONCURRENT = 4  # OLLAMA_NUM_PARALLELと揃える

async def generate_with_semaphore(
    prompt: str,
    idx: int,
    client: httpx.AsyncClient,
    sem: asyncio.Semaphore
) -> tuple[int, str]:
    async with sem:  # ここでMAX_CONCURRENT以上の同時実行をブロック
        payload = {"model": MODEL, "prompt": prompt, "stream": False}
        response = await client.post(OLLAMA_URL, json=payload, timeout=120.0)
        return idx, json.loads(response.text)["response"]

async def run_parallel(prompts: list[str]) -> list[str]:
    sem = asyncio.Semaphore(MAX_CONCURRENT)
    async with httpx.AsyncClient() as client:
        tasks = [
            generate_with_semaphore(p, i, client, sem)
            for i, p in enumerate(prompts)
        ]
        results = await asyncio.gather(*tasks)
    return [answer for _, answer in sorted(results)]

async def generate_with_semaphore(
    prompt: str,
    idx: int,
    client: httpx.AsyncClient,
    sem: asyncio.Semaphore
) -> tuple[int, str]:
    async with sem:
        try:
            payload = {"model": MODEL, "prompt": prompt, "stream": False}
            response = await client.post(OLLAMA_URL, json=payload, timeout=120.0)
            response.raise_for_status()
            return idx, json.loads(response.text)["response"]
        except httpx.TimeoutException:
            return idx, "ERROR: タイムアウト"
        except httpx.HTTPStatusError as e:
            return idx, f"ERROR: HTTP {e.response.status_code}"
        except Exception as e:
            return idx, f"ERROR: {e}"

#!/usr/bin/env python3
"""社内ドキュメント一括要約スクリプト"""
import asyncio
import httpx
import json
import sys
from pathlib import Path

OLLAMA_URL = "http://localhost:11434/api/generate"
MODEL = "mistral:7b-instruct-q4_0"
MAX_CONCURRENT = 4
SUMMARY_PROMPT = "以下の文章を3行以内で簡潔に要約してください。\n\n---\n{text}"

async def summarize(
    text: str, idx: int, fname: str,
    client: httpx.AsyncClient,
    sem: asyncio.Semaphore
) -> dict:
    async with sem:
        prompt = SUMMARY_PROMPT.format(text=text[:3000])
        payload = {"model": MODEL, "prompt": prompt, "stream": False}
        try:
            r = await client.post(OLLAMA_URL, json=payload, timeout=180.0)
            r.raise_for_status()
            summary = json.loads(r.text)["response"].strip()
        except Exception as e:
            summary = f"ERROR: {e}"
        return {"idx": idx, "file": fname, "summary": summary}

async def main(input_dir: str):
    files = list(Path(input_dir).glob("*.txt"))
    if not files:
        print(f"対象ファイルなし: {input_dir}")
        return
    texts = [(f.name, f.read_text(encoding="utf-8")) for f in files]
    sem = asyncio.Semaphore(MAX_CONCURRENT)
    async with httpx.AsyncClient() as client:
        tasks = [summarize(t, i, n, client, sem)
                 for i, (n, t) in enumerate(texts)]
        results = await asyncio.gather(*tasks)
    for r in sorted(results, key=lambda x: x["idx"]):
        print(f"\n【{r['file']}】\n{r['summary']}")

if __name__ == "__main__":
    asyncio.run(main(sys.argv[1] if len(sys.argv) > 1 else "."))

# /docs配下のtxtファイルを一括要約
python3 bulk_summarize.py /docs

# 実行例の出力
【incident_report_20260901.txt】
2026年9月1日にWebサーバーが高負荷でダウン。原因はcronジョブの二重起動。
flockコマンドと冪等化スクリプトで再発防止措置を実施した。

【meeting_20260905.txt】
9月5日の定例会議で四半期KPIを確認。
コンバージョン率改善施策3件を翌週までに設計チームが提案する。

ローカルLLMの並列処理と業務バッチ自動化を2日間のハンズオンで習得する

asyncioとOllamaを組み合わせた並列推論の設計は、実際に動く環境で試してこそ定着する。実機GPU環境で手を動かしながら習得したい方向けに、「ローカルAIマスターセミナー」を開催しています。
少人数（最大8名）ZOOMハンズオン形式で実施しています。