wild_eval_bot.py

# Copyright (c) 2017-present, Moscow Institute of Physics and Technology.
# All rights reserved.
# This source code is licensed under the BSD-style license found in the
# LICENSE file in the root directory of this source tree. An additional grant
# of patent rights can be found in the PATENTS file in the same directory.

from parlai.core.params import ParlaiParser
from parlai.core.agents import Agent
from parlai.core.utils import display_messages
from projects.convai2.models.ftlm.wild_eval_world import ConvAIWorld

import random

class ConvAISampleAgent(Agent):

    def __init__(self, opt, shared=None):
        super().__init__(opt, shared)
        self.id = 'ConvAISampleAgent'
        self.text = 'Nothing to say yet!'
        self.episode_done = False

    def observe(self, observation):
        print('\t' + display_messages([observation]))
        self.observation = observation
        self.episode_done = observation['episode_done']

        if self.episode_done:
            self.text = '/end'
        else:
            self.text = random.choice([
                'I love you!',
                'Wow!',
                'Really?',
                'Nice!',
                'Hi',
                'Hello',
                'This is not very interesting. Let\'s change the subject of the conversation and talk about cats.',
                '/end'])

    def act(self):
        reply = {
            'id': self.getID(),
            'text': self.text,
            'episode_done': self.episode_done
        }
        print('\t' + display_messages([reply]))
        return reply


def main():
    parser = ParlaiParser(True, True)
    ConvAIWorld.add_cmdline_args(parser)
    opt = parser.parse_args()

    agent = ConvAISampleAgent(opt)
    world = ConvAIWorld(opt, [agent])

    while True:
        try:
            world.parley()
        except Exception as e:
            print('Exception: {}'.format(e))

if __name__ == '__main__':
    main()

wild_eval_world.py

# Copyright (c) 2017-present, Moscow Institute of Physics and Technology.
# All rights reserved.
# This source code is licensed under the BSD-style license found in the
# LICENSE file in the root directory of this source tree. An additional grant
# of patent rights can be found in the PATENTS file in the same directory.

from parlai.core.worlds import World
from parlai.core.utils import display_messages
from parlai.core.agents import create_agent_from_shared

import requests
import json
import time


class ConvAIWorld(World):
    """ConvAIWorld provides conversations with participants in the convai.io
    competition.
    This world takes in exactly one agent which will converse with a partner
    over a remote connection.
    For each remote conversation being maintained by this world, a copy of the
    original agent will be instantiated from the original agent's `share()`
    method.
    """

    @staticmethod
    def add_cmdline_args(argparser):
        convai = argparser.add_argument_group('ConvAI Arguments')
        convai.add_argument('-bi', '--bot-id', required=True,
                            help='Id of local bot used to communicate with RouterBot')
        convai.add_argument('-bc', '--bot-capacity', type=int, default=-1,
                            help='The maximum number of open dialogs. Use -1 ' +
                                 'for unlimited number of open dialogs')
        convai.add_argument('-rbu', '--router-bot-url', required=True,
                            help='Url of RouterBot')
        convai.add_argument('-rbpd', '--router-bot-pull-delay', type=int,
                            default=1,
                            help='Delay before new request to RouterBot: minimum 1 sec')

    def __init__(self, opt, agents, shared=None):
        super().__init__(opt, shared)

        if len(agents) != 1:
            raise RuntimeError('Need agent to talk to.')
        self.agent = agents[0]
        self.shared = agents[0].share()
        # Current chat id
        self.curr_chatID = None
        # All active and finished (but not cleared yet) chats
        self.chats = {}
        # Finished chats
        self.finished_chats = set()
        # Pairs of exchanges between remote and local agents (for printing)
        self.last_exchanges = dict()
        # Pool of messages from RouterBot
        self.messages = []
        # Url of RouterBot
        self.router_bot_url = opt['router_bot_url']
        # Delay before new request to RouterBot: minimum 1 sec
        self.router_bot_pull_delay = opt['router_bot_pull_delay']
        if self.router_bot_pull_delay < 1:
            self.router_bot_pull_delay = 1
        # Id of local bot used to communicate with RouterBot
        self.bot_id = opt['bot_id']
        # The maximum number of open dialogs. Use -1 for unlimited number of open dialogs
        self.bot_capacity = opt['bot_capacity']
        # RouterBot url with current bot id
        self.bot_url = self.router_bot_url + self.bot_id

    def _get_updates(self):
        """Make HTTP request to Router Bot for new messages
        :return: list of new messages received since last request
        """
        res = requests.get(self.bot_url + '/getUpdates')
        if res.status_code != 200:
            print(res.text)
            res.raise_for_status()
        return res.json()

    def _send_message(self, observation, chatID):
        """Make HTTP request to Router Bot to post new message
        :param observation: message that will be sent to server
        :param chatID: id of chat
        :return: None
        """
        if self._is_end_of_conversation(observation['text']):
            data = {
                'text': '/end',
                'evaluation': {
                    'quality': 0,
                    'breadth': 0,
                    'engagement': 0
                }
            }
        else:
            data = {
                'text': observation['text'],
                'evaluation': 0
            }
        message = {
            'chat_id': chatID,
            'text': json.dumps(data)
        }

        headers = {
            'Content-Type': 'application/json'
        }

        res = requests.post(self.bot_url + '/sendMessage', json=message, headers=headers)
        if res.status_code != 200:
            print(res.text)
            res.raise_for_status()

    @staticmethod
    def _is_begin_of_conversation(message):
        return message.startswith('/start ')

    @staticmethod
    def _is_end_of_conversation(message):
        return message == '/end'

    @staticmethod
    def _is_skip_response(message):
        return message == ''

    @staticmethod
    def _get_chat_id(message):
        return message['message']['chat']['id']

    @staticmethod
    def _get_message_text(message):
        return message['message']['text']

    @staticmethod
    def _strip_start_message(message):
        return message.replace('/start ', '')

    def _init_chat(self, chatID):
        """Create new chat for new dialog.
        Sets up a new instantiation of the agent so that each chat has its own
        local state.

        :param chatID: chat id
        :return: new instance of your local agent
        """
        agent_info = self.shared

        # Add refs to current world instance and chat id to agent 'opt' parameter
        if 'opt' not in agent_info.keys() or agent_info['opt'] is None:
            agent_info['opt'] = {}
        agent_info['opt']['convai_chatID'] = chatID

        local_agent = create_agent_from_shared(agent_info)
        self.chats[chatID] = local_agent
        return self.chats[chatID]

    def cleanup_finished_chat(self, chatID):
        """Shutdown specified chat.
        :param chatID: chat id
        :return: None
        """
        if chatID in self.finished_chats:
            self.chats.pop(chatID).shutdown()
            self.finished_chats.remove(chatID)

    def cleanup_finished_chats(self):
        """Shutdown all finished chats.
        :return: None
        """
        for chatID in self.finished_chats.copy():
            self.cleanup_finished_chat(chatID)

    def pull_new_messages(self):
        """Requests server for new messages and processes every message.
        If message starts with '/start' then will create new chat and adds message to stack.
        If message has same id as already existing chat then will add to message stack.
        Other messages will be ignored.
        If after processing all messages message stack is still empty then new request to server will be performed.
        :return: None
        """
        print('Waiting for new messages from server...', flush=True)
        while True:
            time.sleep(self.router_bot_pull_delay)
            msgs = self._get_updates()
            if len(msgs) > 0:
                for msg in msgs:
                    print('\nProceed message: %s' % msg)
                    text = self._get_message_text(msg)
                    chatID = self._get_chat_id(msg)

                    if self.chats.get(chatID, None) is not None:
                        print('Message was recognized as part of chat #%s' % chatID)
                        self.messages.append((chatID, text))
                    elif self._is_begin_of_conversation(text):
                        print('Message was recognised as start of new chat #%s' % chatID)
                        if self.bot_capacity == -1 or 0 <= self.bot_capacity > (len(self.chats) - len(self.finished_chats)):
                            self._init_chat(chatID)
                            text = self._strip_start_message(text)
                            self.messages.append((chatID, text))
                            print('New world and agents for chat #%s created.' % chatID)
                        else:
                            print('Cannot start new chat #%s due to bot capacity limit reached.' % chatID)
                    else:
                        print('Message was not recognized as part of any chat. Message skipped.')
                if len(self.messages) > 0:
                    break
                else:
                    print('Waiting for new messages from server...', flush=True)

    def parley(self):
        """Pops next message from stack, gets corresponding chat, agents, world
        and performs communication between agents.
        Result of communication will be send to server.
        If message stack is empty then server will be requested for new messages.
        :return: None
        """
        print('Try to cleanup finished chat before new parley.')
        self.cleanup_finished_chats()

        if len(self.messages) == 0:
            print('Message stack is empty. Try to request new messages from server.')
            self.pull_new_messages()

        print('Pop next message from stack')

        (chatID, text) = self.messages.pop(0)
        episode_done = self._is_end_of_conversation(text)
        local_agent = self.chats.get(chatID, None)

        if local_agent is not None:
            self.curr_chatID = chatID
            msg = {
                'id': 'MasterBot#%s' % chatID,
                'text': text,
                'episode_done': episode_done
            }
            local_agent.observe(msg)
            reply = local_agent.act()
            self.last_exchanges[chatID] = [msg, reply]
            if self._is_end_of_conversation(reply['text']) or reply['episode_done']:
                episode_done = True 

            if self._is_skip_response(reply['text']):
                print('Skip response from agent for chat #%s' % chatID)
            else:
                print('Send response from agent to chat #%s: %s' % (chatID, reply))
                self._send_message(reply, chatID)
        else:
            print('Message was not recognized as part of any chat. Message skipped.')

        if episode_done:
            self.finished_chats.add(chatID)

    def display(self):
        if self.curr_chatID in self.chats.keys():
            return display_messages(self.last_exchanges[self.curr_chatID])
        else:
            return ''

    def shutdown(self):
        for chatID in self.chats.keys():
            self.chats[chatID].shutdown()
            if chatID not in self.finished_chats:
                self._send_message({'text': '/end'}, chatID)

    def get_chats(self):
        return self.chats.keys()

    def get_finished_chats(self):
        return self.finished_chats