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
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