Skip to content

ConsoleUI

fastagency.ui.console.ConsoleUI #

ConsoleUI(super_conversation: Optional[ConsoleUI] = None)

Bases: IOMessageVisitor

Initialize the console UI object.

PARAMETER DESCRIPTION
super_conversation

The super conversation. Defaults to None.

TYPE: Optional[UI] DEFAULT: None

Source code in fastagency/ui/console/base.py 
33
34
35
36
37
38
39
40
41
42
43
def __init__(
    self,
    super_conversation: Optional["ConsoleUI"] = None,
) -> None:
    """Initialize the console UI object.

    Args:
        super_conversation (Optional[UI], optional): The super conversation. Defaults to None.
    """
    self.super_conversation: Optional[ConsoleUI] = super_conversation
    self.sub_conversations: list[ConsoleUI] = []

level property #

level: int

sub_conversations instance-attribute #

sub_conversations: list[ConsoleUI] = []

super_conversation instance-attribute #

super_conversation: Optional[ConsoleUI] = super_conversation

ConsoleMessage dataclass #

ConsoleMessage(
    sender: Optional[str],
    recipient: Optional[str],
    heading: Optional[str],
    body: Optional[str],
)

A console message.

body instance-attribute #

body: Optional[str]

heading instance-attribute #

heading: Optional[str]

recipient instance-attribute #

recipient: Optional[str]

sender instance-attribute #

sender: Optional[str]

create #

create(app: Runnable, import_string: str) -> Iterator[None]
Source code in fastagency/ui/console/base.py 
45
46
47
@contextmanager
def create(self, app: Runnable, import_string: str) -> Iterator[None]:
    yield

create_subconversation #

create_subconversation() -> ConsoleUI
Source code in fastagency/ui/console/base.py 
172
173
174
175
176
def create_subconversation(self) -> "ConsoleUI":
    sub_conversation = ConsoleUI(self)
    self.sub_conversations.append(sub_conversation)

    return sub_conversation

process_message #

process_message(message: IOMessage) -> Optional[str]
Source code in fastagency/ui/console/base.py 
165
166
167
def process_message(self, message: IOMessage) -> Optional[str]:
    # logger.info(f"process_message(): {message=}")
    return self.visit(message)

start #

start(
    *,
    app: Runnable,
    import_string: str,
    name: Optional[str] = None,
    initial_message: Optional[str] = None,
    single_run: bool = False
) -> None
Source code in fastagency/ui/console/base.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def start(
    self,
    *,
    app: Runnable,
    import_string: str,
    name: Optional[str] = None,
    initial_message: Optional[str] = None,
    single_run: bool = False,
) -> None:
    run_workflow(
        wf=app.wf,
        ui=self,
        name=name,
        initial_message=initial_message,
        single_run=single_run,
    )

visit #

visit(message: IOMessage) -> Optional[str]
Source code in fastagency/base.py 
173
174
175
176
def visit(self, message: IOMessage) -> Optional[str]:
    method_name = f"visit_{message.type}"
    method = getattr(self, method_name, self.visit_default)
    return method(message)

visit_default #

visit_default(message: IOMessage) -> None
Source code in fastagency/ui/console/base.py 
106
107
108
109
110
111
112
113
114
def visit_default(self, message: IOMessage) -> None:
    content = message.model_dump()["content"]
    console_msg = self.ConsoleMessage(
        sender=message.sender,
        recipient=message.recipient,
        heading=message.type,
        body=json.dumps(content, indent=2),
    )
    self._format_and_print(console_msg)

visit_error #

visit_error(message: Error) -> Optional[str]
Source code in fastagency/base.py 
206
207
def visit_error(self, message: Error) -> Optional[str]:
    return self.visit_default(message)

visit_function_call_execution #

visit_function_call_execution(
    message: FunctionCallExecution,
) -> Optional[str]
Source code in fastagency/base.py 
189
190
191
192
def visit_function_call_execution(
    self, message: FunctionCallExecution
) -> Optional[str]:
    return self.visit_default(message)

visit_multiple_choice #

visit_multiple_choice(message: MultipleChoice) -> str
Source code in fastagency/ui/console/base.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def visit_multiple_choice(self, message: MultipleChoice) -> str:
    console_msg = self.ConsoleMessage(
        sender=message.sender,
        recipient=message.recipient,
        heading=message.type,
        body=f"{message.prompt} (choices: {', '.join(message.choices)}, default: {message.default})",
    )

    prompt = self._format_message(console_msg)
    prompt = self._indent(prompt)
    while True:
        # logger.info(f"visit_multiple_choice(): {prompt=}")
        retval = input(prompt)
        if retval in message.choices:
            return retval
        elif retval == "" and message.default:
            return message.default
        else:
            print(f"Invalid choice ('{retval}'). Please try again.")  # noqa: T201 `print` found

visit_suggested_function_call #

visit_suggested_function_call(
    message: SuggestedFunctionCall,
) -> Optional[str]
Source code in fastagency/base.py 
184
185
186
187
def visit_suggested_function_call(
    self, message: SuggestedFunctionCall
) -> Optional[str]:
    return self.visit_default(message)

visit_system_message #

visit_system_message(
    message: SystemMessage,
) -> Optional[str]
Source code in fastagency/base.py 
200
201
def visit_system_message(self, message: SystemMessage) -> Optional[str]:
    return self.visit_default(message)

visit_text_input #

visit_text_input(message: TextInput) -> str
Source code in fastagency/ui/console/base.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def visit_text_input(self, message: TextInput) -> str:
    suggestions = (
        f" (suggestions: {', '.join(message.suggestions)})"
        if message.suggestions
        else ""
    )
    console_msg = self.ConsoleMessage(
        sender=message.sender,
        recipient=message.recipient,
        heading=message.type,
        body=f"{message.prompt}{suggestions}:",
    )

    prompt = self._format_message(console_msg)
    prompt = self._indent(prompt)
    if message.password:
        return getpass.getpass(prompt)
    else:
        return input(prompt)

visit_text_message #

visit_text_message(message: TextMessage) -> None
Source code in fastagency/ui/console/base.py 
116
117
118
119
120
121
122
123
def visit_text_message(self, message: TextMessage) -> None:
    console_msg = self.ConsoleMessage(
        sender=message.sender,
        recipient=message.recipient,
        heading=message.type,
        body=message.body,
    )
    self._format_and_print(console_msg)

visit_workflow_completed #

visit_workflow_completed(
    message: WorkflowCompleted,
) -> Optional[str]
Source code in fastagency/base.py 
203
204
def visit_workflow_completed(self, message: WorkflowCompleted) -> Optional[str]:
    return self.visit_default(message)