Skip to content

Chatbot

ChatConversation

Bases: SessionFunction

Base implementation of a chat bot component

A chatbot component should
  • handle internal state, including history messages
  • return output for a given input
Source code in libs/kotaemon/kotaemon/chatbot/base.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
class ChatConversation(SessionFunction):
    """Base implementation of a chat bot component

    A chatbot component should:
        - handle internal state, including history messages
        - return output for a given input
    """

    class Config:
        store_result = session_chat_storage

    system_message: str = ""
    bot: BaseChatBot

    def __init__(self, *args, **kwargs):
        self._history: List[BaseMessage] = []
        self._store_result = (
            f"{self.__module__}.{self.__class__.__name__},uninitiated_bot"
        )
        super().__init__(*args, **kwargs)

    def run(self, message: HumanMessage) -> Optional[BaseMessage]:
        """Chat, given a message, return a response

        Args:
            message: The message to respond to

        Returns:
            The response to the message. If None, no response is sent.
        """
        user_message = (
            HumanMessage(content=message) if isinstance(message, str) else message
        )
        self.history.append(user_message)

        output = self.bot(self.history).text
        output_message = None
        if output is not None:
            output_message = AIMessage(content=output)
            self.history.append(output_message)

        return output_message

    def start_session(self):
        self._store_result = self.bot.config.store_result
        super().start_session()
        if not self.history and self.system_message:
            system_message = SystemMessage(content=self.system_message)
            self.history.append(system_message)

    def end_session(self):
        super().end_session()
        self._history = []

    def check_end(
        self,
        history: Optional[List[BaseMessage]] = None,
        user_message: Optional[HumanMessage] = None,
        bot_message: Optional[AIMessage] = None,
    ) -> bool:
        """Check if a conversation should end"""
        if user_message is not None and user_message.content == "":
            return True

        return False

    def terminal_session(self):
        """Create a terminal session"""
        self.start_session()
        print(">> Start chat:")

        while True:
            human = HumanMessage(content=input("Human: "))
            if self.check_end(history=self.history, user_message=human):
                break

            output = self(human)
            if output is None:
                print("AI: <No response>")
            else:
                print("AI:", output.content)

            if self.check_end(history=self.history, bot_message=output):
                break

        self.end_session()

    @property
    def history(self):
        return self._history

    @history.setter
    def history(self, value):
        self._history = value
        self._variablex()

run 

1
run(message)

Chat, given a message, return a response

Parameters:

Name Type Description Default
message HumanMessage

The message to respond to

 required

Returns:

Type Description
Optional[BaseMessage]

The response to the message. If None, no response is sent.
Source code in libs/kotaemon/kotaemon/chatbot/base.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def run(self, message: HumanMessage) -> Optional[BaseMessage]:
    """Chat, given a message, return a response

    Args:
        message: The message to respond to

    Returns:
        The response to the message. If None, no response is sent.
    """
    user_message = (
        HumanMessage(content=message) if isinstance(message, str) else message
    )
    self.history.append(user_message)

    output = self.bot(self.history).text
    output_message = None
    if output is not None:
        output_message = AIMessage(content=output)
        self.history.append(output_message)

    return output_message

check_end 

1
2
3
check_end(
    history=None, user_message=None, bot_message=None
)

Check if a conversation should end

Source code in libs/kotaemon/kotaemon/chatbot/base.py 
75
76
77
78
79
80
81
82
83
84
85
def check_end(
    self,
    history: Optional[List[BaseMessage]] = None,
    user_message: Optional[HumanMessage] = None,
    bot_message: Optional[AIMessage] = None,
) -> bool:
    """Check if a conversation should end"""
    if user_message is not None and user_message.content == "":
        return True

    return False

terminal_session 

1
terminal_session()

Create a terminal session

Source code in libs/kotaemon/kotaemon/chatbot/base.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def terminal_session(self):
    """Create a terminal session"""
    self.start_session()
    print(">> Start chat:")

    while True:
        human = HumanMessage(content=input("Human: "))
        if self.check_end(history=self.history, user_message=human):
            break

        output = self(human)
        if output is None:
            print("AI: <No response>")
        else:
            print("AI:", output.content)

        if self.check_end(history=self.history, bot_message=output):
            break

    self.end_session()

SimpleRespondentChatbot

Bases: BaseChatBot

Simple text respondent chatbot that essentially wraps around a chat LLM

Source code in libs/kotaemon/kotaemon/chatbot/simple_respondent.py 
 5
 6
 7
 8
 9
10
11
class SimpleRespondentChatbot(BaseChatBot):
    """Simple text respondent chatbot that essentially wraps around a chat LLM"""

    llm: ChatLLM

    def _get_message(self) -> str:
        return self.llm(self.history).text