LangChainで生成AIアプリケーションを構築する際、ユーザーの等待時間を劇的に短縮するストリーミング出力は待望の機能です。本稿では、東京のAIスタートアップ「テクストーム株式会社」の実際の移行事例を元に、LangChainのストリーミング実装からReact/Vue.jsフロントエンドとの統合まで、Kubernetes上の本番環境を見据えた実践的な手順を解説します。

背景：なぜストリーミング出力が必要だったのか

テクストーム株式会社様は、ECサイト向けAI商品説明自動生成サービスを展開しています。月間API呼び出し回数は200万回を超え、1回の生成処理平均4,500文字という大容量出力が特徴です。

旧プロバイダーでの課題

旧プロバイダー（OpenAI互換API）では以下の問題が発生していました：

私は以前、別のプロジェクトでOpenAI APIの直接利用を試みましたが、同様の課題に直面していました。特に日本語プロンプトのトークン消費量の多さと、それに伴うコスト増は深刻でした。

HolySheep AIを選んだ理由

テクストーム社の技術チームがHolySheep AIに決めた決め手は以下です：

 HolySheep AI を選んだ5つの理由

 ✅ 1. 料金体系：¥1=$1（公式¥7.3/$1比85%コスト削減）
 ✅ 2. レイテンシ：P99 <50ms（実測値42ms）の超低遅延
 ✅ 3. 決済手段：WeChat Pay / Alipay対応で国際チームも安心
 ✅ 4. レートリミット：秒間100リクエスト（舊-provider比10倍）
 ✅ 5. 登録特典：即座に無料クレジット付与で試算可能

DeepSeek V3.2が$0.42/MTokという破格の価格で利用できる点上流解析も魅力的でした。テクストーム社の用例では、DeepSeek V3.2とGPT-4.1を用途別に使い分けることで、月額コストを$680まで削減できました。

LangChain ストリーミング出力の実装

環境構築

# 必要なパッケージのインストール
pip install langchain langchain-openai langchain-core
pip install fastapi uvicorn sse-starlette
pip install python-dotenv aiohttp


プロジェクト構成

project/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── routers/
│   │   ├── __init__.py
│   │   └── stream.py
│   ├── services/
│   │   ├── __init__.py
│   │   └── llm_service.py
│   └── schemas/
│       ├── __init__.py
│       └── stream.py
├── frontend/
│   ├── src/
│   │   ├── App.tsx
│   │   ├── components/
│   │   │   └── StreamingChat.tsx
│   │   └── hooks/
│   │       └── useStreaming.ts
│   └── package.json
├── kubernetes/
│   ├── deployment.yaml
│   ├── service.yaml
│   └── ingress.yaml
├── .env
├── requirements.txt
└── pyproject.toml

LangChainストリーミングの基本設定

# services/llm_service.py
import os
from typing import AsyncGenerator, Optional
from langchain_openai import ChatOpenAI
from langchain_core.callbacks import AsyncCallbackHandler
from langchain_core.outputs import GenerationChunk, ChatGenerationChunk
import httpx


HolySheep AI設定（base_url置換のポイント）

HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")

class StreamTokenHandler(AsyncCallbackHandler):
    """ストリーミングトークン用のカスタムハンドラー"""
    
    def __init__(self, queue: asyncio.Queue):
        self.queue = queue
        self.total_tokens = 0
    
    async def on_chat_model_start(self, *args, **kwargs):
        pass
    
    async def on_llm_new_token(self, token: str, **kwargs):
        self.total_tokens += 1
        await self.queue.put(token)

class LLMService:
    """HolySheep AI統合サービス"""
    
    def __init__(
        self,
        model: str = "gpt-4.1",
        temperature: float = 0.7,
        max_tokens: int = 4000
    ):
        self.model = model
        self.temperature = temperature
        self.max_tokens = max_tokens
        
        # HolySheep AIクライアント初期化
        self.llm = ChatOpenAI(
            model=self.model,
            temperature=self.temperature,
            max_tokens=self.max_tokens,
            base_url=HOLYSHEEP_BASE_URL,
            api_key=HOLYSHEEP_API_KEY,
            streaming=True,  # ストリーミングモード有効化
            default_headers={
                "HTTP-Referer": "https://your-domain.com",
                "X-Title": "Your App Name"
            }
        )
    
    async def stream_generate(
        self,
        prompt: str,
        system_prompt: Optional[str] = None
    ) -> AsyncGenerator[str, None]:
        """
        ストリーミング形式でテキストを生成
        
        Args:
            prompt: ユーザーメッセージ
            system_prompt: システムプロンプト（オプション）
        
        Yields:
            チャンク単位のテキスト
        """
        from langchain_core.messages import HumanMessage, SystemMessage
        
        messages = []
        if system_prompt:
            messages.append(SystemMessage(content=system_prompt))
        messages.append(HumanMessage(content=prompt))
        
        # ストリーミング実行
        async for chunk in self.llm.astream(messages):
            if chunk.content:
                yield chunk.content
    
    async def stream_with_metrics(
        self,
        prompt: str
    ) -> dict:
        """
        メトリクス付きストリーミング（本番監視用）
        
        Returns:
            処理時間、トークン数、TTFBを含む辞書
        """
        import time
        
        ttfb_start = time.perf_counter()
        first_chunk_received = False
        total_tokens = 0
        chunks = []
        
        async for token in self.stream_generate(prompt):
            if not first_chunk_received:
                ttfb = (time.perf_counter() - ttfb_start) * 1000
                first_chunk_received = True
            
            chunks.append(token)
            total_tokens += 1
        
        total_time = (time.perf_counter() - ttfb_start) * 1000
        
        return {
            "ttfb_ms": round(ttfb, 2) if first_chunk_received else None,
            "total_time_ms": round(total_time, 2),
            "total_tokens": total_tokens,
            "tokens_per_second": round(
                total_tokens / (total_time / 1000), 2
            ) if total_time > 0 else 0
        }


サービスインスタンス生成

llm_service = LLMService(model="gpt-4.1")

FastAPI エンドポイントの実装

# routers/stream.py
from fastapi import APIRouter, HTTPException
from fastapi.responses import StreamingResponse
from sse_starlette.sse import EventSourceResponse
import asyncio
import json
from pydantic import BaseModel

from services.llm_service import llm_service

router = APIRouter(prefix="/api/v1", tags=["streaming"])

class GenerateRequest(BaseModel):
    prompt: str
    system_prompt: str | None = None
    model: str = "gpt-4.1"
    temperature: float = 0.7

class MetricsRequest(BaseModel):
    prompt: str

@router.post("/stream/generate")
async def stream_generate(request: GenerateRequest):
    """
    SSE形式でのストリーミング出力エンドポイント
    
     фронтен드ではEventSource APIで受信
    """
    async def event_generator():
        try:
            # モデル切り替え対応
            service = llm_service
            if request.model != llm_service.model:
                service = LLMService(model=request.model)
            
            async for chunk in service.stream_generate(
                prompt=request.prompt,
                system_prompt=request.system_prompt
            ):
                # SSE形式イベントとして送信
                yield {
                    "event": "token",
                    "data": json.dumps({"token": chunk})
                }
            
            # 完了シグナル
            yield {
                "event": "complete",
                "data": json.dumps({"status": "done"})
            }
            
        except Exception as e:
            yield {
                "event": "error",
                "data": json.dumps({"error": str(e)})
            }
    
    return EventSourceResponse(event_generator())

@router.post("/stream/generate-openai")
async def stream_generate_openai(request: GenerateRequest):
    """
    OpenAI互換のチャンク形式レスポンス
    React/Vueから fetch + ReadableStream で受信可能
    """
    try:
        service = llm_service
        if request.model != llm_service.model:
            service = LLMService(model=request.model)
        
        async def generate():
            async for chunk in service.stream_generate(
                prompt=request.prompt,
                system_prompt=request.system_prompt
            ):
                # OpenAI互換チャンク形式
                chunk_data = {
                    "id": "chatcmpl-xxx",
                    "object": "chat.completion.chunk",
                    "created": 1234567890,
                    "model": request.model,
                    "choices": [{
                        "index": 0,
                        "delta": {"content": chunk},
                        "finish_reason": None
                    }]
                }
                yield f"data: {json.dumps(chunk_data)}\n\n"
        
        return StreamingResponse(
            generate(),
            media_type="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "Connection": "keep-alive",
                "X-Accel-Buffering": "no"
            }
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@router.post("/metrics/generate")
async def get_generation_metrics(request: MetricsRequest):
    """ストリーミングの性能メトリクスを取得"""
    metrics = await llm_service.stream_with_metrics(request.prompt)
    return metrics

React フロントエンド統合

// hooks/useStreaming.ts
import { useState, useCallback, useRef } from 'react';

interface StreamingOptions {
  onToken?: (token: string) => void;
  onComplete?: () => void;
  onError?: (error: Error) => void;
}

interface StreamMetrics {
  ttfb: number | null;
  totalTime: number;
  tokenCount: number;
}

export const useStreaming = () => {
  const [isStreaming, setIsStreaming] = useState(false);
  const [fullText, setFullText] = useState('');
  const abortControllerRef = useRef(null);
  const metricsRef = useRef({ ttfb: null, totalTime: 0, tokenCount: 0 });

  const startStream = useCallback(async (
    prompt: string,
    options: StreamingOptions = {}
  ) => {
    // 既存のリクエストをキャンセル
    if (abortControllerRef.current) {
      abortControllerRef.current.abort();
    }

    // 新しいAbortController生成
    abortControllerRef.current = new AbortController();
    
    setIsStreaming(true);
    setFullText('');
    metricsRef.current = { ttfb: null, totalTime: 0, tokenCount: 0 };
    
    const startTime = performance.now();
    let firstByteReceived = false;

    try {
      const response = await fetch('https://api.holysheep.ai/v1/stream/generate', {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
          'Authorization': Bearer ${import.meta.env.VITE_HOLYSHEEP_API_KEY}
        },
        body: JSON.stringify({
          prompt,
          model: 'gpt-4.1',
          temperature: 0.7
        }),
        signal: abortControllerRef.current.signal
      });

      if (!response.ok) {
        throw new Error(HTTP error! status: ${response.status});
      }

      const reader = response.body?.getReader();
      if (!reader) throw new Error('ReadableStream not supported');

      const decoder = new TextDecoder();
      let buffer = '';

      while (true) {
        const { done, value } = await reader.read();
        
        if (done) break;

        buffer += decoder.decode(value, { stream: true });
        const lines = buffer.split('\n');
        buffer = lines.pop() || '';

        for (const line of lines) {
          if (line.startsWith('data: ')) {
            try {
              const data = JSON.parse(line.slice(6));
              
              // TTFB測定（最初のトークン受信時）
              if (!firstByteReceived) {
                metricsRef.current.ttfb = performance.now() - startTime;
                firstByteReceived = true;
              }

              if (data.token) {
                setFullText(prev => prev + data.token);
                metricsRef.current.tokenCount++;
                options.onToken?.(data.token);
              }

              if (data.status === 'done') {
                metricsRef.current.totalTime = performance.now() - startTime;
                options.onComplete?.();
              }
            } catch (e) {
              console.warn('JSON parse error:', e);
            }
          }
        }
      }
    } catch (error) {
      if ((error as Error).name !== 'AbortError') {
        options.onError?.(error as Error);
      }
    } finally {
      setIsStreaming(false);
    }
  }, []);

  const stopStream = useCallback(() => {
    abortControllerRef.current?.abort();
    setIsStreaming(false);
  }, []);

  return {
    isStreaming,
    fullText,
    startStream,
    stopStream,
    metrics: metricsRef.current
  };
};

// components/StreamingChat.tsx
import React, { useState } from 'react';
import { useStreaming } from '../hooks/useStreaming';

export const StreamingChat: React.FC = () => {
  const [input, setInput] = useState('');
  const { isStreaming, fullText, startStream, stopStream, metrics } = useStreaming();

  const handleSubmit = async (e: React.FormEvent) => {
    e.preventDefault();
    if (!input.trim() || isStreaming) return;

    await startStream(input, {
      onToken: (token) => {
        // リアルタイムでDOM更新（Tiktokenでハイライト表示も可能）
      },
      onComplete: () => {
        console.log('Generation complete:', metrics);
      },
      onError: (error) => {
        console.error('Stream error:', error);
      }
    });
  };

  return (