この実践的なガイドでは、アプリケーション用に LLM が組み込まれた拡張性の高いモデル展開ソリューションを作成する方法を学びます。
例では、Hugging Face の ChatGPT2 モデルを使用しますが、ChatGPT4、Claude などの他のモデルを簡単にプラグインできます。
AI 機能を備えた新しいアプリケーションを設計している場合でも、既存の AI システムを改善している場合でも、このガイドは、強力な LLM 統合を作成するためのステップバイステップに役立ちます。
コードを書き始める前に、本番環境の LLM 統合を構築するために何が必要かを理解しましょう。本番環境に対応した LLM 統合を構築するときに考慮する必要があるのは API 呼び出しだけではなく、信頼性、コスト、安定性なども考慮する必要があります。運用アプリケーションは、コストを管理しながら、サービスの停止、レート制限、応答時間の変動などの問題に対処する必要があります。
私たちが一緒に構築するものは次のとおりです:
コーディングを始める前に、次のものが揃っていることを確認してください。
フォローしてみませんか?完全なコードは GitHub リポジトリで入手できます。
開発環境を準備することから始めましょう。クリーンなプロジェクト構造を作成し、必要なパッケージをすべてインストールします。
まず、プロジェクト ディレクトリを作成し、Python 仮想環境をセットアップしましょう。ターミナルを開いて次を実行します:
mkdir llm_integration && cd llm_integration python3 -m venv env syource env/bin/activate
次に、プロジェクトの依存関係を設定しましょう。次の必須パッケージを含む新しい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
これらの各パッケージが必要な理由を詳しく見てみましょう:
次のコマンドを使用してすべてのパッケージをインストールします:
mkdir llm_integration && cd llm_integration python3 -m venv env syource env/bin/activate
きれいな構造でプロジェクトを整理しましょう。プロジェクト ディレクトリに次のディレクトリとファイルを作成します:
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 クライアントから始めましょう。ここで、ChatGPT モデル (またはその他の任意の LLM) を操作します。次のコード スニペットを core/llm_client.py ファイルに追加します:
pip install -r requirements.txt
LLMClient クラスの最初の部分では、基盤をセットアップします。
次に、モデルと通信するメソッドを追加しましょう。
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
この完了メソッドで何が起こっているのかを詳しく見てみましょう:
次に、LLM の生の出力を解析して構造化するための応答ハンドラーを追加する必要があります。これを 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__)
次に、アプリケーションのパフォーマンスを向上させ、コストを削減するためのキャッシュ システムを作成しましょう。次のコード スニペットを、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 クラスを作成しました。
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 ファイルからプロンプト テンプレートをロードし、フォーマットされたプロンプト テンプレートも取得できるようになります。
すべての LLM 構成を 1 か所に保管し、アプリケーション全体で簡単に再利用するには、構成設定を作成しましょう。以下のコードを config/settings.py ファイルに追加します:
pip install -r requirements.txt
次に、レート制限を実装して、ユーザーがアプリケーションのリソースにアクセスする方法を制御しましょう。これを行うには、次のコードを 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 では、一定期間内に各ユーザーに許可されるリクエストの期間と数を渡すだけでレート制限を処理するために、任意のルートで使用できる再実行可能な check_rate_limit メソッドを実装しました。
次に、API エンドポイントを api/routes.py ファイルに作成して、アプリケーションに LLM を統合しましょう。
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 ルートの編成を担当する APIRouter クラスに /moderate エンドポイントを定義しました。 @lru_cache デコレーターは依存関係注入関数 (get_llm_client、get_response_handler、get_cache_manager、および get_prompt_manager) に適用され、LLMClient、CacheManager、および PromptManager のインスタンスがキャッシュされてパフォーマンスが向上します。 @router.post で修飾されたmoderate_content関数は、コンテンツモデレーションのためのPOSTルートを定義し、FastAPIのDependsメカニズムを利用してこれらの依存関係を注入します。関数内では、設定からレート制限設定を構成した RateLimiter クラスがリクエスト制限を強制します。
最後に、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 アプリとルーターを作成しました。タイムスタンプ付きの情報メッセージを表示するためのログ記録を有効にしました。アプリは、ホットリロードを有効にして、Uvicorn を使用して localhost:8000 を実行します。
これですべてのコンポーネントが整ったので、アプリケーションを起動して実行してみましょう。まず、プロジェクトのルート ディレクトリに .env ファイルを作成し、HUGGGINGFACE_API_KEY と REDIS_URL:
を追加します。
mkdir llm_integration && cd llm_integration python3 -m venv env syource env/bin/activate
次に、マシン上で Redis が実行されていることを確認します。ほとんどの Unix ベースのシステムでは、次のコマンドで起動できます:
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 で実行を開始します。自動 API ドキュメントは http://localhost:8000/docs で入手できます。これはエンドポイントをテストするのに非常に役立ちます!
新しく作成した API を実際のリクエストでテストしてみましょう。新しいターミナルを開き、次のcurlコマンドを実行します:
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
ターミナルに次のような応答が表示されるはずです:
次に、アプリケーションのパフォーマンスと使用されているリソースの量を追跡するために、いくつかの監視機能を追加しましょう。次のコードを 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 クラスは次の操作を実行します:
次に、使用状況統計を計算する新しいメソッドを追加します。
@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 クラスから監視機能を追加します。
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コマンドを実行して/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 エンドポイント上のリクエストの統計を表示します。
このチュートリアルを通じて、実稼働アプリケーションで大規模な言語モデルを使用する方法を学習しました。 API クライアント、キャッシュ、プロンプト管理、エラー処理などの機能を実装しました。これらの概念の一例として、コンテンツ モデレーション システムを開発しました。
強固な基盤ができたので、次の方法でシステムを強化できます。
例では ChatGPT2 モデルを使用しましたが、このシステムを任意の LLM プロバイダーで動作するように適応させることができることを思い出してください。したがって、要件を満たし、予算内でモデルを選択してください。
ご質問がある場合、またはこのシステムで何を構築しているのか知りたい場合は、お気軽にお問い合わせください。
コーディングを楽しんでください! ?
以上が大規模な言語モデルを実稼働アプリケーションに統合するの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。
![[Web フロントエンド] Node.js クイック スタート](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
