221 lines
7.1 KiB
Python
221 lines
7.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script for live LLM annotation with domain/range validation.
|
|
|
|
Tests the full annotation pipeline with real NDE web archives.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
import os
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
# Add src to path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
# Load environment
|
|
from dotenv import load_dotenv
|
|
load_dotenv()
|
|
|
|
from glam_extractor.annotators.llm_annotator import (
|
|
LLMAnnotator,
|
|
LLMAnnotatorConfig,
|
|
LLMProvider,
|
|
RetryConfig,
|
|
)
|
|
|
|
|
|
async def test_luther_museum_annotation():
|
|
"""Test annotation of Luther Museum website with relationship validation."""
|
|
|
|
print("\n" + "="*70)
|
|
print("LIVE LLM ANNOTATION TEST - Luther Museum")
|
|
print("="*70)
|
|
|
|
# Check for API token
|
|
api_token = os.getenv("ZAI_API_TOKEN")
|
|
if not api_token:
|
|
print(" [SKIP] ZAI_API_TOKEN not set in environment")
|
|
return None
|
|
|
|
# Path to Luther Museum HTML
|
|
html_path = Path("data/nde/enriched/entries/web/1600/luthermuseum.nl/rendered.html")
|
|
if not html_path.exists():
|
|
print(f" [SKIP] HTML file not found: {html_path}")
|
|
return None
|
|
|
|
print(f"\n Source: {html_path}")
|
|
print(f" Provider: Z.AI (GLM-4-Flash)")
|
|
|
|
# Configure annotator
|
|
config = LLMAnnotatorConfig(
|
|
provider=LLMProvider.ZAI,
|
|
model="glm-4-flash",
|
|
api_key=api_token,
|
|
context_convention="GLAM-NER v1.7.0-unified",
|
|
retry=RetryConfig(
|
|
max_retries=3,
|
|
base_delay=2.0,
|
|
max_delay=30.0,
|
|
),
|
|
)
|
|
|
|
annotator = LLMAnnotator(config)
|
|
|
|
print("\n Starting annotation...")
|
|
start_time = datetime.now()
|
|
|
|
try:
|
|
session = await annotator.annotate(
|
|
html_path,
|
|
source_url="https://luthermuseum.nl/nl",
|
|
)
|
|
|
|
elapsed = (datetime.now() - start_time).total_seconds()
|
|
print(f" Completed in {elapsed:.2f} seconds")
|
|
|
|
# Report results
|
|
print("\n" + "-"*70)
|
|
print("ANNOTATION RESULTS")
|
|
print("-"*70)
|
|
|
|
print(f"\n Session ID: {session.session_id}")
|
|
print(f" Agent: {session.agent_name} / {session.model_id}")
|
|
|
|
# Entity claims
|
|
print(f"\n ENTITY CLAIMS: {len(session.entity_claims)}")
|
|
for i, claim in enumerate(session.entity_claims[:10]): # Show first 10
|
|
hyponym = claim.hyponym or (claim.hypernym.value if claim.hypernym else "?")
|
|
print(f" [{i+1}] {hyponym}: {claim.claim_value}")
|
|
if claim.class_uri:
|
|
print(f" class_uri: {claim.class_uri}")
|
|
if claim.wikidata_id:
|
|
print(f" wikidata: {claim.wikidata_id}")
|
|
|
|
if len(session.entity_claims) > 10:
|
|
print(f" ... and {len(session.entity_claims) - 10} more entities")
|
|
|
|
# Relationship claims
|
|
print(f"\n RELATIONSHIP CLAIMS: {len(session.relationship_claims)}")
|
|
for i, claim in enumerate(session.relationship_claims[:10]): # Show first 10
|
|
rel_type = claim.relationship_hyponym or "?"
|
|
subject = claim.subject.span_text if claim.subject else "?"
|
|
obj = claim.object.span_text if claim.object else "?"
|
|
print(f" [{i+1}] {rel_type}: {subject} -> {obj}")
|
|
if claim.predicate_uris:
|
|
print(f" predicates: {claim.predicate_uris[:2]}")
|
|
if claim.temporal_scope and claim.temporal_scope.start_date:
|
|
print(f" temporal: {claim.temporal_scope.start_date}")
|
|
|
|
if len(session.relationship_claims) > 10:
|
|
print(f" ... and {len(session.relationship_claims) - 10} more relationships")
|
|
|
|
# Validation errors/warnings
|
|
if session.errors:
|
|
print(f"\n VALIDATION WARNINGS/ERRORS: {len(session.errors)}")
|
|
for error in session.errors[:10]:
|
|
print(f" - {error[:100]}...")
|
|
if len(session.errors) > 10:
|
|
print(f" ... and {len(session.errors) - 10} more")
|
|
else:
|
|
print("\n VALIDATION: No domain/range violations detected")
|
|
|
|
# Export to JSON
|
|
output_path = Path("data/nde/enriched/entries/web/1600/luthermuseum.nl/annotation_session.json")
|
|
with open(output_path, 'w', encoding='utf-8') as f:
|
|
json.dump(session.to_dict(), f, indent=2, ensure_ascii=False)
|
|
print(f"\n Exported to: {output_path}")
|
|
|
|
return session
|
|
|
|
except Exception as e:
|
|
print(f"\n [ERROR] Annotation failed: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return None
|
|
|
|
|
|
async def test_boijmans_annotation():
|
|
"""Test annotation of Boijmans Museum website."""
|
|
|
|
print("\n" + "="*70)
|
|
print("LIVE LLM ANNOTATION TEST - Boijmans Museum")
|
|
print("="*70)
|
|
|
|
api_token = os.getenv("ZAI_API_TOKEN")
|
|
if not api_token:
|
|
print(" [SKIP] ZAI_API_TOKEN not set")
|
|
return None
|
|
|
|
html_path = Path("data/nde/enriched/entries/web/1606/boijmans.nl/rendered.html")
|
|
if not html_path.exists():
|
|
print(f" [SKIP] HTML file not found: {html_path}")
|
|
return None
|
|
|
|
print(f"\n Source: {html_path}")
|
|
|
|
config = LLMAnnotatorConfig(
|
|
provider=LLMProvider.ZAI,
|
|
model="glm-4-flash",
|
|
api_key=api_token,
|
|
context_convention="GLAM-NER v1.7.0-unified",
|
|
)
|
|
|
|
annotator = LLMAnnotator(config)
|
|
|
|
print("\n Starting annotation...")
|
|
start_time = datetime.now()
|
|
|
|
try:
|
|
session = await annotator.annotate(
|
|
html_path,
|
|
source_url="https://boijmans.nl",
|
|
)
|
|
|
|
elapsed = (datetime.now() - start_time).total_seconds()
|
|
print(f" Completed in {elapsed:.2f} seconds")
|
|
|
|
print(f"\n Entity claims: {len(session.entity_claims)}")
|
|
print(f" Relationship claims: {len(session.relationship_claims)}")
|
|
print(f" Validation issues: {len(session.errors)}")
|
|
|
|
# Show entity type distribution
|
|
type_counts = {}
|
|
for claim in session.entity_claims:
|
|
t = claim.hyponym or (claim.hypernym.value if claim.hypernym else "UNK")
|
|
type_counts[t] = type_counts.get(t, 0) + 1
|
|
|
|
print("\n Entity type distribution:")
|
|
for t, count in sorted(type_counts.items(), key=lambda x: -x[1])[:10]:
|
|
print(f" {t}: {count}")
|
|
|
|
return session
|
|
|
|
except Exception as e:
|
|
print(f"\n [ERROR] {e}")
|
|
return None
|
|
|
|
|
|
async def main():
|
|
"""Run all live annotation tests."""
|
|
|
|
print("\n" + "="*70)
|
|
print("LIVE ANNOTATION TEST SUITE")
|
|
print("="*70)
|
|
print(f"\nTimestamp: {datetime.now().isoformat()}")
|
|
|
|
# Test 1: Luther Museum
|
|
session1 = await test_luther_museum_annotation()
|
|
|
|
# Test 2: Boijmans (optional - comment out to save API calls)
|
|
# session2 = await test_boijmans_annotation()
|
|
|
|
print("\n" + "="*70)
|
|
print("TEST SUITE COMPLETE")
|
|
print("="*70 + "\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|