Commit 33b2a18

This example project shows how to implement a simple stateful component to

score docs on semantic poles.

The method here is based on SemAxis from [An et al

2018](https://arxiv.org/abs/1806.05521). The basic idea is that given a set

of word vectors and some seed poles, like "bad-good", it's possible to

calculate reference vectors. The distance of document vectors from those

reference vectors is like a sentiment or polar score of the document. While

not as sophisticated as a trained model, it's easy to test with existing data.

If you use enough poles, you can use the scores as semantic vectors that can

make downstream tasks explainable. This is explored in the SemAxis paper as

well as [Mathew et al 2020](https://arxiv.org/abs/2001.09876), "The Polar

Framework". (Incorporating semantic vectors as features in a spaCy model is

left as an exercise for the reader.)

**Note:** Because the data is hosted on Kaggle, it can't be automatically

downloaded by `spacy project assets`, so you'll have to download it yourself.

See [the assets section of this README](#assets) for the link.

The [`project.yml`](project.yml) defines the data assets required by the

project, as well as the available commands and workflows. For details, see the

[spaCy projects documentation](https://spacy.io/usage/projects).

The following commands are defined by the project. They

can be executed using [`spacy project run [name]`](https://spacy.io/api/cli#project-run).

Commands are only re-run if their inputs have changed.

| Command | Description |

| --- | --- |

| `evaluate` | Check output on sample data |

The following workflows are defined by the project. They

can be executed using [`spacy project run [name]`](https://spacy.io/api/cli#project-run)

and will run the specified commands in order. Commands are only re-run if their

inputs have changed.

| Workflow | Steps |

| --- | --- |

| `all` | `evaluate` |

The following assets are defined by the project. They can

be fetched by running [`spacy project assets`](https://spacy.io/api/cli#project-assets)

in the project directory.

| File | Source | Description |

| --- | --- | --- |

| `assets/IMDB Dataset.csv` | Local | IMDB Review Corpus. Download from [Kaggle](https://www.kaggle.com/lakshmi25npathi/imdb-dataset-of-50k-movie-reviews). |

title: "Polar Component"

description: |

vars:

reviews: "assets/IMDB Dataset.csv"

directories: ["scripts", "assets"]

assets:

- dest: ${vars.reviews}

description: "IMDB Review Corpus. Download from [Kaggle](https://www.kaggle.com/lakshmi25npathi/imdb-dataset-of-50k-movie-reviews)."

workflows:

all:

- evaluate

commands:

- name: "evaluate"

help: "Check output on sample data"

script:

- "python ./scripts/evaluate.py"

deps:

- ${vars.reviews}

import spacy

import polar_pipe

import csv

import textwrap

def create_polar_pipeline():

"""Create a pipeline with a polar component."""

# We only need vectors

nlp = spacy.load(

"en_core_web_md", disable=["tok2vec", "parser", "tagger", "ner", "lemmatizer"]

)

polar = nlp.add_pipe("polar")

# now add some axes

polar.add_axis("science", "magic")

polar.add_axis("serious", "funny")

polar.add_axis("heartwarming", "creepy")

polar.add_axis("bad", "good")

return nlp

def format_review(text, truncate=3):

lines = textwrap.wrap(text)

return '\n'.join(lines[:truncate])

def check_axes(nlp, nn=3):

"""Given a pipeline with a polar component, check the IMDB dataset."""

docs = []

with open("assets/IMDB Dataset.csv") as csvfile:

reader = csv.reader(csvfile)

next(reader)

for row in reader:

docs.append(nlp(row[0]))

# XXX Limit for faster testing iteration

if len(docs) > 10000:

break

# now print the reviews at polar extremes

polar = nlp.get_pipe("polar")

for axis in polar.axes:

docs.sort(key=lambda x: x._.poles[axis.get_key()])

print(f"Top {nn} {axis.neg} reviews:")

print()

for ii in range(nn):

doc = docs[ii]

print(format_review(doc.text))

print()

print()

print(f"Top {nn} {axis.pos} reviews:")

print()

for ii in range(nn):

doc = docs[(ii + 1) * -1]

print(format_review(doc.text))

print()

print("=" * 70)

def main():

nlp = create_polar_pipeline()

check_axes(nlp, 3)

if __name__ == "__main__":

main()

from spacy import Language

from spacy.tokens import Doc

from spacy.pipeline import Pipe

from spacy.vocab import Vocab

from thinc.types import Floats1d, Floats2d

from thinc.api import Model, CosineDistance, get_ops

from dataclasses import dataclass

Doc.set_extension("poles", default={}, force=True)

class Axis:

"""An invididual semantic axis."""

neg: str

pos: str

vector: Floats1d

def get_key(self, sep="-"):

return f"{self.neg}{sep}{self.pos}"

def make_polar_embeddings(

nlp: Language,

name: str,

):

return PolarEmbeddings(

nlp,

name,

)

class PolarEmbeddings(Pipe):

"""PolarEmbeddings let you turn normal word embeddings into embeddings

def __init__(

self,

nlp: Language,

name: str = "polar",

*,

separator: str = "-",

) -> None:

self.nlp = nlp

self.name = name

self.separator = separator

self.ops = get_ops("numpy")

self.cosine = CosineDistance()

self._matrix = None

self.axes = []

self.cfg = {}

def get_average_neighbors(self, query, nn=150):

"""Given a query vector, return the average of the nearest vecs.

# Note that in spaCy pipelines for many languages vectors include case

# variations, so nn should be larger than in the reference paper

seed = self.nlp.vocab[query].vector

vectors = self.nlp.vocab.vectors

qarray = self.ops.asarray2f([seed])

keys, best_rows, scores = vectors.most_similar(qarray, n=nn)

targets = vectors.data[best_rows].squeeze()

return self.ops.xp.mean(targets, axis=0)

def add_axis(self, neg: str, pos: str) -> None:

"""Add a new pole to the pipe. Pass the negative word first."""

notfound = "Anchor word '{}' not found in vocab."

if not self.nlp.vocab.has_vector(neg):

raise KeyError(notfound.format(neg))

if not self.nlp.vocab.has_vector(pos):

raise KeyError(notfound.format(pos))

pv = self.get_average_neighbors(pos)

nv = self.get_average_neighbors(neg)

self.axes.append(Axis(neg, pos, pv - nv))

self._update_axes_matrix()

def _update_axes_matrix(self) -> None:

vecs = [aa.vector for aa in self.axes]

self._matrix = self.ops.asarray2f(vecs).T

def __call__(self, doc: Doc) -> Doc:

docvec = self.ops.xp.expand_dims(doc.vector, axis=1)

docvec = docvec.repeat(len(self.axes), axis=1)

dists = self.cosine.get_similarity(docvec.T, self._matrix.T)

for axis, dist in zip(self.axes, dists):

key = axis.get_key(sep=self.separator)

doc._.poles[key] = float(dist)

return doc