Modelová rodina Gemma predstavuje komplexný prístup spoločnosti Google k otvoreným veľkým jazykovým modelom a multimodálnej AI, čím dokazuje, že dostupné modely môžu dosiahnuť pozoruhodný výkon a zároveň byť nasaditeľné v rôznych scenároch od mobilných zariadení po podnikové pracovné stanice. Je dôležité pochopiť, ako rodina Gemma umožňuje výkonné AI schopnosti s flexibilnými možnosťami nasadenia, pričom si zachováva konkurencieschopný výkon a zodpovedné praktiky AI.
V tomto návode preskúmame modelovú rodinu Gemma od Google a jej základné koncepty. Pokryjeme vývoj rodiny Gemma, inovatívne metodológie tréningu, ktoré robia modely Gemma efektívnymi, kľúčové varianty v rodine a praktické aplikácie v rôznych scenároch nasadenia.
Na konci tohto návodu budete schopní:
- Pochopiť filozofiu dizajnu a vývoj modelovej rodiny Gemma od Google
- Identifikovať kľúčové inovácie, ktoré umožňujú modelom Gemma dosiahnuť vysoký výkon pri rôznych veľkostiach parametrov
- Rozpoznať výhody a obmedzenia rôznych variantov modelov Gemma
- Aplikovať znalosti o modeloch Gemma na výber vhodných variantov pre reálne scenáre
Prostredie AI sa výrazne vyvinulo, pričom rôzne organizácie sledujú rôzne prístupy k vývoju jazykových modelov. Zatiaľ čo niektoré sa zameriavajú na proprietárne uzavreté modely dostupné iba prostredníctvom API, iné zdôrazňujú otvorenú dostupnosť a transparentnosť. Tradičný prístup zahŕňa buď masívne proprietárne modely s priebežnými nákladmi, alebo otvorené modely, ktoré môžu vyžadovať značnú technickú odbornosť na nasadenie.
Tento paradigmatický prístup vytvára výzvy pre organizácie, ktoré hľadajú výkonné AI schopnosti a zároveň si chcú zachovať kontrolu nad svojimi dátami, nákladmi a flexibilitou nasadenia. Konvenčný prístup často vyžaduje voľbu medzi špičkovým výkonom a praktickými úvahami o nasadení.
Potrebnosť kvalitnej, dostupnej AI sa stáva čoraz dôležitejšou v rôznych scenároch. Zvážte aplikácie vyžadujúce flexibilné možnosti nasadenia pre rôzne organizačné potreby, nákladovo efektívne implementácie, kde náklady na API môžu byť významné, multimodálne schopnosti pre komplexné porozumenie alebo špecializované nasadenie na mobilných a okrajových zariadeniach.
Moderné nasadenia AI čelia niekoľkým základným požiadavkám, ktoré obmedzujú praktickú použiteľnosť:
- Dostupnosť: Otvorená dostupnosť pre transparentnosť a prispôsobenie
- Nákladová efektívnosť: Rozumné výpočtové požiadavky pre rôzne rozpočty
- Flexibilita: Viacero veľkostí modelov pre rôzne scenáre nasadenia
- Multimodálne porozumenie: Schopnosti spracovania vizuálnych, textových a zvukových dát
- Nasadenie na okraji: Optimalizovaný výkon na mobilných a zdrojovo obmedzených zariadeniach
Modelová rodina Gemma predstavuje komplexný prístup spoločnosti Google k vývoju AI modelov, ktorý uprednostňuje otvorenú dostupnosť, multimodálne schopnosti a praktické nasadenie, pričom si zachováva konkurencieschopné výkonnostné charakteristiky. Modely Gemma to dosahujú prostredníctvom rôznych veľkostí modelov, kvalitných metodológií tréningu odvodených z výskumu Gemini a špecializovaných variantov pre rôzne domény a scenáre nasadenia.
Rodina Gemma zahŕňa rôzne prístupy navrhnuté tak, aby poskytovali možnosti naprieč spektrom výkonu a efektivity, umožňujúc nasadenie od mobilných zariadení po podnikové servery a zároveň poskytujúc zmysluplné AI schopnosti. Cieľom je demokratizovať prístup k vysoko kvalitnej AI technológii a zároveň poskytovať flexibilitu v možnostiach nasadenia.
Modely Gemma sú postavené na niekoľkých základných princípoch, ktoré ich odlišujú od iných rodín jazykových modelov:
- Otvorený zdroj ako priorita: Kompletná transparentnosť a dostupnosť pre výskum a komerčné použitie
- Vývoj založený na výskume: Postavené na rovnakom výskume a technológii, ktoré poháňajú modely Gemini
- Škálovateľná architektúra: Viacero veľkostí modelov na prispôsobenie rôznym výpočtovým požiadavkám
- Zodpovedná AI: Integrované bezpečnostné opatrenia a zodpovedné praktiky vývoja
Jedným z definujúcich aspektov rodiny Gemma je sofistikovaný tréningový prístup odvodený z výskumu Gemini od Google. Modely Gemma využívajú destiláciu z väčších modelov, posilňovacie učenie z ľudskej spätnej väzby (RLHF) a techniky zlúčenia modelov na dosiahnutie vylepšeného výkonu v matematike, kódovaní a nasledovaní inštrukcií.
Proces tréningu zahŕňa destiláciu z väčších inštruktážnych modelov, posilňovacie učenie z ľudskej spätnej väzby (RLHF) na zosúladenie s ľudskými preferenciami, posilňovacie učenie zo strojovej spätnej väzby (RLMF) pre matematické uvažovanie a posilňovacie učenie z vykonávacej spätnej väzby (RLEF) pre schopnosti kódovania.
Nedávne modely Gemma zahŕňajú sofistikované multimodálne schopnosti, ktoré umožňujú komplexné porozumenie rôznym typom vstupov:
Integrácia videnia a jazyka (Gemma 3): Gemma 3 dokáže spracovať text aj obrázky súčasne, čo jej umožňuje analyzovať obrázky, odpovedať na otázky o vizuálnom obsahu, extrahovať text z obrázkov a rozumieť komplexným vizuálnym dátam.
Spracovanie zvuku (Gemma 3n): Gemma 3n obsahuje pokročilé zvukové schopnosti vrátane automatického rozpoznávania reči (ASR) a automatického prekladu reči (AST), s obzvlášť silným výkonom pri preklade medzi angličtinou a španielčinou, francúzštinou, taliančinou a portugalčinou.
Spracovanie prepletených vstupov: Modely Gemma podporujú prepletené vstupy naprieč modalitami, čo umožňuje porozumenie komplexným multimodálnym interakciám, kde text, obrázky a zvuk môžu byť spracované spoločne.
Rodina Gemma zahŕňa niekoľko architektonických optimalizácií navrhnutých pre výkon aj efektivitu:
Rozšírenie kontextového okna: Modely Gemma 3 majú kontextové okno s kapacitou 128K tokenov, čo je 16x viac ako predchádzajúce modely Gemma, umožňujúc spracovanie obrovského množstva informácií vrátane viacerých dokumentov alebo stoviek obrázkov.
Architektúra orientovaná na mobilné zariadenia (Gemma 3n): Gemma 3n využíva technológiu Per-Layer Embeddings (PLE) a architektúru MatFormer, čo umožňuje väčším modelom fungovať s pamäťovými nárokmi porovnateľnými s menšími tradičnými modelmi.
Schopnosti volania funkcií: Gemma 3 podporuje volanie funkcií, čo umožňuje vývojárom vytvárať rozhrania prirodzeného jazyka pre programovacie rozhrania a vytvárať inteligentné automatizačné systémy.
- Gemma 3 prináša výkonné schopnosti pre vývojárov s pokročilými schopnosťami textového a vizuálneho uvažovania, podporujúc vstupy vo forme obrázkov a textu pre multimodálne porozumenie.
- Gemma 3n sa umiestňuje vysoko medzi populárnymi proprietárnymi a otvorenými modelmi v Chatbot Arena Elo skóre, čo naznačuje silnú preferenciu používateľov.
Úspechy v efektivite:
- Modely Gemma 3 dokážu spracovať vstupy až do 128K tokenov, čo je 16x väčšie kontextové okno ako pri predchádzajúcich modeloch Gemma.
- Gemma 3n využíva Per-Layer Embeddings (PLE), ktoré prinášajú významné zníženie spotreby RAM pri zachovaní schopností väčších modelov.
Optimalizácia pre mobilné zariadenia:
- Gemma 3n E2B funguje s pamäťou len 2GB, zatiaľ čo E4B vyžaduje iba 3GB, napriek tomu, že má počet parametrov 5B a 8B.
- AI schopnosti v reálnom čase priamo na mobilných zariadeniach s dôrazom na súkromie a pripravenosť na offline prevádzku.
Rozsah tréningu:
- Gemma 3 bola trénovaná na 2T tokenoch pre modely 1B, 4T pre 4B, 12T pre 12B a 14T tokenov pre modely 27B pomocou Google TPU a JAX Framework.
| Séria modelov | Rozsah parametrov | Dĺžka kontextu | Kľúčové silné stránky | Najlepšie použitia |
|---|---|---|---|---|
| Gemma 3 | 1B-27B | 128K | Multimodálne porozumenie, volanie funkcií | Všeobecné aplikácie, úlohy v oblasti videnia a jazyka |
| Gemma 3n | E2B (5B), E4B (8B) | Premenná | Optimalizácia pre mobilné zariadenia, spracovanie zvuku | Mobilné aplikácie, edge computing, AI v reálnom čase |
| Gemma 2.5 | 0.5B-72B | 32K-128K | Vyvážený výkon, multijazyčnosť | Produkčné nasadenie, existujúce pracovné postupy |
| Gemma-VL | Rôzne | Premenná | Špecializácia na videnie a jazyk | Analýza obrázkov, vizuálne otázky a odpovede |
- Gemma 3-1B: Jednoduché textové úlohy, jednoduché mobilné aplikácie
- Gemma 3-4B: Vyvážený výkon s multimodálnou podporou pre všeobecné použitie
- Gemma 3-4B/12B: Porozumenie obrázkom, vizuálne otázky a odpovede
- Gemma 3n: Multimodálne mobilné aplikácie so schopnosťami spracovania zvuku
- Gemma 3n E2B: Zariadenia s obmedzenými zdrojmi, mobilná AI v reálnom čase
- Gemma 3n E4B: Vylepšený mobilný výkon so schopnosťami spracovania zvuku
- Gemma 3-12B/27B: Výkonné porozumenie jazyku a videniu
- Schopnosti volania funkcií: Budovanie inteligentných automatizačných systémov
- Akýkoľvek variant Gemma 3: Podpora 140+ jazykov s kultúrnym porozumením
- Gemma 3n: Mobilné aplikácie orientované na globálne použitie so zvukovým prekladom
- Vertex AI: Kompletné MLOps schopnosti s bezserverovým zážitkom
- Google Kubernetes Engine (GKE): Škálovateľné nasadenie kontajnerov pre komplexné pracovné zaťaženia
- Google GenAI API: Priamy prístup k API pre rýchle prototypovanie
- NVIDIA API Catalog: Optimalizovaný výkon na NVIDIA GPU
- Hugging Face Transformers: Štandardná integrácia pre vývoj
- Ollama: Zjednodušené lokálne nasadenie a správa
- vLLM: Výkonné podávanie pre produkciu
- Gemma.cpp: Optimalizované vykonávanie na CPU
- Google AI Edge: Optimalizácia nasadenia pre mobilné a edge zariadenia
- Google AI Studio: Vyskúšajte modely Gemma len niekoľkými kliknutiami
- Kaggle a Hugging Face: Stiahnite si váhy modelov a príklady od komunity
- Technické správy: Komplexná dokumentácia a výskumné práce
- Fóra komunity: Aktívna podpora komunity a diskusie
- Google AI Studio: Začnite s webovým experimentovaním
- Hugging Face Hub: Preskúmajte modely a implementácie od komunity
- Lokálne nasadenie: Použite Ollama alebo Transformers pre vývoj
- Porozumieť základným konceptom: Študujte multimodálne schopnosti a možnosti nasadenia
- Experimentovať s variantmi: Vyskúšajte rôzne veľkosti modelov a špecializované verzie
- Precvičiť implementáciu: Nasadzujte modely v vývojových prostrediach
- Optimalizovať pre produkciu: Doladite pre špecifické prípady použitia a platformy
- Začnite s malým: Začnite s Gemma 3-4B pre počiatočný vývoj a testovanie
- Používajte oficiálne šablóny: Aplikujte správne šablóny pre optimálne výsledky
- Monitorujte zdroje: Sledujte využitie pamäte a výkon inferencie
- Zvážte špecializáciu: Vyberte vhodné varianty pre multimodálne alebo mobilné potreby
from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments
from peft import LoraConfig, get_peft_model
from trl import SFTTrainer
from datasets import load_dataset
import torch
# Load Gemma model for fine-tuning
model_name = "google/gemma-3-8b-it"
model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=torch.bfloat16,
device_map="auto",
trust_remote_code=True
)
tokenizer = AutoTokenizer.from_pretrained(model_name)
tokenizer.pad_token = tokenizer.eos_token
# Configure LoRA for efficient fine-tuning
peft_config = LoraConfig(
r=16,
lora_alpha=32,
lora_dropout=0.1,
bias="none",
task_type="CAUSAL_LM",
target_modules=["q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj"]
)
# Apply LoRA to model
model = get_peft_model(model, peft_config)
# Training configuration optimized for Gemma
training_args = TrainingArguments(
output_dir="./gemma-finetuned",
learning_rate=2e-5,
per_device_train_batch_size=2,
gradient_accumulation_steps=8,
num_train_epochs=3,
warmup_steps=100,
logging_steps=10,
save_steps=500,
evaluation_strategy="steps",
eval_steps=500,
bf16=True,
remove_unused_columns=False,
dataloader_pin_memory=False
)
def format_gemma_instruction(example):
"""Format instruction for Gemma chat template"""
messages = [
{"role": "user", "content": example['instruction']},
{"role": "assistant", "content": example['output']}
]
return {"text": tokenizer.apply_chat_template(messages, tokenize=False)}
# Load and prepare dataset
dataset = load_dataset("your-custom-dataset")
dataset = dataset.map(format_gemma_instruction, remove_columns=dataset["train"].column_names)
# Initialize trainer
trainer = SFTTrainer(
model=model,
args=training_args,
train_dataset=dataset["train"],
eval_dataset=dataset["validation"],
tokenizer=tokenizer,
max_seq_length=2048,
packing=True
)
# Start fine-tuning
trainer.train()
# Save the fine-tuned model
model.save_pretrained("./gemma-custom-model")
tokenizer.save_pretrained("./gemma-custom-model")Pre multimodálne úlohy:
def create_multimodal_prompt(text_query, image_description="", task_type="analysis"):
"""Create structured prompt for multimodal tasks"""
if task_type == "analysis":
system_msg = """You are Gemma, an AI assistant with advanced vision capabilities.
When analyzing images, provide detailed, accurate descriptions and insights.
Structure your analysis with clear sections for visual elements, context, and implications."""
elif task_type == "comparison":
system_msg = """You are Gemma, an AI assistant specialized in visual comparison tasks.
Compare images systematically, highlighting similarities, differences, and key insights.
Organize your comparison with clear categories and specific observations."""
messages = [
{"role": "system", "content": system_msg},
{
"role": "user",
"content": [
{"type": "image"},
{"type": "text", "text": f"{text_query}\n\nAdditional context: {image_description}"}
]
}
]
return messages
# Example usage for image analysis
analysis_prompt = create_multimodal_prompt(
"Analyze this business chart and identify key trends, patterns, and actionable insights.",
"This appears to be a quarterly revenue chart spanning 2 years",
task_type="analysis"
)Pre volanie funkcií s kontextom:
def create_function_calling_prompt(user_query, available_functions, context=""):
"""Create structured prompt for function calling with Gemma 3"""
function_descriptions = []
for func in available_functions:
func_desc = f"- {func['name']}: {func['description']}"
if 'parameters' in func:
params = ", ".join([f"{k} ({v.get('type', 'string')})"
for k, v in func['parameters'].get('properties', {}).items()])
func_desc += f"\n Parameters: {params}"
function_descriptions.append(func_desc)
system_prompt = f"""You are Gemma, an AI assistant with access to specific functions.
Available Functions:
{chr(10).join(function_descriptions)}
When a user request requires function calls:
1. Identify which function(s) to use
2. Extract appropriate parameters from the user's request
3. Respond with function calls in this format:
FUNCTION_CALL: function_name(param1="value1", param2="value2")
If multiple functions are needed, list them separately.
If no function is needed, respond normally.
{f"Context: {context}" if context else ""}"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_query}
]
return messages
# Example function definitions
weather_functions = [
{
"name": "get_weather_forecast",
"description": "Get weather forecast for a specific location and time period",
"parameters": {
"properties": {
"location": {"type": "string", "description": "City and country"},
"days": {"type": "integer", "description": "Number of days (1-7)"},
"units": {"type": "string", "description": "Temperature units (celsius/fahrenheit)"}
}
}
},
{
"name": "get_weather_alerts",
"description": "Get current weather alerts and warnings",
"parameters": {
"properties": {
"location": {"type": "string", "description": "City and country"},
"severity": {"type": "string", "description": "Minimum alert severity"}
}
}
}
]
# Create function calling prompt
user_request = "I'm traveling to Tokyo next week. Can you check the 5-day weather forecast and any severe weather warnings?"
prompt = create_function_calling_prompt(user_request, weather_functions)def create_culturally_aware_prompt(query, target_cultures, response_style="formal"):
"""Create prompts that consider cultural context"""
cultural_guidelines = {
"japanese": "Use respectful honorifics, avoid direct confrontation, emphasize group harmony",
"american": "Be direct and practical, focus on individual benefits, use casual tone",
"german": "Be precise and thorough, provide detailed explanations, maintain professionalism",
"brazilian": "Be warm and personable, use inclusive language, consider social aspects",
"indian": "Show respect for hierarchy, consider diverse regional perspectives, be comprehensive"
}
style_instructions = {
"formal": "Use professional language, proper grammar, and respectful tone",
"casual": "Use conversational language while remaining informative",
"educational": "Explain concepts clearly with examples and context"
}
cultural_notes = []
for culture in target_cultures:
if culture.lower() in cultural_guidelines:
cultural_notes.append(f"For {culture} audience: {cultural_guidelines[culture.lower()]}")
system_prompt = f"""You are Gemma, a culturally-aware AI assistant. Provide responses that are
appropriate for different cultural contexts while maintaining accuracy and helpfulness.
Response Style: {style_instructions.get(response_style, style_instructions['formal'])}
Cultural Considerations:
{chr(10).join(cultural_notes)}
Provide your response in the requested languages with cultural appropriateness."""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": query}
]
return messages
# Example culturally-aware usage
multicultural_query = """Explain the concept of work-life balance and provide practical advice
for maintaining it. Please provide responses suitable for Japanese and American workplace cultures."""
culturally_aware_prompt = create_culturally_aware_prompt(
multicultural_query,
target_cultures=["Japanese", "American"],
response_style="educational"
)import asyncio
import logging
from typing import List, Dict, Optional, Union
from dataclasses import dataclass
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from PIL import Image
import time
@dataclass
class GenerationConfig:
max_tokens: int = 512
temperature: float = 0.7
top_p: float = 0.9
repetition_penalty: float = 1.05
do_sample: bool = True
@dataclass
class MultimodalInput:
text: str
images: Optional[List[Union[str, Image.Image]]] = None
audio_path: Optional[str] = None
class ProductionGemmaService:
"""Production-ready Gemma service with multimodal support"""
def __init__(self, model_name: str, device: str = "auto", enable_multimodal: bool = True):
self.model_name = model_name
self.device = device
self.enable_multimodal = enable_multimodal
self.model = None
self.tokenizer = None
self.processor = None
self.logger = self._setup_logging()
self._load_model()
def _setup_logging(self):
"""Setup production logging"""
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
return logging.getLogger(f"GemmaService-{self.model_name}")
def _load_model(self):
"""Load model and tokenizer with error handling"""
try:
self.logger.info(f"Loading model: {self.model_name}")
self.tokenizer = AutoTokenizer.from_pretrained(
self.model_name,
trust_remote_code=True
)
if self.enable_multimodal and "gemma-3" in self.model_name.lower():
# Load multimodal variant
from transformers import AutoProcessor
self.processor = AutoProcessor.from_pretrained(self.model_name)
self.model = AutoModelForCausalLM.from_pretrained(
self.model_name,
torch_dtype=torch.bfloat16,
device_map=self.device,
trust_remote_code=True,
low_cpu_mem_usage=True
)
else:
# Load text-only model
self.model = AutoModelForCausalLM.from_pretrained(
self.model_name,
torch_dtype=torch.bfloat16,
device_map=self.device,
trust_remote_code=True,
low_cpu_mem_usage=True
)
# Optimize for inference
self.model.eval()
self.logger.info("Model loaded successfully")
except Exception as e:
self.logger.error(f"Failed to load model: {str(e)}")
raise
def _prepare_multimodal_input(self, multimodal_input: MultimodalInput) -> Dict:
"""Prepare multimodal input for processing"""
if not self.enable_multimodal or not self.processor:
# Text-only processing
return {"text": multimodal_input.text}
inputs = {"text": multimodal_input.text}
if multimodal_input.images:
# Process images
processed_images = []
for img in multimodal_input.images:
if isinstance(img, str):
img = Image.open(img)
processed_images.append(img)
inputs["images"] = processed_images
if multimodal_input.audio_path:
# Process audio (Gemma 3n specific)
inputs["audio"] = multimodal_input.audio_path
return inputs
async def generate_async(
self,
messages: List[Dict[str, str]],
config: GenerationConfig = GenerationConfig(),
multimodal_input: Optional[MultimodalInput] = None
) -> str:
"""Async generation with multimodal support"""
start_time = time.time()
try:
if multimodal_input and self.enable_multimodal:
# Multimodal processing
processed_input = self._prepare_multimodal_input(multimodal_input)
if self.processor:
# Use processor for multimodal input
text = self.processor.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=True
)
model_inputs = self.processor(
text=text,
images=processed_input.get("images"),
return_tensors="pt"
).to(self.model.device)
else:
# Fallback to text-only
formatted_text = self.tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=True
)
model_inputs = self.tokenizer(
formatted_text,
return_tensors="pt"
).to(self.model.device)
else:
# Text-only processing
formatted_text = self.tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=True
)
model_inputs = self.tokenizer(
formatted_text,
return_tensors="pt",
truncation=True,
max_length=4096
).to(self.model.device)
# Generate response
with torch.no_grad():
outputs = await asyncio.get_event_loop().run_in_executor(
None,
lambda: self.model.generate(
**model_inputs,
max_new_tokens=config.max_tokens,
temperature=config.temperature,
top_p=config.top_p,
repetition_penalty=config.repetition_penalty,
do_sample=config.do_sample,
pad_token_id=self.tokenizer.eos_token_id
)
)
# Extract generated text
if self.processor and multimodal_input:
generated_text = self.processor.decode(
outputs[0][model_inputs["input_ids"].shape[1]:],
skip_special_tokens=True
)
else:
generated_text = self.tokenizer.decode(
outputs[0][model_inputs["input_ids"].shape[1]:],
skip_special_tokens=True
)
generation_time = time.time() - start_time
self.logger.info(f"Generation completed in {generation_time:.2f}s")
return generated_text.strip()
except Exception as e:
self.logger.error(f"Generation failed: {str(e)}")
raise
def health_check(self) -> Dict[str, Union[str, bool, float]]:
"""Comprehensive health check"""
health_status = {
"status": "healthy",
"model_loaded": False,
"multimodal_enabled": self.enable_multimodal,
"response_time": None,
"memory_usage": None,
"errors": []
}
try:
# Check if model is loaded
if self.model is not None and self.tokenizer is not None:
health_status["model_loaded"] = True
else:
health_status["errors"].append("Model not properly loaded")
health_status["status"] = "unhealthy"
return health_status
# Test basic functionality
start_time = time.time()
test_messages = [{"role": "user", "content": "Hello"}]
# Synchronous test for health check
formatted_text = self.tokenizer.apply_chat_template(
test_messages,
tokenize=False,
add_generation_prompt=True
)
inputs = self.tokenizer(formatted_text, return_tensors="pt").to(self.model.device)
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_new_tokens=10,
do_sample=False,
pad_token_id=self.tokenizer.eos_token_id
)
response_time = time.time() - start_time
health_status["response_time"] = response_time
# Check memory usage
if torch.cuda.is_available():
memory_used = torch.cuda.memory_allocated() / 1024 / 1024 # MB
health_status["memory_usage"] = memory_used
# Evaluate response time
if response_time > 5.0: # seconds
health_status["errors"].append("Response time exceeds threshold")
health_status["status"] = "degraded"
except Exception as e:
health_status["status"] = "unhealthy"
health_status["errors"].append(f"Health check failed: {str(e)}")
return health_status
# Example production usage
async def production_example():
# Initialize production service
gemma_service = ProductionGemmaService(
model_name="google/gemma-3-8b-it",
enable_multimodal=True
)
# Health check
health = gemma_service.health_check()
print(f"Service Health: {health}")
if health["status"] != "healthy":
print("Service not healthy, aborting")
return
# Text-only generation
text_messages = [
{"role": "user", "content": "Explain the importance of sustainable development"}
]
response = await gemma_service.generate_async(text_messages)
print(f"Text Response: {response}")
# Multimodal generation (if supported)
if gemma_service.enable_multimodal:
multimodal_messages = [
{
"role": "user",
"content": [
{"type": "image"},
{"type": "text", "text": "Describe what you see in this image and explain its significance"}
]
}
]
# Example with image
multimodal_input = MultimodalInput(
text="Analyze this business chart",
images=["path/to/chart.jpg"]
)
multimodal_response = await gemma_service.generate_async(
multimodal_messages,
multimodal_input=multimodal_input
)
print(f"Multimodal Response: {multimodal_response}")
# Run production example
# asyncio.run(production_example())from transformers import AutoModelForCausalLM, BitsAndBytesConfig
import torch
# Memory-efficient loading strategies for Gemma models
def load_optimized_gemma(model_name, optimization_level="balanced"):
"""Load Gemma with various optimization levels"""
if optimization_level == "maximum_efficiency":
# 4-bit quantization for maximum memory efficiency
quantization_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_compute_dtype=torch.bfloat16,
bnb_4bit_use_double_quant=True,
bnb_4bit_quant_type="nf4"
)
model = AutoModelForCausalLM.from_pretrained(
model_name,
quantization_config=quantization_config,
device_map="auto",
low_cpu_mem_usage=True,
trust_remote_code=True
)
elif optimization_level == "balanced":
# 8-bit quantization for balanced performance
quantization_config = BitsAndBytesConfig(
load_in_8bit=True,
llm_int8_threshold=6.0,
)
model = AutoModelForCausalLM.from_pretrained(
model_name,
quantization_config=quantization_config,
device_map="auto",
torch_dtype=torch.bfloat16,
trust_remote_code=True
)
elif optimization_level == "performance":
# Full precision for maximum performance
model = AutoModelForCausalLM.from_pretrained(
model_name,
device_map="auto",
torch_dtype=torch.bfloat16,
trust_remote_code=True
)
return model
# Example usage
efficient_model = load_optimized_gemma("google/gemma-3-8b-it", "maximum_efficiency")import torch
from torch.nn.attention import SDPABackend, sdpa_kernel
from transformers import AutoTokenizer
import time
class OptimizedGemmaInference:
"""Optimized inference class for Gemma models"""
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
self._setup_optimizations()
def _setup_optimizations(self):
"""Configure various inference optimizations"""
# Enable optimized attention mechanisms
if torch.cuda.is_available():
torch.backends.cuda.enable_flash_sdp(True)
torch.backends.cuda.enable_math_sdp(True)
torch.backends.cuda.enable_mem_efficient_sdp(True)
# Set optimal threading for CPU operations
torch.set_num_threads(min(8, torch.get_num_threads()))
# Enable JIT compilation for repeated patterns
if hasattr(torch.jit, 'set_fusion_strategy'):
torch.jit.set_fusion_strategy([('STATIC', 3), ('DYNAMIC', 20)])
# Optimize model for inference
self.model.eval()
# Enable torch.compile if available (PyTorch 2.0+)
if hasattr(torch, 'compile'):
try:
self.model = torch.compile(self.model, mode="reduce-overhead")
except Exception as e:
print(f"Torch compile failed: {e}")
def fast_generate(self, messages, max_tokens=256, use_cache=True):
"""Optimized generation with various performance enhancements"""
start_time = time.time()
# Format input
formatted_text = self.tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=True
)
inputs = self.tokenizer(
formatted_text,
return_tensors="pt",
truncation=True,
max_length=2048
).to(self.model.device)
# Use optimized attention backend
with torch.no_grad():
if torch.cuda.is_available():
with sdpa_kernel(SDPABackend.FLASH_ATTENTION):
outputs = self.model.generate(
**inputs,
max_new_tokens=max_tokens,
do_sample=True,
temperature=0.7,
top_p=0.9,
use_cache=use_cache,
pad_token_id=self.tokenizer.eos_token_id,
early_stopping=True
)
else:
outputs = self.model.generate(
**inputs,
max_new_tokens=max_tokens,
do_sample=True,
temperature=0.7,
top_p=0.9,
use_cache=use_cache,
pad_token_id=self.tokenizer.eos_token_id,
early_stopping=True
)
# Extract response
response = self.tokenizer.decode(
outputs[0][inputs.input_ids.shape[1]:],
skip_special_tokens=True
)
generation_time = time.time() - start_time
tokens_generated = outputs.shape[1] - inputs.input_ids.shape[1]
tokens_per_second = tokens_generated / generation_time if generation_time > 0 else 0
return {
"response": response.strip(),
"generation_time": generation_time,
"tokens_per_second": tokens_per_second,
"tokens_generated": tokens_generated
}
def batch_generate(self, batch_messages, max_tokens=256):
"""Optimized batch generation"""
# Format all inputs
formatted_texts = []
for messages in batch_messages:
formatted_text = self.tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=True
)
formatted_texts.append(formatted_text)
# Tokenize batch with padding
inputs = self.tokenizer(
formatted_texts,
return_tensors="pt",
padding=True,
truncation=True,
max_length=2048
).to(self.model.device)
start_time = time.time()
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_new_tokens=max_tokens,
do_sample=True,
temperature=0.7,
top_p=0.9,
use_cache=True,
pad_token_id=self.tokenizer.eos_token_id,
early_stopping=True
)
# Extract all responses
responses = []
for i, output in enumerate(outputs):
response = self.tokenizer.decode(
output[inputs.input_ids[i].shape[0]:],
skip_special_tokens=True
)
responses.append(response.strip())
generation_time = time.time() - start_time
return {
"responses": responses,
"batch_generation_time": generation_time,
"average_time_per_item": generation_time / len(batch_messages)
}
# Example usage
tokenizer = AutoTokenizer.from_pretrained("google/gemma-3-8b-it")
model = load_optimized_gemma("google/gemma-3-8b-it", "balanced")
optimized_inference = OptimizedGemmaInference(model, tokenizer)
# Single optimized generation
test_messages = [{"role": "user", "content": "Explain machine learning in simple terms"}]
result = optimized_inference.fast_generate(test_messages)
print(f"Response: {result['response']}")
print(f"Speed: {result['tokens_per_second']:.1f} tokens/second")import hashlib
import time
import re
from typing import List, Dict, Optional
import logging
class SecureGemmaService:
"""Security-focused Gemma service implementation"""
def __init__(self, model_name: str, max_requests_per_hour: int = 100):
self.model_name = model_name
self.max_requests_per_hour = max_requests_per_hour
self.model = None
self.tokenizer = None
self.request_logs = {}
self.logger = logging.getLogger("SecureGemmaService")
self._load_model()
def _load_model(self):
"""Load model with security considerations"""
from transformers import AutoModelForCausalLM, AutoTokenizer
self.tokenizer = AutoTokenizer.from_pretrained(
self.model_name,
trust_remote_code=True # Only enable for trusted models
)
self.model = AutoModelForCausalLM.from_pretrained(
self.model_name,
torch_dtype=torch.bfloat16,
device_map="auto",
trust_remote_code=True
)
def _sanitize_input(self, text: str) -> str:
"""Comprehensive input sanitization"""
if not isinstance(text, str):
raise ValueError("Input must be a string")
# Remove potentially harmful patterns
dangerous_patterns = [
r"<script[^>]*>.*?</script>",
r"javascript:",
r"data:text/html",
r"<iframe[^>]*>.*?</iframe>",
r"<object[^>]*>.*?</object>",
r"<embed[^>]*>.*?</embed>"
]
sanitized = text
for pattern in dangerous_patterns:
sanitized = re.sub(pattern, "", sanitized, flags=re.IGNORECASE | re.DOTALL)
# Limit length to prevent resource exhaustion
max_length = 8192
if len(sanitized) > max_length:
sanitized = sanitized[:max_length]
self.logger.warning(f"Input truncated to {max_length} characters")
return sanitized
def _rate_limit_check(self, user_id: str) -> bool:
"""Advanced rate limiting with sliding window"""
current_time = time.time()
window_size = 3600 # 1 hour in seconds
if user_id not in self.request_logs:
self.request_logs[user_id] = []
# Clean old requests outside the time window
self.request_logs[user_id] = [
req_time for req_time in self.request_logs[user_id]
if current_time - req_time < window_size
]
# Check if limit exceeded
if len(self.request_logs[user_id]) >= self.max_requests_per_hour:
self.logger.warning(f"Rate limit exceeded for user {user_id[:8]}...")
return False
# Log current request
self.request_logs[user_id].append(current_time)
return True
def _validate_messages(self, messages: List[Dict[str, str]]) -> List[Dict[str, str]]:
"""Validate and sanitize message structure"""
if not isinstance(messages, list):
raise ValueError("Messages must be a list")
if len(messages) == 0:
raise ValueError("Messages list cannot be empty")
if len(messages) > 50: # Reasonable conversation limit
raise ValueError("Too many messages in conversation")
validated_messages = []
for message in messages:
if not isinstance(message, dict):
raise ValueError("Each message must be a dictionary")
if "role" not in message or "content" not in message:
raise ValueError("Each message must have 'role' and 'content' fields")
# Validate role
valid_roles = ["user", "assistant", "system"]
if message["role"] not in valid_roles:
raise ValueError(f"Invalid role: {message['role']}")
# Sanitize content
sanitized_content = self._sanitize_input(message["content"])
validated_messages.append({
"role": message["role"],
"content": sanitized_content
})
return validated_messages
def _content_filter(self, text: str) -> tuple[bool, str]:
"""Basic content filtering for inappropriate content"""
# This is a simplified example - in production, use more sophisticated filtering
prohibited_keywords = [
"violence", "hate", "illegal", "harmful",
# Add more as needed for your use case
]
text_lower = text.lower()
for keyword in prohibited_keywords:
if keyword in text_lower:
return False, f"Content contains prohibited keyword: {keyword}"
return True, ""
def _hash_sensitive_data(self, data: str) -> str:
"""Hash sensitive data for logging"""
return hashlib.sha256(data.encode()).hexdigest()[:16]
def secure_generate(
self,
messages: List[Dict[str, str]],
user_id: str,
max_tokens: int = 512
) -> Dict[str, any]:
"""Generate response with comprehensive security measures"""
try:
# Rate limiting
if not self._rate_limit_check(user_id):
return {
"success": False,
"error": "Rate limit exceeded. Please try again later.",
"error_code": "RATE_LIMIT_EXCEEDED"
}
# Input validation
validated_messages = self._validate_messages(messages)
# Content filtering
for message in validated_messages:
is_safe, filter_message = self._content_filter(message["content"])
if not is_safe:
self.logger.warning(f"Content filtered for user {user_id[:8]}...: {filter_message}")
return {
"success": False,
"error": "Content violates safety guidelines",
"error_code": "CONTENT_FILTERED"
}
# Log request (with hashed content for privacy)
content_hash = self._hash_sensitive_data(str(validated_messages))
self.logger.info(f"Processing request from user {user_id[:8]}... Content hash: {content_hash}")
# Validate token limit
max_allowed_tokens = min(max_tokens, 1024)
# Generate response
start_time = time.time()
formatted_prompt = self.tokenizer.apply_chat_template(
validated_messages,
tokenize=False,
add_generation_prompt=True
)
inputs = self.tokenizer(formatted_prompt, return_tensors="pt").to(self.model.device)
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_new_tokens=max_allowed_tokens,
temperature=0.7,
top_p=0.9,
repetition_penalty=1.05,
do_sample=True,
pad_token_id=self.tokenizer.eos_token_id,
early_stopping=True
)
response = self.tokenizer.decode(
outputs[0][inputs.input_ids.shape[1]:],
skip_special_tokens=True
).strip()
generation_time = time.time() - start_time
# Filter response content
is_safe_response, filter_message = self._content_filter(response)
if not is_safe_response:
self.logger.warning(f"Response filtered for user {user_id[:8]}...: {filter_message}")
return {
"success": False,
"error": "Generated response violates safety guidelines",
"error_code": "RESPONSE_FILTERED"
}
# Log successful generation
self.logger.info(f"Successful generation for user {user_id[:8]}... in {generation_time:.2f}s")
return {
"success": True,
"response": response,
"generation_time": generation_time,
"tokens_used": max_allowed_tokens
}
except ValueError as e:
self.logger.error(f"Validation error for user {user_id[:8]}...: {str(e)}")
return {
"success": False,
"error": f"Input validation failed: {str(e)}",
"error_code": "VALIDATION_ERROR"
}
except Exception as e:
self.logger.error(f"Generation error for user {user_id[:8]}...: {str(e)}")
return {
"success": False,
"error": "An error occurred while processing your request",
"error_code": "GENERATION_ERROR"
}
def get_usage_statistics(self, user_id: str) -> Dict[str, any]:
"""Get usage statistics for a user"""
current_time = time.time()
window_size = 3600 # 1 hour
if user_id not in self.request_logs:
return {
"requests_last_hour": 0,
"remaining_requests": self.max_requests_per_hour,
"reset_time": current_time + window_size
}
# Count recent requests
recent_requests = [
req_time for req_time in self.request_logs[user_id]
if current_time - req_time < window_size
]
requests_last_hour = len(recent_requests)
remaining_requests = max(0, self.max_requests_per_hour - requests_last_hour)
# Calculate reset time (when oldest request expires)
reset_time = min(recent_requests) + window_size if recent_requests else current_time
return {
"requests_last_hour": requests_last_hour,
"remaining_requests": remaining_requests,
"reset_time": reset_time
}
# Example secure usage
secure_service = SecureGemmaService("google/gemma-3-8b-it", max_requests_per_hour=50)
# Safe generation
user_messages = [
{"role": "user", "content": "Explain the benefits of renewable energy"}
]
result = secure_service.secure_generate(user_messages, user_id="user123")
if result["success"]:
print(f"Secure Response: {result['response']}")
else:
print(f"Error: {result['error']} (Code: {result['error_code']})")
# Check usage statistics
usage_stats = secure_service.get_usage_statistics("user123")
print(f"Usage Statistics: {usage_stats}")import time
import psutil
import torch
from dataclasses import dataclass, asdict
from typing import List, Dict, Any, Optional
import json
from datetime import datetime
import statistics
@dataclass
class PerformanceMetrics:
"""Comprehensive performance metrics for Gemma models"""
timestamp: float
response_time: float
memory_usage_mb: float
gpu_memory_mb: float
token_count: int
tokens_per_second: float
model_name: str
input_length: int
success: bool
error_message: Optional[str] = None
@dataclass
class QualityMetrics:
"""Quality assessment metrics"""
relevance_score: float # 0-1
coherence_score: float # 0-1
safety_score: float # 0-1
factual_accuracy: float # 0-1
user_satisfaction: Optional[float] = None # 0-1
class GemmaMonitoringService:
"""Comprehensive monitoring service for Gemma models"""
def __init__(self, model_name: str):
self.model_name = model_name
self.performance_history: List[PerformanceMetrics] = []
self.quality_history: List[QualityMetrics] = []
self.alert_thresholds = {
"max_response_time": 10.0, # seconds
"max_memory_usage": 8192, # MB
"min_tokens_per_second": 5.0,
"min_success_rate": 0.95 # 95%
}
def measure_performance(
self,
model,
tokenizer,
messages: List[Dict[str, str]],
expected_tokens: int = 256
) -> PerformanceMetrics:
"""Comprehensive performance measurement"""
start_time = time.time()
start_memory = psutil.Process().memory_info().rss / 1024 / 1024 # MB
# GPU metrics
gpu_memory = 0
if torch.cuda.is_available():
torch.cuda.reset_peak_memory_stats()
gpu_memory = torch.cuda.memory_allocated() / 1024 / 1024 # MB
success = True
error_message = None
token_count = 0
try:
# Format input
formatted_text = tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=True
)
input_length = len(tokenizer.encode(formatted_text))
inputs = tokenizer(formatted_text, return_tensors="pt").to(model.device)
# Generate response
with torch.no_grad():
outputs = model.generate(
**inputs,
max_new_tokens=expected_tokens,
temperature=0.7,
do_sample=True,
pad_token_id=tokenizer.eos_token_id
)
token_count = outputs.shape[1] - inputs.input_ids.shape[1]
except Exception as e:
success = False
error_message = str(e)
input_length = 0
# Calculate final metrics
end_time = time.time()
end_memory = psutil.Process().memory_info().rss / 1024 / 1024
response_time = end_time - start_time
memory_usage = end_memory - start_memory
if torch.cuda.is_available():
gpu_memory = torch.cuda.max_memory_allocated() / 1024 / 1024
tokens_per_second = token_count / response_time if response_time > 0 and success else 0
metrics = PerformanceMetrics(
timestamp=start_time,
response_time=response_time,
memory_usage_mb=memory_usage,
gpu_memory_mb=gpu_memory,
token_count=token_count,
tokens_per_second=tokens_per_second,
model_name=self.model_name,
input_length=input_length,
success=success,
error_message=error_message
)
self.performance_history.append(metrics)
return metrics
def evaluate_quality(
self,
input_text: str,
generated_text: str,
reference_text: Optional[str] = None
) -> QualityMetrics:
"""Evaluate response quality using various metrics"""
# Simple quality assessment (in production, use more sophisticated methods)
relevance_score = self._assess_relevance(input_text, generated_text)
coherence_score = self._assess_coherence(generated_text)
safety_score = self._assess_safety(generated_text)
factual_accuracy = self._assess_factual_accuracy(generated_text, reference_text)
quality_metrics = QualityMetrics(
relevance_score=relevance_score,
coherence_score=coherence_score,
safety_score=safety_score,
factual_accuracy=factual_accuracy
)
self.quality_history.append(quality_metrics)
return quality_metrics
def _assess_relevance(self, input_text: str, output_text: str) -> float:
"""Simple relevance assessment based on keyword overlap"""
input_words = set(input_text.lower().split())
output_words = set(output_text.lower().split())
if len(input_words) == 0:
return 0.0
overlap = len(input_words.intersection(output_words))
return min(1.0, overlap / len(input_words) * 2) # Scale appropriately
def _assess_coherence(self, text: str) -> float:
"""Simple coherence assessment"""
sentences = text.split('.')
if len(sentences) < 2:
return 0.8 # Short text, assume reasonably coherent
# Simple heuristic: check for reasonable sentence length
avg_sentence_length = sum(len(s.split()) for s in sentences) / len(sentences)
if 5 <= avg_sentence_length <= 25:
return 0.9
elif 3 <= avg_sentence_length <= 30:
return 0.7
else:
return 0.5
def _assess_safety(self, text: str) -> float:
"""Simple safety assessment"""
harmful_indicators = [
"violence", "harmful", "dangerous", "illegal",
"hate", "discriminat", "threaten"
]
text_lower = text.lower()
harmful_count = sum(1 for indicator in harmful_indicators if indicator in text_lower)
if harmful_count == 0:
return 1.0
elif harmful_count <= 2:
return 0.7
else:
return 0.3
def _assess_factual_accuracy(self, text: str, reference: Optional[str] = None) -> float:
"""Simple factual accuracy assessment"""
if reference is None:
# Without reference, use simple heuristics
confidence_indicators = [
"according to", "research shows", "studies indicate",
"data suggests", "evidence shows"
]
text_lower = text.lower()
confidence_count = sum(1 for indicator in confidence_indicators if indicator in text_lower)
return min(0.8, 0.5 + confidence_count * 0.1)
# With reference, calculate similarity (simplified)
ref_words = set(reference.lower().split())
text_words = set(text.lower().split())
if len(ref_words) == 0:
return 0.5
overlap = len(ref_words.intersection(text_words))
return min(1.0, overlap / len(ref_words))
def get_performance_summary(self, last_n: int = 100) -> Dict[str, Any]:
"""Get comprehensive performance summary"""
if not self.performance_history:
return {"error": "No performance data available"}
recent_metrics = self.performance_history[-last_n:]
successful_metrics = [m for m in recent_metrics if m.success]
if not successful_metrics:
return {"error": "No successful operations in recent history"}
summary = {
"total_requests": len(recent_metrics),
"successful_requests": len(successful_metrics),
"success_rate": len(successful_metrics) / len(recent_metrics),
"avg_response_time": statistics.mean(m.response_time for m in successful_metrics),
"avg_memory_usage": statistics.mean(m.memory_usage_mb for m in successful_metrics),
"avg_tokens_per_second": statistics.mean(m.tokens_per_second for m in successful_metrics),
"p95_response_time": statistics.quantiles([m.response_time for m in successful_metrics], n=20)[18],
"total_tokens_generated": sum(m.token_count for m in successful_metrics),
}
if torch.cuda.is_available():
summary["avg_gpu_memory"] = statistics.mean(m.gpu_memory_mb for m in successful_metrics)
return summary
def get_quality_summary(self, last_n: int = 100) -> Dict[str, Any]:
"""Get quality metrics summary"""
if not self.quality_history:
return {"error": "No quality data available"}
recent_quality = self.quality_history[-last_n:]
return {
"total_evaluations": len(recent_quality),
"avg_relevance": statistics.mean(q.relevance_score for q in recent_quality),
"avg_coherence": statistics.mean(q.coherence_score for q in recent_quality),
"avg_safety": statistics.mean(q.safety_score for q in recent_quality),
"avg_factual_accuracy": statistics.mean(q.factual_accuracy for q in recent_quality),
"overall_quality": statistics.mean([
(q.relevance_score + q.coherence_score + q.safety_score + q.factual_accuracy) / 4
for q in recent_quality
])
}
def check_alerts(self) -> List[Dict[str, Any]]:
"""Check for performance alerts"""
alerts = []
if not self.performance_history:
return alerts
recent_metrics = self.performance_history[-10:] # Last 10 requests
successful_metrics = [m for m in recent_metrics if m.success]
if not successful_metrics:
alerts.append({
"type": "error",
"message": "No successful requests in recent history",
"severity": "high",
"timestamp": time.time()
})
return alerts
# Check response time
avg_response_time = statistics.mean(m.response_time for m in successful_metrics)
if avg_response_time > self.alert_thresholds["max_response_time"]:
alerts.append({
"type": "performance",
"message": f"High response time: {avg_response_time:.2f}s (threshold: {self.alert_thresholds['max_response_time']}s)",
"severity": "medium",
"timestamp": time.time()
})
# Check memory usage
avg_memory = statistics.mean(m.memory_usage_mb for m in successful_metrics)
if avg_memory > self.alert_thresholds["max_memory_usage"]:
alerts.append({
"type": "memory",
"message": f"High memory usage: {avg_memory:.1f}MB (threshold: {self.alert_thresholds['max_memory_usage']}MB)",
"severity": "medium",
"timestamp": time.time()
})
# Check tokens per second
avg_tps = statistics.mean(m.tokens_per_second for m in successful_metrics)
if avg_tps < self.alert_thresholds["min_tokens_per_second"]:
alerts.append({
"type": "performance",
"message": f"Low generation speed: {avg_tps:.1f} tokens/s (threshold: {self.alert_thresholds['min_tokens_per_second']} tokens/s)",
"severity": "low",
"timestamp": time.time()
})
# Check success rate
success_rate = len(successful_metrics) / len(recent_metrics)
if success_rate < self.alert_thresholds["min_success_rate"]:
alerts.append({
"type": "reliability",
"message": f"Low success rate: {success_rate:.2%} (threshold: {self.alert_thresholds['min_success_rate']:.2%})",
"severity": "high",
"timestamp": time.time()
})
return alerts
def export_metrics(self, filepath: str):
"""Export metrics to JSON file"""
export_data = {
"model_name": self.model_name,
"export_timestamp": datetime.now().isoformat(),
"performance_metrics": [asdict(m) for m in self.performance_history],
"quality_metrics": [asdict(q) for q in self.quality_history],
"performance_summary": self.get_performance_summary(),
"quality_summary": self.get_quality_summary(),
"current_alerts": self.check_alerts()
}
with open(filepath, 'w') as f:
json.dump(export_data, f, indent=2)
# Example monitoring usage
def monitoring_example():
"""Example of comprehensive Gemma monitoring"""
# Initialize monitoring
monitor = GemmaMonitoringService("google/gemma-3-8b-it")
# Load model for testing
from transformers import AutoModelForCausalLM, AutoTokenizer
model = AutoModelForCausalLM.from_pretrained(
"google/gemma-3-8b-it",
torch_dtype=torch.bfloat16,
device_map="auto"
)
tokenizer = AutoTokenizer.from_pretrained("google/gemma-3-8b-it")
# Test various scenarios
test_scenarios = [
[{"role": "user", "content": "What is machine learning?"}],
[{"role": "user", "content": "Explain quantum computing in simple terms"}],
[{"role": "user", "content": "Write a short story about AI"}],
[{"role": "user", "content": "How does photosynthesis work?"}],
[{"role": "user", "content": "What are the benefits of renewable energy?"}]
]
print("Running performance tests...")
for i, messages in enumerate(test_scenarios):
print(f"Test {i+1}/5: {messages[0]['content'][:50]}...")
# Measure performance
perf_metrics = monitor.measure_performance(model, tokenizer, messages)
print(f" Response time: {perf_metrics.response_time:.2f}s")
print(f" Tokens/sec: {perf_metrics.tokens_per_second:.1f}")
print(f" Success: {perf_metrics.success}")
if perf_metrics.success:
# Generate actual response for quality evaluation
formatted_text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
inputs = tokenizer(formatted_text, return_tensors="pt").to(model.device)
with torch.no_grad():
outputs = model.generate(**inputs, max_new_tokens=100, temperature=0.7)
response = tokenizer.decode(outputs[0][inputs.input_ids.shape[1]:], skip_special_tokens=True)
# Evaluate quality
quality_metrics = monitor.evaluate_quality(messages[0]['content'], response)
print(f" Quality score: {(quality_metrics.relevance_score + quality_metrics.coherence_score + quality_metrics.safety_score + quality_metrics.factual_accuracy) / 4:.2f}")
print()
# Get comprehensive summary
perf_summary = monitor.get_performance_summary()
quality_summary = monitor.get_quality_summary()
alerts = monitor.check_alerts()
print("=== Performance Summary ===")
print(f"Success Rate: {perf_summary.get('success_rate', 0):.2%}")
print(f"Average Response Time: {perf_summary.get('avg_response_time', 0):.2f}s")
print(f"Average Tokens/Second: {perf_summary.get('avg_tokens_per_second', 0):.1f}")
print(f"Total Tokens Generated: {perf_summary.get('total_tokens_generated', 0)}")
print("\n=== Quality Summary ===")
print(f"Overall Quality Score: {quality_summary.get('overall_quality', 0):.2f}")
print(f"Average Relevance: {quality_summary.get('avg_relevance', 0):.2f}")
print(f"Average Safety: {quality_summary.get('avg_safety', 0):.2f}")
print(f"\n=== Alerts ({len(alerts)}) ===")
for alert in alerts:
print(f"[{alert['severity'].upper()}] {alert['type']}: {alert['message']}")
# Export metrics
monitor.export_metrics("gemma_monitoring_report.json")
print("\nMetrics exported to gemma_monitoring_report.json")
# Run monitoring example
# monitoring_example()Rodina modelov Gemma predstavuje komplexný prístup spoločnosti Google k demokratizácii AI technológie pri zachovaní konkurencieschopného výkonu naprieč rôznymi aplikáciami a scenármi nasadenia. Prostredníctvom záväzku k otvorenému prístupu, multimodálnym schopnostiam a inovatívnym architektonickým návrhom umožňuje Gemma organizáciám a vývojárom využívať výkonné AI schopnosti bez ohľadu na ich zdroje alebo špecifické požiadavky.
Excelencia otvoreného zdroja: Gemma dokazuje, že modely otvoreného zdroja môžu dosiahnuť výkon konkurujúci proprietárnym alternatívam, pričom poskytujú transparentnosť, prispôsobenie a kontrolu nad nasadením AI.
Inovácie v multimodálnosti: Integrácia textových, vizuálnych a zvukových schopností v Gemma 3 a Gemma 3n predstavuje významný pokrok v prístupnej multimodálnej AI, umožňujúci komplexné porozumenie rôznym typom vstupov.
Architektúra orientovaná na mobilné zariadenia: Priekopnícka technológia Per-Layer Embeddings (PLE) a optimalizácia pre mobilné zariadenia v Gemma 3n ukazujú, že výkonná AI môže efektívne fungovať na zariadeniach s obmedzenými zdrojmi bez obetovania schopností.
Škálovateľné nasadenie: Rozsah od 1B do 27B parametrov, so špecializovanými mobilnými variantmi, umožňuje nasadenie naprieč celým spektrom výpočtových prostredí pri zachovaní konzistentnej kvality a výkonu.
Zodpovedná integrácia AI: Zabudované bezpečnostné opatrenia prostredníctvom ShieldGemma 2 a zodpovedné postupy vývoja zabezpečujú, že výkonné AI schopnosti môžu byť nasadené bezpečne a eticky.
Ako sa rodina Gemma bude ďalej vyvíjať, môžeme očakávať:
Vylepšené mobilné schopnosti: Ďalšiu optimalizáciu pre mobilné a edge nasadenie s integráciou architektúry Gemma 3n do hlavných platforiem ako Android a Chrome.
Rozšírené multimodálne porozumenie: Pokračujúci pokrok v integrácii videnia, jazyka a zvuku pre komplexnejšie AI zážitky.
Zlepšená efektivita: Neustále architektonické inovácie na dosiahnutie lepších pomerov výkonu na parameter a zníženie výpočtových požiadaviek.
Širšia integrácia ekosystému: Rozšírená podpora naprieč vývojovými rámcami, cloudovými platformami a nástrojmi na nasadenie pre bezproblémovú integráciu do existujúcich pracovných postupov.
Rast komunity: Pokračujúca expanzia Gemmaverse s modelmi, nástrojmi a aplikáciami vytvorenými komunitou, ktoré rozširujú základné schopnosti.
Či už vytvárate mobilné aplikácie s AI schopnosťami v reálnom čase, vyvíjate multimodálne vzdelávacie nástroje, budujete inteligentné automatizačné systémy alebo pracujete na globálnych aplikáciách vyžadujúcich multijazyčnú podporu, rodina Gemma poskytuje škálovateľné riešenia so silnou podporou komunity a komplexnou dokumentáciou.
Odporúčania na začiatok:
- Experimentujte s Google AI Studio pre okamžitú praktickú skúsenosť
- Stiahnite si modely z Hugging Face pre lokálny vývoj a prispôsobenie
- Preskúmajte špecializované varianty ako Gemma 3n pre mobilné aplikácie
- Implementujte multimodálne schopnosti pre komplexné AI zážitky
- Dodržiavajte bezpečnostné postupy pre produkčné nasadenie
Pre mobilný vývoj: Začnite s Gemma 3n E2B pre efektívne nasadenie so schopnosťami spracovania zvuku a videnia.
Pre podnikové aplikácie: Zvážte modely Gemma 3-12B alebo 27B pre maximálne schopnosti s volaním funkcií a pokročilým uvažovaním.
Pre globálne aplikácie: Využite podporu 140+ jazykov Gemma s kultúrne citlivým inžinierstvom promptov.
Pre špecializované prípady použitia: Preskúmajte prístupy doladenia a techniky optimalizácie pre konkrétne oblasti.
Rodina Gemma predstavuje budúcnosť vývoja AI, kde výkonné, schopné modely sú dostupné každému, od individuálnych vývojárov po veľké podniky. Kombináciou špičkového výskumu s prístupnosťou otvoreného zdroja vytvoril Google základ, ktorý umožňuje inovácie naprieč všetkými sektormi a mierkami.
Úspech Gemma s viac ako 100 miliónmi stiahnutí a 60 000+ variantmi vytvorenými komunitou demonštruje silu otvorenej spolupráce pri pokroku AI technológie. Ako sa posúvame vpred, rodina Gemma bude naďalej slúžiť ako katalyzátor AI inovácií, umožňujúci vývoj aplikácií, ktoré boli predtým možné len s proprietárnymi, drahými modelmi.
Budúcnosť AI je otvorená, prístupná a výkonná – a rodina Gemma vedie cestu k realizácii tejto vízie.
Oficiálna dokumentácia a modely:
- Google AI Studio: Vyskúšajte modely Gemma priamo
- Hugging Face Collections:
- Google AI Developer Documentation: Komplexné sprievodce Gemma
- Vertex AI Documentation: Sprievodce nasadením pre podniky
Technické zdroje:
- Výskumné práce a technické správy: Publikácie Google DeepMind
- Blogy pre vývojárov: Najnovšie oznámenia a tutoriály
- Modelové karty: Podrobné technické špecifikácie a výkonnostné benchmarky
Komunita a podpora:
- Hugging Face Community: Aktívne diskusie a príklady od komunity
- GitHub Repositories: Implementácie otvoreného zdroja a nástroje
- Fóra pre vývojárov: Podpora komunity Google AI Developer
- Stack Overflow: Otázky s tagmi a riešenia od komunity
Vývojové nástroje:
- Ollama: Jednoduché lokálne nasadenie
- vLLM: Výkonné podávanie
- Knižnica Transformers: Integrácia Hugging Face
- Google AI Edge: Optimalizácia nasadenia pre mobilné a edge zariadenia
Vzdelávacie cesty:
- Začiatočník: Začnite s Google AI Studio → Príklady Hugging Face → Lokálne nasadenie
- Vývojár: Integrácia Transformers → Vlastné aplikácie → Produkčné nasadenie
- Výskumník: Technické práce → Doladenie → Nové aplikácie
- Podnik: Nasadenie Vertex AI → Implementácia bezpečnosti → Optimalizácia škálovania
Rodina modelov Gemma nepredstavuje len kolekciu AI modelov, ale kompletný ekosystém pre budovanie budúcnosti prístupných, výkonných a zodpovedných AI aplikácií. Začnite objavovať už dnes a pripojte sa k rastúcej komunite vývojárov a výskumníkov, ktorí posúvajú hranice toho, čo je možné s AI otvoreného zdroja.
Upozornenie:
Tento dokument bol preložený pomocou služby AI prekladu Co-op Translator. Hoci sa snažíme o presnosť, prosím, berte na vedomie, že automatizované preklady môžu obsahovať chyby alebo nepresnosti. Pôvodný dokument v jeho rodnom jazyku by mal byť považovaný za autoritatívny zdroj. Pre kritické informácie sa odporúča profesionálny ľudský preklad. Nie sme zodpovední za akékoľvek nedorozumenia alebo nesprávne interpretácie vyplývajúce z použitia tohto prekladu.