Merge pull request #90 from deepmipt/feat/settings.py

Feat/settings.py

deeppavlov_agent/core/cmd_client.py → deeppavlov_agent/cmd_client.py

@@ -1,6 +1,10 @@
+import argparse
 import asyncio
+
 from aioconsole import ainput
 
+from .setup_agent import setup_agent
+
 
 async def message_processor(register_msg):
     user_id = await ainput('Provide user id: ')
@@ -14,7 +18,8 @@ async def message_processor(register_msg):
             print('Bot: ', response['dialog'].utterances[-1].text)
 
 
-def run_cmd(agent, session, workers, debug):
+def run_cmd(pipeline_configs, debug):
+    agent, session, workers = setup_agent(pipeline_configs=pipeline_configs)
     loop = asyncio.get_event_loop()
     loop.set_debug(debug)
     future = asyncio.ensure_future(message_processor(agent.register_msg))
@@ -32,3 +37,13 @@ def run_cmd(agent, session, workers, debug):
             loop.run_until_complete(session.close())
         loop.stop()
         loop.close()
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser()
+    parser.add_argument('-pl', '--pipeline_configs', help='Pipeline config (overwrite value, defined in settings)',
+                        type=str, action='append')
+    parser.add_argument('-d', '--debug', help='run in debug mode', action='store_true')
+    args = parser.parse_args()
+
+    run_cmd(args.pipeline_configs, args.debug)

deeppavlov_agent/core/agent.py

@@ -1,6 +1,6 @@
 import asyncio
 from time import time
-from typing import Any, Hashable
+from typing import Any
 
 from .log import BaseResponseLogger
 from .pipeline import Pipeline
@@ -27,11 +27,9 @@ def flush_record(self, dialog_id: str):
             workflow_record['timeout_response_task'].cancel()
         return workflow_record
 
-    async def register_msg(self, utterance: str, user_telegram_id: Hashable,
-                           user_device_type: Any, location: Any,
-                           channel_type: str, deadline_timestamp=None,
+    async def register_msg(self, utterance, deadline_timestamp=None,
                            require_response=False, **kwargs):
-        dialog = await self.state_manager.get_or_create_dialog_by_tg_id(user_telegram_id, channel_type)
+        dialog = await self.state_manager.get_or_create_dialog(**kwargs)
         dialog_id = str(dialog.id)
         service = self.pipeline.get_service_by_name('input')
         message_attrs = kwargs.pop('message_attrs', {})

deeppavlov_agent/core/connectors.py

@@ -77,12 +77,18 @@ def glue_tasks(self, batch):
 
 class ConfidenceResponseSelectorConnector:
     async def send(self, payload: Dict, callback: Callable):
-        response = payload['payload']['hypotheses']
-        best_skill = sorted(response, key=lambda x: x['confidence'], reverse=True)[0]
-        await callback(
-            task_id=payload['task_id'],
-            response=best_skill
-        )
+        try:
+            response = payload['payload']['utterances'][-1]['hypotheses']
+            best_skill = max(response, key=lambda x: x['confidence'])
+            await callback(
+                task_id=payload['task_id'],
+                response=best_skill
+            )
+        except Exception as e:
+            await callback(
+                task_id=payload['task_id'],
+                response=e
+            )
 
 
 class EventSetOutputConnector:
@@ -148,3 +154,14 @@ async def send(self, payload: Dict, callback: Callable):
             task_id=payload['task_id'],
             response={'text': self.response_text, 'annotations': self.annotations}
         )
+
+
+class PredefinedOutputConnector:
+    def __init__(self, output):
+        self.output = output
+
+    async def send(self, payload: Dict, callback: Callable):
+        await callback(
+            task_id=payload['task_id'],
+            response=self.output
+        )

deeppavlov_agent/core/state_manager.py

@@ -21,6 +21,11 @@ async def add_hypothesis(self, dialog: Dialog, payload: Dict, label: str, **kwar
     async def add_annotation(self, dialog: Dialog, payload: Dict, label: str, **kwargs):
         dialog.utterances[-1].annotations[label] = payload
 
+    async def add_annotation_prev_bot_utt(self, dialog: Dialog, payload: Dict, label: str, **kwargs):
+        if len(dialog.utterances) > 1:
+            dialog.utterances[-2].annotations[label] = payload
+            dialog.utterances[-2].actual = False
+
     async def add_hypothesis_annotation(self, dialog: Dialog, payload: Dict, label: str, **kwargs):
         ind = kwargs['ind']
         dialog.utterances[-1].hypotheses[ind]['annotations'][label] = payload
@@ -75,7 +80,7 @@ async def add_failure_bot_utterance(self, dialog: Dialog, payload: Dict, label:
     async def save_dialog(self, dialog: Dialog, payload: Dict, label: str, **kwargs) -> None:
         await dialog.save(self._db)
 
-    async def get_or_create_dialog_by_tg_id(self, user_telegram_id, channel_type):
+    async def get_or_create_dialog(self, user_telegram_id, channel_type, **kwargs):
         return await Dialog.get_or_create_by_ext_id(self._db, user_telegram_id, channel_type)
 
     async def get_dialog_by_id(self, dialog_id):

deeppavlov_agent/http_api/__init__.py

@@ -0,0 +1,27 @@
+# This module implements http api, which allows to communicate with Agent
+
+from .api import init_app
+
+from ..settings import (
+    TIME_LIMIT, OUTPUT_FORMATTER, DEBUG_OUTPUT_FORMATTER, DEBUG, RESPONSE_LOGGER
+)
+
+from ..setup_agent import setup_agent
+from ..core.log import LocalResponseLogger
+
+
+def app_factory(pipeline_configs=None, debug=None, response_time_limit=None):
+    agent, session, workers = setup_agent(pipeline_configs)
+    response_logger = LocalResponseLogger(RESPONSE_LOGGER)
+    if DEBUG:
+        output_formatter = DEBUG_OUTPUT_FORMATTER
+    else:
+        output_formatter = OUTPUT_FORMATTER
+
+    app = init_app(
+        agent=agent, session=session, consumers=workers,
+        logger_stats=response_logger, output_formatter=output_formatter,
+        debug=debug or DEBUG, response_time_limit=response_time_limit or TIME_LIMIT
+    )
+
+    return app

deeppavlov_agent/http_api/api.py

@@ -4,14 +4,16 @@
 import jinja2
 from aiohttp import web
 
-from .handlers import ApiHandler, PagesHandler, WSstatsHandler
+from .handlers import ApiHandler, PagesHandler, WSstatsHandler, WSChatHandler
 
 
-async def init_app(agent, session, consumers, logger_stats, debug=False, response_time_limit=0):
+async def init_app(agent, session, consumers, logger_stats, output_formatter,
+                   debug=False, response_time_limit=0):
     app = web.Application()
-    handler = ApiHandler(debug, response_time_limit)
+    handler = ApiHandler(output_formatter, response_time_limit)
     pages = PagesHandler(debug)
     stats = WSstatsHandler()
+    chat = WSChatHandler(output_formatter)
     consumers = [asyncio.ensure_future(i.call_service(agent.process)) for i in consumers]
 
     async def on_startup(app):
@@ -36,7 +38,9 @@ async def on_shutdown(app):
     app.router.add_get('/api/user/{user_telegram_id}', handler.dialogs_by_user)
     app.router.add_get('/ping', pages.ping)
     app.router.add_get('/debug/current_load', stats.ws_page)
-    app.router.add_get('/debug/current_load/ws', stats.index)
+    app.router.add_get('/debug/current_load/ws', stats.ws_handler)
+    app.router.add_get('/chat', chat.ws_page)
+    app.router.add_get('/chat/ws', chat.ws_handler)
     app.on_startup.append(on_startup)
     app.on_shutdown.append(on_shutdown)
     aiohttp_jinja2.setup(app, loader=jinja2.PackageLoader('deeppavlov_agent.http_api', 'templates'))

deeppavlov_agent/http_api/handlers.py

@@ -3,24 +3,23 @@
 from string import hexdigits
 from time import time
 
+import aiohttp
 import aiohttp_jinja2
 from aiohttp import web
 
-from ..state_formatters.output_formatters import (http_api_output_formatter,
-                                                  http_debug_output_formatter)
+
+async def handle_command(payload, user_id, state_manager):
+    if payload in {'/start', '/close'} and state_manager:
+        await state_manager.drop_active_dialog(user_id)
+        return True
 
 
 class ApiHandler:
-    def __init__(self, debug=False, response_time_limit=5):
-        self.debug = debug
+    def __init__(self, output_formatter, response_time_limit=5):
+        self.output_formatter = output_formatter
         self.response_time_limit = response_time_limit
 
     async def handle_api_request(self, request):
-        async def handle_command(payload, user_id, state_manager):
-            if payload in {'/start', '/close'} and state_manager:
-                await state_manager.drop_active_dialog(user_id)
-                return True
-
         response = {}
         register_msg = request.app['agent'].register_msg
         if request.method == 'POST':
@@ -55,10 +54,7 @@ async def handle_command(payload, user_id, state_manager):
 
             if response is None:
                 raise RuntimeError('Got None instead of a bot response.')
-            if self.debug:
-                return web.json_response(http_debug_output_formatter(response['dialog'].to_dict()))
-            else:
-                return web.json_response(http_api_output_formatter(response['dialog'].to_dict()))
+            return web.json_response(self.output_formatter(response['dialog'].to_dict()))
 
     async def dialog(self, request):
         state_manager = request.app['agent'].state_manager
@@ -93,7 +89,7 @@ def __init__(self):
     async def ws_page(self, request):
         return {}
 
-    async def index(self, request):
+    async def ws_handler(self, request):
         ws = web.WebSocketResponse()
         await ws.prepare(request)
         request.app['websockets'].append(ws)
@@ -104,3 +100,48 @@ async def index(self, request):
             await asyncio.sleep(self.update_time)
 
         return ws
+
+
+class WSChatHandler:
+    def __init__(self, output_formatter):
+        self.output_formatter = output_formatter
+
+    @aiohttp_jinja2.template('chat.html')
+    async def ws_page(self, request):
+        return {}
+
+    async def ws_handler(self, request):
+        register_msg = request.app['agent'].register_msg
+        ws = web.WebSocketResponse()
+        await ws.prepare(request)
+        while True:
+            msg = await ws.receive()
+            if msg.type == aiohttp.WSMsgType.text:
+                data = msg.json()
+                user_id = data.pop('user_id')
+                payload = data.pop('payload', '')
+                deadline_timestamp = None
+                if not user_id:
+                    raise web.HTTPBadRequest(reason='user_id key is required')
+                command_performed = await handle_command(payload, user_id, request.app['agent'].state_manager)
+                if command_performed:
+                    await ws.send_json('command_performed')
+                    continue
+
+                response = await register_msg(
+                    utterance=payload, user_telegram_id=user_id,
+                    user_device_type=data.pop('user_device_type', 'websocket'),
+                    date_time=datetime.now(),
+                    location=data.pop('location', ''),
+                    channel_type='ws_client',
+                    message_attrs=data, require_response=True,
+                    deadline_timestamp=deadline_timestamp
+                )
+                if response is None:
+                    raise RuntimeError('Got None instead of a bot response.')
+                await ws.send_json(self.output_formatter(response['dialog'].to_dict()))
+            else:
+                await ws.close()
+                break
+
+        return ws

deeppavlov_agent/http_api/templates/chat.html

@@ -0,0 +1,119 @@
+{% extends "base.html" %}
+{% block title %}Dialogs list{% endblock %}
+{% block head %}
+{{ super() }}
+  <link rel="stylesheet" href="https://maxcdn.bootstrapcdn.com/bootstrap/4.0.0/css/bootstrap.min.css" integrity="sha384-Gn5384xqQ1aoWXA+058RXPxPg6fy4IWvTNh0E263XmFcJlSAwiGgFAW/dAiS6JXm" crossorigin="anonymous">
+  <script src="https://code.jquery.com/jquery-3.2.1.slim.min.js" integrity="sha384-KJ3o2DKtIkvYIK3UENzmM7KCkRr/rE9/Qpg6aAZGJwFDMVNA/GpGFF93hXpG5KkN" crossorigin="anonymous"></script>
+  <script src="https://cdnjs.cloudflare.com/ajax/libs/popper.js/1.12.9/umd/popper.min.js" integrity="sha384-ApNbgh9B+Y1QKtv3Rn7W3mgPxhU9K/ScQsAP7hUibX39j7fakFPskvXusvfa0b4Q" crossorigin="anonymous"></script>
+  <script src="https://maxcdn.bootstrapcdn.com/bootstrap/4.0.0/js/bootstrap.min.js" integrity="sha384-JZR6Spejh4U02d8jOt6vLEHfe/JQGiRRSQQxSfFWpi1MquVdAyjUar5+76PVCmYl" crossorigin="anonymous"></script>
+{% endblock %}
+{% block content %}
+<div class="input-group mb-3">
+  <div class="input-group-prepend">
+    <span class="input-group-text" id="basic-addon1">User Id</span>
+  </div>
+  <input type="text" class="form-control" placeholder="User Id" aria-label="User_id" aria-describedby="basic-addon1" id="user_id">
+</div>
+<div class="input-group mb-3">
+  <div class="input-group-prepend">
+    <span class="input-group-text" id="basic-addon2">Text</span>
+  </div>
+  <input type="text" class="form-control" placeholder="Text" aria-label="Text" aria-describedby="basic-addon2" id="payload">
+  <button id="send", type="submit" class="btn btn-primary">Send</button>
+</div>
+
+<div id="dialog">
+</div>
+
+<script type="text/javascript">
+    var socket = new WebSocket('ws://' + window.location.host + '/chat/ws');
+
+    socket.onopen = function() {
+        update_ui();
+    };
+
+    socket.onclose = function() {
+      update_ui();
+      alert('service is unavailable');
+    };
+
+    function update_ui() {
+      if (socket == null) {
+        $('#send').prop("disabled", true);
+      } else {
+        $('#send').prop("disabled", false);
+      }
+    };
+
+    $('#send').on('click', function() {
+      var payload = $('#payload').val();
+      var user_id = $('#user_id').val();
+
+      socket.send(JSON.stringify({"user_id": user_id, "payload": payload}));
+      $('#payload').val('').focus();
+      update_table(payload, user_id);
+      $('#send').prop("disabled", true);
+      $('#user_id').prop("readonly", true);
+      return false;
+    });
+
+    function update_table(text, id, active_skill, debug) {
+        var dialog = document.getElementById('dialog');
+        var index = dialog.childElementCount;
+        var message = id;
+        var collapse = document.createElement('div');
+        collapse.setAttribute("id", "collapse" + index);
+        collapse.setAttribute("class", "collapse");
+        collapse.setAttribute("data-parent", "#dialog");
+        collapse.setAttribute("aria-labelledby", "heading" + index);
+        if (active_skill == undefined) {
+            message = message + ': ' +  text;
+            active_skill = '';
+        } else {
+            message = message + ' (' + active_skill + '): ' +  text;
+        }
+
+        if (debug != undefined) {
+            var cardBody = document.createElement('div');
+            cardBody.setAttribute("class", "card-body");
+            cardBody.innerHTML = JSON.stringify(debug, undefined, 2);
+            collapse.append(cardBody);
+
+        }
+
+        var card = document.createElement('div');
+        card.setAttribute("class", "card");
+
+        var header = document.createElement('div');
+        header.setAttribute("class", "card-header");
+        header.setAttribute("id", "heading" + index);
+        var h = document.createElement('h5');
+        h.setAttribute("class", "mb-0");
+
+        var button = document.createElement('button');
+        button.setAttribute("class", "btn btn-link collapsed");
+        button.setAttribute("data-toggle", "collapse");
+        button.setAttribute("data-target", "#collapse" + index);
+        button.setAttribute("aria-expanded", "true");
+        button.setAttribute("aria-controls", "collapse" + index);
+        button.innerHTML = message;
+
+        h.append(button);
+        header.append(h);
+
+        card.append(header);
+        if (collapse.childElementCount == 1) {
+          card.append(collapse);
+        }
+
+        dialog.prepend(card);
+        $('#send').prop("disabled", false);
+    }
+
+    socket.onmessage = function(event) {
+        data = JSON.parse(event.data);
+        update_table(data.response, 'bot', data.active_skill, data.debug_output)
+    };
+
+</script>
+{% endblock %}