Coverage for docs/docs_src/user_guide/dependency_injection/workflow.py: 33%

1import os 4 ctx1abcd

2from typing import Annotated, Any 4 ctx1abcd

3 

4from autogen import UserProxyAgent, register_function, ConversableAgent, LLMConfig 4 ctx1abcd

5from fastagency import UI 4 ctx1abcd

6from fastagency.api.dependency_injection import inject_params 4 ctx1abcd

7from fastagency.runtimes.ag2 import Workflow 4 ctx1abcd

8 

9account_ballace_dict = { 4 ctx1abcd

10 ("alice", "password123"): 100, 

11 ("bob", "password456"): 200, 

12 ("charlie", "password789"): 300, 

13} 

14 

15 

16def get_balance( 4 ctx1abcd

17 username: Annotated[str, "Username"], 4 ctx1abcd

18 password: Annotated[str, "Password"], 4 ctx1abcd

19) -> str: 4 ctx1abcd

20 if (username, password) not in account_ballace_dict: 

21 return "Invalid username or password" 

22 return f"Your balance is {account_ballace_dict[(username, password)]}$" 

23 

24 

25llm_config = LLMConfig( 4 ctx1abcd

26 model="gpt-4o-mini", 

27 api_key=os.getenv("OPENAI_API_KEY"), 

28 temperature=0.8, 

29) 

30 

31wf = Workflow() 4 ctx1abcd

32 

33 

34@wf.register(name="bank_chat", description="Bank chat") # type: ignore[misc] 4 ctx1abcd

35def bank_workflow(ui: UI, params: dict[str, str]) -> str: 4 ctx1abcd

36 username = ui.text_input( 

37 sender="Workflow", 

38 recipient="User", 

39 prompt="Enter your username:", 

40 ) 

41 password = ui.text_input( 

42 sender="Workflow", 

43 recipient="User", 

44 prompt="Enter your password:", 

45 ) 

46 

47 with llm_config: 

48 user_agent = UserProxyAgent( 

49 name="User_Agent", 

50 system_message="You are a user agent", 

51 human_input_mode="NEVER", 

52 ) 

53 banker_agent = ConversableAgent( 

54 name="Banker_Agent", 

55 system_message="You are a banker agent", 

56 human_input_mode="NEVER", 

57 ) 

58 

59 ctx: dict[str, Any] = { 

60 "username": username, 

61 "password": password, 

62 } 

63 get_balance_with_params = inject_params(get_balance, ctx) 

64 register_function( 

65 f=get_balance_with_params, 

66 caller=banker_agent, 

67 executor=user_agent, 

68 description="Get balance", 

69 ) 

70 

71 response = user_agent.run( 

72 banker_agent, 

73 message="We need to get user's balance.", 

74 summary_method="reflection_with_llm", 

75 max_turns=3, 

76 ) 

77 

78 return ui.process(response) # type: ignore[no-any-return]