Coverage for docs/docs_src/user_guide/custom_user_interactions/main.py: 100%
25 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-19 12:16 +0000
1import os 1bacde
2from typing import Annotated, Any, Optional 1bacde
4from autogen import register_function, ConversableAgent, LLMConfig 1bacde
6from fastagency import UI, FastAgency 1bacde
7from fastagency.messages import MultipleChoice, SystemMessage, TextInput 1bacde
8from fastagency.runtimes.ag2 import Workflow 1bacde
9from fastagency.ui.console import ConsoleUI 1bacde
11llm_config = LLMConfig( 1bacde
12 model="gpt-4o-mini",
13 api_key=os.getenv("OPENAI_API_KEY"),
14 temperature=0.8,
15)
17wf = Workflow() 1bacde
20@wf.register(name="exam_practice", description="Student and teacher chat") 1bacde
21def exam_learning(ui: UI, params: dict[str, Any]) -> str: 1bacde
22 initial_message = ui.text_input( 1a
23 sender="Workflow",
24 recipient="User",
25 prompt="What do you want to learn today?",
26 )
28 def is_termination_msg(msg: dict[str, Any]) -> bool: 1a
29 return msg["content"] is not None and "TERMINATE" in msg["content"] 1a
31 with llm_config: 1a
32 student_agent = ConversableAgent( 1a
33 name="Student_Agent",
34 system_message="You are a student writing a practice test. Your task is as follows:\n"
35 " 1) Retrieve exam questions by calling a function.\n"
36 " 2) Write a draft of proposed answers and engage in dialogue with your tutor.\n"
37 " 3) Once you are done with the dialogue, register the final answers by calling a function.\n"
38 " 4) Retrieve the final grade by calling a function.\n"
39 "Finally, terminate the chat by saying 'TERMINATE'.",
40 human_input_mode="NEVER",
41 is_termination_msg=is_termination_msg,
42 )
43 teacher_agent = ConversableAgent( 1a
44 name="Teacher_Agent",
45 system_message="You are a teacher.",
46 human_input_mode="NEVER",
47 is_termination_msg=is_termination_msg,
48 )
50 def retrieve_exam_questions( 1a
51 message: Annotated[str, "Message for examiner"],
52 ) -> Optional[str]:
53 try: 1a
54 return ui.text_input( 1a
55 sender="student",
56 recipient="teacher",
57 prompt=message,
58 suggestions=[
59 "1) Mona Lisa",
60 "2) Innovations",
61 "3) Florence at the time of Leonardo",
62 "4) The Last Supper",
63 "5) Vitruvian Man",
64 ],
65 )
66 except Exception as e: # pragma: no cover
67 return f"retrieve_exam_questions() FAILED! {e}"
69 def write_final_answers(message: Annotated[str, "Message for examiner"]) -> str: 1a
70 try: 1a
71 ui.system_message( 1a
72 sender="function call logger",
73 recipient="system",
74 message={
75 "operation": "storing final answers",
76 "content": message,
77 },
78 )
79 return "Final answers stored." 1a
80 except Exception as e: # pragma: no cover
81 return f"write_final_answers() FAILED! {e}"
83 def get_final_grade( 1a
84 message: Annotated[str, "Message for examiner"],
85 ) -> Optional[str]:
86 try: 1a
87 return ui.multiple_choice( 1a
88 sender="student",
89 recipient="teacher",
90 prompt=message,
91 choices=["A", "B", "C", "D", "F"],
92 )
93 except Exception as e: # pragma: no cover
94 return f"get_final_grade() FAILED! {e}"
96 register_function( 1a
97 retrieve_exam_questions,
98 caller=student_agent,
99 executor=teacher_agent,
100 name="retrieve_exam_questions",
101 description="Get exam questions from examiner",
102 )
104 register_function( 1a
105 write_final_answers,
106 caller=student_agent,
107 executor=teacher_agent,
108 name="write_final_answers",
109 description="Write a final answers to exam questions to examiner, but only after discussing with the tutor first.",
110 )
112 register_function( 1a
113 get_final_grade,
114 caller=student_agent,
115 executor=teacher_agent,
116 name="get_final_grade",
117 description="Get the final grade after submitting the answers.",
118 )
120 response = teacher_agent.run( 1a
121 student_agent,
122 message=initial_message,
123 summary_method="reflection_with_llm",
124 max_turns=10,
125 )
127 return ui.process(response) # type: ignore[no-any-return] 1a
130app = FastAgency(provider=wf, ui=ConsoleUI()) 1bacde