Integración de API

Integra modelos de IA que se ejecutan en Clore.ai en tus aplicaciones

Integra modelos de IA que se ejecutan en CLORE.AI en tus aplicaciones.

Despliega servidores API en CLORE.AI Marketplace.

Inicio rápido

La mayoría de los servicios de IA en CLORE.AI ofrecen APIs compatibles con OpenAI. Reemplaza la URL base y estarás listo.

from openai import OpenAI

client = OpenAI(
    base_url="http://<tu-servidor-clore>:8000/v1",
    api_key="no-necesario"  # La mayoría de las instalaciones autohospedadas no requieren clave
)

response = client.chat.completions.create(
    model="meta-llama/Llama-3.1-8B-Instruct",
    messages=[{"role": "user", "content": "¡Hola!"}]
)
print(response.choices[0].message.content)

---

APIs LLM

vLLM (Compatible con OpenAI)

Configuración del servidor:

python -m vllm.entrypoints.openai.api_server \
    --model meta-llama/Llama-3.1-8B-Instruct \
    --host 0.0.0.0 --port 8000

Cliente Python:

from openai import OpenAI

client = OpenAI(base_url="http://server:8000/v1", api_key="dummy")

# Completado de chat
response = client.chat.completions.create(
    model="meta-llama/Llama-3.1-8B-Instruct",
    messages=[
        {"role": "system", "content": "Eres un asistente útil."},
        {"role": "user", "content": "Escribe un poema sobre programación"}
    ],
    temperature=0.7,
    max_tokens=500
)
print(response.choices[0].message.content)

# Transmisión
stream = client.chat.completions.create(
    model="meta-llama/Llama-3.1-8B-Instruct",
    messages=[{"role": "user", "content": "Cuéntame una historia"}],
    stream=True
)
for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

Cliente Node.js:

import OpenAI from 'openai';

const client = new OpenAI({
    baseURL: 'http://server:8000/v1',
    apiKey: 'dummy'
});

async function chat(message) {
    const response = await client.chat.completions.create({
        model: 'meta-llama/Llama-3.1-8B-Instruct',
        messages: [{ role: 'user', content: message }]
    });
    return response.choices[0].message.content;
}

// Transmisión
async function streamChat(message) {
    const stream = await client.chat.completions.create({
        model: 'meta-llama/Llama-3.1-8B-Instruct',
        messages: [{ role: 'user', content: message }],
        stream: true
    });

    for await (const chunk of stream) {
        process.stdout.write(chunk.choices[0]?.delta?.content || '');
    }
}

cURL:

curl http://server:8000/v1/chat/completions \
    -H "Content-Type: application/json" \
    -d '{
        "model": "meta-llama/Llama-3.1-8B-Instruct",
        "messages": [{"role": "user", "content": "¡Hola!"}]
    }'

API de Ollama

Python:

import requests

# Generar
response = requests.post('http://server:11434/api/generate', json={
    'model': 'llama3.2',
    'prompt': '¿Por qué el cielo es azul?',
    'stream': False
})
print(response.json()['response'])

# Chat
response = requests.post('http://server:11434/api/chat', json={
    'model': 'llama3.2',
    'messages': [
        {'role': 'user', 'content': '¡Hola!'}
    ],
    'stream': False
})
print(response.json()['message']['content'])

# Transmisión
response = requests.post('http://server:11434/api/chat', json={
    'model': 'llama3.2',
    'messages': [{'role': 'user', 'content': 'Cuéntame una historia'}],
    'stream': True
}, stream=True)

for line in response.iter_lines():
    if line:
        data = json.loads(line)
        print(data['message']['content'], end='', flush=True)

Ollama también admite el formato OpenAI:

from openai import OpenAI

client = OpenAI(base_url='http://server:11434/v1', api_key='ollama')
# Usa el mismo código que en los ejemplos de vLLM

API de TGI

Python:

import requests

# Generar
response = requests.post('http://server:8080/generate', json={
    'inputs': '¿Qué es el aprendizaje automático?',
    'parameters': {
        'max_new_tokens': 200,
        'temperature': 0.7,
        'do_sample': True
    }
})
print(response.json()['generated_text'])

# Transmisión
response = requests.post('http://server:8080/generate_stream', json={
    'inputs': 'Explica la computación cuántica',
    'parameters': {'max_new_tokens': 500}
}, stream=True)

for line in response.iter_lines():
    if line:
        data = json.loads(line.decode().replace('data:', ''))
        print(data.get('token', {}).get('text', ''), end='', flush=True)

---

APIs de generación de imágenes

API de Stable Diffusion WebUI

Habilitar API: Agregar --api al comando de lanzamiento.

Python:

import requests
import base64
from PIL import Image
from io import BytesIO

def txt2img(prompt, negative_prompt="", steps=20, width=512, height=512):
    response = requests.post('http://server:7860/sdapi/v1/txt2img', json={
        'prompt': prompt,
        'negative_prompt': negative_prompt,
        'steps': steps,
        'width': width,
        'height': height,
        'sampler_name': 'DPM++ 2M Karras',
        'cfg_scale': 7
    })

    # Decodificar imagen base64
    image_data = base64.b64decode(response.json()['images'][0])
    return Image.open(BytesIO(image_data))

# Generar
image = txt2img(
    prompt="Un hermoso atardecer sobre montañas, fotorrealista, 8k",
    negative_prompt="borroso, baja calidad"
)
image.save("output.png")

# img2img
def img2img(prompt, image_path, denoising=0.5):
    with open(image_path, 'rb') as f:
        image_b64 = base64.b64encode(f.read()).decode()

    response = requests.post('http://server:7860/sdapi/v1/img2img', json={
        'prompt': prompt,
        'init_images': [image_b64],
        'denoising_strength': denoising,
        'steps': 30
    })

    image_data = base64.b64decode(response.json()['images'][0])
    return Image.open(BytesIO(image_data))

Node.js:

const axios = require('axios');
const fs = require('fs');

async function txt2img(prompt) {
    const response = await axios.post('http://server:7860/sdapi/v1/txt2img', {
        prompt: prompt,
        steps: 20,
        width: 512,
        height: 512
    });

    const imageBuffer = Buffer.from(response.data.images[0], 'base64');
    fs.writeFileSync('output.png', imageBuffer);
}

API de ComfyUI

Python:

import json
import urllib.request
import urllib.parse
import websocket
import uuid

SERVER = "server:8188"

def queue_prompt(workflow):
    """Encolar un flujo de trabajo para su ejecución"""
    data = json.dumps({"prompt": workflow}).encode('utf-8')
    req = urllib.request.Request(f"http://{SERVER}/prompt", data=data)
    return json.loads(urllib.request.urlopen(req).read())

def get_image(filename, subfolder, folder_type):
    """Descargar imagen generada"""
    params = urllib.parse.urlencode({
        "filename": filename,
        "subfolder": subfolder,
        "type": folder_type
    })
    with urllib.request.urlopen(f"http://{SERVER}/view?{params}") as response:
        return response.read()

# Cargar flujo de trabajo desde archivo
with open('workflow.json') as f:
    workflow = json.load(f)

# Modificar prompt
workflow["6"]["inputs"]["text"] = "Un gato con un sombrero"

# Encolar y obtener resultado
result = queue_prompt(workflow)
print(f"Encolado: {result}")

WebSocket para progreso:

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    if data['type'] == 'progress':
        print(f"Progreso: {data['data']['value']}/{data['data']['max']}")
    elif data['type'] == 'executed':
        print("¡Generación completada!")

ws = websocket.WebSocketApp(
    f"ws://{SERVER}/ws",
    on_message=on_message
)
ws.run_forever()

FLUX con Diffusers

import torch
from diffusers import FluxPipeline
import base64
from io import BytesIO

# Cargar modelo (hacer una vez)
pipe = FluxPipeline.from_pretrained(
    "black-forest-labs/FLUX.1-schnell",
    torch_dtype=torch.bfloat16
)
pipe.to("cuda")

def generate_image(prompt, height=1024, width=1024):
    image = pipe(
        prompt,
        height=height,
        width=width,
        num_inference_steps=4,
        guidance_scale=0.0
    ).images[0]
    return image

# Wrapper de API simple con Flask
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/generate', methods=['POST'])
def generate():
    data = request.json
    image = generate_image(data['prompt'])

    # Convertir a base64
    buffer = BytesIO()
    image.save(buffer, format='PNG')
    img_b64 = base64.b64encode(buffer.getvalue()).decode()

    return jsonify({'image': img_b64})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

---

APIs de audio

Whisper Transcription

Usando whisper-asr-webservice:

import requests

def transcribe(audio_path):
    with open(audio_path, 'rb') as f:
        response = requests.post(
            'http://server:9000/asr',
            files={'audio_file': f},
            data={
                'task': 'transcribe',
                'language': 'en',
                'output': 'json'
            }
        )
    return response.json()['text']

text = transcribe('audio.mp3')
print(text)

API directa de Whisper:

import whisper
from flask import Flask, request, jsonify

model = whisper.load_model("large-v3")

app = Flask(__name__)

@app.route('/transcribe', methods=['POST'])
def transcribe():
    audio = request.files['audio']
    audio.save('/tmp/audio.mp3')

    result = model.transcribe('/tmp/audio.mp3')
    return jsonify({'text': result['text']})

Texto a voz (Bark)

from bark import SAMPLE_RATE, generate_audio, preload_models
from scipy.io.wavfile import write as write_wav
import base64
from flask import Flask, request, jsonify

preload_models()

app = Flask(__name__)

@app.route('/tts', methods=['POST'])
def text_to_speech():
    text = request.json['text']
    audio = generate_audio(text)

    # Guardar en archivo
    write_wav('/tmp/output.wav', SAMPLE_RATE, audio)

    # Devolver base64
    with open('/tmp/output.wav', 'rb') as f:
        audio_b64 = base64.b64encode(f.read()).decode()

    return jsonify({'audio': audio_b64})

---

Construyendo aplicaciones

Aplicación de chat

from flask import Flask, request, jsonify, Response
from openai import OpenAI
import json

app = Flask(__name__)
client = OpenAI(base_url="http://localhost:8000/v1", api_key="dummy")

@app.route('/chat', methods=['POST'])
def chat():
    messages = request.json.get('messages', [])

    response = client.chat.completions.create(
        model="meta-llama/Llama-3.1-8B-Instruct",
        messages=messages,
        temperature=0.7
    )

    return jsonify({
        'response': response.choices[0].message.content
    })

@app.route('/chat/stream', methods=['POST'])
def chat_stream():
    messages = request.json.get('messages', [])

    def generate():
        stream = client.chat.completions.create(
            model="meta-llama/Llama-3.1-8B-Instruct",
            messages=messages,
            stream=True
        )
        for chunk in stream:
            if chunk.choices[0].delta.content:
                yield f"data: {json.dumps({'content': chunk.choices[0].delta.content})}\n\n"
        yield "data: [DONE]\n\n"

    return Response(generate(), mimetype='text/event-stream')

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

Servicio de generación de imágenes

from flask import Flask, request, jsonify, send_file
import requests
import base64
from io import BytesIO

app = Flask(__name__)
SD_API = "http://localhost:7860"

@app.route('/generate', methods=['POST'])
def generate():
    data = request.json

    response = requests.post(f'{SD_API}/sdapi/v1/txt2img', json={
        'prompt': data['prompt'],
        'negative_prompt': data.get('negative_prompt', ''),
        'steps': data.get('steps', 20),
        'width': data.get('width', 512),
        'height': data.get('height', 512)
    })

    image_b64 = response.json()['images'][0]

    if data.get('return_base64'):
        return jsonify({'image': image_b64})

    # Devolver como archivo
    image_data = base64.b64decode(image_b64)
    return send_file(BytesIO(image_data), mimetype='image/png')

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

Pipeline multimodal

from flask import Flask, request, jsonify
from openai import OpenAI
import requests
import base64

app = Flask(__name__)
llm_client = OpenAI(base_url="http://localhost:8000/v1", api_key="dummy")
SD_API = "http://localhost:7860"

@app.route('/create-image-from-description', methods=['POST'])
def create_image():
    description = request.json['description']

    # Paso 1: Generar prompt detallado con LLM
    prompt_response = llm_client.chat.completions.create(
        model="meta-llama/Llama-3.1-8B-Instruct",
        messages=[{
            "role": "user",
            "content": f"Crea un prompt detallado para generación de imágenes para: {description}. Incluye estilo, iluminación y detalles de composición. Devuelve solo el prompt, sin explicación."
        }]
    )
    detailed_prompt = prompt_response.choices[0].message.content

    # Paso 2: Generar imagen
    image_response = requests.post(f'{SD_API}/sdapi/v1/txt2img', json={
        'prompt': detailed_prompt,
        'steps': 25,
        'width': 1024,
        'height': 1024
    })

    return jsonify({
        'prompt_used': detailed_prompt,
        'image': image_response.json()['images'][0]
    })

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

---

Manejo de errores

from openai import OpenAI, APIError, APIConnectionError
import time

client = OpenAI(base_url="http://server:8000/v1", api_key="dummy")

def chat_with_retry(messages, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model="meta-llama/Llama-3.1-8B-Instruct",
                messages=messages,
                timeout=60
            )
            return response.choices[0].message.content

        except APIConnectionError as e:
            print(f"Error de conexión (intento {attempt + 1}): {e}")
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # Reintentos exponenciales
            else:
                raise

        except APIError as e:
            print(f"Error de la API: {e}")
            raise

# Uso
try:
    result = chat_with_retry([{"role": "user", "content": "Hola"}])
    print(result)
except Exception as e:
    print(f"Falló después de los reintentos: {e}")

---

Mejores prácticas

1. Agrupación de conexiones - Reutilizar conexiones HTTP

2. Solicitudes asíncronas - Usar aiohttp para llamadas concurrentes

3. Tiempos de espera - Siempre establecer timeouts en las solicitudes

4. Lógica de reintentos - Manejar fallos temporales

5. Limitación de tasa - No sobrecargar el servidor

6. Verificaciones de salud - Monitorizar la disponibilidad del servidor

---

Próximos pasos

• Procesamiento por lotes - Procesar grandes cargas de trabajo

• Configuración multi-GPU - Escalar tu despliegue

• Comparación de LLM - Elegir el servidor adecuado