프로덕션 애플리케이션에 대규모 언어 모델 통합
- Mary-Kate Olsen원래의
- 2025-01-07 06:24:41864검색
이 실무 가이드에서는 애플리케이션에 LLM이 내장되어 확장성이 뛰어난 모델 배포 솔루션을 만드는 방법을 알아봅니다.
예시에서는 Hugging Face의 ChatGPT2 모델을 사용하지만 ChatGPT4, Claude 등을 포함한 다른 모델도 쉽게 연결할 수 있습니다.
AI 기능을 갖춘 새로운 애플리케이션을 설계하든 기존 AI 시스템을 개선하든 이 가이드는 강력한 LLM 통합을 구축하는 데 단계별로 도움이 될 것입니다.
LLM 통합 기본 사항 이해
코드 작성을 시작하기 전에 프로덕션 LLM 통합을 구축하는 데 필요한 것이 무엇인지 알아보겠습니다. 프로덕션에 즉시 사용 가능한 LLM 통합을 구축할 때 고려해야 할 사항은 API 호출뿐 아니라 안정성, 비용, 안정성 등도 고려해야 합니다. 프로덕션 애플리케이션은 비용을 통제하면서 서비스 중단, 속도 제한, 응답 시간 변동성과 같은 문제를 해결해야 합니다.
우리가 함께 만들 항목은 다음과 같습니다.
- 실패를 우아하게 처리하는 강력한 API 클라이언트
- 비용과 속도를 최적화하는 스마트 캐싱 시스템
- 적절한 프롬프트 관리 시스템
- 종합적인 오류 처리 및 모니터링
- 예시 프로젝트로서의 완전한 콘텐츠 조정 시스템
전제 조건
코딩을 시작하기 전에 다음 사항을 확인하세요.
- 컴퓨터에 Python 3.8 이상이 설치되어 있어야 합니다
- Redis 클라우드 계정 또는 로컬로 설치
- 기본 Python 프로그래밍 지식
- REST API에 대한 기본 이해
- Hugging Face API 키(또는 기타 LLM 제공업체 키)
따라가고 싶으신가요? 전체 코드는 GitHub 저장소에서 확인할 수 있습니다.
개발 환경 설정
개발 환경을 준비하는 것부터 시작하겠습니다. 깔끔한 프로젝트 구조를 만들고 필요한 패키지를 모두 설치하겠습니다.
먼저 프로젝트 디렉터리를 만들고 Python 가상 환경을 설정해 보겠습니다. 터미널을 열고 다음을 실행하세요.
mkdir llm_integration && cd llm_integration python3 -m venv env syource env/bin/activate
이제 프로젝트 종속성을 설정해 보겠습니다. 다음 필수 패키지로 새 요구 사항.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
각 패키지가 필요한 이유를 분석해 보겠습니다.
- Transformers: 이것은 Qwen2.5-Coder 모델과의 인터페이스에 사용할 Hugging Face의 강력한 라이브러리입니다.
- Huggingface-hub: 모델 로딩 및 버전 관리를 가능하게 합니다. redis: 요청 캐싱 구현용
- pydantic: 데이터 검증 및 설정에 사용됩니다.
- 강인함: 안정성 향상을 위해 재시도 기능을 담당합니다.
- python-dotenv: 환경 변수 로드용
- fastapi: 적은 양의 코드로 API 엔드포인트 구축
- uvicorn: FastAPI 애플리케이션을 매우 효율적으로 실행하는 데 사용됩니다
- torch: 변환기 모델 실행 및 기계 학습 작업 처리용
- numpy: 수치 계산에 사용됩니다.
다음 명령을 사용하여 모든 패키지를 설치합니다.
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 클라이언트 구축
지원서의 가장 중요한 구성 요소인 LLM 클라이언트부터 시작하겠습니다. 여기에서 ChatGPT 모델(또는 선호하는 다른 LLM)과 상호 작용할 것입니다. core/llm_client.py 파일에 다음 코드 조각을 추가하세요.
pip install -r requirements.txt
LLMClient 클래스의 첫 번째 부분에서는 다음과 같은 기초를 설정합니다.
- 변환기 라이브러리의 AutoModelForCausalLM 및 AutoTokenizer를 사용하여 모델을 로드하고 있습니다
- device_map="auto" 매개변수는 GPU/CPU 할당을 자동으로 처리합니다
- 좋은 성능을 유지하면서 메모리 사용을 최적화하기 위해 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
이 완성 방법에서 어떤 일이 일어나는지 살펴보겠습니다.
- 일시적인 실패를 처리하기 위해 @retry 데코레이터 메서드를 추가했습니다.
- torch.no_grad() 컨텍스트 관리자를 사용하여 그래디언트 계산을 비활성화하여 메모리를 절약했습니다.
- 비용 계산에 매우 중요한 입력 및 출력 모두에서 토큰 사용량을 추적합니다.
- 응답 및 사용 통계가 포함된 구조화된 사전을 반환합니다.
LLM 응답 처리기 만들기
다음으로 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 클래스를 만들었습니다.
- 프롬프트와 매개변수를 기반으로 고유한 캐시 키를 생성하는 _generate_key 메소드
- 주어진 프롬프트에 대해 캐시된 응답이 있는지 확인하는 get_cached_response
- 나중에 사용할 수 있도록 성공적인 응답을 저장하는 캐시_응답
스마트 프롬프트 관리자 생성
LLM 모델의 프롬프트를 관리할 프롬프트 관리자를 만들어 보겠습니다. core/prompt_manager.py에 다음 코드를 추가하세요.
mkdir llm_integration && cd llm_integration python3 -m venv env syource env/bin/activate
그런 다음 코드 조각을 사용하여 프롬프트/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 구성을 한 곳에 보관하고 애플리케이션 전체에서 쉽게 재사용할 수 있도록 구성 설정을 만들어 보겠습니다. 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 파일에 API 엔드포인트를 생성하여 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로 장식된mode_content 함수는 콘텐츠 조정을 위한 POST 경로를 정의하고 FastAPI의 종속 메커니즘을 활용하여 이러한 종속성을 주입합니다. 함수 내에서 설정의 비율 제한 설정으로 구성된 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 파일을 생성하고 HUGGINGFACE_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에서 확인할 수 있습니다. 이는 엔드포인트를 테스트하는 데 매우 유용합니다!
Content Moderation API 테스트
실제 요청으로 새로 생성된 API를 테스트해 보겠습니다. 새 터미널을 열고 다음 컬 명령을 실행하세요.
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 클래스는 다음 작업을 수행합니다.
- 타임스탬프로 모든 API 요청 추적
- 비용 모니터링을 위한 토큰 사용량 기록
- 응답 시간 측정
- 모든 것을 구조화된 로그 파일에 저장(애플리케이션을 프로덕션에 배포하기 전에 이를 데이터베이스로 대체)
다음으로 사용 통계를 계산하는 새로운 방법을 추가합니다.
@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
이제 다음 컬 명령을 실행하여 /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 클라이언트, 캐싱, 프롬프트 관리, 오류 처리와 같은 기능을 구현했습니다. 이러한 개념의 예로 콘텐츠 조정 시스템을 개발하셨습니다.
이제 기초가 탄탄해졌으므로 다음을 통해 시스템을 향상할 수 있습니다.
- 실시간 지원을 위한 스트리밍 응답
- 즉시 개선을 위한 A/B 테스트
- 메시지 관리를 위한 웹 기반 인터페이스
- 커스텀 모델 미세 조정
- 타사 모니터링 서비스와의 통합
예제에서는 ChatGPT2 모델을 사용했지만 이 시스템을 모든 LLM 제공업체와 협력하도록 조정할 수 있습니다. 따라서 귀하의 요구 사항을 충족하고 예산 범위 내에서 모델을 선택하십시오.
궁금한 점이 있거나 이 시스템으로 무엇을 구축하고 있는지 알려주고 싶다면 주저하지 말고 연락주세요.
즐거운 코딩하세요! ?
위 내용은 프로덕션 애플리케이션에 대규모 언어 모델 통합의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!