Anexos
Anexo: el prototipo
Todo lo que el libro afirma (las siete coordenadas, los hechos con signatura, las situaciones reificadas, la bitemporalidad, el motor de consulta) cabe en nueve archivos cortos de Python que corren sin una sola dependencia externa.
Una arquitectura que solo vive en prosa es una promesa. Una que corre es una prueba.
Este anexo reúne el código fuente íntegro de la librería núcleo del prototipo: unas ochocientas cincuenta líneas repartidas en nueve módulos.
No es pseudocódigo ni un boceto. Es Python real, ejecutable, que da cuerpo a cada decisión de diseño del libro y que sirvió de banco de pruebas mientras el modelo tomaba forma, dominio tras dominio.
Conviene fijar el alcance sin letra pequeña. Lo que sigue es solo la librería
núcleo, el paquete wq/.
Los ejemplos por dominio de la Parte V (el spa, el taxi, la clínica, el banco), la batería de pruebas y los scripts de utilidad no aparecen aquí. Pero existen, corren y están publicados.
El proyecto completo vive en el repositorio del libro. Al final de este anexo se indica qué encontrarás allí y cómo ponerlo en marcha en cinco minutos.
Qué demuestra este código
Que el modelo de las preguntas no es una metáfora afortunada, sino una arquitectura coherente y operable.
Ochocientas cincuenta líneas bastan para sostener las siete coordenadas, validar hechos contra sus firmas, dar identidad propia a las situaciones, resolver polisemia y responder preguntas con vigencia. La validación industrial es otro trabajo, el del capítulo 30. Lo que estas líneas prueban es que la idea cierra.
La arquitectura en una sola mirada
El prototipo se organiza en nueve módulos apilados como capas.
En la base, los siete ejes. Sobre ellos, la noción de individuo. Sobre los individuos, el hecho atómico. A un lado, dos estructuras que lo cruzan todo: el catálogo de roles y el lexicon. Por encima, el universo que lo contiene todo. Y en la cima, la superficie pública que expone lo que el usuario va a tocar.
Cada decisión de diseño del libro aterriza en un módulo concreto. El paralelismo es deliberado y casi uno a uno.
wq/, apilados de la base a la cima. El orden de lectura natural va de abajo
hacia arriba: primero axes.py (los siete ejes), luego los individuos, luego
las tres estructuras transversales —catálogo (K), lexicon
(Q) y hecho (T)—, luego el
universo que las une, y por fin la API de ingesta y consulta. Las nueve secciones de este
anexo siguen ese mismo orden.El recorrido es de abajo hacia arriba. Empezamos por los ejes, subimos a los individuos que los pueblan, llegamos al hecho atómico, abrimos las dos piezas transversales, las reunimos en el universo y rematamos con la fachada.
Cada sección abre con una nota breve sobre qué resuelve el módulo y termina con su código íntegro.
Cómo leer el código de este anexo
Los bloques están coloreados con el resaltador del libro. Las palabras
clave de Python, las cadenas, los números y los comentarios se distinguen al vuelo, y los
identificadores de eje (Q, O, L, T,
N, K, M) llevan su color habitual. Si una línea es
larga, el bloque se desplaza en horizontal.
El botón copiar entrega el código tal cual.
§A1 axes.py — los siete ejes
Es la pieza fundacional y la más corta. Declara, como un Enum, los seis ejes de
valor (los que albergan individuos) y el único eje de predicados.
A partir de aquí, todo el modelo se apoya en estos siete identificadores. No hay cadenas mágicas dispersas por el código. Hay un vocabulario cerrado y verificable.
La distinción que codifica este archivo es la que recorre todo el libro: seis ejes contienen cosas y uno conecta cosas.
Q guarda agentes. O, objetos y situaciones. L, lugares. T, momentos. N, magnitudes. Y K, categorías atemporales.
El séptimo, M, no guarda valores. Guarda los cables que enlazan los valores de los otros seis.
"""Los 7 ejes de WQuestions.
Seis ejes de valor (contienen individuos):
Q — quién — agentes
O — qué — objetos / situaciones reificadas
L — dónde — lugares
T — cuándo — momentos / intervalos
N — cuánto — magnitudes con unidad
K — cuál — categorías atemporales (tipos, unidades, estados, vocabularios)
Un eje estructural (contiene etiquetas con signatura):
M — cómo — predicados/cables que conectan individuos. Cada predicado
declara su cardinalidad en la signatura (`functional`: un solo valor
por sujeto, o múltiple). La cardinalidad es un atributo del predicado,
no un eje aparte.
"""
from enum import Enum
class Axis(Enum):
Q = "Q" # quién
O = "O" # qué (objectum)
L = "L" # dónde
T = "T" # cuándo (tempus)
N = "N" # cuánto
K = "K" # cuál (categórico)
M = "M" # cómo / predicados (modus)
V = "V" # comodín de signatura: cualquier eje de valor
VALUE_AXES = {Axis.Q, Axis.O, Axis.L, Axis.T, Axis.N, Axis.K}
PREDICATE_AXES = {Axis.M}
def is_value_axis(axis: Axis) -> bool:
return axis in VALUE_AXES§A2 individual.py — individuos en los ejes
Ochenta y un líneas. Define qué es un individuo: la entidad concreta que habita en alguno de los seis ejes de valor.
Un agente, una situación, un lugar, un instante, una magnitud o una categoría son todos individuos. Lo que cambia es el eje donde viven.
La clase es inmutable, un dataclass congelado, porque en este modelo un
individuo, una vez acuñado, no cambia de identidad ni de eje.
El archivo aporta además tres fábricas de individuos primitivos, esos que en realidad son
datos. time_point crea un instante en T a partir
de una marca ISO 8601. quantity crea una magnitud en
N anclando su unidad en K. Y
category crea una categoría en K.
Que quantity exija que la unidad viva en K no es decorativo. Es la regla del
eje cuantitativo hecha código: un número sin unidad anclada
simplemente no se puede construir.
Individuo
Ejemplar concreto que ocupa uno y solo un eje de valor,
con un identificador estable. El campo payload aloja el valor nativo cuando el
individuo es un dato: el datetime de un instante, el par
{valor, unidad} de una magnitud. Para agentes y situaciones, el identificador
basta y el payload queda vacío.
"""Individuos del modelo: instancias de cualquier eje de valor.
Cada individuo tiene un identificador estable y vive en uno y solo un eje.
Para sujetos sintéticos usamos un mint determinista (id corto, monótono)
en vez de UUID — la observabilidad gana, la pérdida de globalidad no
importa para un prototipo.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Optional, Any, Dict
from itertools import count
from .axes import Axis, is_value_axis
_counter = count(1)
def mint_id(prefix: str = "id") -> str:
"""Genera un id corto y monótono. En producción se reemplazaría por UUID v7."""
n = next(_counter)
return f"{prefix}_{n:06d}"
@dataclass(frozen=True)
class Individual:
"""Individuo en un eje de valor.
`payload` permite alojar valores nativos cuando el "individuo" es en
realidad un dato (un instante T, una magnitud N, una categoría K). Para
Q y O suele ser None; el id basta.
"""
id: str
axis: Axis
label: Optional[str] = None
payload: Any = None
meta: Dict[str, Any] = field(default_factory=dict)
def __post_init__(self):
if not is_value_axis(self.axis):
raise ValueError(
f"Un individuo debe vivir en un eje de valor, no en {self.axis}."
)
def __repr__(self) -> str:
tag = self.label or self.id
return f"<{self.axis.value}:{tag}>"
# --- helpers para individuos "primitivos" -----------------------------------
def time_point(iso: str) -> Individual:
"""Crea un individuo T desde una marca ISO 8601."""
from datetime import datetime
try:
dt = datetime.fromisoformat(iso.replace("Z", "+00:00"))
except ValueError as e:
raise ValueError(f"Marca temporal inválida: {iso}") from e
return Individual(id=f"t_{iso}", axis=Axis.T, label=iso, payload=dt)
def quantity(value: float, unit_k: "Individual") -> Individual:
"""Crea un individuo N con valor numérico y unidad anclada en K."""
if unit_k.axis != Axis.K:
raise ValueError(
f"La unidad debe vivir en K (recibido: {unit_k.axis})."
)
return Individual(
id=mint_id("n"),
axis=Axis.N,
label=f"{value} {unit_k.label or unit_k.id}",
payload={"value": value, "unit": unit_k.id},
)
def category(label: str) -> Individual:
"""Crea un individuo K (categoría) con id derivado del label."""
return Individual(id=label, axis=Axis.K, label=label)§A3 fact.py — el hecho atómico con bitemporalidad
Sesenta y una líneas para la unidad mínima del modelo. Un hecho es una tupla inmutable
(sujeto, rol, valor), donde sujeto y valor son individuos y el rol es una etiqueta
del catálogo.
Esto es D3 en estado puro. Absolutamente todo lo que el modelo sabe se descompone en estas tripletas tipadas.
Lo complejo no se modela con estructuras más ricas. Se apila como más tripletas.
D3 El hecho atómico
La clase Fact es la encarnación literal de D3: una tripleta
(subject, role, value), congelada e inmutable. No hay forma de representar un
hecho fuera de esta forma. La riqueza del mundo se reconstruye apilando tripletas, no
inflando la estructura de cada una.
Sobre esa base, el archivo añade el tiempo que pide D6. Cada hecho lleva un
rango de vigencia opcional [valid_from, valid_to), desde cuándo y hasta cuándo es
cierto en el mundo. Y un tx_time automático que registra cuándo entró
al sistema.
Son dos relojes distintos: el tiempo de los hechos y el tiempo de los registros. Esa separación basta para modelar mudanzas, rediagnósticos o cláusulas que expiran.
El método is_valid_at condensa la lógica de D6 en seis líneas. Si no hay rango
de vigencia, el hecho vale siempre. Si lo hay, el momento consultado tiene que caer en el
intervalo, con el inicio incluido y el fin excluido. Un valid_to nulo significa
«abierto al futuro».
"""Hechos atómicos: la unidad mínima del modelo.
Cada hecho es una tupla `(sujeto, rol, valor)` donde:
- `sujeto` y `valor` son `Individual` (viven en algún eje de valor).
- `rol` es una etiqueta del catálogo canónico (D8) o de su capa lexicon.
D6 — vigencia temporal: cada hecho lleva opcionalmente un rango
`[valid_from, valid_to)` que indica desde cuándo y hasta cuándo es cierto en
el mundo. Si `valid_to is None`, está abierto al futuro.
Adicionalmente registramos `tx_time` (transaction time): el momento en el que
el hecho entró al sistema. Esto da bitemporalidad ligera, suficiente para
auditoría.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Optional
from .individual import Individual
def _utcnow() -> datetime:
return datetime.now(timezone.utc)
@dataclass(frozen=True)
class Fact:
subject: Individual
role: str
value: Individual
valid_from: Optional[datetime] = None
valid_to: Optional[datetime] = None
tx_time: datetime = field(default_factory=_utcnow)
def is_valid_at(self, moment: datetime) -> bool:
"""¿Este hecho es cierto en el mundo en `moment`?
Reglas:
- Si no hay vigencia, el hecho es atemporal: vale siempre.
- `valid_from` inclusivo, `valid_to` exclusivo.
- `valid_to is None` significa "abierto al futuro".
"""
if self.valid_from is None and self.valid_to is None:
return True
if self.valid_from is not None and moment < self.valid_from:
return False
if self.valid_to is not None and moment >= self.valid_to:
return False
return True
def __repr__(self) -> str:
s = self.subject.label or self.subject.id
v = self.value.label or self.value.id
base = f"({s}, {self.role}, {v})"
if self.valid_from or self.valid_to:
f = self.valid_from.isoformat() if self.valid_from else "-∞"
t = self.valid_to.isoformat() if self.valid_to else "+∞"
base += f" [{f} .. {t})"
return baseUn hecho es una tripleta congelada con dos líneas de tiempo: la del mundo y la del registro. Todo lo demás se construye apilando hechos.El corazón del prototipo
§A4 catalog.py — el catálogo canónico de roles
Ciento setenta y nueve líneas. Es el módulo más largo de la librería y donde se materializa D8.
El catálogo declara los treinta y ocho roles canónicos, cada uno con su firma tipada (de qué eje sale y a cuál llega) y su cardinalidad.
Esa firma convierte una tripleta de texto libre en un hecho verificable. Al insertar
(sujeto, rol, valor), el catálogo comprueba que el eje del sujeto y el del valor
sean los que el rol admite.
D8 El catálogo canónico de roles
Cada rol es un predicado del eje M con una
firma origen → destino. La clase RoleSignature la guarda; el método
validate la hace cumplir. Así, agente solo admite un sujeto en
O y un valor en Q.
monto, un valor en N. Y lugar_de,
uno en L. La basura no entra al grafo.
Conviene detenerse en una decisión que el libro defiende y que aquí ocupa cuatro líneas: la política liberal.
Si un rol no está declarado en el catálogo, validate no lo rechaza.
Simplemente no lo valida y lo deja pasar.
Una política estricta abortaría ante cualquier rol desconocido. El prototipo prefiere que se pueda extender. Quien quiera un rol nuevo lo usa de inmediato sin pedir permiso; si más tarde quiere validación, lo registra. La elección está comentada en el propio código.
La cardinalidad (el campo functional) merece una nota. Un rol funcional admite
un solo valor por sujeto: una situación tiene un agente, un
lugar_de, un inicio.
Un rol no funcional admite varios: una situación puede estar causado_por varias
otras, o contiene muchas partes.
Esto es D2 en acción. Propiedades y relaciones no son cosas distintas. Son cables del mismo eje M que solo se diferencian en cuántos valores admiten.
"""Catálogo canónico de roles (D8).
Cada rol declara una signatura tipada `dominio → rango`, ambos ejes de valor,
más su **cardinalidad** (`functional`: un solo valor por sujeto, o múltiple).
Todos los roles son predicados del eje M (cómo); la cardinalidad es un atributo
de la signatura, no un eje aparte. La signatura habilita la validación mecánica
de hechos: al insertar `(s, role, v)` el catálogo verifica que `s.axis` coincida
con el dominio y `v.axis` con el rango.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Dict, Optional
from .axes import Axis
from .individual import Individual
@dataclass(frozen=True)
class RoleSignature:
name: str
domain: Axis # eje del sujeto
range: Axis # eje del valor
functional: bool # True → un valor por sujeto, False → múltiple
description: str = ""
class SignatureError(ValueError):
pass
class Catalog:
"""Catálogo de signaturas canónicas + validación de hechos."""
def __init__(self):
self._roles: Dict[str, RoleSignature] = {}
self._load_canonical()
def register(self, sig: RoleSignature) -> None:
if sig.name in self._roles:
existing = self._roles[sig.name]
if existing != sig:
raise SignatureError(
f"Rol '{sig.name}' ya registrado con signatura distinta: "
f"{existing} vs {sig}"
)
return
self._roles[sig.name] = sig
def get(self, name: str) -> Optional[RoleSignature]:
return self._roles.get(name)
def validate(self, role: str, subject: Individual, value: Individual) -> None:
"""Lanza `SignatureError` si el hecho viola la signatura."""
sig = self._roles.get(role)
if sig is None:
# Rol no declarado: política liberal — se permite, no se valida.
# Una política estricta lo rechazaría; preferimos extensibilidad.
return
if sig.domain != Axis.V and subject.axis != sig.domain:
raise SignatureError(
f"Sujeto en eje incorrecto para '{role}': se esperaba "
f"{sig.domain.value}, recibido {subject.axis.value} "
f"(sujeto={subject})"
)
if sig.range != Axis.V and value.axis != sig.range:
raise SignatureError(
f"Valor en eje incorrecto para '{role}': se esperaba "
f"{sig.range.value}, recibido {value.axis.value} "
f"(valor={value})"
)
def __contains__(self, name: str) -> bool:
return name in self._roles
def __len__(self) -> int:
return len(self._roles)
# ------------------------------------------------------------------
# Carga del catálogo canónico (subset del documento WQuestions.md)
# ------------------------------------------------------------------
def _load_canonical(self) -> None:
canonical = [
# --- estructurales ---
RoleSignature("instancia_de", Axis.V, Axis.K, False,
"sujeto (cualquier eje de valor) pertenece a la categoría"),
RoleSignature("subtipo_de", Axis.K, Axis.K, False,
"subtipo conceptual"),
RoleSignature("parte_de", Axis.O, Axis.O, False,
"subobjeto/subevento de"),
RoleSignature("contiene", Axis.O, Axis.O, False,
"inversa de parte_de"),
# --- participantes (Q es típico) ---
RoleSignature("agente", Axis.O, Axis.Q, True,
"agente principal de la situación"),
RoleSignature("paciente", Axis.O, Axis.Q, True,
"afectado por la situación"),
RoleSignature("tema", Axis.O, Axis.O, True,
"objeto temático (cosa o sub-situación)"),
RoleSignature("beneficiario", Axis.O, Axis.Q, True,
"destinatario o beneficiario"),
RoleSignature("experimentador", Axis.O, Axis.Q, True,
"quien experimenta un estado mental"),
RoleSignature("instrumento", Axis.O, Axis.O, True,
"objeto usado para ejecutar la acción"),
RoleSignature("comprador", Axis.O, Axis.Q, True,
"comprador en una venta"),
RoleSignature("cliente", Axis.O, Axis.Q, True,
"cliente de un servicio (alias frecuente de agente)"),
# --- lugar / tiempo ---
RoleSignature("lugar_de", Axis.O, Axis.L, True,
"lugar donde ocurre la situación"),
RoleSignature("origen", Axis.O, Axis.L, True,
"lugar de origen"),
RoleSignature("destino", Axis.O, Axis.L, True,
"lugar de destino"),
RoleSignature("lugar_destino", Axis.O, Axis.L, True,
"alias de destino"),
RoleSignature("momento", Axis.O, Axis.T, True,
"momento puntual"),
RoleSignature("inicio", Axis.O, Axis.T, True,
"instante de inicio"),
RoleSignature("fin", Axis.O, Axis.T, True,
"instante de fin"),
# --- cuantitativos ---
RoleSignature("monto", Axis.O, Axis.N, True,
"cantidad numérica con unidad"),
RoleSignature("cantidad", Axis.O, Axis.N, True,
"alias de monto"),
RoleSignature("por_cuanto", Axis.O, Axis.N, True,
"precio o medida asociada"),
RoleSignature("unidad", Axis.O, Axis.K, True,
"unidad de medida (QUDT)"),
# --- clasificatorios ---
RoleSignature("estatus_factual", Axis.O, Axis.K, True,
"real / intencionado / no_realizable / ..."),
RoleSignature("modalidad", Axis.O, Axis.K, True,
"volitiva / deóntica / alética / epistémica"),
RoleSignature("polaridad", Axis.O, Axis.K, True,
"afirmativa / negativa"),
RoleSignature("calificacion", Axis.O, Axis.K, True,
"atributo cualitativo"),
# --- "por qué" (D7, capítulo 10) ---
RoleSignature("causado_por", Axis.O, Axis.O, False,
"causalidad mecánica"),
RoleSignature("motivado_por", Axis.O, Axis.O, False,
"motivación intencional"),
RoleSignature("con_finalidad", Axis.O, Axis.O, False,
"propósito"),
RoleSignature("justificado_por", Axis.O, Axis.O, False,
"regla que autoriza"),
# --- inter-situacionales ---
RoleSignature("precede", Axis.O, Axis.O, False,
"orden lógico/temporal"),
RoleSignature("sigue_a", Axis.O, Axis.O, False,
"inversa de precede"),
RoleSignature("cumple", Axis.O, Axis.O, False,
"cumple una obligación"),
RoleSignature("cancela", Axis.O, Axis.O, False,
"deja sin efecto"),
RoleSignature("rectifica", Axis.O, Axis.O, False,
"corrige otra situación"),
RoleSignature("contrasta_con", Axis.O, Axis.O, False,
"relación adversativa (\"pero\")"),
# --- atributos del sujeto Q ---
RoleSignature("nombre", Axis.Q, Axis.K, True,
"nombre de un agente"),
RoleSignature("identificador", Axis.Q, Axis.K, True,
"id documental"),
]
for sig in canonical:
self.register(sig)§A5 lexicon.py — el lexicon con resolución de polisemia
Ciento dieciséis líneas. Aquí vive la capa que traduce entre el lenguaje del usuario y los roles canónicos, la materialización de D9.
Cada entrada del lexicon es un LexiconEntry. Declara un verbo, el tipo de
situación que produce, sus roles obligatorios y opcionales, los apodos naturales por rol y por
dominio, sus formas nominales y, sobre todo, un pattern de complementos que es el
motor de la desambiguación.
D9 El usuario nunca toca etiquetas canónicas
El lexicon es la interfaz. El catálogo, invisible. Un negocio puede llamar
«socio» a lo que el modelo registra como cliente, o «huésped» a lo que es un
paciente. El método translate_alias y los dialectos de dominio
absorben esa diferencia. El usuario habla su idioma; el sistema traduce por debajo.
La pieza ingeniosa es cómo se resuelve la polisemia. Un mismo verbo puede tener varias
entradas (dar la mano no es dar una conferencia), y el lexicon elige
la más específica que coincide.
La especificidad se mide por la longitud del patrón de complementos: una entrada con patrón
('la_mano',) es más específica que una genérica sin patrón.
Las entradas se mantienen ordenadas de más específica a más general, así que
resolve devuelve la primera que encaja con los complementos de la oración.
El módulo también resuelve los sustantivos que esconden un verbo («la llegada del avión»
remite a llegar) con un índice de formas nominales, y guarda los dialectos de
dominio en una tabla aparte.
Tres mecanismos distintos, una sola idea. Que la riqueza y la ambigüedad del lenguaje natural se resuelvan antes de tocar el grafo, no dentro de él.
"""Lexicon: diccionario verbo → tipo de situación + roles + aliases.
Implementa el paradigma del capítulo 14: cada entrada declara una
signatura (qué roles obligatorios y opcionales), un tipo de situación en K,
y aliases naturales por rol y por dominio. La resolución de polisemia
elige la entrada más específica que coincide con el patrón de complemento.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
@dataclass
class LexiconEntry:
verb: str # "vender", "dar", ...
situation_type: str # K id, ej. "accion_vender"
obligatory: List[str] = field(default_factory=list)
optional: List[str] = field(default_factory=list)
aliases: Dict[str, List[str]] = field(default_factory=dict)
nominal_forms: List[str] = field(default_factory=list)
pattern: Optional[Tuple[str, ...]] = None # complementos clave para polisemia
notes: str = ""
example: str = ""
def specificity(self) -> int:
"""Cuanto más largo el patrón, más específica la entrada."""
return 0 if self.pattern is None else len(self.pattern)
def matches(self, observed_complements: List[str]) -> bool:
"""¿Esta entrada coincide con los complementos vistos en la oración?
Una entrada con `pattern=None` siempre coincide (entrada genérica).
Una entrada con patrón coincide si TODOS sus elementos aparecen
entre los complementos.
"""
if self.pattern is None:
return True
obs = set(observed_complements)
return all(p in obs for p in self.pattern)
class Lexicon:
"""Lexicon con resolución de polisemia.
Las entradas se registran por verbo; varias entradas pueden compartir
verbo (polisemia). `resolve(verb, complements)` devuelve la entrada
más específica que coincide.
"""
def __init__(self):
self._by_verb: Dict[str, List[LexiconEntry]] = {}
# Aliases globales de dominio: nombre_usuario → rol_canonico
self._domain_aliases: Dict[str, Dict[str, str]] = {}
# Aliases nominales globales: forma_nominal → verbo
self._nominal_index: Dict[str, str] = {}
# --- registro ----------------------------------------------------------
def register(self, entry: LexiconEntry) -> None:
self._by_verb.setdefault(entry.verb, []).append(entry)
# Orden por especificidad decreciente: más específico primero.
self._by_verb[entry.verb].sort(key=lambda e: -e.specificity())
for nf in entry.nominal_forms:
self._nominal_index[nf] = entry.verb
def register_domain_dialect(self, domain: str,
aliases: Dict[str, str]) -> None:
"""Agrega un dialecto de dominio: nombre_usuario → rol_canónico."""
self._domain_aliases.setdefault(domain, {}).update(aliases)
# --- resolución --------------------------------------------------------
def resolve(self, verb: str,
complements: Optional[List[str]] = None) -> Optional[LexiconEntry]:
"""Devuelve la entrada más específica que coincide con el verbo
y los complementos observados. None si no hay match.
"""
candidates = self._by_verb.get(verb, [])
comps = complements or []
for entry in candidates: # ya ordenadas más específico → más general
if entry.matches(comps):
return entry
return None
def resolve_nominal(self, nominal_form: str,
complements: Optional[List[str]] = None
) -> Optional[LexiconEntry]:
"""Resolución por forma nominal (nominalización).
*"la llegada del avión"* → verbo `llegar` → entrada de llegar.
"""
verb = self._nominal_index.get(nominal_form)
if verb is None:
return None
return self.resolve(verb, complements)
def translate_alias(self, domain: str, user_term: str) -> Optional[str]:
"""Traduce un término de usuario al rol canónico vía dialecto."""
return self._domain_aliases.get(domain, {}).get(user_term)
def alias_for_role(self, verb: str, role: str) -> List[str]:
"""Aliases declarados para un rol de una entrada específica."""
entries = self._by_verb.get(verb, [])
for e in entries:
if role in e.aliases:
return e.aliases[role]
return []
# --- introspección -----------------------------------------------------
def verbs(self) -> List[str]:
return list(self._by_verb.keys())
def entries_for(self, verb: str) -> List[LexiconEntry]:
return list(self._by_verb.get(verb, []))§A6 universe.py — el universo V
Ciento treinta líneas. El universo es el contenedor que aloja todos los individuos y todos los hechos. Por dentro mantiene tres índices (por sujeto, por valor y por rol) para que las consultas no degeneren en un barrido lineal.
Aquí es donde se conecta el catálogo: cada assert_fact le delega la validación de
la firma, si se le inyectó uno.
Sin catálogo, el universo acepta cualquier hecho bien formado. Con catálogo, rechaza los que violan una firma declarada.
Los hechos no se sobreescriben jamás
El universo solo añade. Un dato que cambia no se edita: se modela
con un rango de vigencia (D6) o con una situación nueva que rectifica o
cancela a la anterior. La historia completa queda en el grafo, y el parámetro
at= de las consultas reconstruye cualquier momento del pasado.
El método add_individual protege una regla sutil y crucial: un mismo
identificador no puede vivir en dos ejes distintos. Si alguien intenta registrar
sede_central como lugar y luego como agente, el universo aborta.
Es la garantía de que cada cosa ocupa un lugar único en el espacio de coordenadas.
Los métodos facts_about, facts_with_role y
facts_with_value son las tres puertas de recuperación, todas con un parámetro
at opcional que filtra por vigencia. Y summary da un censo rápido:
cuántos individuos por eje y cuántos hechos en total.
"""El universo V — unión disjunta de los ejes de valor.
`Universe` es el almacenamiento en memoria del prototipo: una lista de
individuos y una lista de hechos, con índices para consulta eficiente.
Diseño:
- Se accede vía `add_individual`, `assert_fact`, `query`.
- La validación de signatura ocurre al insertar el hecho (delegada al Catalog).
- Los hechos son inmutables; "cambios" se modelan como nuevas situaciones
o como rangos de vigencia (D6).
No persiste a disco — el prototipo es en memoria. Para tests y demos basta.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional, Iterator, Tuple, Any
from .axes import Axis
from .individual import Individual
from .fact import Fact
@dataclass
class Universe:
name: str = "default"
individuals: Dict[str, Individual] = field(default_factory=dict)
facts: List[Fact] = field(default_factory=list)
catalog: Any = None # Catalog inyectado opcionalmente para validación
# Índices
_by_subject: Dict[str, List[int]] = field(default_factory=dict)
_by_value: Dict[str, List[int]] = field(default_factory=dict)
_by_role: Dict[str, List[int]] = field(default_factory=dict)
# --- registro de individuos --------------------------------------------
def add_individual(self, ind: Individual) -> Individual:
existing = self.individuals.get(ind.id)
if existing is not None:
if existing.axis != ind.axis:
raise ValueError(
f"Conflicto de eje para {ind.id}: ya existe en {existing.axis} "
f"y se intenta registrar en {ind.axis}."
)
return existing
self.individuals[ind.id] = ind
return ind
def ind(self, id_or_label: str) -> Individual:
"""Recupera un individuo por id."""
if id_or_label in self.individuals:
return self.individuals[id_or_label]
raise KeyError(f"Individuo no registrado: {id_or_label}")
# --- afirmación de hechos ----------------------------------------------
def assert_fact(
self,
subject: Individual,
role: str,
value: Individual,
valid_from: Optional[datetime] = None,
valid_to: Optional[datetime] = None,
) -> Fact:
"""Afirma un hecho atómico. Valida la signatura si hay catálogo."""
self.add_individual(subject)
self.add_individual(value)
if self.catalog is not None:
self.catalog.validate(role, subject, value)
fact = Fact(
subject=subject,
role=role,
value=value,
valid_from=valid_from,
valid_to=valid_to,
)
idx = len(self.facts)
self.facts.append(fact)
self._by_subject.setdefault(subject.id, []).append(idx)
self._by_value.setdefault(value.id, []).append(idx)
self._by_role.setdefault(role, []).append(idx)
return fact
# --- recuperación ------------------------------------------------------
def facts_about(self, individual: Individual,
at: Optional[datetime] = None) -> List[Fact]:
"""Todos los hechos donde `individual` es el sujeto."""
idxs = self._by_subject.get(individual.id, [])
result = [self.facts[i] for i in idxs]
if at is not None:
result = [f for f in result if f.is_valid_at(at)]
return result
def facts_with_role(self, role: str,
at: Optional[datetime] = None) -> List[Fact]:
idxs = self._by_role.get(role, [])
result = [self.facts[i] for i in idxs]
if at is not None:
result = [f for f in result if f.is_valid_at(at)]
return result
def facts_with_value(self, individual: Individual,
at: Optional[datetime] = None) -> List[Fact]:
idxs = self._by_value.get(individual.id, [])
result = [self.facts[i] for i in idxs]
if at is not None:
result = [f for f in result if f.is_valid_at(at)]
return result
# --- utilidades --------------------------------------------------------
def __len__(self) -> int:
return len(self.facts)
def summary(self) -> str:
n_by_axis: Dict[Axis, int] = {}
for ind in self.individuals.values():
n_by_axis[ind.axis] = n_by_axis.get(ind.axis, 0) + 1
parts = [f"Universe '{self.name}'",
f" individuos: {len(self.individuals)}"]
for ax in Axis:
if ax in n_by_axis:
parts.append(f" {ax.value}: {n_by_axis[ax]}")
parts.append(f" hechos: {len(self.facts)}")
return "\n".join(parts)§A7 ingest.py — el pipeline de ingesta
Noventa y seis líneas. Aquí está la función que más se usa en los ejemplos de la
Parte V: ingest_situation.
Toma un verbo y un diccionario de roles ya identificados, y aplica seis pasos que producen una situación con identidad propia (D4) y todos sus hechos colgando de ella.
El análisis lingüístico se da por hecho fuera: lo hace un LLM, o un analizador dedicado. La ingesta no entiende español. Modela lo que alguien ya entendió.
D4 Reificar la situación
El paso clave es acuñar un individuo nuevo en O que representa la situación entera (la venta, la consulta, el préstamo) y colgar de él cada rol como un hecho.
Un evento deja de ser una fila con columnas y pasa a ser un nodo con cables. Eso es lo que permite que una situación tenga, a su vez, modalidad, causa o vigencia propias.
La secuencia se lee casi como su propia documentación. Resuelve el lexicon, con polisemia o
forma nominal. Da identidad a la situación en O. Asienta su
instancia_de apuntando al tipo. Recorre los roles y asienta cada uno como un
hecho. Agrega los extras (modalidad, estatus factual, calificación).
Y antes de devolver, comprueba que no falte ningún rol obligatorio. El valor de retorno es la situación, para que quien llama pueda seguir enriqueciéndola.
Fíjate en la coherencia con la política liberal del catálogo. La ingesta no exige que cada rol pasado esté declarado en la entrada del lexicon. La entrada informa, no aprieta, y los roles de más se admiten.
Lo único que se verifica con dureza es que estén los obligatorios (sin agente no hay venta) y,
vía assert_fact, la firma de cada hecho contra el catálogo.
"""Ingesta: traduce una "oración" (estructurada) a hechos en el universo.
Para el prototipo no parsear español real — eso es trabajo del LLM.
En su lugar exponemos una API de ingesta que recibe verbo + roles ya
identificados y aplica el pipeline:
1. Resolver lexicon (con polisemia / forma nominal)
2. Reificar la situación en O
3. Asentar instancia_de = tipo_situacion
4. Asentar cada rol como hecho atómico
5. Validar contra el catálogo D8
6. Verificar obligatorios
Devuelve la situación reificada para que el caller pueda enriquecerla.
"""
from __future__ import annotations
from datetime import datetime
from typing import Dict, Optional, Any, List
from .axes import Axis
from .individual import Individual, mint_id, category
from .universe import Universe
from .lexicon import Lexicon, LexiconEntry
class IngestError(ValueError):
pass
def ingest_situation(
universe: Universe,
lexicon: Lexicon,
verb: str,
roles: Dict[str, Individual],
*,
complements: Optional[List[str]] = None,
nominal: bool = False,
valid_from: Optional[datetime] = None,
valid_to: Optional[datetime] = None,
extra: Optional[Dict[str, Individual]] = None,
sit_id: Optional[str] = None,
) -> Individual:
"""Ingesta una situación a partir de un verbo y sus roles.
`roles` mapea NOMBRE_CANÓNICO → Individual (el caller ya resolvió aliases).
`complements` lista patrones de complemento que disparan polisemia
(p. ej. ['la_mano'] para `dar la mano`).
`nominal=True` indica que el "verbo" es en realidad una forma nominal
("llegada") y debe buscarse vía `resolve_nominal`.
`extra` agrega hechos adicionales que no están en la signatura (p. ej.
`modalidad`, `estatus_factual`, `calificacion`, ...).
Devuelve la situación reificada en O.
"""
entry = (
lexicon.resolve_nominal(verb, complements)
if nominal else
lexicon.resolve(verb, complements)
)
if entry is None:
raise IngestError(
f"Lexicon no tiene entrada para '{verb}' "
f"con complementos {complements}."
)
# Verificar obligatorios
missing = [r for r in entry.obligatory if r not in roles]
if missing:
raise IngestError(
f"Faltan roles obligatorios para '{verb}': {missing}"
)
# Reificar la situación
sid = sit_id or mint_id(entry.situation_type)
situ = Individual(id=sid, axis=Axis.O, label=sid)
universe.add_individual(situ)
# instancia_de
tipo = category(entry.situation_type)
universe.assert_fact(situ, "instancia_de", tipo,
valid_from=valid_from, valid_to=valid_to)
# Roles obligatorios y opcionales
for role, value in roles.items():
# No exigimos que esté declarado en la entrada — la entrada es
# informativa; los extras se admiten. (Política liberal.)
universe.assert_fact(situ, role, value,
valid_from=valid_from, valid_to=valid_to)
# Extras (modalidad, calificación, ...)
if extra:
for role, value in extra.items():
universe.assert_fact(situ, role, value,
valid_from=valid_from, valid_to=valid_to)
return situ§A8 query.py — el motor de consulta
Ciento veinticuatro líneas. Aquí las preguntas se vuelven operaciones concretas.
Una consulta es un Pattern: un diccionario de roles fixed con
valores conocidos, lo que ya sabemos, y uno o más roles en ask marcados con un
Var, lo que queremos descubrir.
«¿Quién vendió el circuito de hidroterapia?» se escribe fijando el rol tema en el
circuito y pidiendo el rol agente. El motor busca las situaciones que cumplen los
roles fijos y devuelve el valor del rol preguntado.
La pregunta como proyección
Preguntar es fijar algunas coordenadas y dejar otra libre. ¿Quién? deja libre el rol con valor en Q. ¿Cuándo?, el de T. ¿Cuánto?, el de N. El motor recorre las situaciones que casan con lo fijo y devuelve lo que ocupa la coordenada libre. La geometría del libro se vuelve aquí un bucle.
El algoritmo tiene dos fases.
Primero elige un ancla para no recorrer todo el universo. Si el patrón restringe por tipo, parte de ahí. Si no, toma el primer rol fijo como punto de entrada y usa su índice.
Después filtra las candidatas comprobando que cada una cumpla todos los roles fijos, verifica el tipo si se pidió y extrae los valores de los roles preguntados. Si un rol tiene varios valores, devuelve la lista. Si tiene uno, el individuo suelto.
Toda consulta admite el parámetro at: el momento en que se quiere evaluar la
vigencia. Es lo que permite preguntar «¿quién era el dueño en aquella fecha?» y obtener la
respuesta correcta para ese instante, no para el presente. Es la consulta con dos
tiempos de D6.
Y count es query devolviendo cuántos hay en vez de cuáles. El caso
de «¿cuántos?».
"""Motor de consulta: las preguntas-WH como proyecciones.
Una consulta es un `Pattern`: un diccionario de roles fijos con valores
conocidos, y al menos un rol marcado como `Var(...)` (la pregunta).
El motor busca todas las situaciones del universo que satisfacen los
roles fijos y proyecta el valor del rol pregunta.
Soporta:
- Consultas puntuales: ¿quién vendió X?
- Consultas temporales: ¿quién era el dueño de X en T0? (D6)
- Filtros por tipo (instancia_de = K).
- Agregaciones: count, list.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional, Any
from .individual import Individual
from .universe import Universe
@dataclass
class Var:
"""Marcador de variable en un patrón de consulta."""
name: str = "?"
@dataclass
class Pattern:
"""Patrón de consulta sobre una situación.
`fixed`: roles cuyo valor está dado (Individual).
`ask`: uno o más roles cuyo valor queremos descubrir (Var).
"""
fixed: Dict[str, Individual] = field(default_factory=dict)
ask: Dict[str, Var] = field(default_factory=dict)
type_constraint: Optional[Individual] = None # filtra por instancia_de = K
def __post_init__(self):
# `ask` vacío es válido: cuenta/lista candidatos sin proyección.
# En ese caso el binding contiene solo `_subject`.
pass
def query(universe: Universe, pattern: Pattern,
at: Optional[datetime] = None) -> List[Dict[str, Any]]:
"""Ejecuta el patrón contra el universo. Devuelve una lista de bindings.
Cada binding es un dict con las claves de `pattern.ask` y los valores
encontrados (Individual). Las situaciones-candidatas son los sujetos
que tienen *todos* los roles fijos del patrón (con sus valores) y, si
aplica, instancia_de = type_constraint.
"""
# Punto 1: buscar candidatas — sujetos en O que tienen todos los roles
# del patrón. Tomamos el primer rol fijo (o type_constraint) como ancla
# para reducir el espacio.
if pattern.type_constraint is not None:
# Sujetos cuya instancia_de == type_constraint
candidate_subjects = {
f.subject.id
for f in universe.facts_with_role("instancia_de", at=at)
if f.value.id == pattern.type_constraint.id
}
elif pattern.fixed:
# Tomamos el primer fixed como ancla.
role0, val0 = next(iter(pattern.fixed.items()))
candidate_subjects = {
f.subject.id
for f in universe.facts_with_role(role0, at=at)
if f.value.id == val0.id
}
else:
# Si solo hay ask, buscamos sobre todas las situaciones (raro).
candidate_subjects = set(universe.individuals.keys())
# Punto 2: filtrar candidatas por todos los roles fijos
results: List[Dict[str, Any]] = []
for sid in candidate_subjects:
subject = universe.individuals[sid]
sit_facts = universe.facts_about(subject, at=at)
# Map role → list of values for this subject
roles_map: Dict[str, List[Individual]] = {}
for f in sit_facts:
roles_map.setdefault(f.role, []).append(f.value)
# Chequear roles fijos
ok = True
for role, expected_val in pattern.fixed.items():
vals = roles_map.get(role, [])
if not any(v.id == expected_val.id for v in vals):
ok = False
break
if not ok:
continue
# Chequear type_constraint si está
if pattern.type_constraint is not None:
instancia_vals = roles_map.get("instancia_de", [])
if not any(v.id == pattern.type_constraint.id for v in instancia_vals):
continue
# Extraer valores para los roles preguntados
binding: Dict[str, Any] = {"_subject": subject}
all_present = True
for ask_role in pattern.ask:
vals = roles_map.get(ask_role, [])
if not vals:
all_present = False
break
# Si hay más de uno, devolvemos lista; si uno, el individuo.
binding[ask_role] = vals[0] if len(vals) == 1 else list(vals)
if all_present:
results.append(binding)
return results
def count(universe: Universe, pattern: Pattern,
at: Optional[datetime] = None) -> int:
"""Cuenta sujetos que satisfacen el patrón."""
return len(query(universe, pattern, at=at))§A9 __init__.py — la superficie pública
Treinta y dos líneas. Es lo último que ve quien importa la librería y lo primero que toca al
escribir from wq import *.
Reexporta cada clase y función que el usuario va a usar, con una documentación que resume el modelo entero en una pantalla. Esta fachada es el contrato del paquete: si un nombre está aquí, es público y estable. Si no, es interno.
Vale la pena leer el __all__ como un índice del propio modelo. Están los
Axis. Los constructores de individuos (category,
quantity, time_point). El Fact y el
Universe. El Catalog con sus firmas. El Lexicon. El
motor de consulta (Pattern, Var, query,
count). Y la ingesta.
Nada más. Toda la potencia del prototipo cabe en estos diecisiete nombres.
"""WQuestions — prototipo de validación en Python.
Implementa el modelo descrito en `WQuestions.md` y discutido en el libro:
- 7 ejes (Q, O, L, T, N, K, M)
- hechos atómicos con signatura tipada
- situaciones reificadas (D4, D5)
- catálogo canónico de roles (D8)
- lexicon con resolución de polisemia (D9)
- vigencia temporal por reificación (D6)
- consultas como proyecciones sobre roles (preguntas-WH)
Se prioriza claridad sobre rendimiento. La meta es validar la arquitectura,
no servir producción.
"""
from .axes import Axis
from .individual import Individual, mint_id, category, quantity, time_point
from .fact import Fact
from .universe import Universe
from .catalog import Catalog, RoleSignature, SignatureError
from .lexicon import Lexicon, LexiconEntry
from .query import Pattern, Var, query, count
from .ingest import ingest_situation, IngestError
__all__ = [
"Axis", "Individual", "mint_id", "category", "quantity", "time_point",
"Fact", "Universe",
"Catalog", "RoleSignature", "SignatureError",
"Lexicon", "LexiconEntry",
"Pattern", "Var", "query", "count",
"ingest_situation", "IngestError",
]Seis observaciones sobre el código completo
Visto de una sola pasada, de axes.py a __init__.py, el código deja
seis impresiones que vale la pena nombrar. No son conclusiones nuevas: son la confirmación,
en líneas de Python, de lo que el libro defiende en prosa.
1 Poco de todo
Nueve archivos, ~850 líneas, una clase principal por archivo. Sin infraestructura inútil ni abstracciones especulativas. Cada pieza existe porque un capítulo entero la justifica.
2 Las decisiones se ven
D3 vive en fact.py, D4 en
ingest.py, D6 en is_valid_at, D8 en catalog.py, D9
en lexicon.py. Una decisión, un módulo: el paralelismo es directo.
3 La política liberal es explícita
En catalog.validate, un rol no declarado
no se valida. Cuatro líneas que materializan la decisión de mantener el modelo extensible
sin pelear con el catálogo central.
4 La bitemporalidad es ligera
No hay un sistema bitemporal completo. Hay
valid_from/valid_to opcionales, un tx_time
automático y un parámetro at= en las consultas. Suficiente para mudanzas,
rediagnósticos y cláusulas que expiran.
5 No hay parser de lenguaje
El prototipo asume el parser externo (un LLM o un
módulo dedicado) que se enchufa por ingest_situation. Separar «entender» de
«modelar» es lo que lo hace reutilizable en cualquier idioma.
6 El motor es ingenuo
Un barrido con tres índices simples. Valida el modelo en dominios reales; no escala a producción. Los reemplazos serios (Datalog, SHACL, RDF, columnar) están en el capítulo 31.
Juntas, estas seis observaciones justifican el enfoque: no construir infraestructura, sino prueba de concepto.
Lo que estas ochocientas cincuenta líneas demuestran es que el modelo es coherente y operable. La validación industrial es otro trabajo: el del capítulo 31 y de quien lo continúe.
Lo que queda en el repositorio
Para que este anexo siga siendo legible, incluí solo la librería núcleo. El resto del prototipo (alrededor de 2.400 líneas adicionales) está publicado completo en el repositorio del proyecto. Este es el mapa de lo que encontrarás allí.
github.com/joseabantomarin/WQuestions
└── prototipo/
├── wq/ ← incluida en este anexo (850 líneas)
│ ├── axes.py
│ ├── individual.py
│ ├── fact.py
│ ├── catalog.py
│ ├── lexicon.py
│ ├── universe.py
│ ├── ingest.py
│ ├── query.py
│ └── __init__.py
│
├── ejemplos/ ← no incluida — está en el repo
│ ├── spa.py (532 líneas)
│ ├── dominios_previos.py (522 líneas — receta, gol, canción, noticia)
│ ├── banco.py (465 líneas)
│ ├── clinica.py (312 líneas)
│ └── taxi.py (256 líneas)
│
└── tests/ ← no incluida — está en el repo
└── test_wq.py (349 líneas)Cada archivo de ejemplos/ es un script ejecutable que modela un dominio
completo. Declara los individuos, registra el lexicon, ingresa todas las situaciones, corre las
consultas del libro y valida los resultados.
Ejecutar cualquiera de ellos (por ejemplo python -m prototipo.ejemplos.banco)
reproduce las escenas de la Parte V tal como aparecen en el texto.
El archivo test_wq.py ejercita las reglas del modelo. Que el catálogo valide
firmas. Que las consultas con fecha devuelvan lo correcto en cada momento. Que la polisemia del
lexicon elija la entrada más específica. Que dar identidad a una situación produzca el grafo
esperado.
Es el cinturón de seguridad detrás de la promesa de que «todas las pruebas pasan».
Correr el prototipo en cinco minutos
git clone https://github.com/joseabantomarin/WQuestions.git
cd WQuestions
python -m prototipo.tests.test_wq # corre la batería de tests
python -m prototipo.ejemplos.spa # ejecuta el dominio del Spa
python -m prototipo.ejemplos.banco # ejecuta el dominio bancarioNo hace falta instalar dependencias: el prototipo solo usa la
librería estándar de Python. Los identificadores y las APIs (Universe,
ingest_situation, Pattern, query) son los mismos que
leíste aquí, así que pasar de leer a ejecutar es inmediato.
El prototipo, los ejemplos y los tests son el espejo operable del libro: lo que se afirma en cualquier capítulo de la Parte V se puede ejecutar línea por línea.Del texto a la infraestructura
Si encuentras una afirmación del libro que no puedas reproducir, el repositorio tiene un registro de incidencias abierto. Es exactamente esa clase de respuesta la que el capítulo 31 reclama para que la propuesta deje de ser un texto y empiece a ser infraestructura de uso diario.
Clónalo, córrelo, rómpelo.