[1] Tutorial de implementación de Llama 3.1 (muy detallado) desde cero hasta avanzado, una guía suficiente [2] Instaalción correcta de Ollama [3] Configuración de fuentes de espejo nacionales en Linux para acelerar la descarga de modelos [4] Introducción a Llama 3.1, flujo de implementación y ajuste eficiente
Servidor de implementación: H100 80G Modelo: Llama-3.1-8B-Instrucción
I. Implementación local del modelo
Descarga del modelo mediante huggingface: https://huggingface.co/meta-llama/Llama-3.1-8B
- Crear entorno virtual conda (versión de python 3.10 o superior)
conda create -n nombre_del_entorno python==3.11
- Activar entorno
- Instalar Pytorch en el entorno virtual
nvidia-smi # Verificar versión de CUDA Visitar sitio oficial de Pytorch: https://pytorch.org/get-started/previous-versions/Seleccionar versión adecuada de Pytorch, preferiblemente cercana pero no superior a la máxima versión soportada por CUDA del host, luego usar fuente espejo Por ejemplo: # Usar fuente espejo de Tsinghua pip install torch2.6.0 torchvision0.21.0 torchaudio==2.6.0 -i https://pypi.tuna.tsinghua.edu.cn/simple
- Actualizar pip
python -m pip install --upgrade pip
- Verificar y actualizar wget y md5sum
wget --version md5sum --version
Si no tiene estas herramientas, instálelas con:
apt-get install wget apt-get install md5sum
- Instalar transformers
pip install --upgrade transformers pip install accelerate -i https://pypi.tuna.tsinghua.edu.cn/simple
- Prueba:
import transformers
import torch
modelo_id = "meta-llama/Meta-Llama-3.1-8B-Instruct" # Modificar ruta
flujo = transformers.pipeline(
"generacion-texto",
modelo=modelo_id,
model_kwargs={"torch_dtype": torch.bfloat16},
device_map="auto",
)
mensajes = [
{"role": "system", "content": "Eres un chatbot pirata que siempre responde en lenguaje pirata!"},
{"role": "user", "content": "¿Quién eres?"},
]
resultados = flujo(
mensajes,
max_new_tokens=256,
)
print(resultados[0]["generated_text"][-1])
II. Implementación de un servicio API local de ejecución prolongada
- Instalar paquetes necesarios
pip install fastapi uvicorn pydantic -i https://pypi.tuna.tsinghua.edu.cn/simple
- Crear archivo de servicio API
# servicio_api.py
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import uvicorn
import logging
# Configurar registros
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Ruta del modelo
RUTA_MODELO = "/.../.../.../Llama-3.1-8B-Instruct" # Modificar con su propia ruta
# Definir modelos de solicitud/respuesta
class MensajeChat(BaseModel):
role: str
content: str
class SolicitudChat(BaseModel):
messages: List[MensajeChat]
max_tokens: Optional[int] = 200
temperature: Optional[float] = 0.7
top_p: Optional[float] = 0.9
class RespuestaChat(BaseModel):
response: str
usage: dict
# Inicializar aplicación FastAPI
app = FastAPI(title="API de Llama 3.1", version="1.0")
# Variables globales del modelo
modelo = None
tokenizer = None
flujo = None
@app.on_event("startup")
async def evento_inicio():
"""Cargar modelo al iniciar"""
global modelo, tokenizer, flujo
logger.info("Cargando modelo...")
try:
# Cargar tokenizer y modelo
tokenizer = AutoTokenizer.from_pretrained(RUTA_MODELO)
modelo = AutoModelForCausalLM.from_pretrained(
RUTA_MODELO,
torch_dtype=torch.float16,
device_map="auto",
low_cpu_mem_usage=True
)
# Crear pipeline
flujo = pipeline(
"generacion-texto",
model=modelo,
tokenizer=tokenizer,
device=0 if torch.cuda.is_available() else -1
)
logger.info(f"¡Modelo cargado! Dispositivo: {modelo.device}")
logger.info(f"CUDA disponible: {torch.cuda.is_available()}")
except Exception as e:
logger.error(f"Fallo al cargar modelo: {e}")
raise
@app.get("/")
async def raiz():
"""Punto raíz, devuelve estado del servicio"""
return {
"service": "API de Llama 3.1",
"status": "running",
"model": "Llama-3.1-8B-Instruct",
"device": str(modelo.device) if modelo else "no cargado"
}
@app.get("/health")
async def verificacion_salud():
"""Verificación de salud"""
return {"status": "healthy"}
@app.post("/chat/completions", response_model=RespuestaChat)
async def completar_chat(solicitud: SolicitudChat):
"""Interfaz de finalización de chat (formato similar a OpenAI API)"""
try:
# Formatear con plantilla de chat
texto = tokenizer.apply_chat_template(
[msg.dict() for msg in solicitud.messages],
tokenize=False,
add_generation_prompt=True
)
# Generar respuesta
resultados = flujo(
texto,
max_new_tokens=solicitud.max_tokens,
temperature=solicitud.temperature,
top_p=solicitud.top_p,
do_sample=True,
pad_token_id=tokenizer.eos_token_id
)
texto_respuesta = resultados[0]['generated_text']
# Extraer respuesta del asistente
if "assistant" in texto_respuesta:
texto_respuesta = texto_respuesta.split("assistant")[-1].strip()
else:
# Eliminar texto de entrada
texto_respuesta = texto_respuesta.replace(texto, "").strip()
# Calcular uso de tokens
tokens_entrada = len(tokenizer.encode(texto))
tokens_salida = len(tokenizer.encode(texto_respuesta))
return RespuestaChat(
response=texto_respuesta,
usage={
"prompt_tokens": tokens_entrada,
"completion_tokens": tokens_salida,
"total_tokens": tokens_entrada + tokens_salida
}
)
except Exception as e:
logger.error(f"Fallo en generación: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/generate")
async def generar_texto(prompt: str, max_tokens: int = 100):
"""Interfaz simple de generación de texto"""
try:
resultados = flujo(
prompt,
max_new_tokens=max_tokens,
temperature=0.7,
do_sample=True
)
return {
"text": resultados[0]['generated_text'],
"prompt": prompt
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
# Iniciar servidor
uvicorn.run(
app,
host="0.0.0.0", # Permitir acceso externo
port=8000,
log_level="info"
)
- Iniciar servicio API
Ejecución en primer plano (ver registros) python servicio_api.py
- Probar API
# Probar punto final
curl http://localhost:8000/
# Verificación de salud
curl http://localhost:8000/health
# Usar interfaz de chat
curl -X POST "http://localhost:8000/chat/completions" \
-H "Content-Type: application/json" \
-d '{
"messages": [
{"role": "system", "content": "Eres un asistente AI útil"},
{"role": "user", "content": "¿Cuál es la capital de China?"}
],
"max_tokens": 100
}'
# Interfaz de generación simple
curl -X POST "http://localhost:8000/generate?prompt=Hola&max_tokens=50"
III. Implementación y uso de servicios API entre servidores
1. Servidor
Servidor H100 (implementación de API)
# servidor_h100.py
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, field_validator, model_validator, ConfigDict
from typing import List, Optional
import uvicorn
import logging
import time
import os
from contextlib import asynccontextmanager
# Configuración de registros
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Ruta del modelo
RUTA_MODELO = "/dddd/ddddd/LLMs/Llama-3.1-8B-Instruct"
# Variables globales del modelo
modelo = None
tokenizer = None
@asynccontextmanager
async def lifespan(app: FastAPI):
# Cargar modelo al iniciar
global modelo, tokenizer
logger.info("Cargando modelo Llama-3.1-8B-Instruct en H100...")
try:
tokenizer = AutoTokenizer.from_pretrained(RUTA_MODELO)
if tokenizer.pad_token is None:
tokenizer.pad_token = tokenizer.eos_token
modelo = AutoModelForCausalLM.from_pretrained(
RUTA_MODELO,
torch_dtype=torch.float16,
device_map="auto",
low_cpu_mem_usage=True
)
logger.info(f"✅ ¡Modelo cargado exitosamente! Dispositivo: {modelo.device}")
yield
except Exception as e:
logger.error(f"❌ Fallo al cargar modelo: {e}")
raise RuntimeError("Falló la inicialización del modelo") from e
finally:
# Limpiar recursos
if modelo is not None:
del modelo
torch.cuda.empty_cache()
app = FastAPI(
title="API de Llama 3.1 Instruct",
description="API compatible con OpenAI para Llama-3.1-8B-Instruct en H100",
version="1.0",
lifespan=lifespan
)
# ======================
# Modelos Pydantic
# ======================
class MensajeChat(BaseModel):
role: str
content: str
class SolicitudCompletacionChat(BaseModel):
model_config = ConfigDict(extra="ignore")
messages: Optional[List[MensajeChat]] = None
prompt: Optional[str] = None
model: str = "llama-3.1-8b"
max_tokens: int = 200
temperature: float = 0.7
top_p: float = 0.9
stream: bool = False
@model_validator(mode='after')
def validar_entrada(self):
if not self.messages and not self.prompt:
raise ValueError("Debe proporcionar 'messages' o 'prompt'")
if self.prompt and not self.messages:
# Compatibilidad con versión anterior: convertir prompt a messages
self.messages = [MensajeChat(role="user", content=self.prompt)]
return self
@field_validator('max_tokens')
@classmethod
def validar_max_tokens(cls, v):
if v < 1 or v > 2048:
raise ValueError("max_tokens debe estar entre 1~2048")
return v
class OpcionRespuestaCompletacionChat(BaseModel):
index: int = 0
message: MensajeChat
finish_reason: str = "stop"
class RespuestaCompletacionChat(BaseModel):
id: str
object: str = "chat.completion"
created: int
model: str
choices: List[OpcionRespuestaCompletacionChat]
usage: dict
# ======================
# Rutas
# ======================
@app.get("/")
async def raiz():
return {
"service": "API de Llama 3.1 Instruct",
"status": "running",
"device": str(modelo.device) if modelo else "uninitialized"
}
@app.get("/health")
async def verificacion_salud():
return {"status": "healthy", "model_loaded": modelo is not None}
@app.post("/v1/chat/completions")
async def completar_chat(solicitud: SolicitudCompletacionChat):
if modelo is None or tokenizer is None:
raise HTTPException(status_code=503, detail="El modelo aún no se ha cargado completamente")
if solicitud.stream:
raise HTTPException(status_code=400, detail="La salida en streaming aún no es compatible")
try:
# Preparar entrada
if solicitud.messages is None:
raise ValueError("messages no puede estar vacío")
# Asegurar que messages es una lista de diccionarios
mensajes_dict = [msg.model_dump() for msg in solicitud.messages]
# Usar apply_chat_template
encoding = tokenizer.apply_chat_template(
mensajes_dict,
add_generation_prompt=True,
return_tensors="pt"
)
# Extraer input_ids
input_ids = encoding.input_ids.to(modelo.device)
input_length = input_ids.shape[1]
# Generar
start_time = time.time()
with torch.no_grad():
outputs = modelo.generate(
input_ids=input_ids,
max_new_tokens=solicitud.max_tokens,
temperature=solicitud.temperature,
top_p=solicitud.top_p,
do_sample=True,
pad_token_id=tokenizer.pad_token_id,
eos_token_id=tokenizer.eos_token_id,
use_cache=True
)
gen_time = time.time() - start_time
# Decodificar la parte recién generada
new_tokens = outputs[0][input_length:]
response_text = tokenizer.decode(new_tokens, skip_special_tokens=True).strip()
# Construir respuesta
response = RespuestaCompletacionChat(
id=f"cmpl-{int(time.time())}",
created=int(time.time()),
model=solicitud.model,
choices=[
OpcionRespuestaCompletacionChat(
message=MensajeChat(role="assistant", content=response_text)
)
],
usage={
"prompt_tokens": input_length,
"completion_tokens": len(new_tokens),
"total_tokens": input_length + len(new_tokens)
}
)
logger.info(f"✅ Generación completada | Entrada: {len(solicitud.messages)} mensajes | Salida: {len(response_text)} caracteres | Tiempo: {gen_time:.2f}s")
return response
except Exception as e:
logger.error(f"❌ Error de generación: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
port = int(os.getenv("PORT", 8000))
uvicorn.run(
app,
host="0.0.0.0",
port=port,
log_level="info",
workers=1
)
Iniciar servicio API en H100
# Ejecución en primer plano (ver registros)
python servidor_h100.py
Prueba local en servidor
# prueba_rapida.py
from transformers import AutoTokenizer
RUTA_MODELO = "/ddd/ddddd/LLMs/Llama-3.1-8B-Instruct"
tokenizer = AutoTokenizer.from_pretrained(RUTA_MODELO)
mensajes = [
{"role": "system", "content": "Eres un asistente AI útil."},
{"role": "user", "content": "Hola, por favor preséntate."}
]
print("Probando la lógica corregida...")
encoding = tokenizer.apply_chat_template(mensajes, add_generation_prompt=True, return_tensors="pt")
print(f"tipo de encoding: {type(encoding)}")
print(f"tipo de encoding.input_ids: {type(encoding.input_ids)}")
print(f"encoding.input_ids.shape: {encoding.input_ids.shape}")
print("✅ ¡Prueba superada!")
(llama_3_1) root@f2osdcap97m-0:/ddd/qddyj/LLMs# python prueba.py Probando la lógica corregida… tipo de encoding: <class 'transformers.tokenization_utils_base.BatchEncoding'> tipo de encoding.input_ids: <class 'torch.Tensor'> encoding.input_ids.shape: torch.Size([1, 51]) ✅ ¡Prueba superada!
2. Cliente
Entorno de implementación
conda create -n llama3 python=3.11
conda activate llama3
(1) En una terminal, establecer túnel SSH (reenvío de puertos)
ssh -L 8000:localhost:8000 root@115.11.11.111 -p 30304 # 115.11.11.111 es la IP del servidor remoto
Probar si se conecta al servidor
netstat -tuln | grep 8000
Si hay salida, la conexión es exitosa
tcp 0 0 127.0.0.1:8000 0.0.0.0:* LISTEN
tcp6 0 0 ::1:8000 :::* LISTEN
(2) En otra terminal, ejecutar script Python
python preguntar_llm.py
# cliente_llama3.py
from openai import OpenAI
cliente = OpenAI(
base_url="http://localhost:8000/v1", # Mantener /v1 aquí
api_key="not-needed"
)
respuesta = cliente.chat.completions.create(
model="llama-3.1-8b",
messages=[
{"role": "system", "content": "Eres un asistente AI útil."},
{"role": "user", "content": "Hola, por favor preséntate."}
],
max_tokens=150
)
print(respuesta.choices[0].message.content)
Respuesta:
(llama3) rd-111s@rd-111s-Z790-PG-ITX-TB4:~/document/ddd/LLM$ python cliente_llama3.py
¡Hola! Soy LLaMA, un asistente de inteligencia artificial. Mi nombre proviene de "Large Language Model Application", soy un modelo de lenguaje que puede entender y generar lenguaje humano.
Puedo responder a diversas preguntas, proporcionar información, ayudarte a realizar tareas e incluso realizar creación y conversación. Mi conocimiento es amplio, cubriendo varios campos, incluyendo但不限于 tecnología, historia, cultura, entretenimiento, etc.
Soy un modelo de aprendizaje que puede mejorar constantemente. A través de la interacción con los usuarios, puedo mejorar mi capacidad de comprensión y generación, proporcionando un mejor servicio.
Espero ser tu asistente útil, ayudándote a resolver problemas, encontrar respuestas e incluso traer algo de diversión
Nuevo problema 1: el puerto del servidor está ocupado o el cliente no puede conectarse al servidor a través de ssh
Solución 1: Cambiar puerto en el servidor
if __name__ == "__main__":
# Iniciar servidor
uvicorn.run(
app,
host="0.0.0.0", # Permitir acceso externo
port=8000,
log_level="info"
)
Solución 2: Detener todos los procesos relacionados
# Verificar puerto 8000
(llama_3_1) root@f2osdcap97m-0:/ddd/dd/LLMs# netstat -tulpn | grep :8000
tcp 0 0 127.0.0.1:8000 0.0.0.0:* LISTEN 1845141/ssh
Matar proceso
> kill -9 1845141
Cambiar método de creación de terminal: Abrir una nueva ventana de terminal y ejecutar
# Ejecutar en una nueva terminal local
ssh -4 -N -L 8000:localhost:8000 root@115.24.15.218 -p 30304
# Es decir, agregar -4 -N -L 8000:localhost:8000 al ssh original root@115.24.15.218 -p 30304
Luego de ingresar la contraseña, puede no haber contenido de salida, no cerrar esta terminal, abrir otra para probar
(llama3) rd-111s@rd-111s-Z790-PG-ITX-TB4:~/document/ddd/code/SGRL (v2)$ netstat -tuln | grep 8000
tcp 0 0 127.0.0.1:8000 0.0.0.0:* LISTEN
(llama3) rd-111s@rd-111s-Z790-PG-ITX-TB4:~/document/ddd/code/SGRL (v2)$
Conexión exitosa
Nuevo problema 2: la conexión ssh es inestable
Solución: crear un script para reconectar si se interrumpe Método de uso:
# Establecer variable de entorno
export SSH_TUNNEL_PASSWORD="su_contraseña"
# Ejecutar script
./tunel_ssh.sh
Contenido del script:
#!/bin/bash
# Parámetros del túnel SSH
HOST_SSH="root@115.24.15.111" # nombre de usuario@dirección IP del servidor
PUERTO_SSH="111111" # número de puerto del servicio SSH
PUERTO_LOCAL="8000" # puerto local que escuchará
HOST_REMOTO="localhost" # aquí localhost se refiere al servidor remoto en sí
PUERTO_REMOTO="8000" # puerto del servicio de destino en el servidor remoto
# Leer contraseña desde variable de entorno
CONTRASEÑA_SSH="${SSH_TUNNEL_PASSWORD}"
# Verificar si la contraseña está configurada
if [ -z "$CONTRASEÑA_SSH" ]; then
echo "Error: configure la variable de entorno SSH_TUNNEL_PASSWORD"
echo "Uso: export SSH_TUNNEL_PASSWORD='su_contraseña'"
exit 1
fi
# Archivo de registro
ARCHIVO_LOG="tunel_ssh.log" # configure su propia ruta
ARCHIVO_PID="tunel_ssh.pid" # configure su propia ruta
# Función: verificar si el puerto está escuchando
verificar_puerto() {
if command -v nc &> /dev/null; then
nc -z localhost $PUERTO_LOCAL
elif command -v telnet &> /dev/null; then
echo "" | telnet localhost $PUERTO_LOCAL 2>&1 | grep -q "Connected"
else
ss -tln | grep -q ":$PUERTO_LOCAL"
fi
}
# Función: iniciar túnel SSH
iniciar_tunel() {
echo "$(date): Iniciando túnel SSH..." >> "$ARCHIVO_LOG"
if ! command -v sshpass &> /dev/null; then
echo "$(date): Error: sshpass no instalado" >> "$ARCHIVO_LOG"
return 1
fi
if command -v autossh &> /dev/null; then
sshpass -p "$CONTRASEÑA_SSH" autossh -M 0 -4 -N -L ${PUERTO_LOCAL}:${HOST_REMOTO}:${PUERTO_REMOTO} ${HOST_SSH} -p ${PUERTO_SSH} \
-o "ServerAliveInterval=30" \
-o "ServerAliveCountMax=3" \
-o "ExitOnForwardFailure=yes" \
-o "StrictHostKeyChecking=no" &
echo $! > "$ARCHIVO_PID"
else
sshpass -p "$CONTRASEÑA_SSH" ssh -4 -N -L ${PUERTO_LOCAL}:${HOST_REMOTO}:${PUERTO_REMOTO} ${HOST_SSH} -p ${PUERTO_SSH} \
-o "ServerAliveInterval=30" \
-o "ServerAliveCountMax=3" \
-o "ExitOnForwardFailure=yes" \
-o "StrictHostKeyChecking=no" &
echo $! > "$ARCHIVO_PID"
fi
sleep 3
}
# Función: detener túnel SSH
detener_tunel() {
if [ -f "$ARCHIVO_PID" ]; then
PID=$(cat "$ARCHIVO_PID")
if kill -0 $PID 2>/dev/null; then
echo "$(date): Deteniendo túnel SSH existente (PID: $PID)" >> "$ARCHIVO_LOG"
kill $PID
sleep 2
fi
rm -f "$ARCHIVO_PID"
fi
# Asegurarse de que todos los procesos ssh se limpien
pkill -f "ssh.*-L ${PUERTO_LOCAL}:${HOST_REMOTO}:${PUERTO_REMOTO}" 2>/dev/null
}
# Función de limpieza
limpiar() {
echo "$(date): Señal de recepción de salida, limpiando procesos..." >> "$ARCHIVO_LOG"
detener_tunel
exit 0
}
# Configurar manejo de señales
trap limpiar SIGINT SIGTERM
# Bucle principal
echo "$(date): Script de monitoreo de túnel SSH iniciado" >> "$ARCHIVO_LOG"
echo "$(date): Destino: ${HOST_SSH}:${PUERTO_SSH}" >> "$ARCHIVO_LOG"
echo "$(date): Reenvío local: ${PUERTO_LOCAL} -> ${HOST_REMOTO}:${PUERTO_REMOTO}" >> "$ARCHIVO_LOG"
while true; do
# Verificar si el túnel está funcionando
if ! verificar_puerto; then
echo "$(date): Túnel no en ejecución o puerto no disponible, reiniciando..." >> "$ARCHIVO_LOG"
detener_tunel
iniciar_tunel
fi
# Verificar cada 10 segundos
sleep 10
done