Coverage for docs/docs_src/user_guide/runtimes/ag2/whatsapp.py: 42%
12 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-19 12:16 +0000
1import os 1abcd
2from typing import Any 1abcd
4from autogen import UserProxyAgent, LLMConfig 1abcd
6from fastagency import UI, FastAgency 1abcd
7from fastagency.runtimes.ag2 import Workflow 1abcd
8from fastagency.runtimes.ag2.agents.whatsapp import WhatsAppAgent 1abcd
9from fastagency.ui.console import ConsoleUI 1abcd
11llm_config = LLMConfig( 1abcd
12 model="gpt-4o-mini",
13 api_key=os.getenv("OPENAI_API_KEY"),
14 temperature=0.8,
15)
17wf = Workflow() 1abcd
20@wf.register(name="simple_whatsapp", description="WhatsApp chat") # type: ignore[type-var] 1abcd
21def whatsapp_workflow(ui: UI, params: dict[str, Any]) -> str: 1abcd
22 def is_termination_msg(msg: dict[str, Any]) -> bool:
23 return msg["content"] is not None and "TERMINATE" in msg["content"]
25 initial_message = ui.text_input(
26 sender="Workflow",
27 recipient="User",
28 prompt="I can help you with sending a message over whatsapp, what would you like to send?",
29 )
30 user_agent = UserProxyAgent(
31 name="User_Agent",
32 system_message="You are a user agent, when the message is successfully sent, you can end the conversation by sending 'TERMINATE'",
33 llm_config=llm_config,
34 human_input_mode="NEVER",
35 is_termination_msg=is_termination_msg,
36 )
38 whatsapp_agent = WhatsAppAgent(
39 name="Assistant_Agent",
40 llm_config=llm_config,
41 human_input_mode="NEVER",
42 executor=user_agent,
43 # This is the default sender number for Infobip.
44 # If you want to use your own sender, please update the value below:
45 sender="447860099299",
46 whatsapp_api_key=os.getenv("WHATSAPP_API_KEY", ""),
47 is_termination_msg=is_termination_msg,
48 )
50 response = user_agent.run(
51 whatsapp_agent,
52 message=initial_message,
53 summary_method="reflection_with_llm",
54 max_turns=5,
55 )
57 return ui.process(response) # type: ignore[no-any-return]
60app = FastAgency(provider=wf, ui=ConsoleUI()) 1abcd