A interactive chatbot that performs NER on a text and specified entity type and provides a wikipedia summary for each identified entity.
from pathlib import Path

from chainlit.cli import run_chainlit


def run():
    """Python entrypoint to run chainlit server.

    Used to provide a script via poetry.

    """
    run_chainlit(Path(Path(__file__).parent, "server.py").as_posix())

wikilinker/cli.py
sebastianschramm/wikilinker
import json

from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline


def prepare_model_pipeline(model_name: str) -> pipeline:
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    model = AutoModelForCausalLM.from_pretrained(model_name, device_map="auto")

    ner_gen = pipeline(
        "text-generation",
        model=model,
        tokenizer=tokenizer,
        do_sample=False,
        max_new_tokens=100,
        return_full_text=False,
    )
    return ner_gen


def get_prompt_template():
    pt = """A virtual assistant answers questions from
         a user based on the provided text.
         USER: Text: {input_text}
         ASSISTANT: I’ve read this text.
         USER: What describes {entity_name} in the text?
         ASSISTANT:
         """
    return pt


def get_ner(input_text: str, entity_type: str, gen_ner):
    prompt = get_prompt_template().format_map(
        {"input_text": input_text, "entity_name": entity_type.lower()}
    )
    try:
        ner_list = json.loads(gen_ner(prompt)[0]["generated_text"])
    except Exception:
        ner_list = []
    return ner_list

wikilinker/ner.py
sebastianschramm/wikilinker
from typing import List

import chainlit as cl

from wikilinker.ner import get_ner, prepare_model_pipeline
from wikilinker.wiki import get_wiki_summary

AUTHOR_LINKER = "WIKILINKER"
AUTHOR_NER = "NER"
AUTHOR_USER = "User"
NER_MODEL_NAME = "SebastianSchramm/UniNER-7B-type-GPTQ-4bit-128g-actorder_True"


async def create_avatars():
    await cl.Avatar(
        name=AUTHOR_LINKER,
        path="avatars/linker.png",
    ).send()
    await cl.Avatar(
        name=AUTHOR_NER,
        path="avatars/llama.png",
    ).send()
    await cl.Avatar(
        name=AUTHOR_USER,
        path="avatars/user.png",
    ).send()


async def display_summaries(named_entities: List[str]):
    for entity in named_entities:
        await cl.Text(
            name=entity, content=get_wiki_summary(entity), display="side"
        ).send()


async def display_ner(entity_type: str, named_entities: List[str], parent_message_id):
    await cl.Message(
        content=f"Recognized the following entities of type {entity_type}:"
        f"\n\t{named_entities}",
        parent_id=parent_message_id,
        author=AUTHOR_NER,
    ).send()


async def process_message(message: dict, entity_type: str):
    named_entities = get_ner(message, entity_type=entity_type, gen_ner=gen_ner)
    await display_ner(
        entity_type=entity_type,
        named_entities=named_entities,
        parent_message_id=message["id"],
    )
    await display_summaries(named_entities=named_entities)


gen_ner = prepare_model_pipeline(NER_MODEL_NAME)


@cl.on_chat_start
async def main():
    await create_avatars()

    entity_type = await cl.AskUserMessage(
        content="Which entity type do you want to link?",
        timeout=30,
        author=AUTHOR_LINKER,
    ).send()
    if entity_type:
        message = await cl.AskUserMessage(
            content="Which text do you want to process?",
            timeout=30,
            author=AUTHOR_LINKER,
        ).send()
        if message:
            await process_message(message=message, entity_type=entity_type["content"])

wikilinker/server.py
sebastianschramm/wikilinker
import requests


def get_wiki_summary(title: str):
    backup_response = "No details available"
    url = (
        f"https://en.wikipedia.org/w/api.php?action=query&prop=extracts"
        f"&exsentences=5&exsectionformat=plain&explaintext=&titles={title}&format=json"
    )
    response = requests.get(url, headers={"Content-Type": "application/JSON"})
    if response.status_code == 200:
        try:
            response_obj = response.json()
            return "\n\n".join(
                [
                    f"Wikipedia page ID {key}:\n\n{item['extract']}"
                    for key, item in response_obj["query"]["pages"].items()
                ]
            )
        except Exception:
            return backup_response
    return backup_response

wikilinker/wiki.py
sebastianschramm/wikilinker