Question-Answering Chatbot via Transformer

Question-Answering Chatbot via Transformer - 图1

Susana @ Jina AI

Question-Answering Chatbot via Transformer - 图2June 15, 2021

We will use the hello world chatbot for this tutorial. You can find the complete code here and we will go step by step.

At the end of this tutorial, you will have your own chatbot. You will use text as an input and get a text result as output. For this example, we will use a covid dataset. You will understand how every part of this example works and how you can create new apps with different datasets on your own.

Define data and work directories

We can start by creating an empty folder, I’ll call mine tutorial and that’s the name you’ll see through the tutorial but feel free to use whatever you wish.

We will display our results in our browser, so download the static folder from here, and paste it into your tutorial folder. This is only the CSS and HTML files to render our results. We will use a dataset in a .csv format. We’ll use the COVID dataset from Kaggle.

Download it under your tutorial directory:

  1. wget https://static.jina.ai/chatbot/dataset.csv

Create Documents from a csv file

To create a Document in Jina, we do it like this:

  1. doc = Document(content='hello, world!')

In our case, the content of our Document needs to be the dataset we want to use:

  1. from jina.types.document.generators import from_csv
  2. with open('dataset.csv') as fp:
  3.     docs = from_csv(fp, field_resolver={'question': 'text'})

So what happened there? We created a generator of Documents docs, and we used from_csv to load our dataset. We use field_resolver to map the text from our dataset to the Document attributes.

Finally, we can combine the 2 previous steps (loading the dataset into Documents and starting the context) and index like this:

  1. from jina.types.document.generators import from_csv
  2. with flow, open('dataset.csv') as fp:
  3.     flow.index(from_csv(fp, field_resolver={'question': 'text'}))

See Also

from_csv is a function that belongs to the jina.types.document.generators module. Feel free to check it to find more generators.

Important

flow.index will send the data to the /index endpoint. However, both of the added Executors do not have an /index endpoint. In fact, MyTransformer and MyIndexer only expose endpoints /foo and /bar respectively:

  1. class MyTransformer(Executor):
  2.     @requests(on='/foo')
  3.     def foo(self, **kwargs):
  4.         print(f'foo is doing cool stuff: {kwargs}')
  6. class MyIndexer(Executor):
  7.     @requests(on='/bar')
  8.     def bar(self, **kwargs):
  9.         print(f'bar is doing cool stuff: {kwargs}')

This simply means that no endpoint will be triggered by flow.index. Besides, our Executors are dummy and still do not have logic to index data. Later, we will modify Executors so that calling flow.index does indeed store the dataset.

Create Flow

Let’s put the Executors and the Flow together and re-organize our code a little bit. First, we should import everything we need:

  1. import os
  2. import webbrowser
  3. from pathlib import Path
  4. from jina import Flow, Executor, requests
  5. from jina.logging.predefined import default_logger
  6. from jina.types.document.generators import from_csv

Then we should have our main and a tutorial function that contains all the code that we’ve done so far. tutorial accepts one parameter that we’ll need later: port_expose (the port used to expose our Flow)

  1. def tutorial(port_expose):
  2.     class MyTransformer(Executor):
  3.         @requests(on='/foo')
  4.         def foo(self, **kwargs):
  5.             print(f'foo is doing cool stuff: {kwargs}')
  7.     class MyIndexer(Executor):
  8.         @requests(on='/bar')
  9.         def bar(self, **kwargs):
  10.             print(f'bar is doing cool stuff: {kwargs}')
  12.     flow = (
  13.         Flow()
  14.             .add(name='MyTransformer', uses=MyTransformer)
  15.             .add(name='MyIndexer', uses=MyIndexer)
  16.     )
  17.     with flow, open('dataset.csv') as fp:
  18.         flow.index(from_csv(fp, field_resolver={'question': 'text'}))
  21. if __name__ == '__main__':
  22.     tutorial(8080)

If you run this, it should finish without errors. You won’t see much yet because we are not showing anything after we index.

To actually see something we need to specify how we will display it. For our tutorial we will do so in our browser. After indexing, we will open a web browser to serve the static html files. We also need to configure and serve our Flow on a specific port with the HTTP protocol so that the web browser can make requests to the Flow. So, we’ll use the parameter port_expose to configure the Flow and set the protocol to HTTP. Modify the function tutorial like so:

  1. def tutorial(port_expose):
  2.     class MyTransformer(Executor):
  3.         @requests(on='/foo')
  4.         def foo(self, **kwargs):
  5.             print(f'foo is doing cool stuff: {kwargs}')
  7.     class MyIndexer(Executor):
  8.         @requests(on='/bar')
  9.         def bar(self, **kwargs):
  10.             print(f'bar is doing cool stuff: {kwargs}')
  12.     flow = (
  13.         Flow(cors=True)
  14.             .add(name='MyTransformer', uses=MyTransformer)
  15.             .add(name='MyIndexer', uses=MyIndexer)
  16.     )
  17.     with flow, open('dataset.csv') as fp:
  18.         flow.index(from_csv(fp, field_resolver={'question': 'text'}))
  20.         # switches the serving protocol to HTTP at runtime
  21.         flow.protocol = 'http'
  22.         flow.port_expose = port_expose
  23.         url_html_path = 'file://' + os.path.abspath(
  24.             os.path.join(
  25.                 os.path.dirname(os.path.realpath(__file__)), 'static/index.html'
  26.             )
  27.         )
  28.         try:
  29.             webbrowser.open(url_html_path, new=2)
  30.         except:
  31.             pass  # intentional pass, browser support isn't cross-platform
  32.         finally:
  33.             default_logger.success(
  34.                 f'You should see a demo page opened in your browser, '
  35.                 f'if not, you may open {url_html_path} manually'
  36.             )
  37.         flow.block()

See Also

For more information on what the Flow is doing, and how to serve the Flow with f.block() and configure the protocol, check the Flow fundamentals section.

Important

Since we want to call our Flow from the browser, it’s important to enable Cross-Origin Resource Sharing with Flow(cors=True)

Ok, so it seems that we have plenty of work done already. If you run this you will see a new tab open in your browser, and there you will have a text box ready for you to input some text. However, if you try to enter anything you won’t get any results. This is because we are using dummy Executors. Our MyTransformer and MyIndexer aren’t actually doing anything. So far they only print a line when they are called. So we need real Executors.

Create Executors

We will be creating our Executors in a separate file: my_executors.py.

Sentence Transformer

First, let’s import the following:

  1. from typing import Dict
  3. from jina import Executor, DocumentArray, requests
  4. from jina.types.arrays.memmap import DocumentArrayMemmap
  5. from sentence_transformers import SentenceTransformer

Now, let’s implement MyTransformer:

  1. class MyTransformer(Executor):
  2.     """Transformer executor class """
  4.     def __init__(
  5.             self,
  6.             pretrained_model_name_or_path: str = 'paraphrase-mpnet-base-v2',
  7.             device: str = 'cpu',
  8.             *args,
  9.             **kwargs,
  10.     ):
  11.         super().__init__(*args, **kwargs)
  12.         self.model = SentenceTransformer(pretrained_model_name_or_path, device=device)
  13.         self.model.to(device)
  15.     @requests
  16.     def encode(self, docs: 'DocumentArray', *args, **kwargs):
  17.         import torch
  19.         with torch.no_grad():
  20.             texts = docs.get_attributes("text")
  21.             embeddings = self.model.encode(texts, batch_size=32)
  22.             for doc, embedding in zip(docs, embeddings):
  23.                 doc.embedding = embedding

MyTransformer exposes only one endpoint: encode. This will be called whenever we make a request to the Flow, either on query or index. The endpoint will create embeddings for the indexed or query Documents so that they can be used to get the closed matches.

Note

Encoding is a fundamental concept in neural search. It means representing the data in a vectorial form (embeddings).

Encoding is performed through a sentence-transformers model (paraphrase-mpnet-base-v2 by default). We get the text attributes of docs in batch and then compute embeddings. Later, we set the embedding attribute of each Document.

Simple Indexer

Now, let’s implement our indexer (MyIndexer):

  1. class MyIndexer(Executor):
  2.     """Simple indexer class """
  4.     def __init__(self, **kwargs):
  5.         super().__init__(**kwargs)
  6.         self._docs = DocumentArrayMemmap(self.workspace + '/indexer')
  8.     @requests(on='/index')
  9.     def index(self, docs: 'DocumentArray', **kwargs):
  10.         self._docs.extend(docs)
  12.     @requests(on='/search')
  13.     def search(self, docs: 'DocumentArray', **kwargs):
  14.         """Append best matches to each document in docs
  16.         :param docs: documents that are searched
  17.         :param kwargs: other keyword arguments
  18.         """
  19.         docs.match(self._docs, metric='cosine', normalization=(1, 0), limit=1)

MyIndexer exposes 2 endpoints: index and search. To perform indexing, we use DocumentArrayMemmap which is a Jina data type. Indexing is a simple as adding the Documents to the DocumentArrayMemmap.

See Also

Learn more about DocumentArrayMemmap.

To perform the search operation, we use the method match which will return the top match for the query Documents using the cosine similarity.

See Also

.match is a method of both DocumentArray and DocumentArrayMemmap. Learn more about it in this section.

To import the Executors, just add this to the imports:

  1. from my_executors import MyTransformer, MyIndexer

Put all together

Your app.py should now look like this:

  1. import os
  2. import webbrowser
  3. from pathlib import Path
  4. from jina import Flow, Executor
  5. from jina.logging.predefined import default_logger
  6. from jina.types.document.generators import from_csv
  7. from my_executors import MyTransformer, MyIndexer
  10. def tutorial(port_expose):
  11.     flow = (
  12.         Flow(cors=True)
  13.             .add(name='MyTransformer', uses=MyTransformer)
  14.             .add(name='MyIndexer', uses=MyIndexer)
  15.     )
  16.     with flow, open('dataset.csv') as fp:
  17.         flow.index(from_csv(fp, field_resolver={'question': 'text'}))
  19.         # switch to REST gateway at runtime
  20.         flow.protocol = 'http'
  21.         flow.port_expose = port_expose
  22.         url_html_path = 'file://' + os.path.abspath(
  23.             os.path.join(
  24.                 os.path.dirname(os.path.realpath(__file__)), 'static/index.html'
  25.             )
  26.         )
  27.         try:
  28.             webbrowser.open(url_html_path, new=2)
  29.         except:
  30.             pass  # intentional pass, browser support isn't cross-platform
  31.         finally:
  32.             default_logger.success(
  33.                 f'You should see a demo page opened in your browser, '
  34.                 f'if not, you may open {url_html_path} manually'
  35.             )
  36.         flow.block()
  39. if __name__ == '__main__':
  40.     tutorial(8080)

And your directory should be:

  1. .
  2. └── tutorial                    
  3.     ├── app.py          
  4.     ├── my_executors.py         
  5.     ├── static/         
  6.     ├── our_flow.svg #This will be here if you used the .plot() function       
  7.     └── dataset.csv

And we are done! If you followed all the steps, now you should have something like this in your browser:

../../_images/results.png