Skip to content

chat

Chat

Source code in griptape/utils/chat.py 
14
15
16
17
18
19
20
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
@define
class Chat:
    structure: Structure = field()
    exit_keywords: list[str] = field(default=["exit"], kw_only=True)
    exiting_text: str = field(default="Exiting...", kw_only=True)
    processing_text: str = field(default="Thinking...", kw_only=True)
    intro_text: Optional[str] = field(default=None, kw_only=True)
    prompt_prefix: str = field(default="User: ", kw_only=True)
    response_prefix: str = field(default="Assistant: ", kw_only=True)
    output_fn: Callable[[str], None] = field(
        default=Factory(lambda self: self.default_output_fn, takes_self=True),
        kw_only=True,
    )
    logger_level: int = field(default=logging.ERROR, kw_only=True)

    def default_output_fn(self, text: str) -> None:
        from griptape.tasks.prompt_task import PromptTask

        streaming_tasks = [
            task for task in self.structure.tasks if isinstance(task, PromptTask) and task.prompt_driver.stream
        ]
        if streaming_tasks:
            print(text, end="", flush=True)  # noqa: T201
        else:
            print(text)  # noqa: T201

    def start(self) -> None:
        from griptape.configs import Defaults

        # Hide Griptape's logging output except for errors
        old_logger_level = logging.getLogger(Defaults.logging_config.logger_name).getEffectiveLevel()
        logging.getLogger(Defaults.logging_config.logger_name).setLevel(self.logger_level)

        if self.intro_text:
            self.output_fn(self.intro_text)
        while True:
            question = input(self.prompt_prefix)

            if question.lower() in self.exit_keywords:
                self.output_fn(self.exiting_text)
                break

            if Defaults.drivers_config.prompt_driver.stream:
                self.output_fn(self.processing_text + "\n")
                stream = Stream(self.structure).run(question)
                first_chunk = next(stream)
                self.output_fn(self.response_prefix + first_chunk.value)
                for chunk in stream:
                    self.output_fn(chunk.value)
            else:
                self.output_fn(self.processing_text)
                self.output_fn(f"{self.response_prefix}{self.structure.run(question).output_task.output.to_text()}")

        # Restore the original logger level
        logging.getLogger(Defaults.logging_config.logger_name).setLevel(old_logger_level)

exit_keywords: list[str] = field(default=['exit'], kw_only=True) class-attribute instance-attribute

exiting_text: str = field(default='Exiting...', kw_only=True) class-attribute instance-attribute

intro_text: Optional[str] = field(default=None, kw_only=True) class-attribute instance-attribute

logger_level: int = field(default=logging.ERROR, kw_only=True) class-attribute instance-attribute

output_fn: Callable[[str], None] = field(default=Factory(lambda self: self.default_output_fn, takes_self=True), kw_only=True) class-attribute instance-attribute

processing_text: str = field(default='Thinking...', kw_only=True) class-attribute instance-attribute

prompt_prefix: str = field(default='User: ', kw_only=True) class-attribute instance-attribute

response_prefix: str = field(default='Assistant: ', kw_only=True) class-attribute instance-attribute

structure: Structure = field() class-attribute instance-attribute

default_output_fn(text)

Source code in griptape/utils/chat.py 
29
30
31
32
33
34
35
36
37
38
def default_output_fn(self, text: str) -> None:
    from griptape.tasks.prompt_task import PromptTask

    streaming_tasks = [
        task for task in self.structure.tasks if isinstance(task, PromptTask) and task.prompt_driver.stream
    ]
    if streaming_tasks:
        print(text, end="", flush=True)  # noqa: T201
    else:
        print(text)  # noqa: T201

start()

Source code in griptape/utils/chat.py 
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
def start(self) -> None:
    from griptape.configs import Defaults

    # Hide Griptape's logging output except for errors
    old_logger_level = logging.getLogger(Defaults.logging_config.logger_name).getEffectiveLevel()
    logging.getLogger(Defaults.logging_config.logger_name).setLevel(self.logger_level)

    if self.intro_text:
        self.output_fn(self.intro_text)
    while True:
        question = input(self.prompt_prefix)

        if question.lower() in self.exit_keywords:
            self.output_fn(self.exiting_text)
            break

        if Defaults.drivers_config.prompt_driver.stream:
            self.output_fn(self.processing_text + "\n")
            stream = Stream(self.structure).run(question)
            first_chunk = next(stream)
            self.output_fn(self.response_prefix + first_chunk.value)
            for chunk in stream:
                self.output_fn(chunk.value)
        else:
            self.output_fn(self.processing_text)
            self.output_fn(f"{self.response_prefix}{self.structure.run(question).output_task.output.to_text()}")

    # Restore the original logger level
    logging.getLogger(Defaults.logging_config.logger_name).setLevel(old_logger_level)