> **Beta Translation:** This document was translated via Machine Learning and as such may not be 100% accurate. All non-English languages are currently classified as Beta.
Este documento cataloga todas las conexiones entre el código base de TrustGraph y la infraestructura pub/sub. Actualmente, el sistema está codificado para usar Apache Pulsar. Este análisis identifica todos los puntos de integración para informar futuras refactorizaciones hacia una abstracción pub/sub configurable.
Todos los esquemas de servicio: `schema/services/*.py`
Nomenclatura de temas: `schema/core/topic.py`
4.**Conceptos específicos de Pulsar requeridos:**
Mensajería basada en temas
Sistema de esquemas (Registro, tipos de campo)
Suscripciones compartidas
Reconocimiento de mensajes (positivo/negativo)
Posicionamiento del consumidor (más temprano/más reciente)
Propiedades del mensaje
Posiciones iniciales y tipos de consumidor
Soporte de fragmentación
Temas persistentes frente a no persistentes
### Desafíos de refactorización
La buena noticia: la capa de abstracción (Consumidor, Productor, Publicador, Suscriptor) proporciona una encapsulación limpia de la mayoría de las interacciones de Pulsar.
Los desafíos:
1.**Ubicuidad del sistema de esquemas:** Cada definición de mensaje utiliza `pulsar.schema.Record` y los tipos de campo de Pulsar
2.**Enums específicos de Pulsar:**`InitialPosition`, `ConsumerType`
3.**Excepciones de Pulsar:**`_pulsar.Timeout`, `_pulsar.Interrupted`, `_pulsar.InvalidConfiguration`, `_pulsar.AlreadyClosed`
4.**Fichas de método:**`acknowledge()`, `negative_acknowledge()`, `subscribe()`, `create_producer()`, etc.
5.**Formato de URI de tema:** Estructura de Pulsar `kind://tenant/namespace/topic`
### Próximos pasos
Para hacer que la infraestructura de pub/sub sea configurable, necesitamos:
1. Crear una interfaz de abstracción para el sistema de cliente/esquemas
2. Abstractar enums y excepciones específicas de Pulsar
3. Crear envoltorios de esquema o definiciones de esquema alternativas
4. Implementar la interfaz tanto para Pulsar como para sistemas alternativos (Kafka, RabbitMQ, Redis Streams, etc.)
5. Actualizar `pubsub.py` para que sea configurable y admita varios backends
6. Proporcionar una ruta de migración para implementaciones existentes
## Borrador de enfoque 1: Patrón de adaptador con capa de traducción de esquemas
### Idea clave
El **sistema de esquemas** es el punto de integración más profundo; todo lo demás se deriva de él. Necesitamos resolver esto primero, o tendremos que reescribir todo el código base.
### Estrategia: Interrupción mínima con adaptadores
**1. Mantener los esquemas de Pulsar como la representación interna**
No reescribir todas las definiciones de esquemas.
Los esquemas permanecen `pulsar.schema.Record` internamente.
Utilizar adaptadores para traducir en la frontera entre nuestro código y el backend de publicación/suscripción.
**2. Crear una capa de abstracción de publicación/suscripción:**
```
┌─────────────────────────────────────┐
│ Existing Code (unchanged) │
│ - Uses Pulsar schemas internally │
│ - Consumer/Producer/Publisher │
└──────────────┬──────────────────────┘
│
┌──────────────┴──────────────────────┐
│ PubSubFactory (configurable) │
│ - Creates backend-specific client │
└──────────────┬──────────────────────┘
│
┌──────┴──────┐
│ │
┌───────▼─────┐ ┌────▼─────────┐
│ PulsarAdapter│ │ KafkaAdapter │ etc...
│ (passthrough)│ │ (translates) │
└──────────────┘ └──────────────┘
```
**3. Defina interfaces abstractas:**
`PubSubClient` - conexión del cliente
`PubSubProducer` - envío de mensajes
`PubSubConsumer` - recepción de mensajes
`SchemaAdapter` - traducción de esquemas de Pulsar a/desde JSON o formatos específicos del backend
**4. Detalles de implementación:**
Para el **adaptador de Pulsar**: Casi una transmisión directa, traducción mínima.
Para **otros backends** (Kafka, RabbitMQ, etc.):
Serializar objetos de registro de Pulsar a JSON/bytes.
Mapear conceptos como:
`InitialPosition.Earliest/Latest` → auto.offset.reset de Kafka
`acknowledge()` → confirmación de Kafka
`negative_acknowledge()` → patrón de re-cola o cola de mensajes no entregados (DLQ).
URIs de temas → nombres de temas específicos del backend.
### Análisis
**Ventajas:**
✅ Cambios mínimos en el código de los servicios existentes.
✅ Los esquemas permanecen sin cambios (sin reescritura masiva).
✅ Ruta de migración gradual.
✅ Los usuarios de Pulsar no notan ninguna diferencia.
✅ Se agregan nuevos backends a través de adaptadores.
**Desventajas:**
⚠️ Aún mantiene la dependencia de Pulsar (para las definiciones de esquemas).
⚠️ Algunos problemas de compatibilidad al traducir conceptos.
### Consideración alternativa
Crear un sistema de esquemas **TrustGraph** que sea independiente de pub/sub (usando dataclasses o Pydantic), y luego generar esquemas de Pulsar/Kafka/etc a partir de él. Esto requiere reescribir cada archivo de esquema y podría provocar cambios importantes.
### Recomendación para la versión preliminar 1
Comience con el **enfoque de adaptador** porque:
1. Es práctico: funciona con el código existente.
2. Demuestra el concepto con un riesgo mínimo.
3. Puede evolucionar hacia un sistema de esquemas nativo más adelante, si es necesario.
4. Impulsado por la configuración: una variable de entorno cambia entre backends.
## Enfoque de la versión preliminar 2: Sistema de esquemas independiente del backend con dataclasses
### Concepto central
Utilice **dataclasses** de Python como el formato de definición de esquema neutral. Cada backend de pub/sub proporciona su propia serialización/deserialización para dataclasses, eliminando la necesidad de que los esquemas de Pulsar permanezcan en el código base.
### Polimorfismo de esquema a nivel de fábrica
En lugar de traducir esquemas de Pulsar, **cada backend proporciona su propia gestión de esquemas** que funciona con dataclasses de Python estándar.
### Flujo del publicador
```python
# 1. Get the configured backend from factory
pubsub = get_pubsub() # Returns PulsarBackend, MQTTBackend, etc.
# 2. Get schema class from the backend
# (Can be imported directly - backend-agnostic)
from trustgraph.schema.services.llm import TextCompletionRequest
# 3. Create a producer/publisher for a specific topic
producer = pubsub.create_producer(
topic="text-completion-requests",
schema=TextCompletionRequest # Tells backend what schema to use
)
# 4. Create message instances (same API regardless of backend)
`create_producer()` → crea un productor de Pulsar con un esquema JSON o un registro generado dinámicamente.
`send(request)` → serializa la clase de datos a formato JSON/Pulsar y lo envía a Pulsar.
`receive()` → recibe un mensaje de Pulsar, lo deserializa de nuevo a la clase de datos.
**Para el backend de MQTT:**
`create_producer()` → se conecta a un broker de MQTT, no es necesario registrar ningún esquema.
`send(request)` → convierte la clase de datos a JSON y lo publica en un tema de MQTT.
`receive()` → se suscribe a un tema de MQTT y deserializa el JSON a la clase de datos.
**Para el backend de Kafka:**
`create_producer()` → crea un productor de Kafka y registra el esquema Avro si es necesario.
`send(request)` → serializa la clase de datos a formato Avro y lo envía a Kafka.
`receive()` → recibe un mensaje de Kafka y lo deserializa de Avro de nuevo a la clase de datos.
### Puntos Clave del Diseño
1.**Creación del objeto de esquema**: La instancia de la clase de datos (`TextCompletionRequest(...)`) es idéntica independientemente del backend.
2.**El backend se encarga de la codificación**: Cada backend sabe cómo serializar su clase de datos al formato de cable.
3.**Definición del esquema en la creación**: Al crear el productor/consumidor, se especifica el tipo de esquema.
4.**Se mantiene la seguridad de tipos**: Se obtiene un objeto `TextCompletionRequest` adecuado, no un diccionario.
5.**Sin filtración del backend**: El código de la aplicación nunca importa bibliotecas específicas del backend.
### Ejemplo de Transformación
**Actual (específico de Pulsar):**
```python
# schema/services/llm.py
from pulsar.schema import Record, String, Boolean, Integer
class TextCompletionRequest(Record):
system = String()
prompt = String()
streaming = Boolean()
```
**Nuevo (Independiente del backend):**
```python
# schema/services/llm.py
from dataclasses import dataclass
@dataclass
class TextCompletionRequest:
system: str
prompt: str
streaming: bool = False
```
### Integración con el Backend
Cada backend se encarga de la serialización/deserialización de dataclasses:
**Backend de Pulsar:**
Genera clases `pulsar.schema.Record` dinámicamente a partir de dataclasses
O serializa dataclasses a JSON y utiliza el esquema JSON de Pulsar
Mantiene la compatibilidad con implementaciones de Pulsar existentes
**Backend de MQTT/Redis:**
Serialización directa de instancias de dataclass a JSON
Utiliza `dataclasses.asdict()` / `from_dict()`
Ligero, no se necesita un registro de esquemas
**Backend de Kafka:**
Genera esquemas Avro a partir de definiciones de dataclass
Utiliza el registro de esquemas de Confluent
Serialización con seguridad de tipos y soporte para la evolución del esquema
### Arquitectura
```
┌─────────────────────────────────────┐
│ Application Code │
│ - Uses dataclass schemas │
│ - Backend-agnostic │
└──────────────┬──────────────────────┘
│
┌──────────────┴──────────────────────┐
│ PubSubFactory (configurable) │
│ - get_pubsub() returns backend │
└──────────────┬──────────────────────┘
│
┌──────┴──────┐
│ │
┌───────▼─────────┐ ┌────▼──────────────┐
│ PulsarBackend │ │ MQTTBackend │
│ - JSON schema │ │ - JSON serialize │
│ - or dynamic │ │ - Simple queues │
│ Record gen │ │ │
└─────────────────┘ └───────────────────┘
```
### Detalles de implementación
**1. Definiciones de esquema:** Clases de datos simples con sugerencias de tipo
`str`, `int`, `bool`, `float` para tipos primitivos
`list[T]` para arreglos
`dict[str, T]` para mapas
Clases de datos anidadas para tipos complejos
**2. Cada backend proporciona:**
Serializador: `dataclass → bytes/wire format`
Deserializador: `bytes/wire format → dataclass`
Registro de esquema (si es necesario, como Pulsar/Kafka)
**3. Abstracción de consumidor/productor:**
Ya existe (consumer.py, producer.py)
Actualizar para usar la serialización del backend
Eliminar importaciones directas de Pulsar
**4. Mapeos de tipo:**
Pulsar `String()` → Python `str`
Pulsar `Integer()` → Python `int`
Pulsar `Boolean()` → Python `bool`
Pulsar `Array(T)` → Python `list[T]`
Pulsar `Map(K, V)` → Python `dict[K, V]`
Pulsar `Double()` → Python `float`
Pulsar `Bytes()` → Python `bytes`
### Ruta de migración
1.**Crear versiones de clases de datos** de todos los esquemas en `trustgraph/schema/`
2.**Actualizar clases de backend** (Consumidor, Productor, Publicador, Suscriptor) para usar la serialización proporcionada por el backend
3.**Implementar PulsarBackend** con esquema JSON o generación dinámica de registros
4.**Probar con Pulsar** para garantizar la compatibilidad hacia atrás con las implementaciones existentes
5.**Agregar nuevos backends** (MQTT, Kafka, Redis, etc.) según sea necesario
6.**Eliminar importaciones de Pulsar** de los archivos de esquema
### Beneficios
✅ **Sin dependencia de pub/sub** en las definiciones de esquema
✅ **Python estándar** - fácil de entender, tipificar, documentar
✅ **Herramientas modernas** - funciona con mypy, autocompletado de IDE, analizadores
✅ **Optimizada para el backend** - cada backend utiliza la serialización nativa
✅ **Sin sobrecarga de traducción** - serialización directa, sin adaptadores
✅ **Seguridad de tipos** - objetos reales con tipos adecuados
✅ **Validación fácil** - se puede usar Pydantic si es necesario
### Desafíos y soluciones
**Desafío:** El `Record` de Pulsar tiene validación de campo en tiempo de ejecución
**Solución:** Usar clases de datos de Pydantic para la validación si es necesario, o características de clase de datos de Python 3.10+ con `__post_init__`
**Desafío:** Algunas características específicas de Pulsar (como el tipo `Bytes`)
**Solución:** Mapear al tipo `bytes` en la clase de datos, el backend se encarga de la codificación apropiadamente
**Desafío:** Nombres de temas (`persistent://tenant/namespace/topic`)
**Solución:** Abstracto los nombres de los temas en las definiciones de esquema, el backend convierte al formato adecuado
**Desafío:** Evolución y versionado del esquema
**Solución:** Cada backend maneja esto de acuerdo con sus capacidades (versiones de esquema de Pulsar, registro de esquema de Kafka, etc.)
**Desafío:** Tipos complejos anidados
**Solución:** Usar clases de datos anidadas, los backends serializan/deserializan recursivamente
### Decisiones de diseño
1.**¿Clases de datos simples o Pydantic?**
✅ **Decisión: Usar clases de datos de Python simples**
Más simple, sin dependencias adicionales
La validación no es necesaria en la práctica
Más fácil de entender y mantener
2.**Evolución del esquema:**
✅ **Decisión: No se necesita un mecanismo de versionado**
Los esquemas son estables y duraderos
Las actualizaciones normalmente agregan nuevos campos (compatible con versiones anteriores)
Los backends manejan la evolución del esquema según sus capacidades
3.**Compatibilidad hacia atrás:**
✅ **Decisión: Cambio de versión importante, no se requiere compatibilidad hacia atrás**
Será un cambio importante con instrucciones de migración
La ruptura limpia permite un mejor diseño
Se proporcionará una guía de migración para las implementaciones existentes
4.**Tipos anidados y estructuras complejas:**
✅ **Decisión: Usar clases de datos anidadas de forma natural**
Las clases de datos de Python manejan el anidamiento perfectamente
`list[T]` para arreglos, `dict[K, V]` para mapas
Los backends serializan/deserializan recursivamente
Ejemplo:
```python
@dataclass
class Value:
value: str
is_uri: bool
@dataclass
class Triple:
s: Value # Nested dataclass
p: Value
o: Value
@dataclass
class GraphQuery:
triples: list[Triple] # Array of nested dataclasses
metadata: dict[str, str]
```
5.**Valores predeterminados y campos opcionales:**
✅ **Decisión: Combinación de campos obligatorios, valores predeterminados y campos opcionales**
Campos obligatorios: Sin valor predeterminado
Campos con valores predeterminados: Siempre presentes, tienen un valor predeterminado razonable
Campos verdaderamente opcionales: `T | None = None`, omitidos de la serialización cuando `None`
Ejemplo:
```python
@dataclass
class TextCompletionRequest:
system: str # Required, no default
prompt: str # Required, no default
streaming: bool = False # Optional with default value
metadata: dict | None = None # Truly optional, can be absent
```
**Semántica de serialización importante:**
Cuando `metadata = None`:
```json
{
"system": "...",
"prompt": "...",
"streaming": false
// metadata field NOT PRESENT
}
```
Cuando `metadata = {}` (explícitamente vacío):
```json
{
"system": "...",
"prompt": "...",
"streaming": false,
"metadata": {} // Field PRESENT but empty
}
```
**Diferencia clave:**
`None` → campo ausente en JSON (no se serializa)
Valor vacío (`{}`, `[]`, `""`) → campo presente con valor vacío
Esto es importante semánticamente: "no proporcionado" vs "explícitamente vacío"
Los sistemas de serialización deben omitir los campos `None`, no codificarlos como `null`
## Esquema de Implementación Borrador 3: Detalles de Implementación
### Formato Genérico para Nombres de Colas
Reemplace los nombres de colas específicos del sistema de respaldo con un formato genérico que los sistemas de respaldo puedan mapear adecuadamente.
