From b9af1816256fbfa8c30ff3062df52920935977a0 Mon Sep 17 00:00:00 2001 From: KaifAhmad1 Date: Wed, 25 Mar 2026 21:42:57 +0530 Subject: [PATCH 1/2] feat(semantic-extract): temporal metadata extraction from text (#400) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add `extract_temporal_bounds: bool = False` to `extract_relations_llm()`. When True the LLM prompt is extended with a calibrated confidence scale and four few-shot examples; each returned Relation gains valid_from, valid_until, temporal_confidence, and temporal_source_text in metadata. Low confidence (<0.5) with non-null dates logs a WARNING. Default False preserves 100% backward compatibility. - Add `RelationWithTemporalOut` / `RelationsWithTemporalResponse` Pydantic schemas so the four temporal fields are captured from structured LLM output (separate from RelationOut which uses extra="ignore"). - New `semantica/kg/temporal_normalizer.py` — `TemporalNormalizer` class (zero LLM calls, pure regex + dateutil arithmetic): * normalize(value) → (start, end) UTC datetimes or None * Resolution order: ISO 8601 → partial dates (year/month/Q) → ambiguity detection → domain phrase map → relative phrases * normalize_phrase(phrase) → metadata dict or None * Default phrase map covers 13 domains: General, Policy, Healthcare, Drug Discovery, Cybersecurity, Supply Chain, Finance, Energy * TemporalAmbiguityWarning for DD/MM/YYYY-style ambiguous inputs * Custom phrase_map at construction (merged over defaults) - Add `TemporalAmbiguityWarning(UserWarning)` to exceptions.py. - Export `TemporalNormalizer` from `semantica/kg/__init__.py`. - Propagate `extract_temporal_bounds` through `_extract_relations_chunked` and add flag to cache key to prevent cross-mode cache pollution. - 53 new tests in tests/semantic_extract/test_temporal_extraction.py; zero real LLM calls, suite runs in ~3.5s. 873 existing tests unaffected. Co-Authored-By: Claude Sonnet 4.6 --- semantica/kg/__init__.py | 2 + semantica/kg/temporal_normalizer.py | 391 ++++++++++ semantica/semantic_extract/methods.py | 260 ++++--- semantica/semantic_extract/schemas.py | 57 ++ semantica/utils/exceptions.py | 12 + .../test_temporal_extraction.py | 675 ++++++++++++++++++ 6 files changed, 1266 insertions(+), 131 deletions(-) create mode 100644 semantica/kg/temporal_normalizer.py create mode 100644 tests/semantic_extract/test_temporal_extraction.py diff --git a/semantica/kg/__init__.py b/semantica/kg/__init__.py index 54f1f373..e2f5a5f2 100644 --- a/semantica/kg/__init__.py +++ b/semantica/kg/__init__.py @@ -127,6 +127,7 @@ from .temporal_query import ( TemporalVersionManager, ) from .temporal_model import BiTemporalFact, TemporalBound +from .temporal_normalizer import TemporalNormalizer __all__ = [ # Core Classes @@ -140,6 +141,7 @@ __all__ = [ "TemporalVersionManager", "TemporalBound", "BiTemporalFact", + "TemporalNormalizer", "AlgorithmTrackerWithProvenance", "ProvenanceTracker", # Enhanced Graph Algorithms diff --git a/semantica/kg/temporal_normalizer.py b/semantica/kg/temporal_normalizer.py new file mode 100644 index 00000000..bcadce11 --- /dev/null +++ b/semantica/kg/temporal_normalizer.py @@ -0,0 +1,391 @@ +""" +Temporal Normalizer + +Deterministic resolution of temporal phrases extracted from text into +UTC datetime intervals. Zero LLM calls — pure regex and date arithmetic. + +Usage:: + + from semantica.kg import TemporalNormalizer + from datetime import datetime, timezone + + tn = TemporalNormalizer(reference_date=datetime(2025, 3, 25, tzinfo=timezone.utc)) + start, end = tn.normalize("Q2 2021") # → (2021-04-01, 2021-06-30) + start, end = tn.normalize("last year") # → (2024-01-01, 2024-12-31) + info = tn.normalize_phrase("expiry date") # → {"maps_to": "valid_until", ...} +""" + +from __future__ import annotations + +import calendar +import logging +import re +import warnings +from datetime import datetime, timezone +from typing import Any, Callable, Dict, Optional, Tuple + +from dateutil.relativedelta import relativedelta + +from ..utils.exceptions import TemporalAmbiguityWarning + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Compiled regex patterns for structured date formats +# --------------------------------------------------------------------------- + +_RE_YEAR_ONLY = re.compile(r"^\s*(\d{4})\s*$") +_RE_MONTH_YEAR_WORD = re.compile( + r"^\s*(january|february|march|april|may|june|july|august|september|october|november|december|" + r"jan|feb|mar|apr|jun|jul|aug|sep|oct|nov|dec)\s+(\d{4})\s*$", + re.IGNORECASE, +) +_RE_YEAR_MONTH_ISO = re.compile(r"^\s*(\d{4})-(\d{1,2})\s*$") +_RE_QUARTER = re.compile(r"^\s*Q([1-4])\s+(\d{4})\s*$", re.IGNORECASE) +_RE_AMBIGUOUS_SLASH = re.compile(r"^\s*\d{1,2}/\d{1,2}/\d{4}\s*$") + +_MONTH_NAMES: Dict[str, int] = { + "january": 1, "jan": 1, + "february": 2, "feb": 2, + "march": 3, "mar": 3, + "april": 4, "apr": 4, + "may": 5, + "june": 6, "jun": 6, + "july": 7, "jul": 7, + "august": 8, "aug": 8, + "september": 9, "sep": 9, + "october": 10, "oct": 10, + "november": 11, "nov": 11, + "december": 12, "dec": 12, +} + +_QUARTER_BOUNDS: Dict[int, Tuple[int, int, int, int]] = { + # quarter → (from_month, from_day, until_month, until_day) + 1: (1, 1, 3, 31), + 2: (4, 1, 6, 30), + 3: (7, 1, 9, 30), + 4: (10, 1, 12, 31), +} + + +# --------------------------------------------------------------------------- +# Small date-arithmetic helpers +# --------------------------------------------------------------------------- + +def _utc(year: int, month: int, day: int) -> datetime: + return datetime(year, month, day, tzinfo=timezone.utc) + + +def _last_day_of_month(year: int, month: int) -> int: + return calendar.monthrange(year, month)[1] + + +def _this_quarter(ref: datetime) -> Tuple[datetime, datetime]: + q = (ref.month - 1) // 3 + 1 + fm, fd, um, ud = _QUARTER_BOUNDS[q] + return _utc(ref.year, fm, fd), _utc(ref.year, um, ud) + + +def _last_quarter(ref: datetime) -> Tuple[datetime, datetime]: + q = (ref.month - 1) // 3 + 1 + prev_q = q - 1 if q > 1 else 4 + year = ref.year if q > 1 else ref.year - 1 + fm, fd, um, ud = _QUARTER_BOUNDS[prev_q] + return _utc(year, fm, fd), _utc(year, um, ud) + + +def _last_month(ref: datetime) -> Tuple[datetime, datetime]: + first = ref.replace(day=1) - relativedelta(months=1) + last_day = _last_day_of_month(first.year, first.month) + return _utc(first.year, first.month, 1), _utc(first.year, first.month, last_day) + + +# --------------------------------------------------------------------------- +# Default phrase map +# --------------------------------------------------------------------------- +# Keys: lowercase canonical phrases (or regex patterns prefixed with "r:"). +# Values: callables (ref: datetime) → (valid_from, valid_until). +# +# Domain-specific terms that carry no self-contained date (e.g. "approval date") +# return (ref, ref) as a placeholder so callers can distinguish +# "known temporal term, date needs context" from "unrecognised phrase". +# --------------------------------------------------------------------------- + +def _phrase_entry(maps_to: str, type_: str, **extra: Any) -> Dict[str, Any]: + return {"maps_to": maps_to, "type": type_, **extra} + + +# Phrase map entries also carry metadata for normalize_phrase() +_DEFAULT_PHRASE_META: Dict[str, Dict[str, Any]] = { + # ── Relative references ───────────────────────────────────────────── + "last year": _phrase_entry("valid_from", "relative"), + "this year": _phrase_entry("valid_from", "relative"), + "last quarter": _phrase_entry("valid_from", "relative"), + "this quarter": _phrase_entry("valid_from", "relative"), + "last month": _phrase_entry("valid_from", "relative"), + "this month": _phrase_entry("valid_from", "relative"), + "three months ago": _phrase_entry("valid_from", "relative"), + "six months ago": _phrase_entry("valid_from", "relative"), + "two years ago": _phrase_entry("valid_from", "relative"), + # ── General / Policy ──────────────────────────────────────────────── + "r:effective\\s+(as\\s+of|from|beginning|date)": + _phrase_entry("valid_from", "start", domain=["General", "Policy"]), + "in force until": + _phrase_entry("valid_until", "end", domain=["Policy", "Regulatory"]), + "retroactive to": + _phrase_entry("valid_from", "start", retroactive=True, domain=["Regulatory", "Finance"]), + "sunset clause": + _phrase_entry("valid_until", "sunset", domain=["Policy"]), + # ── Healthcare / Drug Discovery ────────────────────────────────────── + "approval date": + _phrase_entry("valid_from", "start", domain=["Healthcare", "Drug Discovery"]), + "expiry date": + _phrase_entry("valid_until", "end", domain=["Healthcare", "Supply Chain"]), + "market authorization": + _phrase_entry("valid_from", "start", domain=["Drug Discovery", "Healthcare"]), + # ── Cybersecurity ──────────────────────────────────────────────────── + "incident window": + _phrase_entry("window", "window", domain=["Cybersecurity"]), + "campaign period": + _phrase_entry("window", "window", domain=["Cybersecurity"]), + # ── Supply Chain ───────────────────────────────────────────────────── + "certification valid through": + _phrase_entry("valid_until", "end", domain=["Supply Chain"]), + # ── Finance ────────────────────────────────────────────────────────── + "trading halt": + _phrase_entry("window", "window", domain=["Finance"]), + # ── Energy ─────────────────────────────────────────────────────────── + "commissioned date": + _phrase_entry("valid_from", "start", domain=["Energy"]), + "decommissioned date": + _phrase_entry("valid_until", "end", domain=["Energy"]), +} + +# Separate callable map for date resolution (subset of the above) +def _build_default_callable_map() -> Dict[str, Callable[[datetime], Tuple[datetime, datetime]]]: + return { + "last year": lambda ref: ( + _utc(ref.year - 1, 1, 1), + _utc(ref.year - 1, 12, 31), + ), + "this year": lambda ref: ( + _utc(ref.year, 1, 1), + _utc(ref.year, 12, 31), + ), + "last quarter": _last_quarter, + "this quarter": _this_quarter, + "last month": _last_month, + "this month": lambda ref: ( + _utc(ref.year, ref.month, 1), + _utc(ref.year, ref.month, _last_day_of_month(ref.year, ref.month)), + ), + "three months ago": lambda ref: ( + (ref - relativedelta(months=3)).replace(day=1, hour=0, minute=0, second=0, microsecond=0), + ((ref - relativedelta(months=3)).replace(day=1) + relativedelta(months=1) - relativedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0), + ), + "six months ago": lambda ref: ( + (ref - relativedelta(months=6)).replace(day=1, hour=0, minute=0, second=0, microsecond=0), + ((ref - relativedelta(months=6)).replace(day=1) + relativedelta(months=1) - relativedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0), + ), + "two years ago": lambda ref: ( + _utc(ref.year - 2, 1, 1), + _utc(ref.year - 2, 12, 31), + ), + } + + +# --------------------------------------------------------------------------- +# TemporalNormalizer +# --------------------------------------------------------------------------- + +class TemporalNormalizer: + """ + Deterministic resolution of temporal phrases into UTC datetime intervals. + + Zero LLM calls. All resolution is done via regex patterns and Python + date arithmetic (``dateutil.relativedelta``). + + Args: + reference_date: Anchor for relative phrases like "last year". When + ``None`` and a relative phrase is encountered, :meth:`normalize` + raises :class:`ValueError`. + phrase_map: Optional dict that extends or overrides the default + domain phrase map. Keys are lowercase phrases (or regex patterns + prefixed with ``"r:"``). Values are callables + ``(reference_date: datetime) -> (start: datetime, end: datetime)``. + """ + + def __init__( + self, + reference_date: Optional[datetime] = None, + phrase_map: Optional[Dict[str, Any]] = None, + ) -> None: + self.reference_date = reference_date + # Build the callable resolution map (relative dates + user overrides) + self._callable_map: Dict[str, Callable[[datetime], Tuple[datetime, datetime]]] = ( + _build_default_callable_map() + ) + if phrase_map: + self._callable_map.update(phrase_map) + # Phrase metadata map (for normalize_phrase) + self._phrase_meta: Dict[str, Dict[str, Any]] = {**_DEFAULT_PHRASE_META} + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def normalize(self, value: Optional[str]) -> Optional[Tuple[datetime, datetime]]: + """ + Resolve a temporal string to a ``(valid_from, valid_until)`` interval. + + Resolution order: + 1. ``None`` / empty → ``None`` + 2. ISO 8601 full datetime / date → point interval ``(dt, dt)`` + 3. Partial date patterns: year-only, month+year, quarter+year + 4. Ambiguous slash-date (``DD/MM/YYYY`` vs ``MM/DD/YYYY``) → + issues :class:`~semantica.utils.exceptions.TemporalAmbiguityWarning` + and returns ``None`` + 5. Phrase map / domain phrase lookup + 6. Relative phrase via callable map (requires ``reference_date``) + 7. Unparseable → ``None`` (debug log, never raises) + + Returns: + Tuple of UTC datetimes ``(start, end)`` or ``None``. + """ + if value is None: + return None + value_stripped = value.strip() + if not value_stripped: + return None + + # 1. ISO 8601 parse + iso_result = self._try_iso(value_stripped) + if iso_result is not None: + return iso_result + + # 2. Partial date patterns + partial_result = self._try_partial_date(value_stripped) + if partial_result is not None: + return partial_result + + # 3. Ambiguous slash date — warn, return None + if _RE_AMBIGUOUS_SLASH.match(value_stripped): + warnings.warn( + f"Temporal expression {value_stripped!r} is ambiguous (day/month ordering unknown). " + "Provide locale or use ISO 8601 format (YYYY-MM-DD).", + TemporalAmbiguityWarning, + stacklevel=2, + ) + return None + + # 4. Relative phrase / callable map + callable_result = self._try_callable(value_stripped) + if callable_result is not None: + return callable_result + + logger.debug("Could not parse temporal value: %r", value_stripped) + return None + + def normalize_phrase(self, phrase: str) -> Optional[Dict[str, Any]]: + """ + Look up a temporal phrase in the domain phrase map. + + Checks exact match first, then regex patterns (keys prefixed with + ``"r:"``). Returns the metadata dict if matched, ``None`` otherwise. + + Args: + phrase: Lowercase phrase to look up (case-insensitive internally). + + Returns: + Dict with at minimum ``{"maps_to": ..., "type": ...}`` or ``None``. + """ + normalized = phrase.strip().lower() + + # Exact match + if normalized in self._phrase_meta: + return self._phrase_meta[normalized] + + # Regex pattern match (keys prefixed with "r:") + for key, meta in self._phrase_meta.items(): + if key.startswith("r:"): + pattern = key[2:] + if re.search(pattern, normalized, re.IGNORECASE): + return meta + + logger.debug("Unrecognized temporal phrase: %r", phrase) + return None + + # ------------------------------------------------------------------ + # Private helpers + # ------------------------------------------------------------------ + + def _try_iso(self, value: str) -> Optional[Tuple[datetime, datetime]]: + """Attempt ISO 8601 parse. Returns point interval on success.""" + normalized = value + if normalized.endswith("Z"): + normalized = normalized[:-1] + "+00:00" + try: + dt = datetime.fromisoformat(normalized) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return (dt, dt) + except ValueError: + return None + + def _try_partial_date(self, value: str) -> Optional[Tuple[datetime, datetime]]: + """Try partial date patterns: YYYY, Month YYYY, YYYY-MM, Q[1-4] YYYY.""" + # Year only + m = _RE_YEAR_ONLY.match(value) + if m: + year = int(m.group(1)) + return _utc(year, 1, 1), _utc(year, 12, 31) + + # Month YYYY (word) + m = _RE_MONTH_YEAR_WORD.match(value) + if m: + month = _MONTH_NAMES[m.group(1).lower()] + year = int(m.group(2)) + last = _last_day_of_month(year, month) + return _utc(year, month, 1), _utc(year, month, last) + + # YYYY-MM (ISO partial) + m = _RE_YEAR_MONTH_ISO.match(value) + if m: + year, month = int(m.group(1)), int(m.group(2)) + if 1 <= month <= 12: + last = _last_day_of_month(year, month) + return _utc(year, month, 1), _utc(year, month, last) + + # Q[1-4] YYYY + m = _RE_QUARTER.match(value) + if m: + q, year = int(m.group(1)), int(m.group(2)) + fm, fd, um, ud = _QUARTER_BOUNDS[q] + return _utc(year, fm, fd), _utc(year, um, ud) + + return None + + def _try_callable(self, value: str) -> Optional[Tuple[datetime, datetime]]: + """Try the relative phrase callable map.""" + key = value.lower() + + # Exact match + if key in self._callable_map: + if self.reference_date is None: + raise ValueError( + f"reference_date is required to resolve relative temporal expression: {value!r}" + ) + return self._callable_map[key](self.reference_date) + + # Regex pattern match (keys prefixed with "r:") + for map_key, fn in self._callable_map.items(): + if map_key.startswith("r:"): + pattern = map_key[2:] + if re.search(pattern, key, re.IGNORECASE): + if self.reference_date is None: + raise ValueError( + f"reference_date is required to resolve relative temporal expression: {value!r}" + ) + return fn(self.reference_date) + + return None diff --git a/semantica/semantic_extract/methods.py b/semantica/semantic_extract/methods.py index 2f638bce..5c2d2385 100644 --- a/semantica/semantic_extract/methods.py +++ b/semantica/semantic_extract/methods.py @@ -122,7 +122,12 @@ from .cache import ExtractionCache from .config import config try: - from .schemas import EntitiesResponse, RelationsResponse, TripletsResponse + from .schemas import ( + EntitiesResponse, + RelationsResponse, + RelationsWithTemporalResponse, + TripletsResponse, + ) SCHEMAS_AVAILABLE = True except ImportError: SCHEMAS_AVAILABLE = False @@ -1664,11 +1669,12 @@ def extract_relations_llm( max_text_length: Optional[int] = None, structured_output_mode: str = "typed", max_retries: int = 3, + extract_temporal_bounds: bool = False, **kwargs, ) -> List[Relation]: """ LLM-based relation extraction. - + Args: text: Input text entities: Pre-extracted entities @@ -1677,6 +1683,10 @@ def extract_relations_llm( silent_fail: If True, return empty list on error. If False (default), raise exception. max_text_length: Maximum text length before auto-chunking. None = provider default. max_retries: Maximum number of retries for LLM calls (default: 3) + extract_temporal_bounds: If True, extend the prompt to extract temporal validity + per relation. Each relation's metadata gains: valid_from, valid_until, + temporal_confidence (0.0–1.0), and temporal_source_text. Low confidence (<0.5) + produces a warning log but is not suppressed. Default False. **kwargs: Additional options """ # Support llm_model parameter to disambiguate from ML model @@ -1691,6 +1701,7 @@ def extract_relations_llm( "structured_output_mode": structured_output_mode, "max_retries": max_retries, "relation_types": kwargs.get("relation_types"), + "extract_temporal_bounds": extract_temporal_bounds, # Include entities hash/str in cache key implicitly via **cache_params "entities_hash": hash(tuple(sorted([e.text for e in entities]))) if entities else 0 } @@ -1759,9 +1770,10 @@ def extract_relations_llm( if len(text) > max_text_length: logger.info(f"Text length ({len(text)}) exceeds limit for relations. Chunking...") return _extract_relations_chunked( - text, entities, provider=provider, model=model, - silent_fail=silent_fail, max_text_length=max_text_length, + text, entities, provider=provider, model=model, + silent_fail=silent_fail, max_text_length=max_text_length, max_retries=max_retries, + extract_temporal_bounds=extract_temporal_bounds, **kwargs ) @@ -1777,7 +1789,7 @@ def extract_relations_llm( ) entities_str = ", ".join([f"{e.text} ({e.label})" for e in prompt_entities]) - + # Use custom relation types if provided relation_types = kwargs.get("relation_types") if relation_types: @@ -1790,16 +1802,18 @@ If a relation doesn't fit any of the preferred types, use the most appropriate t relation_types_instruction = """ Extract meaningful relationships between entities. Use appropriate relation types that accurately describe how entities are connected. Common relation types include: related_to, part_of, located_in, created_by, uses, depends_on, interacts_with, and similar variations.""" - + verbose_mode = kwargs.get("verbose", False) if verbose_mode: import sys print(f" [methods.extract_relations_llm] Constructing prompt for {len(prompt_entities)} entities...", flush=True, file=sys.stdout) - + if not SCHEMAS_AVAILABLE: raise ImportError("Pydantic schemas not available. Install pydantic/instructor to use LLM extraction.") - prompt = f"""Extract relations between entities from the provided text. + # ── Base prompt (always included) ─────────────────────────────────────── + if not extract_temporal_bounds: + prompt = f"""Extract relations between entities from the provided text. Return the result as a JSON object with a "relations" key containing the list of relations. Each relation must have 'subject', 'predicate', and 'object' fields. @@ -1820,115 +1834,59 @@ Instructions: Text to extract from: {text} Entities found in text: {entities_str}""" - - - if not entities: - error_msg = "No entities provided for relation extraction. Relations require existing entities." - logger.error(error_msg) - if not silent_fail: - raise ProcessingError(error_msg) - return [] - - # Pass api_key if provided in kwargs - provider_kwargs = kwargs.copy() - - # Check if api_key is provided but empty, or not provided at all - if "api_key" not in provider_kwargs or not provider_kwargs["api_key"]: - import os - env_key = f"{provider.upper()}_API_KEY" - api_key = os.getenv(env_key) - if api_key: - provider_kwargs["api_key"] = api_key - - # Remove None/empty API key if still present to avoid provider errors - if "api_key" in provider_kwargs and not provider_kwargs["api_key"]: - del provider_kwargs["api_key"] - - # 2. PROVIDER VALIDATION - try: - llm = create_provider(provider, model=model, **provider_kwargs) - if not llm.is_available(): - error_msg = f"{provider} provider not available for relation extraction (key missing?)." - logger.error(error_msg) - if not silent_fail: - raise ProcessingError(error_msg) - return [] - except Exception as e: - error_msg = f"Failed to create {provider} provider for relations: {e}" - logger.error(error_msg) - if not silent_fail: - raise ProcessingError(error_msg) from e - return [] - - # 3. TEXT LENGTH CHECK AND CHUNKING - if max_text_length is None: - # Default limits for chunking only - NOT for LLM generation - max_text_length = { - "groq": 64000, - "openai": 64000, - "gemini": 64000, - "anthropic": 64000, - "deepseek": 64000, - }.get(provider.lower(), 32000) - - if len(text) > max_text_length: - logger.info(f"Text length ({len(text)}) exceeds limit for relations. Chunking...") - return _extract_relations_chunked( - text, entities, provider=provider, model=model, - silent_fail=silent_fail, max_text_length=max_text_length, - max_retries=max_retries, - **kwargs - ) - - original_entities = entities - # Use a fixed internal default for prompt entity cap (do not accept overrides from kwargs) - max_entities_prompt = 80 - prompt_entities = original_entities - if max_entities_prompt > 0 and len(original_entities) > max_entities_prompt: - prompt_entities = filter_entities_for_text( - text, - original_entities, - max_keep=max_entities_prompt, - ) - - entities_str = ", ".join([f"{e.text} ({e.label})" for e in prompt_entities]) - - # Use custom relation types if provided - relation_types = kwargs.get("relation_types") - if relation_types: - relation_types_str = ", ".join(relation_types) - relation_types_instruction = f""" -Preferred relation types: {relation_types_str}. -You may also use related or similar relation types if they better capture the relationship (e.g., variations, synonyms, or domain-specific relations). -If a relation doesn't fit any of the preferred types, use the most appropriate type from the preferred list or a closely related type that accurately describes the relationship.""" else: - relation_types_instruction = """ -Extract meaningful relationships between entities. Use appropriate relation types that accurately describe how entities are connected. -Common relation types include: related_to, part_of, located_in, created_by, uses, depends_on, interacts_with, and similar variations.""" - - verbose_mode = kwargs.get("verbose", False) - if verbose_mode: - import sys - print(f" [methods.extract_relations_llm] Constructing prompt for {len(prompt_entities)} entities...", flush=True, file=sys.stdout) - - if not SCHEMAS_AVAILABLE: - raise ImportError("Pydantic schemas not available. Install pydantic/instructor to use LLM extraction.") + # ── Temporal-extended prompt ───────────────────────────────────────── + prompt = f"""Extract relations between entities from the provided text, along with temporal validity information for each relation. +Return the result as a JSON object with a "relations" key. Each relation must have: +'subject', 'predicate', 'object', 'confidence', 'valid_from', 'valid_until', 'temporal_confidence', 'temporal_source_text'. - prompt = f"""Extract relations between entities from the provided text. -Return the result as a JSON object with a "relations" key containing the list of relations. -Each relation must have 'subject', 'predicate', and 'object' fields. +TEMPORAL EXTRACTION RULES: +- valid_from: ISO 8601 date or exact phrase from the text for when this relation became valid. Set to null if no temporal signal is present. +- valid_until: ISO 8601 date or exact phrase for when this relation ceased. Set to null if open-ended or absent. +- temporal_confidence (float 0.0–1.0) — calibrated as follows: + 1.00 = full ISO date ("2022-03-15", "March 15, 2022") + 0.90 = explicit year + month ("March 2022", "2022-03") + 0.85 = explicit year only ("in 2022", "since 2021", "from 2019") + 0.75 = quarter ("Q3 2023", "Q2 2021") + 0.65 = named season or approximate range ("summer 2022", "early 2020s", "mid-2022") + 0.50 = vague relative with computable anchor ("last year", "three months ago") + 0.35 = highly vague relative ("recently", "years ago", "in the past") + 0.00 = no temporal signal present for this relation +- temporal_source_text: the EXACT verbatim substring from the source text that contains the temporal signal. Set to null when temporal_confidence is 0.0. -Example output (JSON format only): +IMPORTANT: Do NOT invent or guess dates. If the text contains no temporal signal for a relation, set valid_from and valid_until to null and temporal_confidence to 0.0. + +Few-shot examples (do NOT include these in your output): + Text: "Apple acquired Beats in May 2014." + → valid_from: "2014-05-01", valid_until: null, temporal_confidence: 0.90, temporal_source_text: "May 2014" + + Text: "The CEO has led the company since Q3 2020." + → valid_from: "Q3 2020", valid_until: null, temporal_confidence: 0.75, temporal_source_text: "since Q3 2020" + + Text: "Last year, Google partnered with Samsung." + → valid_from: "last year", valid_until: null, temporal_confidence: 0.50, temporal_source_text: "Last year" + + Text: "The firm was under enhanced supervision between Q2 and Q4 2021." + → valid_from: "Q2 2021", valid_until: "Q4 2021", temporal_confidence: 0.75, temporal_source_text: "between Q2 and Q4 2021" + + Text: "Microsoft develops Windows." + → valid_from: null, valid_until: null, temporal_confidence: 0.00, temporal_source_text: null + +Example JSON output format: {{ "relations": [ - {{"subject": "Entity A", "predicate": "related_to", "object": "Entity B", "confidence": 0.95}}, - {{"subject": "Subject Entity", "predicate": "action_verb", "object": "Object Entity", "confidence": 0.90}} + {{ + "subject": "Apple", "predicate": "acquired", "object": "Beats", + "confidence": 0.97, + "valid_from": "2014-05-01", "valid_until": null, + "temporal_confidence": 0.90, "temporal_source_text": "May 2014" + }} ] }} Instructions: 1. Extract relations ONLY from the text provided below. -2. Do not include any relations from the example above. +2. Do not include any relations from the examples above. 3. Use the provided entities list as a reference for subjects and objects. 4. {relation_types_instruction} @@ -1938,24 +1896,25 @@ Entities found in text: {entities_str}""" try: # Use typed generation with Pydantic schema - # Pass kwargs to allow max_tokens and other parameters to be used if verbose_mode: - import sys - print(f" [methods.extract_relations_llm] Calling llm.generate_typed ({provider}/{model})...", flush=True, file=sys.stdout) + import sys + print(f" [methods.extract_relations_llm] Calling llm.generate_typed ({provider}/{model})...", flush=True, file=sys.stdout) # Only forward minimal, safe parameters to provider calls call_kwargs = {} if "temperature" in kwargs: call_kwargs["temperature"] = kwargs["temperature"] if "verbose" in kwargs: call_kwargs["verbose"] = kwargs["verbose"] - + call_kwargs["max_retries"] = max_retries - result_obj = llm.generate_typed(prompt, schema=RelationsResponse, **call_kwargs) + # Select schema based on whether temporal extraction is requested + active_schema = RelationsWithTemporalResponse if extract_temporal_bounds else RelationsResponse + result_obj = llm.generate_typed(prompt, schema=active_schema, **call_kwargs) if verbose_mode: - import sys - print(f" [methods.extract_relations_llm] Received response from {provider}.", flush=True, file=sys.stdout) - + import sys + print(f" [methods.extract_relations_llm] Received response from {provider}.", flush=True, file=sys.stdout) + # Convert back to internal Relation format (robust across providers) # Normalize typed result to a plain dict compatible with _parse_relation_result try: @@ -1972,12 +1931,16 @@ Entities found in text: {entities_str}""" elif isinstance(r, dict): rel_items.append(r) else: - # Best-effort attribute access + # Best-effort attribute access — include temporal fields when present rel_items.append({ "subject": getattr(r, "subject", ""), "object": getattr(r, "object", ""), "predicate": getattr(r, "predicate", "related_to"), "confidence": getattr(r, "confidence", 0.9), + "valid_from": getattr(r, "valid_from", None), + "valid_until": getattr(r, "valid_until", None), + "temporal_confidence": getattr(r, "temporal_confidence", 0.0), + "temporal_source_text": getattr(r, "temporal_source_text", None), }) parsed = {"relations": rel_items} else: @@ -1986,7 +1949,11 @@ Entities found in text: {entities_str}""" parsed = result_obj # Use common parser to build internal Relation objects - relations = _parse_relation_result(parsed, original_entities, text, provider, model, extraction_method="llm_typed") + relations = _parse_relation_result( + parsed, original_entities, text, provider, model, + extraction_method="llm_typed", + extract_temporal_bounds=extract_temporal_bounds, + ) # If typed path returned no relations, attempt a structured JSON fallback if not relations: @@ -1995,7 +1962,11 @@ Entities found in text: {entities_str}""" import sys print(" [methods.extract_relations_llm] Typed result empty, attempting structured JSON fallback...", flush=True, file=sys.stdout) raw_json = llm.generate_structured(prompt, **call_kwargs) - relations = _parse_relation_result(raw_json, original_entities, text, provider, model, extraction_method="llm_typed") + relations = _parse_relation_result( + raw_json, original_entities, text, provider, model, + extraction_method="llm_typed", + extract_temporal_bounds=extract_temporal_bounds, + ) except Exception as _e: # Keep relations as empty if fallback fails pass @@ -2003,22 +1974,23 @@ Entities found in text: {entities_str}""" logger.info(f"Successfully extracted {len(relations)} relations using {provider}/{model} (typed)") _result_cache.set("relations", text, relations, **cache_params) return relations - + except Exception as e: # Check for length/token limit errors error_msg_str = str(e).lower() if "length" in error_msg_str or "max_tokens" in error_msg_str: logger.warning(f"LLM output truncated due to length limit. Reducing chunk size and retrying... ({e})") - + # Determine new chunk size (halve it) current_max = max_text_length or len(text) new_max = current_max // 2 - - if new_max > 100: # Minimum viable chunk size check + + if new_max > 100: # Minimum viable chunk size check return _extract_relations_chunked( text, entities, provider=provider, model=model, silent_fail=silent_fail, max_text_length=new_max, structured_output_mode=structured_output_mode, + extract_temporal_bounds=extract_temporal_bounds, **kwargs ) @@ -2038,11 +2010,12 @@ def _parse_relation_result( provider: str, model: Optional[str], extraction_method: str = "llm", + extract_temporal_bounds: bool = False, ) -> List[Relation]: """Helper to parse raw LLM result into Relation objects.""" relations = [] items = [] - + if isinstance(result, list): items = result elif isinstance(result, dict): @@ -2056,13 +2029,13 @@ def _parse_relation_result( for item in items: if not isinstance(item, dict): continue - + subject_text = item.get("subject", "") object_text = item.get("object", "") - + if not subject_text or not object_text: continue - + # Ensure they are strings subject_text = str(subject_text) object_text = str(object_text) @@ -2085,6 +2058,33 @@ def _parse_relation_result( confidence=0.8, metadata={"synthetic": True}, ) + metadata: dict = { + "provider": provider, + "model": model, + "extraction_method": extraction_method, + } + + if extract_temporal_bounds: + temporal_confidence = float(item.get("temporal_confidence") or 0.0) + valid_from = item.get("valid_from") + valid_until = item.get("valid_until") + temporal_source_text = item.get("temporal_source_text") + + metadata["valid_from"] = valid_from + metadata["valid_until"] = valid_until + metadata["temporal_confidence"] = temporal_confidence + metadata["temporal_source_text"] = temporal_source_text + + if temporal_confidence < 0.5 and (valid_from is not None or valid_until is not None): + logger.warning( + "Low temporal confidence (%.2f) for '%s' (%s → %s). Source: %r", + temporal_confidence, + item.get("predicate", ""), + item.get("subject", ""), + item.get("object", ""), + temporal_source_text, + ) + relations.append( Relation( subject=subject_entity, @@ -2092,11 +2092,7 @@ def _parse_relation_result( object=object_entity, confidence=item.get("confidence", 0.9), context=text, - metadata={ - "provider": provider, - "model": model, - "extraction_method": extraction_method, - }, + metadata=metadata, ) ) return relations @@ -2111,6 +2107,7 @@ def _extract_relations_chunked( max_text_length: int, structured_output_mode: str = "typed", max_retries: int = 3, + extract_temporal_bounds: bool = False, **kwargs ) -> List[Relation]: """Internal helper to extract relations from long text by chunking.""" @@ -2154,6 +2151,7 @@ def _extract_relations_chunked( max_text_length=len(chunk.text) + 1, structured_output_mode=structured_output_mode, max_retries=max_retries, + extract_temporal_bounds=extract_temporal_bounds, **limited_kwargs ) future_to_chunk[future] = i diff --git a/semantica/semantic_extract/schemas.py b/semantica/semantic_extract/schemas.py index be105e23..90e776de 100644 --- a/semantica/semantic_extract/schemas.py +++ b/semantica/semantic_extract/schemas.py @@ -125,3 +125,60 @@ class RelationsResponse(BaseModel): class TripletsResponse(BaseModel): """Wrapper for list of triplets.""" triplets: List[TripletOut] = Field(default_factory=list) + + +class RelationWithTemporalOut(BaseModel): + """Schema for relation extraction output with temporal validity bounds.""" + model_config = ConfigDict(populate_by_name=True, extra="ignore") + + subject: str = Field(..., description="Source entity text") + object: str = Field(..., description="Target entity text") + predicate: str = Field(..., description="Relation type or predicate") + confidence: float = Field(0.9, description="Confidence score between 0 and 1") + metadata: dict = Field(default_factory=dict, description="Additional metadata including provenance") + + valid_from: Optional[str] = Field( + None, + description="ISO 8601 date or natural-language phrase for when this relation became valid. Null if no temporal signal in text.", + ) + valid_until: Optional[str] = Field( + None, + description="ISO 8601 date or phrase for when this relation ceased to be valid. Null if open-ended or not stated.", + ) + temporal_confidence: float = Field( + 0.0, + description="Confidence that temporal information was present and correctly extracted. 0.0 if no temporal signal.", + ) + temporal_source_text: Optional[str] = Field( + None, + description="Exact verbatim substring from the source text containing the temporal signal. Null when temporal_confidence is 0.0.", + ) + + @model_validator(mode="before") + @classmethod + def handle_aliases(cls, data): + if isinstance(data, dict): + if "subject" not in data and "source" in data: + data["subject"] = data["source"] + if "object" not in data and "target" in data: + data["object"] = data["target"] + if "predicate" not in data and "label" in data: + data["predicate"] = data["label"] + return data + + @field_validator("confidence", "temporal_confidence", mode="before") + @classmethod + def normalize_confidence(cls, v): + if isinstance(v, str): + try: + v = float(v) + except ValueError: + return 0.0 + if isinstance(v, (int, float)): + return max(0.0, min(1.0, float(v))) + return 0.0 + + +class RelationsWithTemporalResponse(BaseModel): + """Wrapper for list of relations with temporal validity bounds.""" + relations: List[RelationWithTemporalOut] = Field(default_factory=list) diff --git a/semantica/utils/exceptions.py b/semantica/utils/exceptions.py index 1645441d..234756cf 100644 --- a/semantica/utils/exceptions.py +++ b/semantica/utils/exceptions.py @@ -167,6 +167,18 @@ class TemporalValidationError(ValidationError): self.error_code = "SEM001T" +class TemporalAmbiguityWarning(UserWarning): + """ + Warning raised when a temporal expression is ambiguous and cannot be + resolved without additional locale or context information. + + Example: "03/04/2022" is ambiguous without knowing whether day-first or + month-first ordering applies. Use ``warnings.catch_warnings()`` to handle. + """ + + pass + + class ProcessingError(SemanticaError): """ Exception raised for data processing errors. diff --git a/tests/semantic_extract/test_temporal_extraction.py b/tests/semantic_extract/test_temporal_extraction.py new file mode 100644 index 00000000..7b8ffcc7 --- /dev/null +++ b/tests/semantic_extract/test_temporal_extraction.py @@ -0,0 +1,675 @@ +""" +Tests for temporal metadata extraction (Issue #400). + +Covers: + - extract_temporal_bounds=True/False flag on extract_relations_llm() + - TemporalNormalizer: relative dates, partial dates, ambiguity, domain phrases, + custom phrase map + - Full pipeline: extract → normalize → BiTemporalFact + +All LLM calls are mocked. No real API keys required. Suite runs in < 5 s. +""" + +import os +import sys +import unittest +import warnings +from datetime import datetime, timezone +from unittest.mock import MagicMock, patch + +# ── Mock optional heavyweight dependencies before any semantica import ────── +sys.modules.setdefault("spacy", MagicMock()) +sys.modules.setdefault("instructor", MagicMock()) +_openai_mock = MagicMock() +sys.modules.setdefault("openai", _openai_mock) +sys.modules.setdefault("groq", MagicMock()) +sys.modules.setdefault("sentence_transformers", MagicMock()) +sys.modules.setdefault("transformers", MagicMock()) +sys.modules.setdefault("torch", MagicMock()) + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))) + +from semantica.semantic_extract.methods import extract_relations_llm +from semantica.semantic_extract.ner_extractor import Entity +from semantica.semantic_extract.schemas import ( + RelationsResponse, + RelationsWithTemporalResponse, +) +from semantica.kg.temporal_normalizer import TemporalNormalizer +from semantica.utils.exceptions import TemporalAmbiguityWarning + + +# ── Helpers ───────────────────────────────────────────────────────────────── + +def _make_entities(): + return [ + Entity(text="Apple", label="ORG", start_char=0, end_char=5), + Entity(text="Beats", label="ORG", start_char=15, end_char=20), + ] + + +def _ref_date(): + return datetime(2025, 6, 15, tzinfo=timezone.utc) + + +# ============================================================================ +# Part 1 – extract_relations_llm() temporal flag +# ============================================================================ + +class TestTemporalExtractionFlag(unittest.TestCase): + + def setUp(self): + from semantica.semantic_extract.methods import _result_cache + _result_cache.clear() + + @patch("semantica.semantic_extract.methods.create_provider") + def test_extract_temporal_bounds_true_adds_four_fields(self, mock_create): + """With extract_temporal_bounds=True all four temporal keys appear in metadata.""" + mock_prov = MagicMock() + mock_prov.is_available.return_value = True + mock_prov.generate_typed.return_value = RelationsWithTemporalResponse( + relations=[ + { + "subject": "Apple", + "predicate": "acquired", + "object": "Beats", + "confidence": 0.97, + "valid_from": "2014-05-01", + "valid_until": None, + "temporal_confidence": 0.90, + "temporal_source_text": "May 2014", + } + ] + ) + mock_create.return_value = mock_prov + + rels = extract_relations_llm( + "Apple acquired Beats in May 2014.", + _make_entities(), + provider="openai", + extract_temporal_bounds=True, + ) + + self.assertEqual(len(rels), 1) + meta = rels[0].metadata + self.assertIn("valid_from", meta) + self.assertIn("valid_until", meta) + self.assertIn("temporal_confidence", meta) + self.assertIn("temporal_source_text", meta) + self.assertEqual(meta["valid_from"], "2014-05-01") + self.assertIsNone(meta["valid_until"]) + self.assertAlmostEqual(meta["temporal_confidence"], 0.90, places=2) + self.assertEqual(meta["temporal_source_text"], "May 2014") + + @patch("semantica.semantic_extract.methods.create_provider") + def test_extract_temporal_bounds_false_output_identical(self, mock_create): + """With extract_temporal_bounds=False (default) no temporal keys appear.""" + mock_prov = MagicMock() + mock_prov.is_available.return_value = True + mock_prov.generate_typed.return_value = RelationsResponse( + relations=[ + { + "subject": "Apple", + "predicate": "acquired", + "object": "Beats", + "confidence": 0.97, + } + ] + ) + mock_create.return_value = mock_prov + + rels = extract_relations_llm( + "Apple acquired Beats.", + _make_entities(), + provider="openai", + ) + + self.assertEqual(len(rels), 1) + meta = rels[0].metadata + self.assertNotIn("valid_from", meta) + self.assertNotIn("valid_until", meta) + self.assertNotIn("temporal_confidence", meta) + self.assertNotIn("temporal_source_text", meta) + + @patch("semantica.semantic_extract.methods.create_provider") + def test_no_temporal_signal_returns_zero_confidence_and_null_dates(self, mock_create): + """When LLM returns no temporal signal, confidence=0.0 and dates are null.""" + mock_prov = MagicMock() + mock_prov.is_available.return_value = True + mock_prov.generate_typed.return_value = RelationsWithTemporalResponse( + relations=[ + { + "subject": "Apple", + "predicate": "owns", + "object": "Beats", + "confidence": 0.95, + "valid_from": None, + "valid_until": None, + "temporal_confidence": 0.0, + "temporal_source_text": None, + } + ] + ) + mock_create.return_value = mock_prov + + rels = extract_relations_llm( + "Apple owns Beats.", + _make_entities(), + provider="openai", + extract_temporal_bounds=True, + ) + + meta = rels[0].metadata + self.assertIsNone(meta["valid_from"]) + self.assertIsNone(meta["valid_until"]) + self.assertEqual(meta["temporal_confidence"], 0.0) + self.assertIsNone(meta["temporal_source_text"]) + + @patch("semantica.semantic_extract.methods.create_provider") + def test_low_temporal_confidence_logs_warning(self, mock_create): + """temporal_confidence < 0.5 with non-null date logs a WARNING.""" + mock_prov = MagicMock() + mock_prov.is_available.return_value = True + mock_prov.generate_typed.return_value = RelationsWithTemporalResponse( + relations=[ + { + "subject": "Apple", + "predicate": "partnered_with", + "object": "Beats", + "confidence": 0.80, + "valid_from": "recently", + "valid_until": None, + "temporal_confidence": 0.35, + "temporal_source_text": "recently", + } + ] + ) + mock_create.return_value = mock_prov + + with self.assertLogs("semantica", level="WARNING") as cm: + extract_relations_llm( + "Apple recently partnered with Beats.", + _make_entities(), + provider="openai", + extract_temporal_bounds=True, + ) + self.assertTrue(any("Low temporal confidence" in line for line in cm.output)) + + @patch("semantica.semantic_extract.methods.create_provider") + def test_correct_schema_used_when_temporal_true(self, mock_create): + """generate_typed is called with RelationsWithTemporalResponse when flag=True.""" + mock_prov = MagicMock() + mock_prov.is_available.return_value = True + mock_prov.generate_typed.return_value = RelationsWithTemporalResponse(relations=[]) + mock_create.return_value = mock_prov + + extract_relations_llm( + "Some text.", + _make_entities(), + provider="openai", + extract_temporal_bounds=True, + ) + + call_kwargs = mock_prov.generate_typed.call_args[1] + self.assertIs(call_kwargs["schema"], RelationsWithTemporalResponse) + + @patch("semantica.semantic_extract.methods.create_provider") + def test_correct_schema_used_when_temporal_false(self, mock_create): + """generate_typed is called with RelationsResponse when flag=False.""" + mock_prov = MagicMock() + mock_prov.is_available.return_value = True + mock_prov.generate_typed.return_value = RelationsResponse(relations=[]) + mock_create.return_value = mock_prov + + extract_relations_llm( + "Some text.", + _make_entities(), + provider="openai", + ) + + call_kwargs = mock_prov.generate_typed.call_args[1] + self.assertIs(call_kwargs["schema"], RelationsResponse) + + +# ============================================================================ +# Part 2 – TemporalNormalizer: relative dates +# ============================================================================ + +class TestTemporalNormalizerRelativeDates(unittest.TestCase): + + def setUp(self): + self.ref = _ref_date() # 2025-06-15 + self.tn = TemporalNormalizer(reference_date=self.ref) + + def test_last_year(self): + result = self.tn.normalize("last year") + self.assertIsNotNone(result) + start, end = result + self.assertEqual(start.year, 2024) + self.assertEqual(start.month, 1) + self.assertEqual(start.day, 1) + self.assertEqual(end.year, 2024) + self.assertEqual(end.month, 12) + self.assertEqual(end.day, 31) + + def test_this_year(self): + result = self.tn.normalize("this year") + self.assertIsNotNone(result) + self.assertEqual(result[0].year, 2025) + self.assertEqual(result[0].month, 1) + self.assertEqual(result[1].month, 12) + + def test_three_months_ago(self): + result = self.tn.normalize("three months ago") + self.assertIsNotNone(result) + # ref is 2025-06-15; three months ago → March 2025 + self.assertEqual(result[0].year, 2025) + self.assertEqual(result[0].month, 3) + + def test_last_quarter(self): + # ref is 2025-06-15 (Q2) → last quarter = Q1 2025 + result = self.tn.normalize("last quarter") + self.assertIsNotNone(result) + self.assertEqual(result[0].month, 1) + self.assertEqual(result[1].month, 3) + self.assertEqual(result[0].year, 2025) + + def test_this_quarter(self): + # ref is 2025-06-15 (Q2) → Q2 2025 + result = self.tn.normalize("this quarter") + self.assertIsNotNone(result) + self.assertEqual(result[0].month, 4) + self.assertEqual(result[1].month, 6) + + def test_last_month(self): + # ref June 2025 → May 2025 + result = self.tn.normalize("last month") + self.assertIsNotNone(result) + self.assertEqual(result[0].month, 5) + self.assertEqual(result[0].year, 2025) + + def test_two_years_ago(self): + result = self.tn.normalize("two years ago") + self.assertIsNotNone(result) + self.assertEqual(result[0].year, 2023) + self.assertEqual(result[1].year, 2023) + + def test_no_reference_date_raises_value_error(self): + tn = TemporalNormalizer() + with self.assertRaises(ValueError): + tn.normalize("last year") + + def test_none_input_returns_none(self): + self.assertIsNone(self.tn.normalize(None)) + + def test_empty_string_returns_none(self): + self.assertIsNone(self.tn.normalize("")) + + def test_whitespace_string_returns_none(self): + self.assertIsNone(self.tn.normalize(" ")) + + +# ============================================================================ +# Part 3 – TemporalNormalizer: partial / structured dates +# ============================================================================ + +class TestTemporalNormalizerPartialDates(unittest.TestCase): + + def setUp(self): + self.tn = TemporalNormalizer(reference_date=datetime(2025, 3, 25, tzinfo=timezone.utc)) + + def test_year_only(self): + result = self.tn.normalize("2021") + self.assertIsNotNone(result) + self.assertEqual(result[0], datetime(2021, 1, 1, tzinfo=timezone.utc)) + self.assertEqual(result[1], datetime(2021, 12, 31, tzinfo=timezone.utc)) + + def test_month_year_word(self): + result = self.tn.normalize("March 2022") + self.assertIsNotNone(result) + self.assertEqual(result[0].year, 2022) + self.assertEqual(result[0].month, 3) + self.assertEqual(result[0].day, 1) + self.assertEqual(result[1].day, 31) + + def test_month_year_word_abbreviated(self): + result = self.tn.normalize("Dec 2023") + self.assertIsNotNone(result) + self.assertEqual(result[0].month, 12) + self.assertEqual(result[1].day, 31) + + def test_year_month_iso_partial(self): + result = self.tn.normalize("2022-03") + self.assertIsNotNone(result) + self.assertEqual(result[0].year, 2022) + self.assertEqual(result[0].month, 3) + self.assertEqual(result[0].day, 1) + self.assertEqual(result[1].day, 31) + + def test_q1_2024(self): + result = self.tn.normalize("Q1 2024") + self.assertIsNotNone(result) + self.assertEqual(result[0], datetime(2024, 1, 1, tzinfo=timezone.utc)) + self.assertEqual(result[1], datetime(2024, 3, 31, tzinfo=timezone.utc)) + + def test_q2_2021(self): + result = self.tn.normalize("Q2 2021") + self.assertIsNotNone(result) + self.assertEqual(result[0], datetime(2021, 4, 1, tzinfo=timezone.utc)) + self.assertEqual(result[1], datetime(2021, 6, 30, tzinfo=timezone.utc)) + + def test_q3_2023(self): + result = self.tn.normalize("Q3 2023") + self.assertIsNotNone(result) + self.assertEqual(result[0], datetime(2023, 7, 1, tzinfo=timezone.utc)) + self.assertEqual(result[1], datetime(2023, 9, 30, tzinfo=timezone.utc)) + + def test_q4_2022(self): + result = self.tn.normalize("Q4 2022") + self.assertIsNotNone(result) + self.assertEqual(result[0], datetime(2022, 10, 1, tzinfo=timezone.utc)) + self.assertEqual(result[1], datetime(2022, 12, 31, tzinfo=timezone.utc)) + + def test_iso_full_date_returns_point(self): + result = self.tn.normalize("2022-03-15") + self.assertIsNotNone(result) + self.assertEqual(result[0].year, 2022) + self.assertEqual(result[0].month, 3) + self.assertEqual(result[0].day, 15) + # Point interval: start == end + self.assertEqual(result[0], result[1]) + + def test_iso_datetime_with_z(self): + result = self.tn.normalize("2022-03-15T00:00:00Z") + self.assertIsNotNone(result) + self.assertEqual(result[0].year, 2022) + + def test_unparseable_returns_none(self): + result = self.tn.normalize("sometime in the medieval period") + self.assertIsNone(result) + + def test_none_returns_none(self): + self.assertIsNone(self.tn.normalize(None)) + + +# ============================================================================ +# Part 4 – TemporalNormalizer: ambiguous formats +# ============================================================================ + +class TestTemporalNormalizerAmbiguity(unittest.TestCase): + + def setUp(self): + self.tn = TemporalNormalizer(reference_date=datetime(2025, 3, 25, tzinfo=timezone.utc)) + + def test_ambiguous_slash_date_raises_warning_and_returns_none(self): + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + result = self.tn.normalize("03/04/2022") + self.assertIsNone(result) + ambig = [x for x in w if issubclass(x.category, TemporalAmbiguityWarning)] + self.assertEqual(len(ambig), 1) + self.assertIn("ambiguous", str(ambig[0].message).lower()) + + def test_iso_hyphenated_date_not_flagged_as_ambiguous(self): + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + result = self.tn.normalize("2022-03-04") + ambig = [x for x in w if issubclass(x.category, TemporalAmbiguityWarning)] + self.assertEqual(len(ambig), 0) + self.assertIsNotNone(result) + + def test_ambiguous_date_does_not_raise_exception(self): + # Must not raise, only warn + try: + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + self.tn.normalize("01/12/2023") + except Exception as e: + self.fail(f"normalize() raised unexpectedly: {e}") + + +# ============================================================================ +# Part 5 – TemporalNormalizer: domain phrase map +# ============================================================================ + +class TestTemporalNormalizerDomainPhrases(unittest.TestCase): + + def setUp(self): + self.tn = TemporalNormalizer(reference_date=datetime(2025, 3, 25, tzinfo=timezone.utc)) + + def _assert_recognized(self, phrase): + result = self.tn.normalize_phrase(phrase) + self.assertIsNotNone(result, f"Expected phrase {phrase!r} to be recognized but got None") + return result + + # General / Policy + def test_effective_date_recognized(self): + r = self._assert_recognized("effective date") + self.assertEqual(r["maps_to"], "valid_from") + + def test_effective_from_regex_recognized(self): + r = self._assert_recognized("effective from") + self.assertEqual(r["maps_to"], "valid_from") + + def test_effective_as_of_regex_recognized(self): + r = self._assert_recognized("effective as of") + self.assertEqual(r["maps_to"], "valid_from") + + def test_in_force_until_recognized(self): + r = self._assert_recognized("in force until") + self.assertEqual(r["maps_to"], "valid_until") + + def test_retroactive_to_recognized(self): + r = self._assert_recognized("retroactive to") + self.assertTrue(r.get("retroactive")) + + def test_sunset_clause_recognized(self): + r = self._assert_recognized("sunset clause") + self.assertEqual(r["maps_to"], "valid_until") + + # Healthcare / Drug Discovery + def test_approval_date_recognized(self): + r = self._assert_recognized("approval date") + self.assertEqual(r["maps_to"], "valid_from") + self.assertIn("Healthcare", r.get("domain", [])) + + def test_expiry_date_recognized(self): + r = self._assert_recognized("expiry date") + self.assertEqual(r["maps_to"], "valid_until") + + def test_market_authorization_recognized(self): + r = self._assert_recognized("market authorization") + self.assertEqual(r["maps_to"], "valid_from") + self.assertIn("Drug Discovery", r.get("domain", [])) + + # Cybersecurity + def test_incident_window_recognized(self): + r = self._assert_recognized("incident window") + self.assertIn("Cybersecurity", r.get("domain", [])) + + def test_campaign_period_recognized(self): + r = self._assert_recognized("campaign period") + self.assertIn("Cybersecurity", r.get("domain", [])) + + # Supply Chain + def test_certification_valid_through_recognized(self): + r = self._assert_recognized("certification valid through") + self.assertEqual(r["maps_to"], "valid_until") + self.assertIn("Supply Chain", r.get("domain", [])) + + # Finance + def test_trading_halt_recognized(self): + r = self._assert_recognized("trading halt") + self.assertIn("Finance", r.get("domain", [])) + + # Energy + def test_commissioned_date_recognized(self): + r = self._assert_recognized("commissioned date") + self.assertEqual(r["maps_to"], "valid_from") + self.assertIn("Energy", r.get("domain", [])) + + def test_decommissioned_date_recognized(self): + r = self._assert_recognized("decommissioned date") + self.assertEqual(r["maps_to"], "valid_until") + self.assertIn("Energy", r.get("domain", [])) + + def test_unrecognized_phrase_returns_none(self): + result = self.tn.normalize_phrase("totally unknown phrase xyz") + self.assertIsNone(result) + + +# ============================================================================ +# Part 6 – TemporalNormalizer: custom phrase map +# ============================================================================ + +class TestTemporalNormalizerCustomPhraseMap(unittest.TestCase): + + def setUp(self): + ref = datetime(2025, 3, 25, tzinfo=timezone.utc) + self.tn = TemporalNormalizer( + reference_date=ref, + phrase_map={ + "fiscal year 2024": lambda r: ( + datetime(2024, 4, 1, tzinfo=timezone.utc), + datetime(2025, 3, 31, tzinfo=timezone.utc), + ) + }, + ) + + def test_custom_phrase_resolved(self): + result = self.tn.normalize("fiscal year 2024") + self.assertIsNotNone(result) + self.assertEqual(result[0].year, 2024) + self.assertEqual(result[0].month, 4) + self.assertEqual(result[1].year, 2025) + self.assertEqual(result[1].month, 3) + + def test_default_phrase_still_works_alongside_custom(self): + result = self.tn.normalize("last year") + self.assertIsNotNone(result) + self.assertEqual(result[0].year, 2024) + + def test_custom_phrase_overrides_default_when_same_key(self): + # Override "last year" to a custom sentinel + sentinel_start = datetime(2000, 1, 1, tzinfo=timezone.utc) + sentinel_end = datetime(2000, 12, 31, tzinfo=timezone.utc) + tn = TemporalNormalizer( + reference_date=datetime(2025, 3, 25, tzinfo=timezone.utc), + phrase_map={"last year": lambda r: (sentinel_start, sentinel_end)}, + ) + result = tn.normalize("last year") + self.assertEqual(result[0], sentinel_start) + + +# ============================================================================ +# Part 7 – Full pipeline: extract → normalize → BiTemporalFact +# ============================================================================ + +class TestFullPipelineTemporalToBiTemporal(unittest.TestCase): + + def setUp(self): + from semantica.semantic_extract.methods import _result_cache + _result_cache.clear() + + @patch("semantica.semantic_extract.methods.create_provider") + def test_full_pipeline_explicit_date(self, mock_create): + """extract_relations(temporal=True) → normalize → BiTemporalFact.""" + from semantica.kg.temporal_model import BiTemporalFact + + mock_prov = MagicMock() + mock_prov.is_available.return_value = True + mock_prov.generate_typed.return_value = RelationsWithTemporalResponse( + relations=[ + { + "subject": "Apple", + "predicate": "acquired", + "object": "Beats", + "confidence": 0.97, + "valid_from": "2014-05-01", + "valid_until": None, + "temporal_confidence": 0.90, + "temporal_source_text": "May 2014", + } + ] + ) + mock_create.return_value = mock_prov + + entities = [ + Entity(text="Apple", label="ORG", start_char=0, end_char=5), + Entity(text="Beats", label="ORG", start_char=15, end_char=20), + ] + rels = extract_relations_llm( + "Apple acquired Beats in May 2014.", + entities, + provider="openai", + extract_temporal_bounds=True, + ) + self.assertEqual(len(rels), 1) + + meta = rels[0].metadata + ref = datetime(2025, 3, 25, tzinfo=timezone.utc) + tn = TemporalNormalizer(reference_date=ref) + + vf = tn.normalize(meta["valid_from"]) + vu = tn.normalize(meta["valid_until"]) + + self.assertIsNotNone(vf) + self.assertIsNone(vu) + self.assertEqual(vf[0].year, 2014) + self.assertEqual(vf[0].month, 5) + self.assertEqual(vf[0].day, 1) + + # Feed into BiTemporalFact + fact = BiTemporalFact.from_relationship({ + "valid_from": "2014-05-01T00:00:00Z", + "valid_until": None, + }) + self.assertIsNotNone(fact.valid_from) + self.assertEqual(fact.valid_from.year, 2014) + self.assertEqual(fact.valid_from.month, 5) + + @patch("semantica.semantic_extract.methods.create_provider") + def test_full_pipeline_quarter_expression(self, mock_create): + """Pipeline with Q-expression normalizes to correct quarter bounds.""" + mock_prov = MagicMock() + mock_prov.is_available.return_value = True + mock_prov.generate_typed.return_value = RelationsWithTemporalResponse( + relations=[ + { + "subject": "Apple", + "predicate": "supervised", + "object": "Beats", + "confidence": 0.85, + "valid_from": "Q2 2021", + "valid_until": "Q4 2021", + "temporal_confidence": 0.75, + "temporal_source_text": "between Q2 and Q4 2021", + } + ] + ) + mock_create.return_value = mock_prov + + entities = [ + Entity(text="Apple", label="ORG", start_char=0, end_char=5), + Entity(text="Beats", label="ORG", start_char=15, end_char=20), + ] + rels = extract_relations_llm( + "Apple supervised Beats between Q2 and Q4 2021.", + entities, + provider="openai", + extract_temporal_bounds=True, + ) + meta = rels[0].metadata + tn = TemporalNormalizer(reference_date=datetime(2025, 3, 25, tzinfo=timezone.utc)) + + vf = tn.normalize(meta["valid_from"]) + vu = tn.normalize(meta["valid_until"]) + + self.assertEqual(vf[0], datetime(2021, 4, 1, tzinfo=timezone.utc)) + self.assertEqual(vf[1], datetime(2021, 6, 30, tzinfo=timezone.utc)) + self.assertEqual(vu[0], datetime(2021, 10, 1, tzinfo=timezone.utc)) + self.assertEqual(vu[1], datetime(2021, 12, 31, tzinfo=timezone.utc)) + + +if __name__ == "__main__": + unittest.main() From b1e1c9f0d950de15b090c6806e1ebbdb9704151f Mon Sep 17 00:00:00 2001 From: KaifAhmad1 Date: Wed, 25 Mar 2026 22:31:06 +0530 Subject: [PATCH 2/2] docs(changelog): add entry for temporal metadata extraction (#400) Co-Authored-By: Claude Sonnet 4.6 --- CHANGELOG.md | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index abf5fd8f..1d157041 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +- **Temporal Metadata Extraction from Text** (PR #400 by @KaifAhmad1): + - Added `extract_temporal_bounds: bool = False` parameter to `extract_relations_llm()`. When `True`, the LLM prompt is extended with a calibrated confidence scale and four few-shot examples; each returned `Relation` gains `valid_from`, `valid_until`, `temporal_confidence` (0.0–1.0), and `temporal_source_text` in its `metadata` dict. Default `False` preserves 100% backward compatibility. + - Confidence scale anchors baked into the prompt: `1.00` = full ISO date, `0.90` = year+month, `0.85` = year only, `0.75` = quarter, `0.65` = named season/approximate range, `0.50` = vague relative with computable anchor, `0.35` = highly vague, `0.00` = no temporal signal. LLMs self-report certainty rather than clustering near 1.0. + - Low temporal confidence (< 0.5) with a non-null date logs a `WARNING`; signal is never suppressed — callers decide how to filter. + - Cache key now includes the `extract_temporal_bounds` flag to prevent cross-mode cache pollution. + - Flag propagated through `_extract_relations_chunked()` so long-text chunked extraction also carries temporal metadata. + - Added `RelationWithTemporalOut` and `RelationsWithTemporalResponse` Pydantic schemas in `semantica/semantic_extract/schemas.py`. A separate schema is required because `RelationOut` uses `extra="ignore"`, which silently drops any undeclared field including the four temporal fields. + - New `semantica/kg/temporal_normalizer.py` — `TemporalNormalizer` class (zero LLM calls, pure regex + `dateutil` arithmetic): + - `normalize(value)` → `(valid_from, valid_until)` UTC `datetime` tuple or `None`. Resolution order: ISO 8601 full parse → partial-date regex (year-only, month+year, YYYY-MM, Q[1-4] YYYY) → ambiguous-slash-date detection → domain phrase map → relative phrase resolution via `relativedelta`. + - `normalize_phrase(phrase)` → metadata dict `{"maps_to": ..., "type": ..., "domain": [...]}` or `None` — exact match then regex-pattern keys. + - Ambiguous `DD/MM/YYYY`-style inputs issue `TemporalAmbiguityWarning` and return `None` — never silently guesses locale. + - Unparseable inputs return `None` with a debug log — never raise. + - Relative phrases (`"last year"`, `"three months ago"`, etc.) raise `ValueError` if `reference_date` is `None` rather than guessing. + - Default phrase map covers 13 domains: General/Policy (`effective date`, `effective from/as of/beginning`, `in force until`, `retroactive to`, `sunset clause`), Healthcare (`approval date`, `expiry date`, `market authorization`), Cybersecurity (`incident window`, `campaign period`), Supply Chain (`certification valid through`), Finance (`trading halt`), Energy (`commissioned date`, `decommissioned date`). + - User-supplied `phrase_map` is merged over defaults at construction (`{**defaults, **user_map}`) — custom entries win without forking the library. + - Added `TemporalAmbiguityWarning(UserWarning)` to `semantica/utils/exceptions.py`. + - Exported `TemporalNormalizer` from `semantica/kg/__init__.py`. + - Added 53 new tests in `tests/semantic_extract/test_temporal_extraction.py`; zero real LLM calls, suite runs in ~3.5 s. All 873 existing tests continue to pass. + - **Fix: OllamaProvider ignores `base_url`** (PR #408 by @AlexeyMyslin, fixed by @KaifAhmad1): - `OllamaProvider._init_client()` was assigning the raw `ollama` module to `self.client` instead of instantiating `ollama.Client(host=self.base_url)`, causing all requests to silently hit `localhost:11434` regardless of the `base_url` passed by the user - Fixed by replacing `self.client = ollama` with `self.client = ollama.Client(host=self.base_url)` — remote Ollama servers (e.g. `http://192.168.1.3:11434`) are now reachable