👋

ローカルLLMのストリームAPI

2025/04/10に公開

ローカルLLMでAPIを作成するPythonコードをご紹介しようと思います。
今回は通常の出力及びストリーム出力の2つをご紹介します。

最後にコード全文を載せますので、ご参照ください。

通常の出力

fastapiを使用してAPIを作成します。

サーバー

まずはサーバー側のコードから見ていきましょう。

from fastapi import FastAPI
from fastapi.responses import PlainTextResponse
from pydantic import BaseModel
...
@app.post("/call")
async def call_llm_api(item: Item):
    return PlainTextResponse(content=call_llm(tokenizer, model, item.data))

fastapiを利用して、POSTリクエストを受け取り、ローカルLLMの出力を返すAPIです。

クライアント

次にクライアント側のコードを見ていきましょう。postメソッドを使用して、サーバーにリクエストを送信します。

import httpx
...
with httpx.Client() as client:
    response = client.post(url_call, json=data, timeout=timeout)
    print(response.text)

ストリーム出力

次にストリーム出力の場合を見ていきましょう。
ストリーム出力にはServer-Sent Events(SSE)を使用します。
Server-Sent Events(SSE)は、Webブラウザーとサーバー間の一方向の非同期通信方法です。頻繁なリクエスト送信を抑えることができるため、ストリーミング処理に向いています。

サーバー

先ほどとコードはほとんど変わりません。
StreamingResponseを使用すること、media_typeをtext/event-streamにすることで、ストリーム出力が可能になります。

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
...
async def event_stream(items: Item):
    streamer = stream_response_generator(tokenizer, model, items)
    for new_text in streamer:
        if new_text != "":
            yield f"data: {new_text}\n\n"

@app.post("/stream", response_class=StreamingResponse)
async def stream_items(item: Item):
    return StreamingResponse(event_stream(item.data), media_type="text/event-stream")

クライアント

クライアント側のコードもほとんど代わりません。streamを使用して、サーバーにリクエストを送信します。

import httpx
...
with httpx.Client() as client:
    # POSTリクエストを送信
    with client.stream("POST", url_stream, json=data, timeout=timeout) as response:
        # レスポンスのステータスコードを確認
        if response.status_code == 200:
            # ストリーミングされたレスポンスを逐次処理
            for line in response.iter_lines():
                if line:
                    # SSE形式のデータを解析
                    if line.startswith("data: "):
                        event_data = line[6:]
                        print(f"受信したイベント: {event_data}")
        else:
            print(f"エラー: {response.status_code}")

コード全文

最後にコード全文を載せます。