ConsoleUI

fastagency.ui.console.base.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[Chatable] DEFAULT: None

Source code in fastagency/ui/console/base.py

def __init__(self, super_conversation: Optional["ConsoleUI"] = None) -> None:
    """Initialize the console UI object.

    Args:
        super_conversation (Optional[Chatable], 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

@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

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

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,
) -> None

Source code in fastagency/ui/console/base.py

def start(
    self,
    app: Runnable,
    import_string: str,
    name: Optional[str] = None,
    initial_message: Optional[str] = None,
) -> None:
    run_workflow(app.wf, self, name, initial_message)

visit

visit(message: IOMessage) -> Optional[str]

Source code in fastagency/base.py

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

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_function_call_execution

visit_function_call_execution(
    message: FunctionCallExecution,
) -> Optional[str]

Source code in fastagency/base.py

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

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

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

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

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

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

def visit_workflow_completed(self, message: WorkflowCompleted) -> Optional[str]:
    return self.visit_default(message)