File size: 3,677 Bytes
78f0f06 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 |
##########################################################################
#
#
# Waiting on https://github.com/microsoft/autogen/issues/527 to be solved
#
#
##########################################################################
from typing import Dict, Optional, Union
from autogen import Agent, AssistantAgent, UserProxyAgent, config_list_from_json
import chainlit as cl
from dotenv import load_dotenv
import os
# Load environment variables from .env file
load_dotenv()
TASK = "Plot a chart of NVDA stock price change YTD and save it on disk."
async def ask_helper(func, **kwargs):
res = await func(**kwargs).send()
while not res:
res = await func(**kwargs).send()
return res
class ChainlitAssistantAgent(AssistantAgent):
async def a_send(
self,
message: Union[Dict, str],
recipient: Agent,
request_reply: Optional[bool] = None,
silent: Optional[bool] = False,
) -> bool:
await cl.Message(
content=f'*Sending message to "{recipient.name}":*\n\n{message}',
author="AssistantAgent",
).send()
await super(ChainlitAssistantAgent, self).a_send(
message=message,
recipient=recipient,
request_reply=request_reply,
silent=silent,
)
class ChainlitUserProxyAgent(UserProxyAgent):
async def get_human_input(self, prompt: str) -> str:
if prompt.startswith(
"Provide feedback to assistant. Press enter to skip and use auto-reply"
):
res = await ask_helper(
cl.AskActionMessage,
content="Continue or provide feedback?",
actions=[
cl.Action(
name="continue", value="continue", label="✅ Continue"
),
cl.Action(
name="feedback",
value="feedback",
label="💬 Provide feedback",
),
cl.Action(
name="exit",
value="exit",
label="🔚 Exit Conversation"
),
],
)
if res.get("value") == "continue":
return ""
if res.get("value") == "exit":
return "exit"
reply = await ask_helper(
cl.AskUserMessage, content=prompt, timeout=60)
return reply["content"].strip()
async def a_send(
self,
message: Union[Dict, str],
recipient: Agent,
request_reply: Optional[bool] = None,
silent: Optional[bool] = False,
):
await cl.Message(
content=f'*Sending message to "{recipient.name}"*:\n\n{message}',
author="UserProxyAgent",
).send()
await super(ChainlitUserProxyAgent, self).a_send(
message=message,
recipient=recipient,
request_reply=request_reply,
silent=silent,
)
@cl.on_chat_start
async def on_chat_start():
config_list = config_list_from_json(env_or_file="OAI_CONFIG_LIST")
assistant = ChainlitAssistantAgent(
"assistant", llm_config={"config_list": config_list}
)
user_proxy = ChainlitUserProxyAgent(
"user_proxy",
code_execution_config={
"work_dir": "workspace",
"use_docker": False,
},
)
await cl.Message(content=f"Starting agents on task: {TASK}...").send()
await user_proxy.a_initiate_chat(
assistant,
message=TASK,
) |