Semantic Extraction

from semantica.semantic_extract import NamedEntityRecognizer

# A finished intelligence report excerpt
report = """
ASSESSMENT (SECRET//NOFORN): GAMMA-7 threat actor, assessed with HIGH confidence
to operate from COUNTRY_X, conducted OPERATION NIGHTFALL between Jan–Mar 2025.
The group deployed HAMMERTOSS malware targeting NATO logistics networks via
spear-phishing lures referencing Exercise STEADFAST DEFENDER 2025.
Infrastructure cluster 185.220.101.0/24 was active throughout the campaign.
A second actor, DELTA-3, provided GAMMA-7 with zero-day exploits (CVE-2025-1234,
CVE-2025-5678) for Ivanti Connect Secure VPN appliances.
"""

# LLM-backed NER handles custom intelligence entity types that spaCy won't know.
# Fallback chain: if llm returns nothing, try ml (spaCy), then pattern matching.
ner = NamedEntityRecognizer(
    methods=["llm", "ml", "pattern"],
    confidence_threshold=0.75,
    provider="anthropic",
    llm_model="claude-sonnet-4-6",
)
entities = ner.extract_entities(report)

for e in entities:
    print("[{:>5.2f}]  {:15s}  {}".format(e.confidence, e.label, e.text))

# Expected output (abbreviated):
# [ 0.94]  THREAT_ACTOR    GAMMA-7
# [ 0.91]  THREAT_ACTOR    DELTA-3
# [ 0.97]  MALWARE         HAMMERTOSS
# [ 0.99]  CVE             CVE-2025-1234
# [ 0.99]  CVE             CVE-2025-5678
# [ 0.88]  NETWORK         185.220.101.0/24
# [ 0.85]  ORG             NATO

# classify_entities groups the flat list into a dict keyed by label
grouped = ner.classify_entities(entities)

print("Threat actors:", [e.text for e in grouped.get("THREAT_ACTOR", [])])
# Threat actors: ['GAMMA-7', 'DELTA-3']

print("Malware:      ", [e.text for e in grouped.get("MALWARE", [])])
# Malware:       ['HAMMERTOSS']

print("CVEs:         ", [e.text for e in grouped.get("CVE", [])])
# CVEs:          ['CVE-2025-1234', 'CVE-2025-5678']

# score_confidence re-scores with a statistical model — useful when your primary
# method doesn't return calibrated probabilities (e.g., pattern matching)
rescored = ner.score_confidence(entities)
high_conf = [e for e in rescored if e.confidence >= 0.85]
print("High-confidence entities: {}".format(len(high_conf)))

from semantica.semantic_extract import RelationExtractor

# dependency parsing picks up syntactic relations ("DELTA-3 provided GAMMA-7 with...")
# pattern matching catches explicit verb phrases as a fallback
rel = RelationExtractor(
    method=["dependency", "pattern"],
    relation_types=["operates_from", "deployed", "targets", "provided_to", "exploits"],
    bidirectional=True,      # also try object→subject direction
    confidence_threshold=0.65,
)

# extract_relations takes the original text plus the entities you already found.
# Passing entities avoids re-running NER and constrains the search space.
relations = rel.extract_relations(report, entities)

for r in relations:
    print("[{:.2f}]  {} --[{}]--> {}".format(
        r.confidence, r.subject.text, r.predicate, r.object.text
    ))

# Example output:
# [0.88]  GAMMA-7 --[deployed]--> HAMMERTOSS
# [0.82]  DELTA-3 --[provided_to]--> GAMMA-7
# [0.79]  CVE-2025-1234 --[exploits]--> Ivanti Connect Secure
# [0.85]  GAMMA-7 --[targets]--> NATO

from semantica.semantic_extract import EventDetector

# method="llm" is the default and gives the best recall for implicit events.
# method="pattern" is appropriate for offline, high-throughput pipelines.
detector = EventDetector(method="llm", provider="anthropic")

events = detector.detect_events(report)

for ev in events:
    print("[{}]  time={}  location={}".format(
        ev.event_type, ev.time or "unknown", ev.location or "n/a"
    ))
    print("   participants: {}".format(ev.participants))
    print("   text: {}".format(ev.text[:80]))

# Example output:
# [campaign]  time=Jan–Mar 2025  location=n/a
#    participants: ['GAMMA-7', 'NATO']
#    text: GAMMA-7 conducted OPERATION NIGHTFALL between Jan–Mar 2025...
# [supply_chain]  time=unknown  location=n/a
#    participants: ['DELTA-3', 'GAMMA-7']
#    text: DELTA-3 provided GAMMA-7 with zero-day exploits...

# extract() handles concurrent processing internally
batch_events = detector.extract([
    {"content": report, "id": "FINTEL_2025_0192"},
    {"content": advisory_2, "id": "FINTEL_2025_0193"},
    # ... up to 200 documents
])

# batch_events is List[List[Event]] — one inner list per document
for doc_idx, doc_events in enumerate(batch_events):
    print("Doc {}: {} events".format(doc_idx, len(doc_events)))

from semantica.semantic_extract import CoreferenceResolver

resolver = CoreferenceResolver(method="llm", provider="anthropic")

# Pass the already-extracted entities so the resolver can anchor pronouns
# to the correct named entities rather than guessing from scratch
chains = resolver.resolve_coreferences(report, entities=entities)

for chain in chains:
    print("Representative: {}".format(chain.representative.text))
    mentions = [m.text for m in chain.mentions if m.text != chain.representative.text]
    print("   Also called: {}".format(mentions))

# Example output:
# Representative: GAMMA-7
#    Also called: ['the group', 'they', 'the threat actor']

from semantica.semantic_extract import TripletExtractor

tri = TripletExtractor(
    method="llm",
    provider="anthropic",
    llm_model="claude-sonnet-4-6",
    include_temporal=True,       # attach time context to triplets when available
    include_provenance=True,     # embed source document reference in each triplet
)

# Feed in the entities and relations you already extracted — the extractor
# uses them to constrain and validate what it produces
triplets = tri.extract_triplets(report, entities, relations)

# Filter malformed triplets before serialisation
valid = tri.validate_triplets(triplets)
print("Valid: {}/{}".format(len(valid), len(triplets)))

for t in valid:
    print("({}, {}, {})  conf={:.2f}".format(
        t.subject, t.predicate, t.object, t.confidence
    ))

# Serialise to Turtle RDF for graph ingestion or STIX export
turtle_rdf = tri.serialize_triplets(valid, format="turtle")
print(turtle_rdf[:600])

# formats: "turtle" | "ntriples" | "jsonld" | "xml"

from semantica.semantic_extract import (
    NamedEntityRecognizer, RelationExtractor,
    EventDetector, CoreferenceResolver, TripletExtractor,
)
from semantica.provenance import ProvenanceManager
from semantica.context import AgentContext, ContextGraph
from semantica.vector_store import VectorStore


def ingest_intel_report(
    text: str,
    doc_id: str,
    agent: AgentContext,
    prov: ProvenanceManager,
    method: str = "llm",
) -> dict:
    """
    Full extraction pipeline: NER → Coref → Relations → Events → Triplets → Graph.
    Returns a summary dict with counts for monitoring dashboards.
    """

    # 1. Named entity recognition
    ner = NamedEntityRecognizer(
        methods=[method, "pattern"],
        confidence_threshold=0.70,
        provider="anthropic",
        llm_model="claude-sonnet-4-6",
    )
    entities = ner.extract_entities(text)
    classified = ner.classify_entities(entities)

    # 2. Coreference resolution — before relation extraction
    resolver = CoreferenceResolver(method=method, provider="anthropic")
    chains = resolver.resolve_coreferences(text, entities=entities)

    # 3. Relation extraction
    rel = RelationExtractor(
        method=[method, "pattern"],
        relation_types=["deployed", "targets", "exploits", "operates_from", "provided_to"],
        confidence_threshold=0.65,
        provider="anthropic",
        llm_model="claude-sonnet-4-6",
    )
    relations = rel.extract_relations(text, entities)

    # 4. Event detection
    detector = EventDetector(method=method, provider="anthropic")
    events = detector.detect_events(text)

    # 5. Triplet extraction and RDF serialisation
    tri = TripletExtractor(
        method=method,
        provider="anthropic",
        llm_model="claude-sonnet-4-6",
        include_temporal=True,
        include_provenance=True,
    )
    triplets = tri.extract_triplets(text, entities, relations)
    valid    = tri.validate_triplets(triplets)
    turtle   = tri.serialize_triplets(valid, format="turtle")

    # 6. Store in agent memory + knowledge graph
    graph_stats = agent.store(
        [{"content": text, "metadata": {"source": doc_id, "classification": "SECRET//NOFORN"}}],
        extract_entities=True,
        extract_relationships=True,
        link_entities=True,
    )

    # 7. Provenance tracking — every entity linked back to its source document
    prov.track_entities_batch(
        [{"id": e.text.lower().replace(" ", "_"), "confidence": e.confidence}
         for e in entities],
        source=doc_id,
        activity_id="intel_extraction_pipeline",
    )

    return {
        "entities":       len(entities),
        "entity_types":   {k: len(v) for k, v in classified.items()},
        "coref_chains":   len(chains),
        "relations":      len(relations),
        "events":         len(events),
        "triplets_valid": len(valid),
        "graph_nodes":    graph_stats.get("graph_nodes", 0),
        "graph_edges":    graph_stats.get("graph_edges", 0),
        "rdf_turtle":     turtle,
    }


# Process all 200 reports
intel_graph = ContextGraph(advanced_analytics=True)
intel_agent = AgentContext(
    vector_store=VectorStore(backend="faiss", dimension=768),
    knowledge_graph=intel_graph,
    decision_tracking=True,
)
prov = ProvenanceManager(storage_path="intel_provenance.db")

# reports is a list of (text, doc_id) tuples loaded from your 200 documents
for text, doc_id in reports:
    summary = ingest_intel_report(text, doc_id, intel_agent, prov)
    print("{}: {} entities, {} relations, {} events, {}/{} triplets valid".format(
        doc_id,
        summary["entities"],
        summary["relations"],
        summary["events"],
        summary["triplets_valid"],
        len(summary["rdf_turtle"]),
    ))