---

1.1. Propósito y audiencia

Este notebook ofrece una guía práctica para construir grafos de conocimiento con conciencia temporal y realizar recuperación multi-salto directamente sobre esos grafos.

Está diseñado para ingenieros, arquitectos y analistas que trabajan con grafos de conocimiento con conciencia temporal. Ya sea que estés creando prototipos, implementando a escala o explorando nuevas formas de usar datos estructurados, encontrarás flujos de trabajo prácticos, mejores prácticas y marcos de decisión para acelerar tu trabajo.

Este manual presenta dos flujos de trabajo prácticos que puedes usar, extender e implementar de inmediato:

1. Construcción de grafos de conocimiento (KG) con conciencia temporal
   

   Un desafío clave en el desarrollo de sistemas de IA basados en conocimiento es mantener una base de datos que se mantenga actualizada y relevante. Si bien se presta mucha atención a mejorar la precisión de la recuperación con técnicas como la similitud semántica y el re-ranking, esta guía se centra en un aspecto fundamental, pero frecuentemente pasado por alto: actualizar y validar sistemáticamente tu base de conocimiento a medida que llegan nuevos datos.

   No importa cuán avanzados sean tus algoritmos de recuperación, su efectividad está limitada por la calidad y la frescura de tu base de datos. Este manual demuestra cómo validar y actualizar rutinariamente las entradas del grafo de conocimiento a medida que llegan nuevos datos, ayudando a asegurar que tu base de conocimiento permanezca precisa y actualizada.

2. Recuperación multi-salto usando grafos de conocimiento
   

   Aprende a combinar modelos de OpenAI (como o3, o4-mini, GPT-4.1 y GPT-4.1-mini) con consultas de grafos estructuradas a través de llamadas a herramientas, permitiendo que el modelo recorra tu grafo en múltiples pasos a través de entidades y relaciones.

   Este método permite que tu sistema responda preguntas complejas y multifacéticas que requieren razonamiento sobre varios hechos vinculados, yendo mucho más allá de lo que la recuperación de un solo salto puede lograr.

Dentro, descubrirás:

• Marcos de decisión prácticos para elegir modelos y técnicas de prompt en cada etapa
• Ejemplos de código plug-and-play para una fácil integración en tus pipelines de ML y datos
• Enlaces a recursos detallados sobre el uso de herramientas de OpenAI, fine-tuning, selección de backend de grafos y más
• Un camino claro del prototipo a la producción, con mejores prácticas accionables para la escalabilidad y la fiabilidad

Nota: Todas las referencias y recomendaciones se basan en los mejores modelos y prácticas disponibles a junio de 2025. A medida que el ecosistema evoluciona, revisa periódicamente tu enfoque para mantenerte al día con las nuevas capacidades y mejoras.

1.2. Puntos clave

Creación de un grafo de conocimiento con conciencia temporal con un agente temporal

1. ¿Por qué hacer que tu grafo de conocimiento sea temporal?
   

   Los grafos de conocimiento tradicionales tratan los hechos como estáticos, pero la información del mundo real evoluciona constantemente. Lo que era cierto el trimestre pasado puede estar obsoleto hoy, lo que arriesga errores o decisiones desinformadas si el grafo no captura el cambio a lo largo del tiempo. Los grafos de conocimiento temporales te permiten responder con precisión preguntas como "¿Qué era cierto en una fecha determinada?" o analizar cómo los hechos y las relaciones han cambiado, asegurando que las decisiones siempre se basen en el contexto más relevante.

2. ¿Qué es un agente temporal?
   

   Un agente temporal es un componente de pipeline que ingiere datos sin procesar y produce tripletas con marca de tiempo para tu grafo de conocimiento. Esto permite consultas precisas basadas en el tiempo, construcción de líneas de tiempo, análisis de tendencias y más.

3. ¿Cómo funciona el pipeline?
   

   El pipeline comienza dividiendo semánticamente tus documentos sin procesar. Estos fragmentos se descomponen en declaraciones listas para nuestro agente temporal, que luego crea tripletas con conciencia del tiempo. Un agente de invalidación puede entonces realizar verificaciones de validez temporal, detectando y manejando cualquier declaración que sea invalidada por nuevas declaraciones que inciden en el grafo.

Recuperación multi-paso sobre un grafo de conocimiento

1. ¿Por qué usar la recuperación multi-paso?
   

   Las consultas directas de un solo salto con frecuencia omiten hechos relevantes distribuidos a través de la topología de un grafo. La recuperación multi-paso (multi-salto) permite el recorrido iterativo, siguiendo relaciones y agregando evidencia a través de varios saltos. Esta metodología saca a la luz dependencias complejas y conexiones latentes que permanecerían ocultas con búsquedas de un solo intento, proporcionando respuestas más completas y matizadas a consultas sofisticadas.

2. Planificadores
   

   Los planificadores orquestan el proceso de recuperación. Los planificadores orientados a tareas descomponen las consultas en subtareas concretas y secuenciales. Los planificadores orientados a hipótesis, por el contrario, proponen afirmaciones para confirmar, refutar o evolucionar. Elegir la estrategia óptima depende de dónde se encuentre el problema en el espectro desde la elaboración de informes deterministas (rutas bien definidas) hasta la investigación exploratoria (inferencia abierta).

3. Paradigmas de diseño de herramientas
   

   El diseño de herramientas abarca un continuo: las herramientas fijas proporcionan resultados consistentes y predecibles para consultas específicas (por ejemplo, un servicio que siempre devuelve el clima de hoy para San Francisco). En el otro extremo, las herramientas de forma libre ofrecen una amplia flexibilidad, como la ejecución de código o la recuperación de datos abiertos. Las herramientas semiestructuradas se encuentran entre estos extremos, restringiendo ciertas acciones mientras permiten una flexibilidad personalizada; los subagentes especializados son un ejemplo típico. Seleccionar el paradigma apropiado es una compensación entre control, adaptabilidad y complejidad.

4. Evaluación de sistemas de recuperación
   

   La evaluación de alta fidelidad depende de respuestas "doradas" curadas por expertos, aunque estas son costosas y laboriosas de producir. Los juicios automatizados, como los de los LLM o los rastros de herramientas, se pueden generar rápidamente para complementar o preseleccionar, pero pueden carecer de la precisión de la evaluación humana. A medida que tu sistema madura, haz la transición hacia el aprovechamiento de los comentarios de los usuarios reales para medir y optimizar la calidad de la recuperación en producción.

   Un flujo de trabajo probado: comienza con pruebas sintéticas, compara con tu conjunto de datos "dorado" curado y anotado por humanos, y refina iterativamente utilizando los comentarios y calificaciones de los usuarios en vivo.

Del prototipo a la producción

1. Mantén el grafo ligero
   

   Establece políticas de archivo y asigna puntuaciones de relevancia numérica a cada arista (por ejemplo, actualidad x confianza x frecuencia de consulta). Automatiza el archivo o la esparsificación de nodos y aristas de bajo valor, asegurando que solo los hechos más críticos y frecuentemente accedidos permanezcan para una recuperación rápida.

2. Paraleliza el pipeline de ingesta
   

   Haz la transición de un pipeline lineal documento → fragmento → extracción → resolución a una arquitectura por etapas y asíncrona. Asigna a cada fase de procesamiento su propia cola y un grupo de trabajadores dedicado. Aplica agrupamiento o procesamiento por lotes basado en red para trabajos de invalidación para maximizar la eficiencia. Procesa por lotes las solicitudes de API externas (por ejemplo, OpenAI) y las escrituras en la base de datos siempre que sea posible. Este diseño aumenta el rendimiento, introduce contrapresión para la fiabilidad y te permite escalar cada etapa del pipeline de forma independiente.

3. Integra salvaguardas de producción robustas
   

   Aplica una validación rigurosa de la salida: estandariza los campos temporales (por ejemplo, formato de fecha ISO-8601), restringe los tipos de entidad a tu vocabulario controlado y aplica verificaciones de cordura ligeras basadas en modelos para la consistencia de la salida. Emplea el registro estructurado con identificadores rastreables y monitorea la calidad en tiempo real y las métricas de rendimiento en tiempo real para detectar proactivamente la deriva de datos, las regresiones o las anomalías del pipeline antes de que impacten en las aplicaciones posteriores.

2. Cómo usar este manual

---

Este manual está diseñado para una interacción flexible:

1. Úsalo como una guía técnica completa: léelo de principio a fin para obtener una comprensión profunda de los sistemas de grafos de conocimiento con conciencia temporal.
2. Echa un vistazo a los conceptos avanzados, metodologías y patrones de implementación si prefieres una visión general de alto nivel.
3. Salta a cualquiera de las tres secciones modulares; cada una es autónoma y directamente aplicable a escenarios del mundo real.

Dentro, encontrarás:

1. Creación de un grafo de conocimiento con conciencia temporal con un agente temporal
   

   Construye un pipeline que extrae entidades y relaciones de texto no estructurado, resuelve conflictos temporales y mantiene tu grafo actualizado a medida que llega nueva información.

2. Recuperación multi-paso sobre un grafo de conocimiento
   

   Usa consultas estructuradas y razonamiento de modelos de lenguaje para encadenar múltiples saltos a través de tu grafo y responder preguntas complejas.

3. Del prototipo a la producción
   

   Pasa de la experimentación a la implementación. Esta sección cubre consejos de arquitectura, patrones de integración y consideraciones para escalar de manera confiable.

2.1. Requisitos previos

Antes de sumergirte en la construcción de agentes temporales y grafos de conocimiento, configuremos tu entorno. Instala todas las dependencias requeridas con pip y establece tu clave de API de OpenAI como una variable de entorno. Se requiere Python 3.12 o posterior.

!python -V
%pip install --upgrade pip
%pip install -qU chonkie datetime ipykernel jinja2 matplotlib networkx numpy openai plotly pydantic rapidfuzz scipy tenacity tiktoken pandas
%pip install -q "datasets<3.0"

Python 3.12.8
Requirement already satisfied: pip in ./.venv/lib/python3.12/site-packages (25.1.1)
Note: you may need to restart the kernel to use updated packages.
Note: you may need to restart the kernel to use updated packages.
Note: you may need to restart the kernel to use updated packages.

import os

if "OPENAI_API_KEY" not in os.environ:
    import getpass
    os.environ["OPENAI_API_KEY"] = getpass.getpass("Paste your OpenAI API key here: ")

3. Creación de un grafo de conocimiento con conciencia temporal con un agente temporal

---

Los datos precisos son la base de cualquier buena decisión empresarial. Los últimos modelos de OpenAI como o3, o4-mini y la familia GPT-4.1 están permitiendo a las empresas construir sistemas de recuperación de última generación para sus flujos de trabajo más importantes. Sin embargo, la información evoluciona rápidamente: los hechos ingeridos con confianza ayer pueden estar ya desactualizados hoy.

Sin la capacidad de rastrear cuándo fue válido cada hecho, los sistemas de recuperación corren el riesgo de devolver respuestas desactualizadas, no conformes o engañosas. Las consecuencias de perder el contexto temporal pueden ser graves en cualquier industria, como lo ilustran los siguientes ejemplos.

Si bien hemos señalado algunos ejemplos específicos aquí, este tema es cierto en muchas industrias, incluidas las farmacéuticas, el derecho, los bienes de consumo y más.

Más allá de la recuperación estándar

Un grafo de conocimiento con conciencia temporal te permite ir más allá de la búsqueda de hechos estáticos. Permite flujos de trabajo de recuperación más ricos, como preguntas y respuestas fácticas basadas en el tiempo, generación de líneas de tiempo, seguimiento de cambios, análisis contrafactual y más. Profundizamos en estos con más detalle en nuestra sección de recuperación más adelante en el manual.

3.1. Presentamos nuestro agente temporal

---

Un agente temporal es un pipeline especializado que convierte declaraciones sin formato y de forma libre en tripletas con conciencia del tiempo listas para ser ingeridas en un grafo de conocimiento que luego puede ser consultado con preguntas del tipo "¿Qué era cierto en el momento T?".

Las tripletas son los bloques de construcción básicos de los grafos de conocimiento. Es una forma de representar un solo hecho o pieza de conocimiento usando tres partes (de ahí, "tripletas"):

• Sujeto - la entidad de la que estás hablando
• Predicado - el tipo de relación o propiedad
• Objeto - el valor u otra entidad a la que está conectado el sujeto

Puedes pensar en esto como una oración con una estructura [Subject] - [Predicate] - [Object]. Como un ejemplo más claro:

"London" - "isCapitalOf" - "United Kingdom"

El agente temporal implementado en este manual se inspira en Zep y Graphiti, al tiempo que introduce un control más estricto sobre la invalidación de hechos y un enfoque más matizado para la tipificación episódica.

3.1.1. Mejoras clave introducidas en este manual

1. Extracción de validez temporal
   

   Se basa en el diseño de prompt de Graphiti para identificar lapsos temporales y contexto episódico sin requerir declaraciones de referencia auxiliares.

2. Lógica de invalidación de hechos
   

   Introduce verificaciones de bidireccionalidad y restringe las comparaciones por tipo episódico. Esto conserva el enfoque sin pérdidas de Zep al tiempo que reduce las evaluaciones innecesarias.

3. Tipificación temporal y episódica
   

   Diferencia entre Fact, Opinion, Prediction, así como entre clases temporales Static, Dynamic, Atemporal.

4. Extracción de múltiples eventos
   

   Maneja oraciones compuestas y referencias de fecha anidadas en una sola pasada.

Este proceso nos permite actualizar nuestras fuentes de verdad de manera eficiente y confiable:

Nota: Si bien la implementación en este manual se centra en una implementación basada en grafos, este enfoque es generalizable a otras estructuras de bases de conocimiento, por ejemplo, sistemas basados en pgvector.

---

3.1.2. El pipeline del agente temporal

El agente temporal procesa las declaraciones entrantes a través de un pipeline de tres etapas:

1. Clasificación temporal
   

   Etiqueta cada declaración como Atemporal, Estática o Dinámica:

   • Las declaraciones Atemporales nunca cambian (por ejemplo, "La velocidad de la luz en el vacío es ≈3×10⁸ m s⁻¹").
   • Las declaraciones Estáticas son válidas desde un punto en el tiempo pero no cambian después (por ejemplo, "La persona YY fue CEO de la Compañía XX el 23 de octubre de 2014.").
   • Las declaraciones Dinámicas evolucionan (por ejemplo, "La persona YY es CEO de la Compañía XX.").
2. Extracción de eventos temporales
   

   Identifica fechas relativas o parciales (por ejemplo, "martes", "hace tres meses") y las resuelve a una fecha absoluta utilizando la marca de tiempo del documento o heurísticas de respaldo (por ejemplo, por defecto el 1 o el último día del mes si solo se conoce el mes).

3. Verificación de validez temporal
   

   Asegura que cada declaración incluya una marca de tiempo t_created y, cuando corresponda, una marca de tiempo t_expired. Luego, el agente compara la tripleta candidata con las entradas existentes del grafo de conocimiento para:

   • Detectar contradicciones y marcar entradas obsoletas con t_invalid
   • Vincular declaraciones más nuevas con aquellas que invalidan con invalidated_by

3.1.3. Selección del modelo adecuado para un agente temporal

Al construir sistemas con LLM, es una buena práctica comenzar con modelos más grandes y luego buscar optimizar y reducir.

La serie GPT-4.1 es particularmente adecuada para construir agentes temporales debido a su gran capacidad para seguir instrucciones. En benchmarks como Scale’s MultiChallenge, GPT-4.1 supera a GPT-4o en un $10.5\%_{abs}$, demostrando una capacidad superior para mantener el contexto, razonar en la conversación y adherirse a las instrucciones, rasgos clave para extraer tripletas con marca de tiempo. Estas capacidades lo convierten en una excelente opción para prototipar agentes que dependen de la extracción de datos con conciencia del tiempo.

Flujo de trabajo de desarrollo recomendado

1. Prototipa con GPT-4.1
   

   Maximiza la corrección y reduce el tiempo de depuración de prompts mientras construyes la lógica central del pipeline.

2. Cambia a GPT-4.1-mini o GPT-4.1-nano
   

   Una vez que los prompts y la lógica sean estables, cambia a variantes más pequeñas para una menor latencia y una inferencia rentable.

3. Destila en GPT-4.1-mini o GPT-4.1-nano
   

   Usa la destilación de modelos de OpenAI para entrenar modelos más pequeños con salidas de alta calidad de un modelo "maestro" más grande como GPT-4.1, preservando (o incluso mejorando) el rendimiento en relación con GPT-4.1.

En la práctica, esto se ve así: prototipa con GPT-4.1 → mide la calidad → baja en la escala hasta que las compensaciones ya no satisfagan tus necesidades.

3.2. Construyendo nuestro pipeline de agente temporal

---

Antes de sumergirnos en los detalles de la implementación, es útil comprender el pipeline de ingesta a un alto nivel:

1. Cargar transcripciones
   
2. Creación de un segmentador semántico
   
3. Sentando las bases para nuestro agente temporal
   
4. Extracción de declaraciones
   
5. Extracción de rango temporal
   
6. Creación de nuestras tripletas
   
7. Eventos temporales
   
8. Definición de nuestro agente temporal
   
9. Resolución de entidades
   
10. Agente de invalidación
    
11. Construyendo nuestro pipeline
    

Diagrama de arquitectura

3.2.1. Cargar transcripciones

Para los propósitos de este manual, hemos seleccionado el "Earnings Calls Dataset" (jlh-ibm/earnings_call), que está disponible bajo la licencia Creative Commons Zero v1.0. Este conjunto de datos contiene una colección de 188 transcripciones de llamadas de ganancias originadas en el período 2016-2020 en relación con el mercado de valores NASDAQ. Creemos que este conjunto de datos es una buena opción para este manual, ya que extraer información de, y posteriormente consultar información de, las transcripciones de llamadas de ganancias es un problema común en muchas instituciones financieras de todo el mundo.

Además, el carácter a menudo variable de las declaraciones y los temas de la misma empresa en múltiples llamadas de ganancias proporciona un vector útil para demostrar el concepto de grafo de conocimiento temporal.

A pesar de que este conjunto de datos se centra en el mundo financiero, construimos el Agente Temporal con una estructura general, por lo que será rápido de adaptar a problemas similares en otras industrias como la farmacéutica, legal, automotriz y más.

Para los propósitos de este manual, estamos limitando el procesamiento a dos empresas —AMD y Nvidia—, aunque en la práctica esta pipeline se puede escalar fácilmente a cualquier empresa.

Comencemos cargando el conjunto de datos desde HuggingFace.

from datasets import load_dataset

hf_dataset_name = "jlh-ibm/earnings_call"
subset_options = ["stock_prices", "transcript-sentiment", "transcripts"]

hf_dataset = load_dataset(hf_dataset_name, subset_options[2])
my_dataset = hf_dataset["train"]

my_dataset

Dataset({
    features: ['company', 'date', 'transcript'],
    num_rows: 150
})

row = my_dataset[0]
row["company"], row["date"], row["transcript"][:200]

from collections import Counter

company_counts = Counter(my_dataset["company"])
company_counts

Configuración de la base de datos

Antes de procesar estos datos, configuremos nuestra base de datos.

Para mayor comodidad dentro de un formato de notebook, hemos elegido SQLite como nuestra base de datos para esta implementación. En la sección "Prototipo a Producción" y en la sección del Apéndice A.1, "Almacenamiento y recuperación de datos de grafos de alto volumen", discutimos consideraciones para diferentes opciones de conjuntos de datos en un entorno de producción.

Si estás ejecutando este manual localmente, puedes optar por configurar memory = False para guardar la base de datos en el almacenamiento. Se utilizará la ruta de archivo predeterminada, my_database.db, o puedes pasar tu propio argumento db_path a make_connection.

Configuraremos varias tablas para almacenar la siguiente información:

• Transcripciones
• Fragmentos (Chunks)
• Eventos temporales
• Tríadas (Triplets)
• Entidades (incluyendo mapeos canónicos)

Este código está abstraído detrás de un método make_connection que crea la nueva base de datos SQLite. Los detalles de este método se pueden encontrar en el script db_interface.py en el repositorio de GitHub para este manual.

from db_interface import make_connection

sqlite_conn = make_connection(memory=False, refresh=True)

3.2.2. Creación de un Chunker Semántico

Antes de sumergirnos en la construcción de la clase Chunker en sí, comenzamos definiendo nuestros primeros modelos de datos. Como se considera una buena práctica al trabajar con Python, se utiliza Pydantic para garantizar la seguridad de tipos y la claridad en nuestras definiciones de modelos. Pydantic proporciona una forma limpia y declarativa de definir estructuras de datos, al tiempo que valida y analiza automáticamente los datos de entrada, lo que hace que nuestros modelos de datos sean robustos y fáciles de usar.

Modelo de fragmento (Chunk model)

Este es un modelo de datos central que usaremos para almacenar segmentos individuales de texto extraídos de transcripciones, junto con cualquier metadato asociado. A medida que procesamos las transcripciones dividiéndolas en fragmentos semánticamente significativos, cada pieza se guardará como un Chunk separado.

Cada Chunk contiene:

• id: Un identificador único generado automáticamente para cada fragmento. Esto nos ayuda a identificar y rastrear fragmentos de texto a lo largo del proceso.
• text: Un campo de cadena que contiene el contenido de texto del fragmento.
• metadata: Un diccionario para permitir un almacenamiento flexible de metadatos.

import uuid
from typing import Any

from pydantic import BaseModel, Field


class Chunk(BaseModel):
    """A chunk of text from an earnings call."""

    id: uuid.UUID = Field(default_factory=uuid.uuid4)
    text: str
    metadata: dict[str, Any]

Modelo de transcripción

Como su nombre indica, utilizaremos el modelo Transcript para representar el contenido completo de una transcripción de llamada de ganancias. Captura varias piezas clave de información:

• id: Análogo a Chunk, esto nos da un identificador único.
• text: El texto completo de la transcripción.
• company: El nombre de la empresa sobre la que se realizó la llamada de ganancias.
• date: La fecha de la llamada de ganancias.
• quarter: El trimestre fiscal en el que se realizó la llamada de ganancias.
• chunks: Una lista de objetos Chunk, cada uno representando un segmento significativo de la transcripción completa.

Para asegurar que el campo date se maneje correctamente, se utiliza el validador to_datetime para convertir el valor al formato datetime.

from datetime import datetime

from pydantic import field_validator


class Transcript(BaseModel):
    """A transcript of a company earnings call."""

    id: uuid.UUID = Field(default_factory=uuid.uuid4)
    text: str
    company: str
    date: datetime
    quarter: str | None = None
    chunks: list[Chunk] | None = None

    @field_validator("date", mode="before")
    @classmethod
    def to_datetime(cls, d: Any) -> datetime:
        """Convert input to a datetime object."""
        if isinstance(d, datetime):
            return d
        if hasattr(d, "isoformat"):
            return datetime.fromisoformat(d.isoformat())
        return datetime.fromisoformat(str(d))

Clase Chunker

Ahora, definimos la clase Chunker para dividir cada transcripción en fragmentos semánticamente significativos. En lugar de depender de reglas arbitrarias como el recuento de caracteres o los saltos de línea, aplicamos el chunking semántico para preservar más la integridad contextual de la transcripción original. Esto asegura que cada fragmento sea una unidad autocontenida que mantenga juntas las ideas contextualmente vinculadas. Esto es particularmente útil para tareas posteriores como la extracción de declaraciones, donde el contexto influye en gran medida en la precisión.

La clase chunker contiene dos métodos:

• find_quarter

  Este método intenta extraer el trimestre fiscal (por ejemplo, "Q1 2023") directamente del texto de la transcripción utilizando una expresión regular simple. En este caso, esto es sencillo ya que el formato de datos de los trimestres en las transcripciones es consistente y está bien definido.

  Sin embargo, en escenarios del mundo real, detectar el trimestre de manera confiable puede requerir más trabajo. En múltiples fuentes o tipos de documentos, los detalles del trimestre probablemente serán diferentes. Los LLM son excelentes herramientas para ayudar a aliviar este problema. Intenta usar GPT-4.1-mini con un prompt específicamente para extraer el trimestre dado un contexto más amplio del documento.

• generate_transcripts_and_chunks

  Este es el método principal que toma un conjunto de datos (como un iterable de diccionarios) y devuelve una lista de objetos Transcript, cada uno poblado con Chunks derivados semánticamente. Realiza los siguientes pasos:

  1. Creación de transcripciones: Inicializa objetos Transcript utilizando los campos de texto, empresa y fecha proporcionados.
  2. Filtrado: Utiliza el SemanticChunker de chonkie junto con el modelo text-embedding-3-small de OpenAI para dividir la transcripción en segmentos lógicos.
  3. Asignación de fragmentos: Envuelve cada segmento semántico en un modelo Chunk, adjuntando metadatos relevantes como índices de inicio y fin.

El chunker se encuentra en esta parte de nuestra pipeline:

import re
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Any

from chonkie import OpenAIEmbeddings, SemanticChunker
from tqdm import tqdm


class Chunker:
    """
    Takes in transcripts of earnings calls and extracts quarter information and splits
    the transcript into semantically meaningful chunks using embedding-based similarity.
    """

    def __init__(self, model: str = "text-embedding-3-small"):
        self.model = model

    def find_quarter(self, text: str) -> str | None:
        """Extract the quarter (e.g., 'Q1 2023') from the input text if present, otherwise return None."""
        # In this dataset we can just use regex to find the quarter as it is consistently defined
        search_results = re.findall(r"[Q]\d\s\d{4}", text)

        if search_results:
            quarter = str(search_results[0])
            return quarter

        return None


    def generate_transcripts_and_chunks(
        self,
        dataset: Any,
        company: list[str] | None = None,
        text_key: str = "transcript",
        company_key: str = "company",
        date_key: str = "date",
        threshold_value: float = 0.7,
        min_sentences: int = 3,
        num_workers: int = 50,
    ) -> list[Transcript]:
        """Populate Transcript objects with semantic chunks."""
        # Populate the Transcript objects with the passed data on the transcripts
        transcripts = [
            Transcript(
                text=d[text_key],
                company=d[company_key],
                date=d[date_key],
                quarter=self.find_quarter(d[text_key]),
            )
            for d in dataset
        ]

        if company:
            transcripts = [t for t in transcripts if t.company in company]

        def _process(t: Transcript) -> Transcript:
            if not hasattr(_process, "chunker"):
                embed_model = OpenAIEmbeddings(self.model)
                _process.chunker = SemanticChunker(
                    embedding_model=embed_model,
                    threshold=threshold_value,
                    min_sentences=max(min_sentences, 1),
                )
            semantic_chunks = _process.chunker.chunk(t.text)
            t.chunks = [
                Chunk(
                    text=c.text,
                    metadata={
                        "start_index": getattr(c, "start_index", None),
                        "end_index": getattr(c, "end_index", None),
                    },
                )
                for c in semantic_chunks
            ]
            return t

        # Create the semantic chunks and add them to their respective Transcript object using a thread pool
        with ThreadPoolExecutor(max_workers=num_workers) as pool:
            futures = [pool.submit(_process, t) for t in transcripts]
            transcripts = [
                f.result()
                for f in tqdm(
                    as_completed(futures),
                    total=len(futures),
                    desc="Generating Semantic Chunks",
                )
            ]

        return transcripts

raw_data = list(my_dataset)

chunker = Chunker()
transcripts = chunker.generate_transcripts_and_chunks(raw_data)

Alternativamente, podemos cargar solo las transcripciones AMD y NVDA pre-fragmentadas de archivos preprocesados en transcripts/

import pickle
from pathlib import Path


def load_transcripts_from_pickle(directory_path: str = "transcripts/") -> list[Transcript]:
    """Load all pickle files from a directory into a dictionary."""
    loaded_transcripts = []
    dir_path = Path(directory_path).resolve()


    for pkl_file in sorted(dir_path.glob("*.pkl")):
        try:
            with open(pkl_file, "rb") as f:
                transcript = pickle.load(f)
                # Ensure it's a Transcript object
                if not isinstance(transcript, Transcript):
                    transcript = Transcript(**transcript)
                loaded_transcripts.append(transcript)
                print(f"✅ Loaded transcript from {pkl_file.name}")
        except Exception as e:
            print(f"❌ Error loading {pkl_file.name}: {e}")

    return loaded_transcripts

# transcripts = load_transcripts_from_pickle()

Ahora podemos inspeccionar un par de fragmentos:

chunks = transcripts[0].chunks
if chunks is not None:
    for i, chunk in enumerate(chunks[21:23]):
        print(f"Chunk {i+21}:")
        print(f"  ID: {chunk.id}")
        print(f"  Text: {repr(chunk.text[:200])}{'...' if len(chunk.text) > 100 else ''}")
        print(f"  Metadata: {chunk.metadata}")
        print()
else:
    print("No chunks found for the first transcript.")

Con esto, hemos dividido con éxito nuestras transcripciones en fragmentos seccionados semánticamente. Ahora podemos pasar a los siguientes pasos de nuestra pipeline.

3.2.3. Sentando las bases para nuestro Agente Temporal

Antes de pasar a definir la clase TemporalAgent, primero definiremos los prompts y modelos de datos necesarios para su funcionamiento.

Formalizando nuestras definiciones de etiquetas

Para que nuestro agente temporal pueda extraer con precisión los tipos de declaración y temporales, necesitamos proporcionarle un contexto suficientemente detallado y específico. Para mayor comodidad, los definimos a continuación en un formato estructurado.

Cada etiqueta contiene tres piezas cruciales de información que luego pasaremos a nuestros LLM en los prompts.

• definition
  

  Proporciona una descripción concisa de lo que representa la etiqueta. Establece los límites conceptuales del tipo de declaración o temporal y asegura la consistencia en la interpretación a través de ejemplos.

• date_handling_guidance
  

  Explica cómo interpretar la validez temporal de una declaración asociada con la etiqueta. Describe cómo las fechas valid_at y invalid_at deben derivarse al procesar instancias de esa etiqueta.

• date_handling_examples
  

  Incluye ejemplos ilustrativos de cómo las declaraciones del mundo real serían etiquetadas y anotadas temporalmente bajo esta etiqueta. Estos se utilizarán como ejemplos de pocas tomas (few-shot examples) para los LLM posteriores.

LABEL_DEFINITIONS: dict[str, dict[str, dict[str, str]]] = {
    "episode_labelling": {
        "FACT": dict(
            definition=(
                "Statements that are objective and can be independently "
                "verified or falsified through evidence."
            ),
            date_handling_guidance=(
                "These statements can be made up of multiple static and "
                "dynamic temporal events marking for example the start, end, "
                "and duration of the fact described statement."
            ),
            date_handling_example=(
                "'Company A owns Company B in 2022', 'X caused Y to happen', "
                "or 'John said X at Event' are verifiable facts which currently "
                "hold true unless we have a contradictory fact."
            ),
        ),
        "OPINION": dict(
            definition=(
                "Statements that contain personal opinions, feelings, values, "
                "or judgments that are not independently verifiable. It also "
                "includes hypothetical and speculative statements."
            ),
            date_handling_guidance=(
                "This statement is always static. It is a record of the date the "
                "opinion was made."
            ),
            date_handling_example=(
                "'I like Company A's strategy', 'X may have caused Y to happen', "
                "or 'The event felt like X' are opinions and down to the reporters "
                "interpretation."
            ),
        ),
        "PREDICTION": dict(
            definition=(
                "Uncertain statements about the future on something that might happen, "
                "a hypothetical outcome, unverified claims. It includes interpretations "
                "and suggestions. If the tense of the statement changed, the statement "
                "would then become a fact."
            ),
            date_handling_guidance=(
                "This statement is always static. It is a record of the date the "
                "prediction was made."
            ),
            date_handling_example=(
                "'It is rumoured that Dave will resign next month', 'Company A expects "
                "X to happen', or 'X suggests Y' are all predictions."
            ),
        ),
    },
    "temporal_labelling": {
        "STATIC": dict(
            definition=(
                "Often past tense, think -ed verbs, describing single points-in-time. "
                "These statements are valid from the day they occurred and are never "
                "invalid. Refer to single points in time at which an event occurred, "
                "the fact X occurred on that date will always hold true."
            ),
            date_handling_guidance=(
                "The valid_at date is the date the event occurred. The invalid_at date "
                "is None."
            ),
            date_handling_example=(
                "'John was appointed CEO on 4th Jan 2024', 'Company A reported X percent "
                "growth from last FY', or 'X resulted in Y to happen' are valid the day "
                "they occurred and are never invalid."
            ),
        ),
        "DYNAMIC": dict(
            definition=(
                "Often present tense, think -ing verbs, describing a period of time. "
                "These statements are valid for a specific period of time and are usually "
                "invalidated by a Static fact marking the end of the event or start of a "
                "contradictory new one. The statement could already be referring to a "
                "discrete time period (invalid) or may be an ongoing relationship (not yet "
                "invalid)."
            ),
            date_handling_guidance=(
                "The valid_at date is the date the event started. The invalid_at date is "
                "the date the event or relationship ended, for ongoing events this is None."
            ),
            date_handling_example=(
                "'John is the CEO', 'Company A remains a market leader', or 'X is continuously "
                "causing Y to decrease' are valid from when the event started and are invalidated "
                "by a new event."
            ),
        ),
        "ATEMPORAL": dict(
            definition=(
                "Statements that will always hold true regardless of time therefore have no "
                "temporal bounds."
            ),
            date_handling_guidance=(
                "These statements are assumed to be atemporal and have no temporal bounds. Both "
                "their valid_at and invalid_at are None."
            ),
            date_handling_example=(
                "'A stock represents a unit of ownership in a company', 'The earth is round', or "
                "'Europe is a continent'. These statements are true regardless of time."
            ),
        ),
    },
}

3.2.4. Extracción de declaraciones

"Extracción de declaraciones" se refiere al proceso de dividir nuestros fragmentos semánticos en los "hechos" atómicos más pequeños posibles. Dentro de nuestro Agente Temporal, esto se logra mediante:

1. Encontrar cada afirmación declarativa e independiente
   

   Extraer declaraciones que puedan valerse por sí mismas como expresiones completas de sujeto-predicado-objeto sin depender del contexto circundante.

2. Asegurar la atomicidad
   

   Descomponer oraciones complejas o compuestas en unidades fácticas mínimas e indivisibles, cada una expresando una única relación.

3. Resolver referencias
   

   Reemplazar pronombres o referencias abstractas (por ejemplo, "él" o "La Compañía") con entidades específicas (por ejemplo, "John Smith", "AMD") utilizando el sujeto principal para la desambiguación.

4. Preservar la precisión temporal y cuantitativa
   

   Retener fechas, duraciones y cantidades explícitas para anclar cada hecho con precisión en el tiempo y la escala.

5. Etiquetar cada declaración extraída
   

   Cada declaración se anota con un StatementType y un TemporalType.

Tipos temporales

El enum TemporalType proporciona un conjunto estandarizado de categorías temporales que facilitan la clasificación y el trabajo con declaraciones extraídas de transcripciones de llamadas de ganancias.

Cada categoría captura un tipo diferente de referencia temporal:

• Atemporal: Declaraciones que son universalmente verdaderas e invariantes en el tiempo (por ejemplo, "La velocidad de la luz en el vacío es ≈3×10⁸ m s⁻¹.").
• Estático: Declaraciones que se hicieron verdaderas en un punto específico en el tiempo y permanecen inalteradas a partir de entonces (por ejemplo, "La Persona YY fue CEO de la Compañía XX el 23 de octubre de 2014.").
• Dinámico: Declaraciones que pueden cambiar con el tiempo y requieren contexto temporal para interpretarse con precisión (por ejemplo, "La Persona YY es CEO de la Compañía XX.").

from enum import StrEnum


class TemporalType(StrEnum):
    """Enumeration of temporal types of statements."""

    ATEMPORAL = "ATEMPORAL"
    STATIC = "STATIC"
    DYNAMIC = "DYNAMIC"

Tipos de declaraciones

De manera similar, el enum StatementType clasifica la naturaleza de cada declaración extraída, capturando sus características epistémicas.

• Hecho: Una declaración que afirma una afirmación verificable considerada verdadera en el momento en que se hizo. Sin embargo, puede ser posteriormente reemplazada o contradicha por otros hechos (por ejemplo, información actualizada o correcciones).
• Opinión: Una declaración subjetiva que refleja la creencia, el sentimiento o el juicio de un orador. Por naturaleza, las opiniones se consideran temporalmente verdaderas en el momento en que se expresan.
• Predicción: Una declaración prospectiva o hipotética sobre un posible evento o resultado futuro. Temporalmente, se asume que una predicción es verdadera desde el momento de la enunciación hasta la conclusión de la ventana de predicción inferida.

class StatementType(StrEnum):
    """Enumeration of statement types for statements."""

    FACT = "FACT"
    OPINION = "OPINION"
    PREDICTION = "PREDICTION"

Declaración en bruto (Raw Statement)

El modelo RawStatement representa una declaración individual extraída por un LLM, anotada con su tipo semántico (StatementType) y clasificación temporal (TemporalType). Estas declaraciones en bruto sirven como representaciones intermedias y están destinadas a ser transformadas en objetos TemporalEvent en etapas de procesamiento posteriores.

Campos principales:

• statement: El contenido textual de la declaración extraída.
• statement_type: El tipo de declaración (Hecho, Opinión, Predicción), basado en el enum StatementType.
• temporal_type: La clasificación temporal de la declaración (Estática, Dinámica, Atemporal), extraída del enum TemporalType.

El modelo incluye validadores a nivel de campo para asegurar que todas las anotaciones de tipo se ajusten a sus respectivos enums, proporcionando una capa de robustez contra entradas inválidas.

El modelo complementario RawStatementList contiene la salida del paso de extracción de declaraciones: una lista de instancias RawStatement.

from pydantic import field_validator


class RawStatement(BaseModel):
    """Model representing a raw statement with type and temporal information."""

    statement: str
    statement_type: StatementType
    temporal_type: TemporalType

    @field_validator("temporal_type", mode="before")
    @classmethod
    def _parse_temporal_label(cls, value: str | None) -> TemporalType:
        if value is None:
            return TemporalType.ATEMPORAL
        cleaned_value = value.strip().upper()
        try:
            return TemporalType(cleaned_value)
        except ValueError as e:
            raise ValueError(f"Invalid temporal type: {value}. Must be one of {[t.value for t in TemporalType]}") from e

    @field_validator("statement_type", mode="before")
    @classmethod
    def _parse_statement_label(cls, value: str | None = None) -> StatementType:
        if value is None:
            return StatementType.FACT
        cleaned_value = value.strip().upper()
        try:
            return StatementType(cleaned_value)
        except ValueError as e:
            raise ValueError(f"Invalid temporal type: {value}. Must be one of {[t.value for t in StatementType]}") from e

class RawStatementList(BaseModel):
    """Model representing a list of raw statements."""

    statements: list[RawStatement]

Prompt de extracción de declaraciones

Este es el prompt central que impulsa la capacidad de nuestro Agente Temporal para extraer y etiquetar declaraciones atómicas. Está escrito en Jinja, lo que nos permite componer modularmente entradas dinámicas sin reescribir la lógica central.

Anatomía del prompt

1. Configurar la tarea de extracción
   

   Instruimos al asistente para que se comporte como un experto en finanzas y definimos claramente las dos subtareas: (i) extraer declaraciones atómicas y declarativas, y (ii) etiquetar cada una con un statement_type y un temporal_type.

2. Aplica estrictas pautas de extracción
   

   Las reglas de extracción ayudan a garantizar la coherencia y la claridad. Las declaraciones deben:

   • Estar estructuradas como tríadas limpias de sujeto-predicado-objeto.
   • Ser autocontenidas e independientes del contexto.
   • Resolver correferencias (por ejemplo, "él" → "John Smith").
   • Incluir calificadores temporales/cuantitativos cuando estén presentes.
   • Dividirse cuando se describan múltiples eventos o temporalidades.
3. Soporta definiciones plug-and-play
   

   El bloque {% if definitions %} facilita la inyección de definiciones estructuradas como categorías de declaraciones, tipos temporales y términos específicos del dominio.

4. Incluye ejemplos de pocas tomas (few-shot examples)
   

   Proporcionamos un fragmento de ejemplo anotado y la salida JSON correspondiente para demostrar al modelo cómo debe comportarse.

statement_extraction_prompt = '''
{% macro tidy(name) -%}
  {{ name.replace('_', ' ')}}
{%- endmacro %}

You are an expert finance professional and information-extraction assistant.

===Inputs===
{% if inputs %}
{% for key, val in inputs.items() %}
- {{ key }}: {{val}}
{% endfor %}
{% endif %}

===Tasks===
1. Identify and extract atomic declarative statements from the chunk given the extraction guidelines
2. Label these (1) as Fact, Opinion, or Prediction and (2) temporally as Static or Dynamic

===Extraction Guidelines===
- Structure statements to clearly show subject-predicate-object relationships
- Each statement should express a single, complete relationship (it is better to have multiple smaller statements to achieve this)
- Avoid complex or compound predicates that combine multiple relationships
- Must be understandable without requiring context of the entire document
- Should be minimally modified from the original text
- Must be understandable without requiring context of the entire document,
    - resolve co-references and pronouns to extract complete statements, if in doubt use main_entity for example:
      "your nearest competitor" -> "main_entity's nearest competitor"
    - There should be no reference to abstract entities such as 'the company', resolve to the actual entity name.
    - expand abbreviations and acronyms to their full form

- Statements are associated with a single temporal event or relationship
- Include any explicit dates, times, or quantitative qualifiers that make the fact precise
- If a statement refers to more than 1 temporal event, it should be broken into multiple statements describing the different temporalities of the event.
- If there is a static and dynamic version of a relationship described, both versions should be extracted

{%- if definitions %}
  {%- for section_key, section_dict in definitions.items() %}
==== {{ tidy(section_key) | upper }} DEFINITIONS & GUIDANCE ====
    {%- for category, details in section_dict.items() %}
{{ loop.index }}. {{ category }}
- Definition: {{ details.get("definition", "") }}
    {% endfor -%}
  {% endfor -%}
{% endif -%}

===Examples===
Example Chunk: """
  TechNova Q1 Transcript (Edited Version)
  Attendees:
  * Matt Taylor
    ABC Ltd - Analyst
  * Taylor Morgan
    BigBank Senior - Coordinator
  ----
  On April 1st, 2024, John Smith was appointed CFO of TechNova Inc. He works alongside the current Senior VP Olivia Doe. He is currently overseeing the company’s global restructuring initiative, which began in May 2024 and is expected to continue into 2025.
  Analysts believe this strategy may boost profitability, though others argue it risks employee morale. One investor stated, “I think Jane has the right vision.”
  According to TechNova’s Q1 report, the company achieved a 10% increase in revenue compared to Q1 2023. It is expected that TechNova will launch its AI-driven product line in Q3 2025.
  Since June 2024, TechNova Inc has been negotiating strategic partnerships in Asia. Meanwhile, it has also been expanding its presence in Europe, starting July 2024. As of September 2025, the company is piloting a remote-first work policy across all departments.
  Competitor SkyTech announced last month they have developed a new AI chip and launched their cloud-based learning platform.
"""

Example Output: {
  "statements": [
    {
      "statement": "Matt Taylor works at ABC Ltd.",
      "statement_type": "FACT",
      "temporal_type": "DYNAMIC"
    },
    {
      "statement": "Matt Taylor is an Analyst.",
      "statement_type": "FACT",
      "temporal_type": "DYNAMIC"
    },
    {
      "statement": "Taylor Morgan works at BigBank.",
      "statement_type": "FACT",
      "temporal_type": "DYNAMIC"
    },
    {
      "statement": "Taylor Morgan is a Senior Coordinator.",
      "statement_type": "FACT",
      "temporal_type": "DYNAMIC"
    },
    {
      "statement": "John Smith was appointed CFO of TechNova Inc on April 1st, 2024.",
      "statement_type": "FACT",
      "temporal_type": "STATIC"
    },
    {
      "statement": "John Smith has held position CFO of TechNova Inc from April 1st, 2024.",
      "statement_type": "FACT",
      "temporal_type": "DYNAMIC"
    },
    {
      "statement": "Olivia Doe is the Senior VP of TechNova Inc.",
      "statement_type": "FACT",
      "temporal_type": "DYNAMIC"
    },
    {
      "statement": "John Smith works with Olivia Doe.",
      "statement_type": "FACT",
      "temporal_type": "DYNAMIC"
    },
    {
      "statement": "John Smith is overseeing TechNova Inc's global restructuring initiative starting May 2024.",
      "statement_type": "FACT",
      "temporal_type": "DYNAMIC"
    },
    {
      "statement": "Analysts believe TechNova Inc's strategy may boost profitability.",
      "statement_type": "OPINION",
      "temporal_type": "STATIC"
    },
    {
      "statement": "Some argue that TechNova Inc's strategy risks employee morale.",
      "statement_type": "OPINION",
      "temporal_type": "STATIC"
    },
    {
      "statement": "An investor stated 'I think John has the right vision' on an unspecified date.",
      "statement_type": "OPINION",
      "temporal_type": "STATIC"
    },
    {
      "statement": "TechNova Inc achieved a 10% increase in revenue in Q1 2024 compared to Q1 2023.",
      "statement_type": "FACT",
      "temporal_type": "DYNAMIC"
    },
    {
      "statement": "It is expected that TechNova Inc will launch its AI-driven product line in Q3 2025.",
      "statement_type": "PREDICTION",
      "temporal_type": "DYNAMIC"
    },
    {
      "statement": "TechNova Inc started negotiating strategic partnerships in Asia in June 2024.",
      "statement_type": "FACT",
      "temporal_type": "STATIC"
    },
    {
      "statement": "TechNova Inc has been negotiating strategic partnerships in Asia since June 2024.",
      "statement_type": "FACT",
      "temporal_type": "DYNAMIC"
    },
    {
      "statement": "TechNova Inc has been expanding its presence in Europe since July 2024.",
      "statement_type": "FACT",
      "temporal_type": "DYNAMIC"
    },
    {
      "statement": "TechNova Inc started expanding its presence in Europe in July 2024.",
      "statement_type": "FACT",
      "temporal_type": "STATIC"
    },
    {
      "statement": "TechNova Inc is going to pilot a remote-first work policy across all departments as of September 2025.",
      "statement_type": "FACT",
      "temporal_type": "STATIC"
    },
    {
      "statement": "SkyTech is a competitor of TechNova.",
      "statement_type": "FACT",
      "temporal_type": "DYNAMIC"
    },
    {
      "statement": "SkyTech developed new AI chip.",
      "statement_type": "FACT",
      "temporal_type": "STATIC"
    },
    {
      "statement": "SkyTech launched cloud-based learning platform.",
      "statement_type": "FACT",
      "temporal_type": "STATIC"
    }
  ]
}
===End of Examples===

**Output format**
Return only a list of extracted labelled statements in the JSON ARRAY of objects that match the schema below:
{{ json_schema }}
'''

3.2.5. Extracción de rango temporal

Rango temporal en bruto

El modelo RawTemporalRange contiene la extracción en bruto de las cadenas de fecha valid_at y invalid_at para una declaración. Ambas utilizan la propiedad de cadena compatible con fecha y hora.

• valid_at representa el inicio del período de validez de una declaración.
• invalid_at representa el final del período de validez de una declaración.

class RawTemporalRange(BaseModel):
    """Model representing the raw temporal validity range as strings."""

    valid_at: str | None = Field(..., json_schema_extra={"format": "date-time"})
    invalid_at: str | None = Field(..., json_schema_extra={"format": "date-time"})

Rango de validez temporal

Mientras que el modelo RawTemporalRange conserva las cadenas de fecha originalmente extraídas, el modelo TemporalValidityRange las transforma en objetos datetime estandarizados para su procesamiento posterior.

Analiza los valores brutos valid_at y invalid_at, convirtiéndolos de cadenas a instancias datetime conscientes de la zona horaria. Esto se maneja a través de un validador a nivel de campo.

from utils import parse_date_str


class TemporalValidityRange(BaseModel):
    """Model representing the parsed temporal validity range as datetimes."""

    valid_at: datetime | None = None
    invalid_at: datetime | None = None

    @field_validator("valid_at", "invalid_at", mode="before")
    @classmethod
    def _parse_date_string(cls, value: str | datetime | None) -> datetime | None:
        if isinstance(value, datetime) or value is None:
            return value
        return parse_date_str(value)

Prompt de extracción de fechas

Ahora, creemos el prompt que guía a nuestro Agente Temporal para determinar con precisión la validez temporal de las declaraciones.

Anatomía del prompt

Este prompt ayuda al Agente Temporal a comprender y extraer con precisión los rangos de validez temporal.

1. Define claramente la tarea de extracción
   

   El prompt instruye a nuestro modelo a determinar cuándo una declaración se hizo verdadera (valid_at) y, opcionalmente, cuándo dejó de serlo (invalid_at).

2. Utiliza guía contextual
   

   Al incorporar dinámicamente {{ inputs.temporal_type }} y {{ inputs.statement_type }}, el prompt guía al modelo en la interpretación de los matices temporales basándose en la naturaleza de cada declaración (como distinguir hechos de predicciones o contextos estáticos de dinámicos).

3. Asegura la consistencia con reglas de formato claras
   

   Para mantener la claridad y la consistencia, el prompt requiere que todas las fechas se conviertan a formatos estandarizados de fecha y hora ISO 8601, normalizados a UTC. Ancla explícitamente expresiones relativas (como "el último trimestre") a fechas de publicación conocidas, haciendo que la información temporal sea precisa y confiable.

4. Se alinea con los ciclos de informes comerciales
   

   Reconociendo la necesidad práctica de razonamiento basado en trimestres, común en contextos comerciales y financieros, el prompt puede interpretar y calcular rangos temporales basados en trimestres comerciales, minimizando la ambigüedad.

5. Se adapta a los tipos de declaraciones para la precisión semántica
   

   Reglas específicas aseguran la integridad semántica de las declaraciones; por ejemplo, las opiniones pueden tener solo una fecha de inicio (valid_at) que refleje el momento en que se expresaron, mientras que las predicciones definirán claramente su ventana de pronóstico utilizando una fecha de finalización (invalid_at).

date_extraction_prompt = """
{#
  This prompt (template) is adapted from [getzep/graphiti]
  Licensed under the Apache License, Version 2.0

  Original work:
    https://github.com/getzep/graphiti/blob/main/graphiti_core/prompts/extract_edge_dates.py

  Modifications made by Tomoro on 2025-04-14
  See the LICENSE file for the full Apache 2.0 license text.
#}

{% macro tidy(name) -%}
  {{ name.replace('_', ' ')}}
{%- endmacro %}

INPUTS:
{% if inputs %}
{% for key, val in inputs.items() %}
- {{ key }}: {{val}}
{% endfor %}
{% endif %}

TASK:
- Analyze the statement and determine the temporal validity range as dates for the temporal event or relationship described.
- Use the temporal information you extracted, guidelines below, and date of when the statement was made or published. Do not use any external knowledge to determine validity ranges.
- Only set dates if they explicitly relate to the validity of the relationship described in the statement. Otherwise ignore the time mentioned.
- If the relationship is not of spanning nature and represents a single point in time, but you are still able to determine the date of occurrence, set the valid_at only.

{{ inputs.get("temporal_type") | upper }} Temporal Type Specific Guidance:
{% for key, guide in temporal_guide.items() %}
- {{ tidy(key) | capitalize }}: {{ guide }}
{% endfor %}

{{ inputs.get("statement_type") | upper }} Statement Type Specific Guidance:
{%for key, guide in statement_guide.items() %}
- {{ tidy(key) | capitalize }}: {{ guide }}
{% endfor %}

Validity Range Definitions:
- `valid_at` is the date and time when the relationship described by the statement became true or was established.
- `invalid_at` is the date and time when the relationship described by the statement stopped being true or ended. This may be None if the event is ongoing.

General Guidelines:
  1. Use ISO 8601 format (YYYY-MM-DDTHH:MM:SS.SSSSSSZ) for datetimes.
  2. Use the reference or publication date as the current time when determining the valid_at and invalid_at dates.
  3. If the fact is written in the present tense without containing temporal information, use the reference or publication date for the valid_at date
  4. Do not infer dates from related events or external knowledge. Only use dates that are directly stated to establish or change the relationship.
  5. Convert relative times (e.g., “two weeks ago”) into absolute ISO 8601 datetimes based on the reference or publication timestamp.
  6. If only a date is mentioned without a specific time, use 00:00:00 (midnight) for that date.
  7. If only year or month is mentioned, use the start or end as appropriate at 00:00:00 e.g. do not select a random date if only the year is mentioned, use YYYY-01-01 or YYYY-12-31.
  8. Always include the time zone offset (use Z for UTC if no specific time zone is mentioned).
{% if inputs.get('quarter') and inputs.get('publication_date') %}
  9. Assume that {{ inputs.quarter }} ends on {{ inputs.publication_date }} and infer dates for any Qx references from there.
{% endif %}

Statement Specific Rules:
- when `statement_type` is **opinion** only valid_at must be set
- when `statement_type` is **prediction** set its `invalid_at` to the **end of the prediction window** explicitly mentioned in the text.

Never invent dates from outside knowledge.

**Output format**
Return only the validity range in the JSON ARRAY of objects that match the schema below:
{{ json_schema }}
"""

3.2.6. Creando nuestros trillizos

Ahora construiremos las definiciones y los prompts para crear nuestros trillizos. Como se mencionó anteriormente, estos son una combinación de:

• Sujeto - la entidad de la que estás hablando
• Predicado - el tipo de relación o propiedad
• Objeto - el valor u otra entidad a la que el sujeto está conectado

Comencemos con nuestro predicado.

Predicado

El enum Predicate proporciona un conjunto estándar de predicados que describen claramente las relaciones extraídas del texto.

Hemos definido el conjunto de predicados a continuación para que sean apropiados para las transcripciones de llamadas de ganancias. Aquí hay algunos ejemplos de cómo cada uno de estos predicados podría encajar en un trillizo en nuestro grafo de conocimiento: Aquí hay más ejemplos anonimizados y generalizados siguiendo tu plantilla:

• IS_A: [Empresa ABC]-[ES_UN]-[Proveedor de Software]
• HAS_A: [Corporación XYZ]-[TIENE_UNA]-[División de Innovación]
• LOCATED_IN: [Fábrica 123]-[UBICADA_EN]-[Alemania]
• HOLDS_ROLE: [Jane Doe]-[OCUPA_EL_CARGO_DE]-[CEO en la Empresa LMN]
• PRODUCES: [Empresa DEF]-[PRODUCE]-[Modelo de Smartphone X]
• SELLS: [Minorista 789]-[VENDE]-[Muebles]
• LAUNCHED: [Empresa UVW]-[LANZÓ]-[Nuevo Servicio de Suscripción]
• DEVELOPED: [Startup GHI]-[DESARROLLÓ]-[Herramienta Basada en la Nube]
• ADOPTED_BY: [Nueva Tecnología]-[ADOPTADA_POR]-[Industria ABC]
• INVESTS_IN: [Firma de Inversión JKL]-[INVierte_EN]-[Startups de Energía Limpia]
• COLLABORATES_WITH: [Empresa PQR]-[COLABORA_CON]-[Universidad XYZ]
• SUPPLIES: [Fabricante STU]-[SUMINISTRA]-[Componentes Automotrices a la Empresa VWX]
• HAS_REVENUE: [Corporación LMN]-[TIENE_INGRESOS_DE]-[€500 Millones]
• INCREASED: [Empresa YZA]-[AUMENTÓ]-[Cuota de Mercado]
• DECREASED: [Firma BCD]-[DISMINUYÓ]-[Gastos Operativos]
• RESULTED_IN: [Iniciativa de Reducción de Costos]-[RESULTÓ_EN]-[Márgenes de Beneficio Mejorados]
• TARGETS: [Campaña de Lanzamiento de Producto]-[DIRIGIDA_A]-[Consumidores Millennial]
• PART_OF: [Subsidiaria EFG]-[PARTE_DE]-[Corporación Matriz HIJ]
• DISCONTINUED: [Empresa KLM]-[DESCONTINUÓ]-[Línea de Productos Heredada]
• SECURED: [Startup NOP]-[ASEGURÓ]-[Financiamiento Serie B]

class Predicate(StrEnum):
    """Enumeration of normalised predicates."""

    IS_A = "IS_A"
    HAS_A = "HAS_A"
    LOCATED_IN = "LOCATED_IN"
    HOLDS_ROLE = "HOLDS_ROLE"
    PRODUCES = "PRODUCES"
    SELLS = "SELLS"
    LAUNCHED = "LAUNCHED"
    DEVELOPED = "DEVELOPED"
    ADOPTED_BY = "ADOPTED_BY"
    INVESTS_IN = "INVESTS_IN"
    COLLABORATES_WITH = "COLLABORATES_WITH"
    SUPPLIES = "SUPPLIES"
    HAS_REVENUE = "HAS_REVENUE"
    INCREASED = "INCREASED"
    DECREASED = "DECREASED"
    RESULTED_IN = "RESULTED_IN"
    TARGETS = "TARGETS"
    PART_OF = "PART_OF"
    DISCONTINUED = "DISCONTINUED"
    SECURED = "SECURED"

También asignamos una definición a cada predicado, que luego pasaremos al prompt de extracción más adelante.

PREDICATE_DEFINITIONS = {
    "IS_A": "Denotes a class-or-type relationship between two entities (e.g., 'Model Y IS_A electric-SUV'). Includes 'is' and 'was'.",
    "HAS_A": "Denotes a part-whole relationship between two entities (e.g., 'Model Y HAS_A electric-engine'). Includes 'has' and 'had'.",
    "LOCATED_IN": "Specifies geographic or organisational containment or proximity (e.g., headquarters LOCATED_IN Berlin).",
    "HOLDS_ROLE": "Connects a person to a formal office or title within an organisation (CEO, Chair, Director, etc.).",
    "PRODUCES": "Indicates that an entity manufactures, builds, or creates a product, service, or infrastructure (includes scale-ups and component inclusion).",
    "SELLS": "Marks a commercial seller-to-customer relationship for a product or service (markets, distributes, sells).",
    "LAUNCHED": "Captures the official first release, shipment, or public start of a product, service, or initiative.",
    "DEVELOPED": "Shows design, R&D, or innovation origin of a technology, product, or capability. Includes 'researched' or 'created'.",
    "ADOPTED_BY": "Indicates that a technology or product has been taken up, deployed, or implemented by another entity.",
    "INVESTS_IN": "Represents the flow of capital or resources from one entity into another (equity, funding rounds, strategic investment).",
    "COLLABORATES_WITH": "Generic partnership, alliance, joint venture, or licensing relationship between entities.",
    "SUPPLIES": "Captures vendor–client supply-chain links or dependencies (provides to, sources from).",
    "HAS_REVENUE": "Associates an entity with a revenue amount or metric—actual, reported, or projected.",
    "INCREASED": "Expresses an upward change in a metric (revenue, market share, output) relative to a prior period or baseline.",
    "DECREASED": "Expresses a downward change in a metric relative to a prior period or baseline.",
    "RESULTED_IN": "Captures a causal relationship where one event or factor leads to a specific outcome (positive or negative).",
    "TARGETS": "Denotes a strategic objective, market segment, or customer group that an entity seeks to reach.",
    "PART_OF": "Expresses hierarchical membership or subset relationships (division, subsidiary, managed by, belongs to).",
    "DISCONTINUED": "Indicates official end-of-life, shutdown, or termination of a product, service, or relationship.",
    "SECURED": "Marks the successful acquisition of funding, contracts, assets, or rights by an entity.",
}

Definiendo tus propios predicados

Cuando trabajes con diferentes fuentes de datos, querrás definir tus propios predicados que sean específicos para tu caso de uso.

Para definir tus propios predicados:

1. Primero, ejecuta tu pipeline con PREDICATE_DEFINITIONS = {} en una muestra representativa de tus documentos. Esta ejecución inicial derivará un grafo ruidoso con muchos predicados no estandarizados y superpuestos.
2. Luego, introduce algunos de tus resultados iniciales en ChatGPT o revísalos manualmente para fusionar clases de predicados similares. Este proceso ayuda a eliminar duplicados como IS_CEO y IS_CEO_OF.
3. Finalmente, revisa y refina cuidadosamente esta lista de predicados para asegurar claridad y precisión. Estas definiciones de predicados finalizadas guiarán tu proceso de extracción y asegurarán un pipeline de extracción consistente.

Trillizo en bruto

Con los predicados ahora bien definidos, podemos comenzar a construir los modelos de datos para nuestros trillizos.

El modelo RawTriplet representa una relación básica sujeto-predicado-objeto que se extrae directamente de datos textuales. Esto sirve como precursor para la representación de trillizos más detallada en Triplet que introduciremos más adelante.

Campos principales:

• subject_name: La representación textual de la entidad sujeto
• subject_id: Identificador numérico para la entidad sujeto
• predicate: El tipo de relación, especificado por el enum Predicate
• object_name: La representación textual de la entidad objeto
• object_id: Identificador numérico para la entidad objeto
• value: Valor numérico asociado a la relación, puede ser None, por ejemplo, Company -> HAS_A -> Revenue con value='$100 mill'

class RawTriplet(BaseModel):
    """Model representing a subject-predicate-object triplet."""

    subject_name: str
    subject_id: int
    predicate: Predicate
    object_name: str
    object_id: int
    value: str | None = None

Trillizo

El modelo Triplet extiende el RawTriplet al incorporar identificadores únicos y, opcionalmente, vincular cada trillizo a un evento específico. Estos identificadores ayudan con la integración en bases de conocimiento estructuradas como nuestro grafo de conocimiento temporal.

class Triplet(BaseModel):
    """Model representing a subject-predicate-object triplet."""

    id: uuid.UUID = Field(default_factory=uuid.uuid4)
    event_id: uuid.UUID | None = None
    subject_name: str
    subject_id: int | uuid.UUID
    predicate: Predicate
    object_name: str
    object_id: int | uuid.UUID
    value: str | None = None

    @classmethod
    def from_raw(cls, raw_triplet: "RawTriplet", event_id: uuid.UUID | None = None) -> "Triplet":
        """Create a Triplet instance from a RawTriplet, optionally associating it with an event_id."""
        return cls(
            id=uuid.uuid4(),
            event_id=event_id,
            subject_name=raw_triplet.subject_name,
            subject_id=raw_triplet.subject_id,
            predicate=raw_triplet.predicate,
            object_name=raw_triplet.object_name,
            object_id=raw_triplet.object_id,
            value=raw_triplet.value,
        )

Entidad en bruto

El modelo RawEntity representa una Entidad tal como se extrae del Statement. Esto sirve como precursor para la representación de trillizos más detallada en Entity que introduciremos a continuación.

Campos principales:

• entity_idx: Un número entero para diferenciar las entidades extraídas de la declaración (se vincula a RawTriplet)
• name: El nombre de la entidad extraída, por ejemplo, AMD
• type: El tipo de entidad extraída, por ejemplo, Company
• description: La descripción textual de la entidad, por ejemplo, Technology company know for manufacturing semiconductors

class RawEntity(BaseModel):
    """Model representing an entity (for entity resolution)."""

    entity_idx: int
    name: str
    type: str = ""
    description: str = ""

Entidad

El modelo Entity extiende el RawEntity al incorporar identificadores únicos y, opcionalmente, vincular cada entidad a un evento específico. Además, contiene resolved_id que se llenará durante la resolución de entidades con el ID de la entidad canónica para eliminar la duplicación de nombres de entidades en la base de datos. Estos identificadores actualizados ayudan con la integración y vinculación de entidades a eventos y trillizos.

class Entity(BaseModel):
    """
    Model representing an entity (for entity resolution).
    'id' is the canonical entity id if this is a canonical entity.
    'resolved_id' is set to the canonical id if this is an alias.
    """

    id: uuid.UUID = Field(default_factory=uuid.uuid4)
    event_id: uuid.UUID | None = None
    name: str
    type: str
    description: str
    resolved_id: uuid.UUID | None = None

    @classmethod
    def from_raw(cls, raw_entity: "RawEntity", event_id: uuid.UUID | None = None) -> "Entity":
        """Create an Entity instance from a RawEntity, optionally associating it with an event_id."""
        return cls(
            id=uuid.uuid4(),
            event_id=event_id,
            name=raw_entity.name,
            type=raw_entity.type,
            description=raw_entity.description,
            resolved_id=None,
        )

Extracción en bruto

Tanto RawTriplet como RawEntity se extraen al mismo tiempo por Statement para reducir las llamadas a LLM y permitir una fácil referencia de Entidades a través de Trillizos.

class RawExtraction(BaseModel):
    """Model representing a triplet extraction."""

    triplets: list[RawTriplet]
    entities: list[RawEntity]

Prompt de extracción de trillizos

El prompt a continuación guía a nuestro Agente Temporal para extraer eficazmente trillizos y entidades de las declaraciones proporcionadas.

Anatomía del prompt

• Evita detalles temporales
  

  Se instruye específicamente al agente para que ignore las relaciones temporales, ya que estas se capturan por separado dentro del TemporalValidityRange. Los Predicates definidos están diseñados deliberadamente para ser neutrales en el tiempo; por ejemplo, HAS_A cubre contextos tanto presentes (HAS_A) como pasados (HAD_A).

• Mantiene salidas estructuradas
  

  El prompt produce salidas RawExtraction estructuradas, respaldadas por ejemplos detallados que ilustran claramente:

  • Cómo extraer información de un Statement dado
  • Cómo vincular Entities con los Triplets correspondientes
  • Cómo manejar los values extraídos
  • Cómo gestionar múltiples Triplets que involucran el mismo Entity

triplet_extraction_prompt = """
You are an information-extraction assistant.

**Task:** You are going to be given a statement. Proceed step by step through the guidelines.

**Statement:** "{{ statement }}"

**Guidelines**
First, NER:
- Identify the entities in the statement, their types, and context independent descriptions.
- Do not include any lengthy quotes from the reports
- Do not include any calendar dates or temporal ranges or temporal expressions
- Numeric values should be extracted as separate entities as an instance_of _Numeric_, where the name is the units as a string and the numeric_value is the value. e.g: £30 -> name: 'GBP', numeric_value: 30, instance_of: 'Numeric'

Second, Triplet extraction:
- Identify the subject entity of that predicate – the main entity carrying out the action or being described.
- Identify the object entity of that predicate – the entity, value, or concept that the predicate affects or describes.
- Identify a predicate between the entities expressed in the statement, such as 'is', 'works at', 'believes', etc. Follow the schema below if given.
- Extract the corresponding (subject, predicate, object, date) knowledge triplet.
- Exclude all temporal expressions (dates, years, seasons, etc.) from every field.
- Repeat until all predicates contained in the statement have been extracted form the statements.

{%- if predicate_instructions -%}
-------------------------------------------------------------------------
Predicate Instructions:
Please try to stick to the following predicates, do not deviate unless you can't find a relevant definition.
{%- for pred, instruction in predicate_instructions.items() -%}
- {{ pred }}: {{ instruction }}
{%- endfor -%}
-------------------------------------------------------------------------
{%- endif -%}

Output:
List the entities and triplets following the JSON schema below. Return ONLY with valid JSON matching this schema.
Do not include any commentary or explanation.
{{ json_schema }}

===Examples===
Example 1 Statement: "Google's revenue increased by 10% from January through March."
Example 1 Output: {
  "triplets": [
    {
      "subject_name": "Google",
      "subject_id": 0,
      "predicate": "INCREASED",
      "object_name": "Revenue",
      "object_id": 1,
      "value": "10%",
    }
  ],
  "entities": [
    {
      "entity_idx": 0,
      "name": "Google",
      "type": "Organization",
      "description": "Technology Company",
    },
    {
      "entity_idx": 1,
      "name": "Revenue",
      "type": "Financial Metric",
      "description": "Income of a Company",
    }
  ]
}

Example 2 Statement: "Amazon developed a new AI chip in 2024."
Example 2 Output:
{
  "triplets": [
    {
      "subject_name": "Amazon",
      "subject_id": 0,
      "predicate": "DEVELOPED",
      "object_name": "AI chip",
      "object_id": 1,
      "value": None,
    },
  ],
  "entities": [
    {
      "entity_idx": 0,
      "name": "Amazon",
      "type": "Organization",
      "description": "E-commerce and cloud computing company"
    },
    {
      "entity_idx": 1,
      "name": "AI chip",
      "type": "Technology",
      "description": "Artificial intelligence accelerator hardware"
    }
  ]
}

Example 3 Statement: "It is expected that TechNova Inc will launch its AI-driven product line in Q3 2025.",
Example 3 Output:{
  "triplets": [
    {
      "subject_name": "TechNova",
      "subject_id": 0,
      "predicate": "LAUNCHED",
      "object_name": "AI-driven Product",
      "object_id": 1,
      "value": "None,
    }
  ],
  "entities": [
    {
      "entity_idx": 0,
      "name": "TechNova",
      "type": "Organization",
      "description": "Technology Company",
    },
    {
      "entity_idx": 1,
      "name": "AI-driven Product",
      "type": "Product",
      "description": "General AI products",
    }
  ]
}

Example 4 Statement: "The SVP, CFO and Treasurer of AMD spoke during the earnings call."
Example 4 Output: {
  "triplets": [],
  "entities":[].
}

===End of Examples===
"""