Mengintegrasikan Model Bahasa Besar dalam Aplikasi Pengeluaran

Dalam panduan praktikal ini, anda akan belajar cara mencipta penyelesaian penggunaan model yang sangat berskala dengan LLM terbina dalam untuk aplikasi anda.
Dalam contoh anda, kami akan menggunakan model ChatGPT2 Hugging Face, tetapi anda boleh dengan mudah memasangkan mana-mana model lain termasuk ChatGPT4, Claude, dsb.
Sama ada anda sedang mereka bentuk aplikasi baharu dengan keupayaan AI atau menambah baik sistem AI sedia ada, panduan ini akan membantu anda langkah demi langkah untuk mencipta integrasi LLM yang kukuh.

Memahami Asas Integrasi LLM

Sebelum kita mula menulis kod, mari kita fikirkan perkara yang diperlukan untuk membina penyepaduan LLM pengeluaran. Panggilan API bukan satu-satunya perkara yang perlu anda pertimbangkan semasa membina integrasi LLM sedia pengeluaran, anda juga perlu mempertimbangkan perkara seperti kebolehpercayaan, kos dan kestabilan. Aplikasi pengeluaran anda mesti menangani isu seperti gangguan perkhidmatan, had kadar dan kebolehubahan dalam masa tindak balas sambil mengekalkan kos terkawal.
Inilah yang akan kita bina bersama:

• Pelanggan API yang mantap yang menangani kegagalan dengan anggun
• Sistem caching pintar untuk mengoptimumkan kos dan kelajuan
• Sistem pengurusan segera yang betul
• Pengendalian dan pemantauan ralat yang komprehensif
• Sistem penyederhanaan kandungan yang lengkap sebagai projek contoh anda

Prasyarat

Sebelum kami memulakan pengekodan, pastikan anda mempunyai:

• Python 3.8 atau lebih baharu dipasang pada mesin anda
• Akaun awan Redis atau dipasang secara tempatan
• Pengetahuan pengaturcaraan Python asas
• Pemahaman asas tentang REST API
• Kunci API Wajah Memeluk (atau mana-mana kunci pembekal LLM lain)

Mahu ikut? Kod lengkap tersedia dalam repositori GitHub anda.

Sediakan Persekitaran Pembangunan anda

Mari mulakan dengan menyediakan persekitaran pembangunan anda. Kami akan mencipta struktur projek yang bersih dan memasang semua pakej yang diperlukan.

Pertama, mari buat direktori projek anda dan sediakan persekitaran maya Python. Buka terminal anda dan jalankan:

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

Sekarang mari sediakan kebergantungan projek anda. Cipta fail requirements.txt baharu dengan pakej penting ini:

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

Mari kita pecahkan mengapa kita memerlukan setiap pakej ini:

• transformer: Ini adalah perpustakaan hebat Hugging Face yang akan kami gunakan untuk antara muka dengan model Qwen2.5-Coder.
• huggingface-hub: Membolehkan kami mengendalikan pemuatan model dan versi redis: Untuk melaksanakan caching permintaan
• pydantic: Digunakan untuk pengesahan dan tetapan data.
• ketabahan: Bertanggungjawab untuk kefungsian mencuba semula anda untuk meningkatkan kebolehpercayaan
• python-dotenv: Untuk memuatkan pembolehubah persekitaran
• fastapi: Membina titik akhir API anda dengan sejumlah kecil kod
• uvicorn: Digunakan untuk menjalankan aplikasi FastAPI anda dengan kecekapan yang hebat
• obor: Untuk menjalankan model pengubah dan mengendalikan operasi pembelajaran mesin
• numpy: Digunakan untuk pengkomputeran berangka.

Pasang semua pakej dengan arahan:

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

Jom susun projek anda dengan struktur yang bersih. Cipta direktori dan fail ini dalam direktori projek anda:

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

Membina Pelanggan LLM

Mari mulakan dengan pelanggan LLM anda yang merupakan komponen terpenting dalam aplikasi anda. Di sinilah kami akan berinteraksi dengan model ChatGPT (atau mana-mana LLM lain yang anda suka). Tambahkan coretan kod berikut pada fail teras/llm_client.py anda:

pip install -r requirements.txt

Dalam bahagian pertama kelas LLMClient anda ini, kami menyediakan asas:

• Kami menggunakan AutoModelForCausalLM dan AutoTokenizer daripada perpustakaan transformer untuk memuatkan model anda
• Parameter device_map="auto" mengendalikan peruntukan GPU/CPU secara automatik
• Kami menggunakan torch.float16 untuk mengoptimumkan penggunaan memori sambil mengekalkan prestasi yang baik

Sekarang mari tambah kaedah yang bercakap dengan model anda:

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       

Mari kita pecahkan apa yang berlaku dalam kaedah penyelesaian ini:

• Menambah kaedah penghias @cuba semula untuk menangani kegagalan sementara.
• Pengurus konteks torch.no_grad() digunakan untuk menyimpan memori dengan melumpuhkan pengiraan kecerunan.
• Menjejaki penggunaan token dalam kedua-dua input dan output yang sangat penting untuk pengekosan.
• Mengembalikan kamus berstruktur dengan respons dan statistik penggunaan.

Mencipta Pengendali Respons LLM anda

Seterusnya, kita perlu menambah pengendali tindak balas untuk menghuraikan dan menstrukturkan output mentah LLM. Lakukan itu dalam fail core/response_handler.py anda dengan coretan kod berikut:

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__)

Menambah Sistem Caching yang Teguh

Sekarang mari buat sistem caching anda untuk meningkatkan prestasi aplikasi dan mengurangkan kos. Tambahkan coretan kod berikut pada fail cache/redis_manager.py anda:

 @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

Dalam coretan kod di atas, kami mencipta kelas CacheManager yang mengendalikan semua operasi caching dengan perkara berikut:

• Kaedah _generate_key, yang mencipta kunci cache unik berdasarkan gesaan dan parameter
• get_cached_response yang menyemak sama ada kami mempunyai respons cache untuk gesaan yang diberikan
• cache_response yang menyimpan respons yang berjaya untuk kegunaan masa hadapan

Mencipta Pengurus Gesaan Pintar

Mari buat pengurus gesaan anda yang akan mengurus gesaan untuk model LLM anda. Tambahkan kod berikut pada teras/prompt_manager.py anda:

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

Kemudian buat contoh templat gesaan untuk penyederhanaan kandungan dalam fail gesaan/content_moderation.json anda dengan coretan kod:

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

Kini pengurus gesaan anda akan dapat memuatkan templat gesaan daripada fail JSON anda dan turut mendapat templat gesaan yang diformatkan.

Menyediakan Pengurus Konfigurasi

Untuk menyimpan semua konfigurasi LLM anda di satu tempat dan menggunakannya semula dengan mudah merentas aplikasi anda, mari buat tetapan konfigurasi. Tambahkan kod di bawah pada fail config/settings.py anda:

pip install -r requirements.txt

Melaksanakan Had Kadar

Seterusnya, mari laksanakan pengehadan kadar untuk mengawal cara pengguna mengakses sumber aplikasi anda. Untuk berbuat demikian, tambahkan kod berikut pada fail utils/rate_limiter.py anda:

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       

Dalam RateLimiter kami melaksanakan kaedah check_rate_limit yang boleh digunakan semula yang boleh digunakan dalam mana-mana laluan untuk mengendalikan pengehadan kadar dengan hanya melepasi tempoh dan bilangan permintaan yang dibenarkan untuk setiap pengguna untuk tempoh masa.

Mencipta Titik Akhir API anda

Sekarang mari buat titik akhir API anda dalam fail api/routes.py untuk menyepadukan LLM anda dalam aplikasi anda:

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__)

Di sini kami menentukan titik akhir /sederhana dalam kelas APIRouter, yang bertanggungjawab untuk mengatur laluan API. Penghias @lru_cache digunakan pada fungsi suntikan kebergantungan (get_llm_client, get_response_handler, get_cache_manager dan get_prompt_manager) untuk memastikan bahawa tika LLMClient, CacheManager dan PromptManager dicache untuk prestasi yang lebih baik. Fungsi moderate_content, dihiasi dengan @router.post, mentakrifkan laluan POST untuk penyederhanaan kandungan dan menggunakan mekanisme FastAPI's Depends untuk menyuntik kebergantungan ini. Di dalam fungsi, kelas RateLimiter, dikonfigurasikan dengan tetapan had kadar daripada tetapan, menguatkuasakan had permintaan.

Akhir sekali, mari kemas kini main.py anda untuk menyatukan segala-galanya:

 @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

Dalam kod di atas, kami telah mencipta apl FastAPI dan penghala menggunakan api.routes di bawah awalan /api/v1. Pengelogan didayakan untuk memaparkan mesej maklumat dengan cap masa. Apl ini akan menjalankan localhost:8000 menggunakan Uvicorn, dengan muat semula panas didayakan.

Menjalankan Aplikasi anda

Kami kini mempunyai semua komponen, mari mula menyediakan aplikasi anda dan berjalan. Mula-mula, buat fail .env dalam direktori akar projek anda dan tambah HUGGINGFACE_API_KEY dan REDIS_URL anda:

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

Kemudian pastikan Redis berjalan pada mesin anda. Pada kebanyakan sistem berasaskan Unix, anda boleh memulakannya dengan arahan:

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

Kini anda boleh memulakan permohonan anda:

pip install -r requirements.txt

pelayan FastAPI anda akan mula berjalan pada http://localhost:8000. Dokumentasi API automatik akan tersedia di http://localhost:8000/docs - ini sangat membantu untuk menguji titik akhir anda!

Menguji API Penyederhanaan Kandungan anda

Mari uji API anda yang baru dibuat dengan permintaan sebenar. Buka terminal baharu dan jalankan arahan curl ini:

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       

Anda sepatutnya melihat respons seperti ini pada terminal anda:

Menambahkan Pemantauan dan Analitis

Sekarang mari tambah beberapa ciri pemantauan untuk menjejak prestasi aplikasi anda dan jumlah sumber semula yang digunakan. Tambahkan kod berikut pada fail utils/monitoring.py anda:

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__)

Kelas UsageMonitor akan melaksanakan operasi berikut:

• Menjejaki setiap permintaan API dengan cap masa
• Merakam penggunaan token untuk pemantauan kos
• Mengukur masa tindak balas
• Menyimpan segala-galanya dalam fail log berstruktur (ganti ini dengan pangkalan data sebelum anda menggunakan aplikasi anda ke pengeluaran)

Seterusnya, tambahkan kaedah baharu untuk mengira statistik penggunaan:

 @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

Kemas kini API anda untuk menambah ciri pemantauan daripada kelas 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

Sekarang, uji titik akhir /stats anda dengan menjalankan arahan curl ini:

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)
        )

Arahan di atas akan menunjukkan kepada anda statistik permintaan anda pada titik akhir /moderate seperti yang ditunjukkan dalam tangkapan skrin di bawah:

Kesimpulan

Sepanjang tutorial ini, telah mempelajari cara menggunakan model bahasa yang besar dalam aplikasi pengeluaran. Anda melaksanakan ciri seperti klien API, caching, pengurusan segera dan pengendalian ralat. Sebagai contoh konsep ini, anda membangunkan sistem penyederhanaan kandungan.

Sekarang anda mempunyai asas yang kukuh, anda boleh meningkatkan sistem anda dengan:

• Menstrim respons untuk aplikasi masa nyata
• Ujian A/B untuk peningkatan segera
• Antara muka berasaskan web untuk mengurus gesaan
• Penalaan halus model tersuai
• Integrasi dengan perkhidmatan pemantauan pihak ketiga

Sila ingat bahawa dalam contoh anda menggunakan model ChatGPT2, tetapi anda boleh menyesuaikan sistem ini untuk berfungsi dengan mana-mana pembekal LLM. Jadi pilih model yang memenuhi keperluan anda dan mengikut bajet anda.

Sila jangan teragak-agak untuk menghubungi saya jika anda mempunyai soalan atau jika anda ingin memberitahu saya apa yang anda sedang bina dengan sistem ini.

Selamat pengekodan! ?

Atas ialah kandungan terperinci Mengintegrasikan Model Bahasa Besar dalam Aplikasi Pengeluaran. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!