この実踐的なガイドでは、アプリケーション用に LLM が組み込まれた拡張性の高いモデル展開ソリューションを作成する方法を?qū)Wびます。
例では、Hugging Face の ChatGPT2 モデルを使用しますが、ChatGPT4、Claude などの他のモデルを簡(jiǎn)単にプラグインできます。
AI 機(jī)能を備えた新しいアプリケーションを設(shè)計(jì)している場(chǎng)合でも、既存の AI システムを改善している場(chǎng)合でも、このガイドは、強(qiáng)力な LLM 統(tǒng)合を作成するためのステップバイステップに役立ちます。

LLM 統(tǒng)合の基礎(chǔ)を理解する

コードを書き始める前に、本番環(huán)境の LLM 統(tǒng)合を構(gòu)築するために何が必要かを理解しましょう。本番環(huán)境に対応した LLM 統(tǒng)合を構(gòu)築するときに考慮する必要があるのは API 呼び出しだけではなく、信頼性、コスト、安??定性なども考慮する必要があります。運(yùn)用アプリケーションは、コストを管理しながら、サービスの停止、レート制限、応答時(shí)間の変動(dòng)などの問題に対処する必要があります。
私たちが一緒に構(gòu)築するものは次のとおりです:

• 障害を適切に処理する堅(jiān)牢な API クライアント
• コストと速度を最適化するスマート キャッシング システム
• 適切かつ迅速な管理體制
• 包括的なエラー処理と監(jiān)視
• サンプル プロジェクトとしての完全なコンテンツ モデレーション システム

前提條件

コーディングを始める前に、次のものが揃っていることを確認(rèn)してください。

• マシンに Python 3.8 以降がインストールされている
• Redis クラウド アカウントまたはローカルにインストールされている
• Python プログラミングの基本的な知識(shí)
• REST API の基本的な理解
• Hugging Face API キー (またはその他の LLM プロバイダー キー)

フォローしてみませんか?完全なコードは GitHub リポジトリで入手できます。

開発環(huán)境のセットアップ

開発環(huán)境を準(zhǔn)備することから始めましょう。クリーンなプロジェクト構(gòu)造を作成し、必要なパッケージをすべてインストールします。

まず、プロジェクト ディレクトリを作成し、Python 仮想環(huán)境をセットアップしましょう。ターミナルを開いて次を?qū)g行します:

mkdir llm_integration && cd llm_integration
python3 -m venv env
syource env/bin/activate

次に、プロジェクトの依存関係を設(shè)定しましょう。次の必須パッケージを含む新しいrequirements.txtファイルを作成します:

transformers==4.36.0
huggingface-hub==0.19.4
redis==4.6.0
pydantic==2.5.0
pydantic-settings==2.1.0
tenacity==8.2.3
python-dotenv==1.0.0
fastapi==0.104.1
uvicorn==0.24.0
torch==2.1.0
numpy==1.24.3

これらの各パッケージが必要な理由を詳しく見てみましょう:

• トランスフォーマー: これは、Qwen2.5-Coder モデルとのインターフェースに使用する Hugging Face の強(qiáng)力なライブラリです。
• ハギングフェイスハブ: モデルの読み込みとバージョン管理を処理できるようにします。 redis: リクエスト キャッシュの実裝用
• pydantic: データの検証と設(shè)定に使用されます。
• Tenacity: 信頼性を高めるための再試行機(jī)能を擔(dān)當(dāng)します
• python-dotenv: 環(huán)境変數(shù)のロード用
• fastapi: 少量のコードで API エンドポイントを構(gòu)築します
• uvicorn: FastAPI アプリケーションを効率的に実行するために使用されます
• torch: 変圧器モデルの実行と機(jī)械學(xué)習(xí)操作の処理用
• numpy: 數(shù)値計(jì)算に使用されます。

次のコマンドを使用してすべてのパッケージをインストールします:

mkdir llm_integration && cd llm_integration
python3 -m venv env
syource env/bin/activate

きれいな構(gòu)造でプロジェクトを整理しましょう。プロジェクト ディレクトリに次のディレクトリとファイルを作成します:

transformers==4.36.0
huggingface-hub==0.19.4
redis==4.6.0
pydantic==2.5.0
pydantic-settings==2.1.0
tenacity==8.2.3
python-dotenv==1.0.0
fastapi==0.104.1
uvicorn==0.24.0
torch==2.1.0
numpy==1.24.3

LLM クライアントの構(gòu)築

アプリケーションの最も重要なコンポーネントである LLM クライアントから始めましょう。ここで、ChatGPT モデル (またはその他の任意の LLM) を操作します。次のコード スニペットを core/llm_client.py ファイルに追加します:

pip install -r requirements.txt

LLMClient クラスの最初の部分では、基盤をセットアップします。

• モデルをロードするために、トランスフォーマー ライブラリの AutoModelForCausalLM と AutoTokenizer を使用しています
• device_map="auto" パラメーターは GPU/CPU 割り當(dāng)てを自動(dòng)的に処理します
• 良好なパフォーマンスを維持しながらメモリ使用量を最適化するために torch.float16 を使用しています

次に、モデルと通信するメソッドを追加しましょう。

llm_integration/
├── core/
│   ├── llm_client.py      # your main LLM interaction code
│   ├── prompt_manager.py  # Handles prompt templates
│   └── response_handler.py # Processes LLM responses
├── cache/
│   └── redis_manager.py   # Manages your caching system
├── config/
│   └── settings.py        # Configuration management
├── api/
│   └── routes.py          # API endpoints
├── utils/
│   ├── monitoring.py      # Usage tracking
│   └── rate_limiter.py    # Rate limiting logic
├── requirements.txt
└── main.py
└── usage_logs.json       

この完了メソッドで何が起こっているのかを詳しく見てみましょう:

• 一時(shí)的な障害に対処するために @retry デコレータ メソッドを追加しました。
• torch.no_grad() コンテキスト マネージャーを使用して、勾配計(jì)算を無効にしてメモリを節(jié)約しました。
• 入力と出力の両方でトークンの使用狀況を追跡します。これはコスト計(jì)算に非常に重要です。
• 応答と使用統(tǒng)計(jì)を含む構(gòu)造化辭書を返します。

LLM 応答ハンドラーの作成

次に、LLM の生の出力を解析して構(gòu)造化するための応答ハンドラーを追加する必要があります。これを core/response_handler.py ファイルで次のコード スニペットを使用して実行します:

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from tenacity import retry, stop_after_attempt, wait_exponential
from typing import Dict, Optional
import logging

class LLMClient:
    def __init__(self, model_name: str = "gpt2", timeout: int = 30):
        try:
            self.tokenizer = AutoTokenizer.from_pretrained(model_name)
            self.model = AutoModelForCausalLM.from_pretrained(
                model_name,
                device_map="auto",
                torch_dtype=torch.float16
            )
        except Exception as e:
            logging.error(f"Error loading model: {str(e)}")
            # Fallback to a simpler model if the specified one fails
            self.tokenizer = AutoTokenizer.from_pretrained("gpt2")
            self.model = AutoModelForCausalLM.from_pretrained("gpt2")

        self.timeout = timeout
        self.logger = logging.getLogger(__name__)

堅(jiān)牢なキャッシュ システムの追加

次に、アプリケーションのパフォーマンスを向上させ、コストを削減するためのキャッシュ システムを作成しましょう。次のコード スニペットを、cache/redis_manager.py ファイルに追加します:

 @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=10),
        reraise=True
    )
    async def complete(self, 
                      prompt: str, 
                      temperature: float = 0.7,
                      max_tokens: Optional[int] = None) -> Dict:
        """Get completion from the model with automatic retries"""
        try:
            inputs = self.tokenizer(prompt, return_tensors="pt").to(
                self.model.device
            )

            with torch.no_grad():
                outputs = self.model.generate(
                    **inputs,
                    max_new_tokens=max_tokens or 100,
                    temperature=temperature,
                    do_sample=True
                )

            response_text = self.tokenizer.decode(
                outputs[0], 
                skip_special_tokens=True
            )

            # Calculate token usage for monitoring
            input_tokens = len(inputs.input_ids[0])
            output_tokens = len(outputs[0]) - input_tokens

            return {
                'content': response_text,
                'usage': {
                    'prompt_tokens': input_tokens,
                    'completion_tokens': output_tokens,
                    'total_tokens': input_tokens + output_tokens
                },
                'model': "gpt2"
            }

        except Exception as e:
            self.logger.error(f"Error in LLM completion: {str(e)}")
            raise

上記のコード スニペットでは、次のようにすべてのキャッシュ操作を処理する CacheManager クラスを作成しました。

• _generate_key メソッド。プロンプトとパラメーターに基づいて一意のキャッシュ キーを作成します
• get_cached_response は、指定されたプロンプトに対するキャッシュされた応答があるかどうかを確認(rèn)します
• 成功した応答を?qū)恧问褂盲韦郡幛吮４妞工隿ache_response

スマート プロンプト マネージャーの作成

LLM モデルのプロンプトを管理するプロンプト マネージャーを作成しましょう。次のコードを core/prompt_manager.py に追加します:

mkdir llm_integration && cd llm_integration
python3 -m venv env
syource env/bin/activate

次に、コード スニペットを使用して、prompts/content_moderation.json ファイルにコンテンツ モデレーション用のサンプル プロンプト テンプレートを作成します。

transformers==4.36.0
huggingface-hub==0.19.4
redis==4.6.0
pydantic==2.5.0
pydantic-settings==2.1.0
tenacity==8.2.3
python-dotenv==1.0.0
fastapi==0.104.1
uvicorn==0.24.0
torch==2.1.0
numpy==1.24.3

これで、プロンプト マネージャーは JSON ファイルからプロンプト テンプレートをロードし、フォーマットされたプロンプト テンプレートも取得できるようになります。

構(gòu)成マネージャーのセットアップ

すべての LLM 構(gòu)成を 1 か所に保管し、アプリケーション全體で簡(jiǎn)単に再利用するには、構(gòu)成設(shè)定を作成しましょう。以下のコードを config/settings.py ファイルに追加します:

pip install -r requirements.txt

レート制限の実裝

次に、レート制限を?qū)g裝して、ユーザーがアプリケーションのリソースにアクセスする方法を制御しましょう。これを行うには、次のコードを utils/rate_limiter.py ファイルに追加します。

llm_integration/
├── core/
│   ├── llm_client.py      # your main LLM interaction code
│   ├── prompt_manager.py  # Handles prompt templates
│   └── response_handler.py # Processes LLM responses
├── cache/
│   └── redis_manager.py   # Manages your caching system
├── config/
│   └── settings.py        # Configuration management
├── api/
│   └── routes.py          # API endpoints
├── utils/
│   ├── monitoring.py      # Usage tracking
│   └── rate_limiter.py    # Rate limiting logic
├── requirements.txt
└── main.py
└── usage_logs.json       

RateLimiter では、一定期間內(nèi)に各ユーザーに許可されるリクエストの期間と數(shù)を渡すだけでレート制限を処理するために、任意のルートで使用できる再実行可能な check_rate_limit メソッドを?qū)g裝しました。

API エンドポイントの作成

次に、API エンドポイントを api/routes.py ファイルに作成して、アプリケーションに LLM を統(tǒng)合しましょう。

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from tenacity import retry, stop_after_attempt, wait_exponential
from typing import Dict, Optional
import logging

class LLMClient:
    def __init__(self, model_name: str = "gpt2", timeout: int = 30):
        try:
            self.tokenizer = AutoTokenizer.from_pretrained(model_name)
            self.model = AutoModelForCausalLM.from_pretrained(
                model_name,
                device_map="auto",
                torch_dtype=torch.float16
            )
        except Exception as e:
            logging.error(f"Error loading model: {str(e)}")
            # Fallback to a simpler model if the specified one fails
            self.tokenizer = AutoTokenizer.from_pretrained("gpt2")
            self.model = AutoModelForCausalLM.from_pretrained("gpt2")

        self.timeout = timeout
        self.logger = logging.getLogger(__name__)

ここでは、API ルートの編成を擔(dān)當(dāng)する APIRouter クラスに /moderate エンドポイントを定義しました。 @lru_cache デコレーターは依存関係注入関數(shù) (get_llm_client、get_response_handler、get_cache_manager、および get_prompt_manager) に適用され、LLMClient、CacheManager、および PromptManager のインスタンスがキャッシュされてパフォーマンスが向上します。 @router.post で修飾されたmoderate_content関數(shù)は、コンテンツモデレーションのためのPOSTルートを定義し、FastAPIのDependsメカニズムを利用してこれらの依存関係を注入します。関數(shù)內(nèi)では、設(shè)定からレート制限設(shè)定を構(gòu)成した RateLimiter クラスがリクエスト制限を強(qiáng)制します。

最後に、main.py を更新してすべてをまとめましょう:

 @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=10),
        reraise=True
    )
    async def complete(self, 
                      prompt: str, 
                      temperature: float = 0.7,
                      max_tokens: Optional[int] = None) -> Dict:
        """Get completion from the model with automatic retries"""
        try:
            inputs = self.tokenizer(prompt, return_tensors="pt").to(
                self.model.device
            )

            with torch.no_grad():
                outputs = self.model.generate(
                    **inputs,
                    max_new_tokens=max_tokens or 100,
                    temperature=temperature,
                    do_sample=True
                )

            response_text = self.tokenizer.decode(
                outputs[0], 
                skip_special_tokens=True
            )

            # Calculate token usage for monitoring
            input_tokens = len(inputs.input_ids[0])
            output_tokens = len(outputs[0]) - input_tokens

            return {
                'content': response_text,
                'usage': {
                    'prompt_tokens': input_tokens,
                    'completion_tokens': output_tokens,
                    'total_tokens': input_tokens + output_tokens
                },
                'model': "gpt2"
            }

        except Exception as e:
            self.logger.error(f"Error in LLM completion: {str(e)}")
            raise

上記のコードでは、/api/v1 プレフィックスの下に api.routes を使用して FastAPI アプリとルーターを作成しました。タイムスタンプ付きの情報(bào)メッセージを表示するためのログ記録を有効にしました。アプリは、ホットリロードを有効にして、Uvicorn を使用して localhost:8000 を?qū)g行します。

アプリケーションを?qū)g行する

これですべてのコンポーネントが整ったので、アプリケーションを起動(dòng)して実行してみましょう。まず、プロジェクトのルート ディレクトリに .env ファイルを作成し、HUGGGINGFACE_API_KEY と REDIS_URL:
を追加します。

mkdir llm_integration && cd llm_integration
python3 -m venv env
syource env/bin/activate

次に、マシン上で Redis が実行されていることを確認(rèn)します。ほとんどの Unix ベースのシステムでは、次のコマンドで起動(dòng)できます:

transformers==4.36.0
huggingface-hub==0.19.4
redis==4.6.0
pydantic==2.5.0
pydantic-settings==2.1.0
tenacity==8.2.3
python-dotenv==1.0.0
fastapi==0.104.1
uvicorn==0.24.0
torch==2.1.0
numpy==1.24.3

これでアプリケーションを開始できます:

pip install -r requirements.txt

FastAPI サーバーは http://localhost:8000 で実行を開始します。自動(dòng) API ドキュメントは http://localhost:8000/docs で入手できます。これはエンドポイントをテストするのに非常に役立ちます!

Content Moderation API のテスト

新しく作成した API を?qū)g際のリクエストでテストしてみましょう。新しいターミナルを開き、次のcurlコマンドを?qū)g行します:

llm_integration/
├── core/
│   ├── llm_client.py      # your main LLM interaction code
│   ├── prompt_manager.py  # Handles prompt templates
│   └── response_handler.py # Processes LLM responses
├── cache/
│   └── redis_manager.py   # Manages your caching system
├── config/
│   └── settings.py        # Configuration management
├── api/
│   └── routes.py          # API endpoints
├── utils/
│   ├── monitoring.py      # Usage tracking
│   └── rate_limiter.py    # Rate limiting logic
├── requirements.txt
└── main.py
└── usage_logs.json       

ターミナルに次のような応答が表示されるはずです:

モニタリングと分析の追加

次に、アプリケーションのパフォーマンスと使用されているリソースの量を追跡するために、いくつかの監(jiān)視機(jī)能を追加しましょう。次のコードを utils/monitoring.py ファイルに追加します:

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from tenacity import retry, stop_after_attempt, wait_exponential
from typing import Dict, Optional
import logging

class LLMClient:
    def __init__(self, model_name: str = "gpt2", timeout: int = 30):
        try:
            self.tokenizer = AutoTokenizer.from_pretrained(model_name)
            self.model = AutoModelForCausalLM.from_pretrained(
                model_name,
                device_map="auto",
                torch_dtype=torch.float16
            )
        except Exception as e:
            logging.error(f"Error loading model: {str(e)}")
            # Fallback to a simpler model if the specified one fails
            self.tokenizer = AutoTokenizer.from_pretrained("gpt2")
            self.model = AutoModelForCausalLM.from_pretrained("gpt2")

        self.timeout = timeout
        self.logger = logging.getLogger(__name__)

UsageMonitor クラスは次の操作を?qū)g行します:

• タイムスタンプによるすべての API リクエストの追跡
• コスト監(jiān)視のためのトークン使用量の記録
• 応答時(shí)間の測(cè)定
• 構(gòu)造化されたログ ファイルにすべてを保存します (アプリケーションを運(yùn)用環(huán)境にデプロイする前に、これをデータベースに置き換えます)

次に、使用狀況統(tǒng)計(jì)を計(jì)算する新しいメソッドを追加します。

 @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=10),
        reraise=True
    )
    async def complete(self, 
                      prompt: str, 
                      temperature: float = 0.7,
                      max_tokens: Optional[int] = None) -> Dict:
        """Get completion from the model with automatic retries"""
        try:
            inputs = self.tokenizer(prompt, return_tensors="pt").to(
                self.model.device
            )

            with torch.no_grad():
                outputs = self.model.generate(
                    **inputs,
                    max_new_tokens=max_tokens or 100,
                    temperature=temperature,
                    do_sample=True
                )

            response_text = self.tokenizer.decode(
                outputs[0], 
                skip_special_tokens=True
            )

            # Calculate token usage for monitoring
            input_tokens = len(inputs.input_ids[0])
            output_tokens = len(outputs[0]) - input_tokens

            return {
                'content': response_text,
                'usage': {
                    'prompt_tokens': input_tokens,
                    'completion_tokens': output_tokens,
                    'total_tokens': input_tokens + output_tokens
                },
                'model': "gpt2"
            }

        except Exception as e:
            self.logger.error(f"Error in LLM completion: {str(e)}")
            raise

API を更新して、UsageMonitor クラスから監(jiān)視機(jī)能を追加します。

from typing import Dict
import logging

class ResponseHandler:
    def __init__(self):
        self.logger = logging.getLogger(__name__)

    def parse_moderation_response(self, raw_response: str) -> Dict:
        """Parse and structure the raw LLM response for moderation"""
        try:
            # Default response structure
            structured_response = {
                "is_appropriate": True,
                "confidence_score": 0.0,
                "reason": None
            }

            # Simple keyword-based analysis
            lower_response = raw_response.lower()

            # Check for inappropriate content signals
            if any(word in lower_response for word in ['inappropriate', 'unsafe', 'offensive', 'harmful']):
                structured_response["is_appropriate"] = False
                structured_response["confidence_score"] = 0.9
                # Extract reason if present
                if "because" in lower_response:
                    reason_start = lower_response.find("because")
                    structured_response["reason"] = raw_response[reason_start:].split('.')[0].strip()
            else:
                structured_response["confidence_score"] = 0.95

            return structured_response

        except Exception as e:
            self.logger.error(f"Error parsing response: {str(e)}")
            return {
                "is_appropriate": True,
                "confidence_score": 0.5,
                "reason": "Failed to parse response"
            }

    def format_response(self, raw_response: Dict) -> Dict:
        """Format the final response with parsed content and usage stats"""
        try:
            return {
                "content": self.parse_moderation_response(raw_response["content"]),
                "usage": raw_response["usage"],
                "model": raw_response["model"]
            }
        except Exception as e:
            self.logger.error(f"Error formatting response: {str(e)}")
            raise

次に、次のcurlコマンドを?qū)g行して/statsエンドポイントをテストします:

import redis
from typing import Optional, Any
import json
import hashlib

class CacheManager:
    def __init__(self, redis_url: str, ttl: int = 3600):
        self.redis = redis.from_url(redis_url)
        self.ttl = ttl

    def _generate_key(self, prompt: str, params: dict) -> str:
        """Generate a unique cache key"""
        cache_data = {
            'prompt': prompt,
            'params': params
        }
        serialized = json.dumps(cache_data, sort_keys=True)
        return hashlib.sha256(serialized.encode()).hexdigest()

    async def get_cached_response(self, 
                                prompt: str, 
                                params: dict) -> Optional[dict]:
        """Retrieve cached LLM response"""
        key = self._generate_key(prompt, params)
        cached = self.redis.get(key)
        return json.loads(cached) if cached else None

    async def cache_response(self, 
                           prompt: str, 
                           params: dict, 
                           response: dict) -> None:
        """Cache LLM response"""
        key = self._generate_key(prompt, params)
        self.redis.setex(
            key,
            self.ttl,
            json.dumps(response)
        )

上記のコマンドは、以下のスクリーンショットに示すように、/moderate エンドポイント上のリクエストの統(tǒng)計(jì)を表示します。

結(jié)論

このチュートリアルを通じて、実稼働アプリケーションで大規(guī)模な言語モデルを使用する方法を?qū)W習(xí)しました。 API クライアント、キャッシュ、プロンプト管理、エラー処理などの機(jī)能を?qū)g裝しました。これらの概念の一例として、コンテンツ モデレーション システムを開発しました。

強(qiáng)固な基盤ができたので、次の方法でシステムを強(qiáng)化できます。

• リアルタイム アプリケーションのストリーミング応答
• 即時(shí)改善のための A/B テスト
• プロンプトを管理するための Web ベースのインターフェイス
• カスタムモデルの微調(diào)整
• サードパーティ監(jiān)視サービスとの統(tǒng)合

例では ChatGPT2 モデルを使用しましたが、このシステムを任意の LLM プロバイダーで動(dòng)作するように適応させることができることを思い出してください。したがって、要件を満たし、予算內(nèi)でモデルを選択してください。

ご質(zhì)問がある場(chǎng)合、またはこのシステムで何を構(gòu)築しているのか知りたい場(chǎng)合は、お?dú)葺Xにお問い合わせください。

コーディングを楽しんでください! ?

以上が大規(guī)模な言語モデルを?qū)g稼働アプリケーションに統(tǒng)合するの詳細(xì)內(nèi)容です。詳細(xì)については、PHP 中國(guó)語 Web サイトの他の関連記事を參照してください。