Custom User Interactions#
In this example, we'll demonstrate how to create custom interaction with the user using UI protocol and its process_message method.
Install#
We strongly recommend using Cookiecutter for setting up the project. Cookiecutter creates the project folder structure, default workflow, automatically installs all the necessary requirements, and creates a devcontainer that can be used with Visual Studio Code.
You can setup the project using Cookiecutter by following the project setup guide.
Alternatively, you can use pip + venv. Before getting started, you need to install FastAgency. You can do this using pip, Python's package installer.
Define Interaction#
This section describes how to define functions for the ConversableAgent instances representing the student and teacher. We will also explain the differences between MultipleChoice, SystemMessage, and TextInput, which are used for communication between the user and agents.
Let's define three functions which will be available to the agents:
Free Textual Tnput#
TextInput is suitable for free-form text messages, ideal for open-ended queries and dialogues. This function allows the student to request exam questions from the teacher and provides some suggestions using TextInput.
def retrieve_exam_questions(
message: Annotated[str, "Message for examiner"],
) -> Optional[str]:
try:
return ui.text_input(
sender="student",
recipient="teacher",
prompt=message,
suggestions=[
"1) Mona Lisa",
"2) Innovations",
"3) Florence at the time of Leonardo",
"4) The Last Supper",
"5) Vitruvian Man",
],
)
except Exception as e: # pragma: no cover
return f"retrieve_exam_questions() FAILED! {e}"
System Info Messages#
SystemMessage is used for operational or system-related instructions, such as logging data, and is not part of the agent dialogue. This function logs the final answers after the student completes the discussion using SystemMessage to log the event.
def write_final_answers(message: Annotated[str, "Message for examiner"]) -> str:
try:
ui.system_message(
sender="function call logger",
recipient="system",
message={
"operation": "storing final answers",
"content": message,
},
)
return "Final answers stored."
except Exception as e: # pragma: no cover
return f"write_final_answers() FAILED! {e}"
Multiple Choice#
MultipleChoice is used for structured responses where the user must select one of several predefined options. This function retrieves the final grade for the student's submitted answers using MultipleChoice, presenting the user with grading options.
def get_final_grade(
message: Annotated[str, "Message for examiner"],
) -> Optional[str]:
try:
return ui.multiple_choice(
sender="student",
recipient="teacher",
prompt=message,
choices=["A", "B", "C", "D", "F"],
)
except Exception as e: # pragma: no cover
return f"get_final_grade() FAILED! {e}"
Other Types of Messages#
All supported messages are subclasses of the IOMessage base class.
Registering the Functions#
We now register these functions with the workflow, linking the student_agent as the caller and the teacher_agent as the executor.
register_function(
retrieve_exam_questions,
caller=student_agent,
executor=teacher_agent,
name="retrieve_exam_questions",
description="Get exam questions from examiner",
)
register_function(
write_final_answers,
caller=student_agent,
executor=teacher_agent,
name="write_final_answers",
description="Write a final answers to exam questions to examiner, but only after discussing with the tutor first.",
)
register_function(
get_final_grade,
caller=student_agent,
executor=teacher_agent,
name="get_final_grade",
description="Get the final grade after submitting the answers.",
)
Complete Application Code#
main.py
import os
from typing import Annotated, Any, Optional
from autogen import register_function
from autogen.agentchat import ConversableAgent
from fastagency import UI, FastAgency
from fastagency.messages import MultipleChoice, SystemMessage, TextInput
from fastagency.runtimes.autogen import AutoGenWorkflows
from fastagency.ui.console import ConsoleUI
llm_config = {
"config_list": [
{
"model": "gpt-4o-mini",
"api_key": os.getenv("OPENAI_API_KEY"),
}
],
"temperature": 0.8,
}
wf = AutoGenWorkflows()
@wf.register(name="exam_practice", description="Student and teacher chat")
def exam_learning(ui: UI, params: dict[str, Any]) -> str:
initial_message = ui.text_input(
sender="Workflow",
recipient="User",
prompt="What do you want to learn today?",
)
def is_termination_msg(msg: dict[str, Any]) -> bool:
return msg["content"] is not None and "TERMINATE" in msg["content"]
student_agent = ConversableAgent(
name="Student_Agent",
system_message="You are a student writing a practice test. Your task is as follows:\n"
" 1) Retrieve exam questions by calling a function.\n"
" 2) Write a draft of proposed answers and engage in dialogue with your tutor.\n"
" 3) Once you are done with the dialogue, register the final answers by calling a function.\n"
" 4) Retrieve the final grade by calling a function.\n"
"Finally, terminate the chat by saying 'TERMINATE'.",
llm_config=llm_config,
human_input_mode="NEVER",
is_termination_msg=is_termination_msg,
)
teacher_agent = ConversableAgent(
name="Teacher_Agent",
system_message="You are a teacher.",
llm_config=llm_config,
human_input_mode="NEVER",
is_termination_msg=is_termination_msg,
)
def retrieve_exam_questions(
message: Annotated[str, "Message for examiner"],
) -> Optional[str]:
try:
return ui.text_input(
sender="student",
recipient="teacher",
prompt=message,
suggestions=[
"1) Mona Lisa",
"2) Innovations",
"3) Florence at the time of Leonardo",
"4) The Last Supper",
"5) Vitruvian Man",
],
)
except Exception as e: # pragma: no cover
return f"retrieve_exam_questions() FAILED! {e}"
def write_final_answers(message: Annotated[str, "Message for examiner"]) -> str:
try:
ui.system_message(
sender="function call logger",
recipient="system",
message={
"operation": "storing final answers",
"content": message,
},
)
return "Final answers stored."
except Exception as e: # pragma: no cover
return f"write_final_answers() FAILED! {e}"
def get_final_grade(
message: Annotated[str, "Message for examiner"],
) -> Optional[str]:
try:
return ui.multiple_choice(
sender="student",
recipient="teacher",
prompt=message,
choices=["A", "B", "C", "D", "F"],
)
except Exception as e: # pragma: no cover
return f"get_final_grade() FAILED! {e}"
register_function(
retrieve_exam_questions,
caller=student_agent,
executor=teacher_agent,
name="retrieve_exam_questions",
description="Get exam questions from examiner",
)
register_function(
write_final_answers,
caller=student_agent,
executor=teacher_agent,
name="write_final_answers",
description="Write a final answers to exam questions to examiner, but only after discussing with the tutor first.",
)
register_function(
get_final_grade,
caller=student_agent,
executor=teacher_agent,
name="get_final_grade",
description="Get the final grade after submitting the answers.",
)
chat_result = teacher_agent.initiate_chat(
student_agent,
message=initial_message,
summary_method="reflection_with_llm",
max_turns=10,
)
return chat_result.summary # type: ignore[no-any-return]
app = FastAgency(provider=wf, ui=ConsoleUI())
Run Application#
Once everything is set up, you can run your FastAgency application using the following command: