Coverage for fastagency/ui/mesop/send_prompt.py: 78%

1from collections.abc import Iterable 3 ctx1bcd

2 

3import mesop as me 3 ctx1bcd

4 

5from ...base import ProviderProtocol 3 ctx1bcd

6from ...logging import get_logger 3 ctx1bcd

7from .data_model import State 3 ctx1bcd

8from .mesop import MesopMessage, MesopUI, run_workflow_mesop 3 ctx1bcd

9 

10logger = get_logger(__name__) 3 ctx1bcd

11 

12 

13def send_prompt_to_autogen( 3 ctx1bcd

14 provider: ProviderProtocol, name: str 

15) -> Iterable[MesopMessage]: 

16 ui = run_workflow_mesop(provider, name=name) 1 ctx1a

17 if not isinstance(ui.ui_base, MesopUI): # pragma: no cover 1 ctx1a

18 logger.error("") 

19 raise RuntimeError(f"Expected MesopUI, got {type(ui.ui_base)}") 

20 mesop_ui: MesopUI = ui.ui_base 1 ctx1a

21 

22 state = me.state(State) 1 ctx1a

23 state.conversation.fastagency = ui.ui_base.id 1 ctx1a

24 return mesop_ui.get_message_stream() 1 ctx1a

25 

26 

27def send_user_feedback_to_autogen(user_response: str) -> Iterable[MesopMessage]: 3 ctx1bcd

28 state = me.state(State) 1 ctx1a

29 mesop_id = state.conversation.fastagency 1 ctx1a

30 mesop_io = MesopUI.get_conversation(mesop_id) 1 ctx1a

31 mesop_io.respond(user_response) 1 ctx1a

32 return mesop_io.get_message_stream() 1 ctx1a

33 

34 

35def get_more_messages() -> Iterable[MesopMessage]: 3 ctx1bcd

36 state = me.state(State) 

37 mesop_id = state.conversation.fastagency 

38 mesop_io = MesopUI.get_conversation(mesop_id) 

39 return mesop_io.get_message_stream()