Integración de API

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

💡 Recomendado: Use el oficial SDK de Python de clore-ai en lugar de solicitudes HTTP sin procesar para gestionar servidores y pedidos de Clore.ai. Límite de tasa incorporado, reintentos, seguridad de tipos y soporte asíncrono.

Integra modelos de IA ejecutándose en CLORE.AI en tus aplicaciones.

Despliega servidores API en Marketplace de CLORE.AI.

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 servicial."},
        {"role": "user", "content": "Escribe un poema sobre programar"}
    ],
    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 soporta el formato de OpenAI:

from openai import OpenAI

client = OpenAI(base_url='http://server:11434/v1', api_key='ollama')
# Usa el mismo código que 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 en base64
    image_data = base64.b64decode(response.json()['images'][0])
    return Image.open(BytesIO(image_data))

# Generar
image = txt2img(
    prompt="Una hermosa puesta de sol 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 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 completa!")

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

Transcripción Whisper

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)

Canal Multi-Modal

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 imagen 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)  # Retroceso exponencial
            else:
                raise

        except APIError as e:
            print(f"Error de 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 reintentos: {e}")

---

Mejores Prácticas

1. Pool de conexiones - Reutiliza conexiones HTTP

2. Peticiones asíncronas - Usa aiohttp para llamadas concurrentes

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

4. Lógica de reintento - Maneja fallos temporales

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

6. Cheques de salud - Monitorea la disponibilidad del servidor

---

Próximos Pasos

• Procesamiento por Lotes - Procesa grandes cargas de trabajo

• Configuración Multi-GPU - Escala tu despliegue

• Comparación de LLM - Elige el servidor adecuado