diff --git a/crawl4ai/extraction_strategy.py b/crawl4ai/extraction_strategy.py
index ed039890e..e0f7011c0 100644
--- a/crawl4ai/extraction_strategy.py
+++ b/crawl4ai/extraction_strategy.py
@@ -3,6 +3,7 @@
from typing import Any, List, Dict, Optional, Tuple, Pattern, Union
from concurrent.futures import ThreadPoolExecutor, as_completed
import json
+import sys
import time
from enum import IntFlag, auto
@@ -1080,10 +1081,158 @@ def __init__(self, schema: Dict[str, Any], **kwargs):
Args:
schema (Dict[str, Any]): The schema defining the extraction rules.
+ health_threshold (float, optional): Enables serve-time health checking. A
+ schema is generated and validated against one snapshot of a page, but it
+ is then applied for months while the site keeps changing. When a selector
+ stops matching, extraction does not fail - it returns fewer records, or
+ records with silently empty fields, and the caller is told nothing. With
+ this set, every extraction is scored (see `_health_report`) and anything
+ below the threshold is reported through `on_degraded`. Costs nothing:
+ the score is computed from the records already extracted, with no second
+ pass over the DOM. Defaults to None (off, no behaviour change).
+ on_degraded (Callable[[dict], None], optional): Called with the health report
+ when a extraction scores below `health_threshold`. Defaults to a warning
+ on the strategy's logger, or stderr when there is none.
+ heal_llm_config (LLMConfig, optional): When set, a degraded extraction also
+ triggers ONE regeneration of the schema from the page that degraded it,
+ through `generate_schema`. The replacement is adopted only if it actually
+ scores better on that same page, so a failed repair leaves the old schema
+ in place rather than swapping one broken extractor for another. The field
+ names of the current schema are carried into the request, because
+ everything downstream is keyed on them. Requires `health_threshold`.
+ max_regenerations (int): How many times this strategy may regenerate over its
+ lifetime. Defaults to 1 - a site that is permanently unreadable must not
+ bill for a model call per page.
"""
super().__init__(**kwargs)
self.schema = schema
self.verbose = kwargs.get("verbose", False)
+ self.health_threshold = kwargs.get("health_threshold", None)
+ self.on_degraded = kwargs.get("on_degraded", None)
+ self.heal_llm_config = kwargs.get("heal_llm_config", None)
+ self.max_regenerations = kwargs.get("max_regenerations", 1)
+ self._regenerations = 0
+
+ def _health_report(self, results: List[Dict[str, Any]]) -> Dict[str, Any]:
+ """Score an extraction that has already happened. No DOM work, no model call.
+
+ `coverage` is the fraction of the schema's declared fields that came back
+ populated in at least one record; a schema whose baseSelector matched nothing
+ scores 0. Both are the shapes site changes actually take: a renamed container
+ empties the result, a renamed field empties one column and leaves the record
+ count untouched - which is the one nothing downstream can see.
+ """
+ declared = [f.get("name") for f in self.schema.get("fields", [])
+ if f.get("name")]
+ declared += [f.get("name") for f in self.schema.get("baseFields", [])
+ if f.get("name")]
+ populated = {name for r in results for name in declared
+ if r.get(name) not in (None, "", [], {})}
+ coverage = (len(populated) / len(declared)) if declared else 1.0
+ return {
+ "records": len(results),
+ "declared_fields": len(declared),
+ "populated_fields": len(populated),
+ "coverage": 0.0 if not results else coverage,
+ "empty_fields": sorted(set(declared) - populated),
+ "healthy": bool(results) and coverage >= (self.health_threshold or 0.0),
+ }
+
+ def _check_health(self, results: List[Dict[str, Any]], url: str = "",
+ html_content: str = "") -> List[Dict[str, Any]]:
+ """Report an extraction that no longer fits the page it is being applied to, and -
+ when a heal config is present - regenerate the schema from that same page.
+
+ Returns the results to serve: the original ones, or the ones a healed schema
+ produced.
+ """
+ if self.health_threshold is None:
+ return results
+ report = self._health_report(results)
+ if report["healthy"]:
+ return results
+ report["url"] = url
+ report["schema_name"] = self.schema.get("name", "")
+ healed = self._try_heal(report, html_content, url)
+ if healed is not None:
+ return healed
+ if self.on_degraded is not None:
+ self.on_degraded(report)
+ return results
+ message = (
+ f"Extraction schema '{report['schema_name']}' looks stale on {url or 'page'}: "
+ f"{report['records']} record(s), coverage {report['coverage']:.0%} "
+ f"(threshold {self.health_threshold:.0%})"
+ + (f", always-empty field(s): {', '.join(report['empty_fields'])}"
+ if report["empty_fields"] else "")
+ + " - the page has probably changed; regenerate the schema."
+ )
+ logger = getattr(self, "logger", None)
+ if logger is not None:
+ logger.warning(message=message, tag="EXTRACT")
+ else:
+ print(f"[EXTRACT] {message}", file=sys.stderr)
+ return results
+
+ def _try_heal(self, report: Dict[str, Any], html_content: str,
+ url: str) -> Optional[List[Dict[str, Any]]]:
+ """Regenerate the schema from the page that broke it. None when not attempted or
+ when the replacement did not turn out better.
+
+ The check after regeneration is the point: `generate_schema` validates a candidate
+ against the HTML it was given, but 'the selectors match something' is not the same
+ as 'this extractor is healthy again', and adopting a replacement on the strength of
+ the generator's own opinion is how a stale extractor becomes a differently-stale
+ one. So the candidate is scored by the same measure that raised the alarm, and it
+ has to beat it.
+ """
+ if (self.heal_llm_config is None or not html_content
+ or self._regenerations >= self.max_regenerations):
+ return None
+ self._regenerations += 1
+ schema_type = "XPATH" if "XPath" in type(self).__name__ else "CSS"
+ # Carry the field names over: everything downstream is keyed on them, so a repair
+ # that renames a column is not a repair.
+ target = {f["name"]: "..." for f in self.schema.get("fields", [])
+ if f.get("name")}
+ try:
+ candidate = JsonElementExtractionStrategy.generate_schema(
+ html=html_content, schema_type=schema_type,
+ target_json_example=json.dumps(target),
+ llm_config=self.heal_llm_config)
+ probe = type(self)(schema=candidate)
+ probe_results = probe.extract(url=url, html_content=html_content)
+ probe.schema, probe.health_threshold = candidate, self.health_threshold
+ new_report = probe._health_report(probe_results)
+ except Exception as e: # noqa: BLE001 - a failed repair must not fail the crawl
+ self._report_heal(report, None, f"regeneration failed: {e}", url)
+ return None
+ if new_report["coverage"] <= report["coverage"] or not new_report["healthy"]:
+ self._report_heal(report, new_report, "regenerated schema is no better - "
+ "keeping the current one", url)
+ return None
+ self.schema = candidate
+ self._report_heal(report, new_report, "schema regenerated and adopted", url)
+ return probe_results
+
+ def _report_heal(self, before: Dict[str, Any], after: Optional[Dict[str, Any]],
+ outcome: str, url: str) -> None:
+ payload = dict(before)
+ payload["heal"] = {"outcome": outcome, "attempt": self._regenerations,
+ "coverage_before": before["coverage"],
+ "coverage_after": after["coverage"] if after else None}
+ if self.on_degraded is not None:
+ self.on_degraded(payload)
+ return
+ message = (f"Extraction schema '{before.get('schema_name', '')}' on "
+ f"{url or 'page'}: {outcome} (coverage "
+ f"{before['coverage']:.0%}"
+ + (f" -> {after['coverage']:.0%}" if after else "") + ")")
+ logger = getattr(self, "logger", None)
+ if logger is not None:
+ logger.warning(message=message, tag="EXTRACT")
+ else:
+ print(f"[EXTRACT] {message}", file=sys.stderr)
def extract(
self, url: str, html_content: str, *q, **kwargs
@@ -1128,7 +1277,7 @@ def extract(
if item:
results.append(item)
- return results
+ return self._check_health(results, url, html_content)
@abstractmethod
def _parse_html(self, html_content: str):
diff --git a/tests/test_schema_health.py b/tests/test_schema_health.py
new file mode 100644
index 000000000..6105e8ad8
--- /dev/null
+++ b/tests/test_schema_health.py
@@ -0,0 +1,204 @@
+"""Serve-time health checking for generated extraction schemas.
+
+A schema is validated once, against the snapshot it was generated from, and then applied
+for months while the site keeps changing. When a selector stops matching, extraction does
+not raise: it returns fewer records, or records with a silently empty column. These tests
+pin both shapes of that failure and the report that now names them.
+"""
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from crawl4ai.extraction_strategy import ( # noqa: E402
+ JsonCssExtractionStrategy,
+ JsonElementExtractionStrategy,
+)
+
+SCHEMA = {
+ "name": "Products",
+ "baseSelector": "article.product",
+ "fields": [
+ {"name": "title", "selector": "h3", "type": "text"},
+ {"name": "price", "selector": "p.price", "type": "text"},
+ ],
+}
+
+PAGE = """
+
+ First
10
+ Second
20
+
+"""
+
+# The two shapes a redesign takes.
+PAGE_CONTAINER_RENAMED = PAGE.replace('class="product"', 'class="product-card"')
+PAGE_FIELD_RENAMED = PAGE.replace('class="price"', 'class="product-price"')
+
+
+def test_off_by_default_and_silent():
+ """No threshold: behaviour is exactly as before, including on a broken page."""
+ strategy = JsonCssExtractionStrategy(schema=SCHEMA)
+ assert len(strategy.extract(url="", html_content=PAGE)) == 2
+ assert strategy.extract(url="", html_content=PAGE_CONTAINER_RENAMED) == []
+
+
+def test_healthy_page_reports_nothing():
+ seen = []
+ strategy = JsonCssExtractionStrategy(
+ schema=SCHEMA, health_threshold=0.9, on_degraded=seen.append)
+ results = strategy.extract(url="http://x/", html_content=PAGE)
+ assert len(results) == 2
+ assert seen == []
+
+
+def test_container_renamed_is_reported():
+ """The obvious rot: nothing matches, the result is empty."""
+ seen = []
+ strategy = JsonCssExtractionStrategy(
+ schema=SCHEMA, health_threshold=0.9, on_degraded=seen.append)
+ results = strategy.extract(url="http://x/", html_content=PAGE_CONTAINER_RENAMED)
+ assert results == []
+ assert len(seen) == 1
+ assert seen[0]["records"] == 0
+ assert seen[0]["coverage"] == 0.0
+ assert seen[0]["url"] == "http://x/"
+
+
+def test_field_renamed_is_reported_although_record_count_is_unchanged():
+ """The dangerous rot: the same number of records, one column silently empty."""
+ seen = []
+ strategy = JsonCssExtractionStrategy(
+ schema=SCHEMA, health_threshold=0.9, on_degraded=seen.append)
+ results = strategy.extract(url="http://x/", html_content=PAGE_FIELD_RENAMED)
+ assert len(results) == 2, "the count is unchanged - this is why it goes unnoticed"
+ assert all("price" not in r or not r["price"] for r in results)
+ assert len(seen) == 1
+ assert seen[0]["empty_fields"] == ["price"]
+ assert seen[0]["coverage"] == pytest.approx(0.5)
+
+
+def test_threshold_is_respected():
+ """Half the fields populated passes at 0.5 and fails at 0.6."""
+ lenient, strict = [], []
+ JsonCssExtractionStrategy(schema=SCHEMA, health_threshold=0.5,
+ on_degraded=lenient.append
+ ).extract(url="", html_content=PAGE_FIELD_RENAMED)
+ JsonCssExtractionStrategy(schema=SCHEMA, health_threshold=0.6,
+ on_degraded=strict.append
+ ).extract(url="", html_content=PAGE_FIELD_RENAMED)
+ assert lenient == []
+ assert len(strict) == 1
+
+
+def test_default_reporter_warns_without_a_callback(capsys):
+ JsonCssExtractionStrategy(schema=SCHEMA, health_threshold=0.9).extract(
+ url="http://x/", html_content=PAGE_FIELD_RENAMED)
+ err = capsys.readouterr().err
+ assert "looks stale" in err
+ assert "price" in err
+
+
+# --- regeneration -----------------------------------------------------------------
+# The generator is stubbed: these pin the POLICY around regeneration - when it fires,
+# what it accepts, what it refuses, how often - which is what can go wrong without a
+# model being involved.
+
+REPAIRED = {
+ "name": "Products",
+ "baseSelector": "article.product",
+ "fields": [
+ {"name": "title", "selector": "h3", "type": "text"},
+ {"name": "price", "selector": "p.product-price", "type": "text"},
+ ],
+}
+USELESS = dict(REPAIRED, fields=[{"name": "title", "selector": "h3", "type": "text"},
+ {"name": "price", "selector": "p.gone", "type": "text"}])
+
+
+@pytest.fixture
+def stub_generator(monkeypatch):
+ calls = []
+
+ def make(schema_to_return):
+ def fake(**kwargs):
+ calls.append(kwargs)
+ return schema_to_return
+ monkeypatch.setattr(JsonElementExtractionStrategy, "generate_schema",
+ staticmethod(fake))
+ return calls
+ return make
+
+
+def test_heal_adopts_a_schema_that_actually_fixes_the_page(stub_generator):
+ calls = stub_generator(REPAIRED)
+ seen = []
+ strategy = JsonCssExtractionStrategy(
+ schema=dict(SCHEMA), health_threshold=0.9, on_degraded=seen.append,
+ heal_llm_config=object())
+ results = strategy.extract(url="http://x/", html_content=PAGE_FIELD_RENAMED)
+ assert len(calls) == 1, "one regeneration, from the page that degraded"
+ assert all(r.get("price") for r in results), "the served records are the healed ones"
+ assert strategy.schema == REPAIRED, "and the strategy keeps the new schema"
+ assert seen[0]["heal"]["outcome"] == "schema regenerated and adopted"
+ assert seen[0]["heal"]["coverage_before"] == pytest.approx(0.5)
+ assert seen[0]["heal"]["coverage_after"] == pytest.approx(1.0)
+
+
+def test_heal_refuses_a_replacement_that_is_no_better(stub_generator):
+ stub_generator(USELESS)
+ seen = []
+ original = dict(SCHEMA)
+ strategy = JsonCssExtractionStrategy(
+ schema=dict(SCHEMA), health_threshold=0.9, on_degraded=seen.append,
+ heal_llm_config=object())
+ strategy.extract(url="http://x/", html_content=PAGE_FIELD_RENAMED)
+ assert strategy.schema == original, "a repair that does not repair is discarded"
+ assert "no better" in seen[0]["heal"]["outcome"]
+
+
+def test_heal_carries_the_field_names_into_the_request(stub_generator):
+ calls = stub_generator(REPAIRED)
+ JsonCssExtractionStrategy(
+ schema=dict(SCHEMA), health_threshold=0.9, heal_llm_config=object()
+ ).extract(url="http://x/", html_content=PAGE_FIELD_RENAMED)
+ example = calls[0]["target_json_example"]
+ assert "title" in example and "price" in example, (
+ "downstream is keyed on the field names - a repair may not rename a column")
+
+
+def test_heal_is_bounded(stub_generator):
+ calls = stub_generator(USELESS)
+ strategy = JsonCssExtractionStrategy(
+ schema=dict(SCHEMA), health_threshold=0.9, heal_llm_config=object(),
+ max_regenerations=1)
+ for _ in range(5):
+ strategy.extract(url="http://x/", html_content=PAGE_FIELD_RENAMED)
+ assert len(calls) == 1, "a permanently unreadable site must not bill per page"
+
+
+def test_heal_failure_does_not_break_the_crawl(monkeypatch):
+ def boom(**kwargs):
+ raise RuntimeError("provider is down")
+ monkeypatch.setattr(JsonElementExtractionStrategy, "generate_schema",
+ staticmethod(boom))
+ seen = []
+ strategy = JsonCssExtractionStrategy(
+ schema=dict(SCHEMA), health_threshold=0.9, on_degraded=seen.append,
+ heal_llm_config=object())
+ results = strategy.extract(url="http://x/", html_content=PAGE_FIELD_RENAMED)
+ assert len(results) == 2, "the original records are still served"
+ assert "regeneration failed" in seen[0]["heal"]["outcome"]
+
+
+def test_no_heal_config_means_no_generation(stub_generator):
+ calls = stub_generator(REPAIRED)
+ JsonCssExtractionStrategy(schema=dict(SCHEMA), health_threshold=0.9).extract(
+ url="http://x/", html_content=PAGE_FIELD_RENAMED)
+ assert calls == [], "reporting stays available without opting into model calls"
+
+
+if __name__ == "__main__":
+ raise SystemExit(pytest.main([__file__, "-v"]))