coreason-tagger

A High-Throughput Biomedical Semantic Extraction Engine

Coreason-Tagger is a high-throughput, latency-aware NLP engine designed to normalize unstructured clinical and biomedical text into structured Knowledge Graph nodes. It utilizes a Strategy Pattern Architecture to dynamically switch extraction engines at runtime based on the complexity of the request.

Features

• Dynamic Extraction Strategies:
  • Speed (GLiNER): Single-pass inference for high-volume ETL and real-time UI highlighting.
  • Precision (NuNER Zero): Token-classification for fixed schemas where boundary precision is critical.
  • Reasoning (Ensemble): LLM-verified candidates for complex, ambiguous text using GLiNER recall + LLM verification.
• Contextualization: Lightweight assertion detection (Negation, Speculation, History, Family) to prevent false positives using rules or DistilBERT.
• Normalization: Maps ambiguous text spans to canonical IDs (e.g., SNOMED, RxNorm) using Vector Retrieval (Bi-Encoder) and Semantic Re-ranking.
• Resilience: Built-in Circuit Breakers and fallback mechanisms (L1/L2 Caching, Offline Mode) for robust operation.

Installation

pip install -r requirements.txt

Usage

Here is a quick example of how to initialize and use the CoreasonTagger:

import asyncio
from coreason_tagger.tagger import CoreasonTagger
from coreason_tagger.ner import ExtractorFactory
from coreason_tagger.assertion_detector import RegexBasedAssertionDetector
from coreason_tagger.codex_real import RealCoreasonCodex
from coreason_tagger.linker import VectorLinker
from coreason_tagger.schema import ExtractionStrategy

async def main():
    # Initialize components
    ner = ExtractorFactory()
    assertion = RegexBasedAssertionDetector()
    codex_client = RealCoreasonCodex(api_url="http://localhost:8000")
    linker = VectorLinker(codex_client=codex_client)

    # Initialize Tagger
    tagger = CoreasonTagger(ner=ner, assertion=assertion, linker=linker)

    # Tag text
    text = "Patient complains of severe migraine and nausea."
    results = await tagger.tag(
        text,
        labels=["Symptom", "Condition"],
        strategy=ExtractionStrategy.SPEED_GLINER
    )

    for entity in results:
        print(f"{entity.text}: {entity.label} (Assertion: {entity.assertion})")

if __name__ == "__main__":
    asyncio.run(main())

For more details, please refer to the Product Requirements.