Chunking: Shallow Parsing for Phrase Identification in NLP

from nltk import pos_tag, word_tokenize
from nltk.chunk import ne_chunk

# Download required NLTK resources (comment out after first run)
# nltk.download("punkt", quiet=True)
# nltk.download("averaged_perceptron_tagger", quiet=True)
# nltk.download("maxent_ne_chunker", quiet=True)
# nltk.download("words", quiet=True)
# nltk.download("conll2000", quiet=True)

# Example sentence with named entities
ne_sentence = "Barack Obama visited New York City yesterday."
ne_tokens = word_tokenize(ne_sentence)
ne_tagged = pos_tag(ne_tokens)
ne_chunked = ne_chunk(ne_tagged)

# Examples of noun phrases with varying internal structure
np_examples = [
    ("the cat", "simple: determiner + noun"),
    ("a big red ball", "adjectives: det + adj + adj + noun"),
    ("President Obama", "proper: title + name"),
    ("three blind mice", "numeral: number + adj + noun"),
    ("my favorite book", "possessive: possessive pronoun + adj + noun"),
]

# Common POS patterns that form noun phrases
np_pos_patterns = [
    "DT NN",  # the cat
    "DT JJ NN",  # the black cat
    "DT JJ JJ NN",  # the big black cat
    "NNP NNP",  # Barack Obama
    "CD NN",  # three cats
    "PRP$ NN",  # my cat
    "JJ NN",  # black cat (no determiner)
    "NN NN",  # computer screen (compound noun)
]

# Examples of verb phrases at different tense/aspect combinations
vp_examples = [
    ("runs", "simple present"),
    ("is running", "progressive: auxiliary + main verb"),
    ("has been running", "perfect progressive: 2 auxiliaries + main"),
    ("can swim", "modal + base form"),
    ("should have called", "modal + perfect auxiliary + past participle"),
    ("was eaten", "passive: auxiliary + past participle"),
]

# POS patterns for verb phrase chunks
vp_pos_patterns = [
    "VBZ",  # runs
    "VBD",  # ran
    "VBG",  # running (bare progressive, less common as VP)
    "MD VB",  # can run
    "VBZ VBG",  # is running
    "VBZ VBN",  # is eaten (passive)
    "MD VB VBN",  # will be eaten
    "VBZ VBN VBN",  # has been eaten (perfect passive)
]

# Examples of prepositional phrases
pp_examples = [
    ("in the house", "location"),
    ("with great care", "manner"),
    ("after the meeting", "time"),
    ("under the old oak tree", "complex location with adjective"),
    ("for my best friend", "beneficiary with possessive NP"),
    ("because of the rain", "causal (complex preposition)"),
]

other_chunk_types = {
    "ADJP": (
        "Adjective phrase",
        "very happy, quite tired, extremely well-read",
    ),
    "ADVP": ("Adverb phrase", "very quickly, rather slowly, almost never"),
    "SBAR": (
        "Subordinate clause",
        "that he left, if it rains, although she tried",
    ),
    "PRT": ("Particle", "give up, turn on, look after"),
    "CONJP": ("Conjunction phrase", "as well as, rather than, not only"),
    "INTJ": ("Interjection", "oh, wow, alas"),
}

# Example sentence with IOB chunk tags
sentence = "The quick brown fox jumps over the lazy dog"
tokens = sentence.split()

# IOB2 tags: B marks the START of every chunk
iob_tags = [
    "B-NP",  # The      (starts NP)
    "I-NP",  # quick    (inside NP)
    "I-NP",  # brown    (inside NP)
    "I-NP",  # fox      (inside NP)
    "B-VP",  # jumps    (starts VP)
    "B-PP",  # over     (starts PP)
    "B-NP",  # the      (starts new NP)
    "I-NP",  # lazy     (inside NP)
    "I-NP",  # dog      (inside NP)
]

def chunks_to_iob(tokens, chunks):
    """
    Convert chunk spans to IOB2 tags.

    Args:
        tokens: List of token strings
        chunks: List of (start_idx, end_idx, chunk_type) tuples
                where end_idx is exclusive (Python slicing convention)

    Returns:
        List of IOB2 tags, one per token
    """
    tags = ["O"] * len(tokens)

    for start, end, chunk_type in sorted(chunks, key=lambda x: x[0]):
        if start < 0 or end > len(tokens) or start >= end:
            continue
        tags[start] = f"B-{chunk_type}"
        for i in range(start + 1, end):
            tags[i] = f"I-{chunk_type}"

    return tags


def iob_to_chunks(tokens, tags):
    """
    Extract chunk spans from IOB2 tags.

    Returns:
        List of (start_idx, end_idx, chunk_type, text) tuples
    """
    chunks = []
    current_chunk = None  # (start_idx, chunk_type)

    for i, (token, tag) in enumerate(zip(tokens, tags)):
        if tag.startswith("B-"):
            # Close any open chunk first
            if current_chunk is not None:
                start, ctype = current_chunk
                chunks.append((start, i, ctype))
            # Start new chunk
            current_chunk = (i, tag[2:])

        elif tag.startswith("I-"):
            chunk_type = tag[2:]
            # Handle orphan I tag (I without preceding B)
            if current_chunk is None:
                current_chunk = (i, chunk_type)
            elif current_chunk[1] != chunk_type:
                # Type mismatch: close old, start new
                start, ctype = current_chunk
                chunks.append((start, i, ctype))
                current_chunk = (i, chunk_type)

        else:  # O tag
            if current_chunk is not None:
                start, ctype = current_chunk
                chunks.append((start, i, ctype))
                current_chunk = None

    # Handle chunk at end of sequence
    if current_chunk is not None:
        start, ctype = current_chunk
        chunks.append((start, len(tokens), ctype))

    # Attach text spans
    result = []
    for start, end, ctype in chunks:
        text = " ".join(tokens[start:end])
        result.append((start, end, ctype, text))

    return result


# Test round-trip conversion
test_tokens = "The quick brown fox jumps over the lazy dog".split()
test_chunks_input = [
    (0, 4, "NP"),  # The quick brown fox
    (4, 5, "VP"),  # jumps
    (5, 6, "PP"),  # over
    (6, 9, "NP"),  # the lazy dog
]

test_iob = chunks_to_iob(test_tokens, test_chunks_input)
recovered = iob_to_chunks(test_tokens, test_iob)

# Illustrate the ambiguity that chunking avoids
ambiguous = "I saw the man with the telescope"

# What chunking produces: flat, unambiguous
chunk_output = [
    ("I", "NP"),
    ("saw", "VP"),
    ("the man", "NP"),
    ("with", "PP"),
    ("the telescope", "NP"),
]

# What full parsing must choose between
parse_option_1 = "VP[saw NP[the man] PP[with NP[the telescope]]]"
parse_option_2 = "VP[saw NP[the man PP[with NP[the telescope]]]]"

parse_interp_1 = "I used the telescope to see the man"
parse_interp_2 = "I saw the man who had the telescope"

# Compare IOB1 and IOB2 for adjacent chunks of the same type
example_tokens = ["John", "Mary", "left"]
gold_chunks = [(0, 1, "NP"), (1, 2, "NP"), (2, 3, "VP")]

# IOB2: B always marks the first token of a chunk
iob2_tags = chunks_to_iob(example_tokens, gold_chunks)

# IOB1: B only used when same-type chunk follows same-type chunk
# Adjacent NPs: first NP uses I (since nothing precedes it of same type yet)
# Second NP uses B (because it follows an NP)
iob1_tags = ["I-NP", "B-NP", "I-VP"]  # IOB1 encoding

from nltk import RegexpParser

# Define a grammar for NP, VP, and PP chunking
np_grammar = r"""
    NP: {<DT|PRP\$>?<JJ>*<NN.*>+}
    VP: {<MD>?<VB.*>+}
    PP: {<IN>}
"""

# Create the parser
chunker = RegexpParser(np_grammar)


# A helper to extract chunks from NLTK trees
def tree_to_chunks(tree):
    """Extract (chunk_type, text) pairs from an NLTK parse tree."""
    chunks = []
    for subtree in tree:
        if hasattr(subtree, "label"):
            chunk_type = subtree.label()
            words = " ".join(word for word, _ in subtree.leaves())
            chunks.append((chunk_type, words))
    return chunks


# Test on several sentences
test_sentences = [
    "The quick brown fox jumps over the lazy dog.",
    "A beautiful sunset illuminated the entire valley.",
    "My old computer finally crashed yesterday.",
    "Scientists discovered a new species in the deep ocean.",
]

regex_results = []
for sent in test_sentences:
    tokens = word_tokenize(sent)
    tagged = pos_tag(tokens)
    tree = chunker.parse(tagged)
    chunks = tree_to_chunks(tree)
    regex_results.append((sent, tagged, chunks))

# Breakdown of the NP pattern components
np_pattern_breakdown = {
    "<DT|PRP\\$>?": "Optional: determiner (the, a, an) OR possessive pronoun (my, your, its)",
    "<JJ>*": "Zero or more adjectives (modifiers before the noun)",
    "<NN.*>+": "One or more nouns: NN (singular), NNS (plural), NNP (proper), NNPS (proper plural)",
}

regex_quantifiers = {
    "?": "zero or one (optional element)",
    "*": "zero or more (any number)",
    "+": "one or more (at least one required)",
    "|": "alternation (either/or)",
}

# Chinking grammar:
# Step 1: {<.*>+} chunks every token into one big NP
# Step 2: }<VB.*|IN|CC|\\.>{ removes (chinks) verbs, prepositions,
#         conjunctions, and punctuation from chunks
chink_grammar = r"""
    NP: {<.*>+}
        }<VB.*|IN|CC|\.>{
"""

chink_chunker = RegexpParser(chink_grammar)

chink_sentences = [
    "The cat and the dog sat on the mat.",
    "Alice ran quickly through the dark forest.",
]

chink_results = []
for sent in chink_sentences:
    tokens = word_tokenize(sent)
    tagged = pos_tag(tokens)
    tree = chink_chunker.parse(tagged)
    chunks = tree_to_chunks(tree)
    chink_results.append((sent, tagged, chunks))

# Cases where regex chunking struggles
hard_cases = [
    {
        "sentence": "The man I saw yesterday left.",
        "issue": "Relative clause ('I saw yesterday') interrupts the NP",
        "expected_np": "The man",
        "regex_np": "The man I saw",  # incorrectly groups relative clause
    },
    {
        "sentence": "The old man the boats.",
        "issue": "Garden path: 'man' is a verb here, not a noun",
        "expected_vp": "man",
        "regex_problem": "POS tagger likely tags 'man' as NN",
    },
    {
        "sentence": "Flying planes can be dangerous.",
        "issue": "Attachment ambiguity: 'flying' modifies 'planes' or starts a VP",
        "expected": "Ambiguous without context",
    },
]

hard_results = []
for case in hard_cases:
    tokens = word_tokenize(case["sentence"])
    tagged = pos_tag(tokens)
    tree = chunker.parse(tagged)
    found_chunks = tree_to_chunks(tree)
    hard_results.append((case, tagged, found_chunks))

from nltk.corpus import conll2000

# Load training and test data
train_sents = conll2000.chunked_sents(
    "train.txt", chunk_types=["NP", "VP", "PP"]
)
test_sents = conll2000.chunked_sents("test.txt", chunk_types=["NP", "VP", "PP"])

# Examine the first sentence
sample_sent = train_sents[0]

def tree_to_iob_triples(tree):
    """Convert an NLTK chunk tree to (word, pos, iob) triples."""
    triples = []
    for subtree in tree:
        if hasattr(subtree, "label"):
            chunk_type = subtree.label()
            for i, (word, pos) in enumerate(subtree.leaves()):
                iob = f"B-{chunk_type}" if i == 0 else f"I-{chunk_type}"
                triples.append((word, pos, iob))
        else:
            # Single token outside any chunk
            word, pos = subtree
            triples.append((word, pos, "O"))
    return triples


# Convert the sample sentence
sample_iob = tree_to_iob_triples(sample_sent)

def evaluate_chunker(chunker_obj, test_trees, max_sents=500):
    """
    Evaluate a chunker at the chunk level.
    Returns precision, recall, F1, and counts.
    """
    tp = fp = fn = 0

    for tree in test_trees[:max_sents]:
        # Gold chunks: set of (type, word_tuple) pairs
        gold = set()
        for subtree in tree:
            if hasattr(subtree, "label"):
                words = tuple(w for w, _ in subtree.leaves())
                gold.add((subtree.label(), words))

        # Predicted chunks
        tagged = list(tree.leaves())
        pred_tree = chunker_obj.parse(tagged)
        pred = set()
        for subtree in pred_tree:
            if hasattr(subtree, "label"):
                words = tuple(w for w, _ in subtree.leaves())
                pred.add((subtree.label(), words))

        tp += len(gold & pred)
        fp += len(pred - gold)
        fn += len(gold - pred)

    precision = tp / (tp + fp) if (tp + fp) > 0 else 0.0
    recall = tp / (tp + fn) if (tp + fn) > 0 else 0.0
    f1 = (
        2 * precision * recall / (precision + recall)
        if (precision + recall) > 0
        else 0.0
    )

    return {
        "precision": precision,
        "recall": recall,
        "f1": f1,
        "tp": tp,
        "fp": fp,
        "fn": fn,
    }


# Evaluate the regex chunker defined earlier
regex_eval = evaluate_chunker(chunker, test_sents)

from nltk.chunk import ChunkParserI
from nltk.tag import BigramTagger, UnigramTagger


class UnigramChunker(ChunkParserI):
    """
    Chunker using a unigram model over (POS -> IOB) mappings.
    For each POS tag, predicts the most frequent IOB label seen in training.
    """

    def __init__(self, train_sents):
        # Convert each tree to (pos, iob) pairs for tagger training
        train_data = []
        for tree in train_sents:
            triples = tree_to_iob_triples(tree)
            train_data.append([(pos, iob) for _, pos, iob in triples])
        self.tagger = UnigramTagger(train_data)

    def parse(self, tagged_sent):
        """Assign IOB labels to a POS-tagged sentence."""
        pos_tags = [pos for _, pos in tagged_sent]
        iob_preds = [self.tagger.tag([pos])[0][1] or "O" for pos in pos_tags]
        iob_triples = [
            (w, p, iob) for (w, p), iob in zip(tagged_sent, iob_preds)
        ]
        return nltk.chunk.conlltags2tree(iob_triples)


class BigramChunker(ChunkParserI):
    """
    Chunker using a bigram model that conditions on the previous POS tag.
    Falls back to unigram for unseen bigrams.
    """

    def __init__(self, train_sents):
        train_data = []
        for tree in train_sents:
            triples = tree_to_iob_triples(tree)
            train_data.append([(pos, iob) for _, pos, iob in triples])
        unigram = UnigramTagger(train_data)
        self.tagger = BigramTagger(train_data, backoff=unigram)

    def parse(self, tagged_sent):
        pos_tags = [pos for _, pos in tagged_sent]
        tagged_preds = self.tagger.tag(pos_tags)
        iob_preds = [iob if iob else "O" for _, iob in tagged_preds]
        iob_triples = [
            (w, p, iob) for (w, p), iob in zip(tagged_sent, iob_preds)
        ]
        return nltk.chunk.conlltags2tree(iob_triples)


# Train both models
unigram_chunker = UnigramChunker(train_sents)
bigram_chunker = BigramChunker(train_sents)

# Evaluate
unigram_eval = evaluate_chunker(unigram_chunker, test_sents)
bigram_eval = evaluate_chunker(bigram_chunker, test_sents)

def extract_svo_triples(sentence, chunker_obj):
    """
    Extract subject-verb-object triples using chunk sequences.
    Looks for NP-VP-NP patterns in the chunk sequence.
    """
    tokens = word_tokenize(sentence)
    tagged = pos_tag(tokens)
    tree = chunker_obj.parse(tagged)

    ordered_chunks = []
    for subtree in tree:
        if hasattr(subtree, "label"):
            text = " ".join(w for w, _ in subtree.leaves())
            ordered_chunks.append((subtree.label(), text))

    triples = []
    for i in range(len(ordered_chunks) - 2):
        if (
            ordered_chunks[i][0] == "NP"
            and ordered_chunks[i + 1][0] == "VP"
            and ordered_chunks[i + 2][0] == "NP"
        ):
            triples.append(
                {
                    "subject": ordered_chunks[i][1],
                    "verb": ordered_chunks[i + 1][1],
                    "object": ordered_chunks[i + 2][1],
                }
            )

    return triples


# Test on news-style sentences
ie_sentences = [
    "The researchers published a landmark study.",
    "The company acquired its largest competitor.",
    "Scientists identified a new protein in the brain.",
    "The president signed the bill into law.",
]

ie_results = [(s, extract_svo_triples(s, chunker)) for s in ie_sentences]

from collections import Counter


def extract_keywords_via_np_chunking(text, top_n=10):
    """
    Extract noun phrase keywords from text using chunking.
    Returns the most frequent multi-word NPs.
    """
    sentences = nltk.sent_tokenize(text)
    all_nps = []

    for sent in sentences:
        tokens = word_tokenize(sent)
        tagged = pos_tag(tokens)
        tree = chunker.parse(tagged)
        for subtree in tree:
            if hasattr(subtree, "label") and subtree.label() == "NP":
                np_text = " ".join(w.lower() for w, _ in subtree.leaves())
                # Only keep multi-word NPs (more specific)
                if len(subtree.leaves()) > 1:
                    all_nps.append(np_text)

    return Counter(all_nps).most_common(top_n)


# Apply to a sample technology text
tech_text = """
Machine learning has transformed natural language processing research.
Deep learning models achieve remarkable performance on many language tasks.
Large language models like GPT and BERT have redefined the state of the art.
Transformer architectures use attention mechanisms to capture long-range dependencies.
Researchers continue to explore new training methods and model architectures.
Language models learn rich text representations from massive training corpora.
"""

keywords = extract_keywords_via_np_chunking(tech_text)

def find_answer_candidates(question, context_sentence, chunker_obj):
    """
    Identify likely answer spans based on question type and chunking.
    """
    q_lower = question.lower()

    # Map question words to likely answer chunk types
    if q_lower.startswith("who"):
        target_type = "NP"  # Named person = noun phrase
    elif q_lower.startswith("where"):
        target_type = "PP"  # Location = prepositional phrase
    elif q_lower.startswith("what"):
        target_type = "NP"  # Thing = noun phrase
    elif q_lower.startswith("when"):
        target_type = "NP"  # Time expressions are often NPs
    else:
        target_type = "NP"

    tokens = word_tokenize(context_sentence)
    tagged = pos_tag(tokens)
    tree = chunker_obj.parse(tagged)

    candidates = []
    for subtree in tree:
        if hasattr(subtree, "label") and subtree.label() == target_type:
            candidates.append(" ".join(w for w, _ in subtree.leaves()))
    return candidates


# Example QA preprocessing
qa_examples = [
    (
        "Who wrote the novel?",
        "The famous author Ernest Hemingway wrote the novel in 1929.",
    ),
    (
        "What did scientists discover?",
        "Scientists announced a new vaccine for the disease.",
    ),
    (
        "Where did the event occur?",
        "The summit took place in the heart of Geneva.",
    ),
]

qa_results = [
    (q, c, find_answer_candidates(q, c, chunker)) for q, c in qa_examples
]

import spacy

try:
    nlp = spacy.load("en_core_web_sm")
except OSError:
    import subprocess

    subprocess.run(
        ["python", "-m", "spacy", "download", "en_core_web_sm"],
        capture_output=True,
    )
    nlp = spacy.load("en_core_web_sm")

# Example sentences that test different NP structures
spacy_sentences = [
    "The quick brown fox jumped over the lazy dog.",
    "A major breakthrough in artificial intelligence was announced yesterday.",
    "The President of the United States addressed the nation.",
    "Machine learning models trained on large corpora show impressive results.",
]

spacy_results = []
for sent in spacy_sentences:
    doc = nlp(sent)
    chunks = [
        (chunk.text, chunk.root.text, chunk.root.dep_)
        for chunk in doc.noun_chunks
    ]
    spacy_results.append((sent, chunks))