jevlaya
Español
Documentación

Patrones de arquitectura y casos de uso en producción

Laya es un motor de decisiones en el dispositivo y no autorregresivo, construido sobre una arquitectura transformer de una sola pasada hacia adelante (ModernBERT-large y mmBERT-base). En lugar de generar tokens de forma secuencial como un LLM (lo que conlleva costos variables de generación de tokens, sobrecarga del bucle de decodificación y esquemas de salida impredecibles), Laya calcula distribuciones de probabilidad calibradas sobre preguntas discretas en una sola pasada.

Este documento sirve como plano de arquitectura para arquitectos de sistemas e ingenieros de backend que integran Laya en entornos de producción.

Motor de decisiones vs. LLM vs. recuperación por embeddings

Elegir la primitiva correcta depende de las restricciones de latencia, la topología de alojamiento y de si la tarea requiere generación libre o clasificación discreta:

Dimensión Recuperación por embeddings Motor de decisiones Laya LLM autorregresivo
Modelo de cómputo Distancia coseno entre vectores Una sola pasada hacia adelante (cabeza enmascarada no autorregresiva) Generación secuencial token a token
Perfil de ejecución Búsqueda en índice de vecinos más cercanos Una sola pasada hacia adelante fija (sin bucle de decodificación) Bucle de decodificación iterativo que escala con la longitud de salida
Alojamiento y topología En proceso o base de datos vectorial En proceso (CPU/GPU local) o daemon HTTP autoalojado API remota alojada o servicio GPU de modelos grandes
Atención de contexto Representación vectorial agrupada Atención cruzada bidireccional profunda sobre toda la entrada Atención secuencial causal
Salida estructurada Fragmentos recuperados sin estructura Distribuciones nativas alineadas al esquema (choice, score, noul) Texto libre que requiere reparación de JSON o muestreo de esquema
Carga de trabajo principal Recuperación amplia de candidatos Clasificación discreta, gating de políticas y enrutamiento Síntesis abierta, traducción y generación

1. Gateway de ingreso inteligente

En arquitecturas por niveles, una gran fracción de las solicitudes entrantes no requiere las capacidades generativas de un LLM autorregresivo. Consultas como preguntas frecuentes estándar, consultas deterministas de estado o decisiones de enrutamiento categóricas pueden evaluarse localmente.

Laya funciona como un gateway de ingreso inteligente: clasifica la intención y la complejidad de la consulta en una sola pasada hacia adelante local. Las solicitudes deterministas se resuelven localmente a través de endpoints internos o respuestas en caché, mientras que las tareas generativas complejas se reenvían a LLMs ascendentes.

Arquitectura

graph TD
    A[User Request] --> B["<b>Laya Gateway Router</b><br/>• query_complexity: simple | moderate | complex<br/>• intent: faq | account_lookup | creative_synthesis"]
    B -->|Simple & High Confidence| C["<b>Local In-Process Resolution</b><br/>Deterministic FAQ / Internal API"]
    B -->|Complex or Low Confidence| D["<b>Upstream Generative LLM</b><br/>Open-ended synthesis & reasoning"]

Implementación

import laya

agent = laya.load("convaiinnovations/laya")

GATEWAY_QUESTIONS = {
    "complexity": {
        "type": "choice",
        "instructions": "How complex is the user's request?",
        "criteria": {
            "canned": "A greeting, standard FAQ, or simple status request.",
            "structured": "A deterministic data query that can be answered by an API.",
            "complex": "Requires creative generation, multi-step code, or complex analysis.",
        },
    },
    "requires_reasoning": {
        "type": "noul",
        "instructions": "Does this query require frontier model reasoning?",
    },
}

def route_request(user_prompt: str):
    res = agent.system_one(user_prompt, GATEWAY_QUESTIONS, min_confidence=0.85)
    answers = res["answers"]

    complexity = answers["complexity"]["choice"]
    low_confidence = answers["complexity"].get("low_confidence", False)

    # Abstain or escalate if complex or unconfident
    if low_confidence or complexity == "complex" or answers["requires_reasoning"]["noul"] > 0.5:
        return call_frontier_llm(user_prompt)

    if complexity == "canned":
        return lookup_faq_response(user_prompt)
    return execute_internal_api(user_prompt)

2. Enrutador de turnos de voz e interrupciones de baja latencia

Los agentes de voz conversacionales (WebRTC, telefonía) operan bajo estrictas restricciones de turnos: los retrasos al detectar cuándo habla o interrumpe un usuario crean silencios antinaturales en la conversación. Esperar a que un modelo generativo completo produzca su primer token introduce un retraso evitable cuando quien llama simplemente asiente o interrumpe.

Laya puede colocarse directamente después de la transcripción de voz a texto (STT) para clasificar el flujo conversacional y la intención del usuario en una sola pasada hacia adelante: activar audio de relleno rápido o detener la reproducción de audio de inmediato cuando se detecta una interrupción, mientras delega las consultas complejas al pipeline de síntesis completo.

Arquitectura

graph TD
    A[User Voice Audio] --> B["<b>Speech-to-Text</b><br/>Streaming Audio Transcription"]
    B --> C["<b>Laya Voice Router</b><br/>• intent: ack | reject | interrupt | inquiry<br/>• is_interruption: noul probability"]
    C -->|Interruption: score &gt; 0.6| D["<b>Halt Audio Playback</b><br/>Immediate playback cutoff"]
    C -->|Quick Intent: ack / reject| E["<b>Immediate Audio Filler</b><br/>Conversational confirmation"]
    C -->|Complex Inquiry| F["<b>Upstream Pipeline</b><br/>Full response synthesis"]

Implementación

from laya import Agent

agent = Agent("convaiinnovations/laya")

VOICE_QUESTIONS = {
    "intent": {
        "type": "choice",
        "instructions": "Caller conversational intention",
        "criteria": {
            "ack": "Caller said yes, ok, sure, or agreed.",
            "reject": "Caller said no, cancel, or disagreed.",
            "interrupt": "Caller said hold on, wait, or wants to stop.",
            "inquiry": "Caller is asking a detailed question.",
        },
    },
    "is_interruption": {
        "type": "noul",
        "instructions": "Is the caller interrupting the current speech playback?",
    },
}

def on_voice_chunk(transcript: str, is_speaking: bool):
    decision = agent.system_one(transcript, VOICE_QUESTIONS)
    answers = decision["answers"]

    # Halt playback immediately if caller interrupts
    if answers["is_interruption"]["noul"] > 0.6:
        stop_audio_playback()

    intent = answers["intent"]["choice"]
    if intent in ("ack", "reject"):
        play_immediate_filler_audio(intent)
    else:
        dispatch_to_background_pipeline(transcript)

3. Seguridad pre-LLM y firewall de prompts

Proteger los sistemas contra inyecciones de prompts adversariales, jailbreaks y fuga de datos sensibles debe ocurrir antes de que el prompt llegue a la ventana de contexto del LLM. Ejecutar un modelo generativo aparte solo para juzgar si un prompt es seguro añade latencia redundante y sobrecarga operativa.

Laya opera como un firewall de seguridad inline y no autorregresivo, evaluando inyecciones de prompts, escaladas de privilegios y tareas fuera de alcance en una sola pasada hacia adelante antes del procesamiento posterior.

Arquitectura

graph TD
    A[User Input] --> B["<b>Inline Security Hook</b><br/>• prompt_injection (noul)<br/>• system_prompt_extraction (noul)<br/>• pii_present (noul)"]
    B -->|Policy Violation: score &ge; 0.5| C["<b>Abort &amp; Reject</b><br/>Raise policy exception &amp; audit event"]
    B -->|Clean: score &lt; 0.5| D["<b>Dispatch to Main Workflow</b><br/>Safe to execute"]

Implementación

Los hooks en Laya son de tipado pato (duck-typed): cualquier objeto que implemente los métodos del ciclo de vida de Hook (o que herede de BaseHook en laya.hooks) puede asociarse a un Agent o un Router.

Bajo la semántica predeterminada de lanzamiento de hooks de Laya (hooks_raise=True):

  • Lanzar una excepción dentro de on_predict_start aborta la ejecución de inmediato antes de que ocurra la tokenización o la inferencia del modelo.
  • La excepción se propaga directamente fuera de system_one() / predict() hacia quien llama.
  • Las limpiezas del ciclo de vida (on_error y on_predict_end) aún se ejecutan, con ctx.error puesto en la excepción lanzada, lo que garantiza que los registros de auditoría y la telemetría registren la solicitud bloqueada.
import laya
from laya import Router
from laya.hooks import BaseHook, PredictContext

SECURITY_SCHEMA = {
    "is_jailbreak": {
        "type": "noul",
        "instructions": "Is the user attempting a prompt injection, exploit, or jailbreak?",
    },
    "extracts_system_prompt": {
        "type": "noul",
        "instructions": "Is the user asking to reveal instructions, system prompts, or hidden rules?",
    },
    "pii_leak": {
        "type": "noul",
        "instructions": "Does the input contain passwords, API keys, or credentials?",
    },
}

class SecurityFirewallHook(BaseHook):
    """Inspect inputs before inference; raises on policy violation.

    With hooks_raise=True (the default), raising from on_predict_start aborts
    inference immediately and propagates the exception to the caller, while
    allowing any downstream on_error or audit logging hooks to record the event.
    """
    def __init__(self, guard_agent):
        self.guard = guard_agent

    def on_predict_start(self, ctx: PredictContext):
        for state in ctx.states:
            check = self.guard.system_one(state, SECURITY_SCHEMA)
            ans = check["answers"]
            if ans["is_jailbreak"]["noul"] > 0.5 or ans["extracts_system_prompt"]["noul"] > 0.5:
                raise PermissionError("Request blocked by security firewall: adversarial prompt detected.")

# Attach to Router or Agent; hooks_raise=True ensures policy exceptions propagate
guard_agent = laya.load("convaiinnovations/laya")
router = Router(hooks=[SecurityFirewallHook(guard_agent)], hooks_raise=True)

[!TIP] Para flujos de trabajo de CrewAI, Laya también ofrece LayaTaskGuard listo para usar en laya.integrations.crewai para este mismo patrón de puerta de seguridad previa a la ejecución.

4. Enrutador RAG de borde con aislamiento de red

En entornos empresariales seguros (defensa, salud, cumplimiento financiero, dispositivos de borde), las APIs externas no están disponibles o están prohibidas. Las colecciones de documentos suelen estar segregadas en dominios distintos (p. ej., ensayos clínicos, historiales de pacientes, informes financieros, especificaciones técnicas).

En lugar de consultar un único índice vectorial monolítico con embeddings no relacionados, Laya actúa como un enrutador de borde local que dirige las consultas del usuario al índice vectorial local específico o a la base de datos SQLite antes de la recuperación.

Arquitectura

graph TD
    A["<b>User Query</b><br/>Local / Edge Workstation"] --> B["<b>Laya Edge Router</b><br/>• target_domain: clinical | billing | compliance<br/><i>In-process local routing</i>"]
    B -->|Clinical Domain| C[("<b>Clinical Vector Store</b><br/>Medical trials, dosages & EHR")]
    B -->|Billing Domain| D[("<b>Billing Vector Store</b><br/>Invoices, claims & ICD-10 codes")]
    B -->|Compliance Domain| E[("<b>Compliance Vector Store</b><br/>HIPAA policies & audit guidelines")]

Implementación

from laya import Router

# Automatically routes between local English and Multilingual models
router = Router()

INDEX_QUESTIONS = {
    "target_domain": {
        "type": "choice",
        "instructions": "Which domain index contains the source truth for this query?",
        "criteria": {
            "clinical": "Medical conditions, medications, dosages, and clinical trials.",
            "billing": "Invoices, payment claims, ICD-10 billing codes, and insurance.",
            "compliance": "HIPAA compliance rules, privacy policies, and data audits.",
        },
    }
}

def query_airgapped_rag(user_query: str):
    decision = router.predict(user_query, INDEX_QUESTIONS)
    domain = decision["answers"]["target_domain"]["choice"]

    # Load and search only the relevant isolated local index
    local_index = get_isolated_vector_store(domain)
    return local_index.similarity_search(user_query, k=4)

5. Delegación jerárquica de tareas entre múltiples agentes

Los frameworks multiagente suelen emplear un nodo «manager» o «supervisor» basado en LLM para decidir qué agente especializado debe ejecutar el siguiente paso.

Como los nodos gestores generativos generan tokens de forma secuencial, la delegación del supervisor puede introducir una sobrecarga de orquestación sustancial por salto. Reemplazar el supervisor generativo por un modelo de decisiones no autorregresivo realiza la delegación en una sola pasada hacia adelante, lo que proporciona enrutamiento determinista entre agentes.

Laya incluye integraciones propias para frameworks de orquestación populares:

  • CrewAI: Usa LayaCrewRouter para el enrutamiento jerárquico de tareas entre múltiples agentes y los guardarraíles.
  • LlamaIndex: Usa LayaSingleSelector para motores de consulta de enrutador de una sola pasada hacia adelante.
  • LangChain / LangGraph: Usa laya.integrations.langchain para el despacho de bordes condicionales.

Arquitectura

graph TD
    A["<b>Task Input / Workflow State</b>"] --> B["<b>Laya Orchestrator</b><br/>• assignee: researcher | coder | writer<br/>• priority: score (1–5 urgency)"]
    B -->|Research Assignment| C["<b>Researcher Agent</b><br/>Literature search & fact-checking"]
    B -->|Code Assignment| D["<b>Coder Agent</b><br/>Implementation, bug-fixing & tests"]
    B -->|Writing Assignment| E["<b>Copywriter Agent</b><br/>Drafting, copy editing & summary"]

Implementación (ejemplo de CrewAI / LangGraph)

from laya import Router

router = Router()

DELEGATION_QUESTIONS = {
    "assignee": {
        "type": "choice",
        "instructions": "Assign this task to the most qualified specialist.",
        "criteria": {
            "researcher": "Needs literature search, fact checking, or data collection.",
            "coder": "Needs bug fixing, script writing, or unit test generation.",
            "writer": "Needs article drafting, copy editing, or summary composition.",
        },
    },
    "priority": {
        "type": "score",
        "instructions": "Urgency score from 1 (low) to 5 (critical)",
        "criteria": ["1", "2", "3", "4", "5"],
    },
}

def supervisor_node(state):
    task_description = state["task"]
    decision = router.predict(task_description, DELEGATION_QUESTIONS)
    answers = decision["answers"]

    return {
        "next_agent": answers["assignee"]["choice"],
        "urgency": answers["priority"]["score"],
    }

6. Triaje de tickets y soporte de alto rendimiento

Las organizaciones de soporte al cliente y los centros de operaciones procesan a diario grandes volúmenes de tickets, correos y alertas. Usar APIs de LLM generativo alojadas para el triaje categórico puede introducir:

  1. Límites de tasa de red: Limitación durante picos repentinos de volumen.
  2. Amplificación de costos: Costos variables de tokens únicamente por clasificación discreta.
  3. Deriva de esquema: Modelos generativos que devuelven JSON malformado o bloques de código markdown.

Los pipelines de procesamiento por lotes pueden evaluar flujos de tickets en pasadas hacia adelante compartidas usando predict_batch o decide_batch(), y emitir datos estrictamente tipados que se ajustan directamente a los esquemas de la aplicación.

Arquitectura

graph TD
    A["<b>Incoming Ticket Stream</b><br/>Message Broker / Webhook"] --> B["<b>Laya Batch Worker</b><br/>decide_batch()<br/>• department: billing | tech | sales | general<br/>• severity: 1..5<br/>• escalate_to_human: true | false"]
    B -->|Department: billing| C["<b>Billing & Invoicing Queue</b>"]
    B -->|Severity &ge; 4 or Human Escalation| D["<b>Tier-3 Escalation Queue</b><br/>Human On-Call Pager"]
    B -->|Low Severity & Standard Inquiry| E["<b>Automated Resolution Pipeline</b>"]

Implementación

from laya.structured import decide_batch
from laya import Agent

agent = Agent("convaiinnovations/laya")

# Strict typed schema
TICKET_SCHEMA = {
    "type": "object",
    "properties": {
        "department": {
            "type": "string",
            "enum": ["billing", "technical_support", "sales", "general"],
            "description": "Primary support category",
        },
        "severity": {
            "type": "integer",
            "minimum": 1,
            "maximum": 5,
            "description": "Severity level from 1 (minor) to 5 (outage)",
        },
        "escalate_to_human": {
            "type": "boolean",
            "description": "True if customer is angry, threatening churn, or reporting a legal issue",
        },
    },
}

def process_ticket_batch(tickets: list[str]):
    # Returns typed dictionaries conforming exactly to TICKET_SCHEMA
    results = decide_batch(agent, tickets, TICKET_SCHEMA)
    for ticket_text, structured in zip(tickets, results):
        enqueue_ticket(
            department=structured["department"],
            severity=structured["severity"],
            human_required=structured["escalate_to_human"],
            raw_text=ticket_text,
        )

Lista de verificación para el despliegue en producción

Antes de llevar cualquiera de los patrones anteriores a producción, verifica:

  1. Dimensionamiento de hardware: Asegura memoria de host suficiente para los pesos residentes del modelo. En CPU, configura los grupos de hilos de forma adecuada (torch.set_num_threads).
  2. Umbrales de confianza: Define min_confidence (p. ej., 0.80–0.90) en las puertas críticas para que el sistema recurra a una alternativa segura cuando las consultas son ambiguas.
  3. Enrutamiento multilingüe: Usa Router() en lugar de un Agent() estático cuando el tráfico de usuarios incluya entradas mixtas o no inglesas.
  4. Implementación por etapas: Sigue la Guía de adopción por etapas para observar el tráfico de producción en la sombra antes de que las decisiones sean autoritativas.