Spaces:
Paused
Paused
| import os | |
| import sys | |
| import threading | |
| from itertools import chain | |
| import anyio | |
| from flaml import autogen | |
| import gradio as gr | |
| from autogen import Agent, AssistantAgent, OpenAIWrapper, UserProxyAgent, ConversableAgent | |
| from autogen.code_utils import extract_code | |
| from gradio import ChatInterface, Request | |
| from gradio.helpers import special_args | |
| LOG_LEVEL = "INFO" | |
| TIMEOUT = 60 | |
| class myChatInterface(ChatInterface): | |
| async def _submit_fn( | |
| self, | |
| message: str, | |
| history_with_input: list[list[str | None]], | |
| request: Request, | |
| *args, | |
| ) -> tuple[list[list[str | None]], list[list[str | None]]]: | |
| history = history_with_input[:-1] | |
| inputs, _, _ = special_args(self.fn, inputs=[message, history, *args], request=request) | |
| if self.is_async: | |
| await self.fn(*inputs) | |
| else: | |
| await anyio.to_thread.run_sync(self.fn, *inputs, limiter=self.limiter) | |
| # history.append([message, response]) | |
| return history, history | |
| with gr.Blocks() as demo: | |
| def flatten_chain(list_of_lists): | |
| return list(chain.from_iterable(list_of_lists)) | |
| class thread_with_trace(threading.Thread): | |
| # https://www.geeksforgeeks.org/python-different-ways-to-kill-a-thread/ | |
| # https://stackoverflow.com/questions/6893968/how-to-get-the-return-value-from-a-thread | |
| def __init__(self, *args, **keywords): | |
| threading.Thread.__init__(self, *args, **keywords) | |
| self.killed = False | |
| self._return = None | |
| def start(self): | |
| self.__run_backup = self.run | |
| self.run = self.__run | |
| threading.Thread.start(self) | |
| def __run(self): | |
| sys.settrace(self.globaltrace) | |
| self.__run_backup() | |
| self.run = self.__run_backup | |
| def run(self): | |
| if self._target is not None: | |
| self._return = self._target(*self._args, **self._kwargs) | |
| def globaltrace(self, frame, event, arg): | |
| if event == "call": | |
| return self.localtrace | |
| else: | |
| return None | |
| def localtrace(self, frame, event, arg): | |
| if self.killed: | |
| if event == "line": | |
| raise SystemExit() | |
| return self.localtrace | |
| def kill(self): | |
| self.killed = True | |
| def join(self, timeout=0): | |
| threading.Thread.join(self, timeout) | |
| return self._return | |
| def update_agent_history(recipient, messages, sender, config): | |
| if config is None: | |
| config = recipient | |
| if messages is None: | |
| messages = recipient._oai_messages[sender] | |
| message = messages[-1] | |
| message.get("content", "") | |
| # config.append(msg) if msg is not None else None # config can be agent_history | |
| return False, None # required to ensure the agent communication flow continues | |
| def _is_termination_msg(message): | |
| """Check if a message is a termination message. | |
| Terminate when no code block is detected. Currently only detect python code blocks. | |
| """ | |
| if isinstance(message, dict): | |
| message = message.get("content") | |
| if message is None: | |
| return False | |
| cb = extract_code(message) | |
| contain_code = False | |
| for c in cb: | |
| # todo: support more languages | |
| if c[0] == "python": | |
| contain_code = True | |
| break | |
| return not contain_code | |
| def initialize_agents(config_list): | |
| assistant = AssistantAgent( | |
| name="assistant", | |
| max_consecutive_auto_reply=5, | |
| llm_config={ | |
| # "seed": 42, | |
| "timeout": TIMEOUT, | |
| "config_list": config_list, | |
| }, | |
| ) | |
| angent_survey = ConversableAgent( | |
| name = "agent_survey", | |
| system_message="you are dedicated Learning Path Advisor. " | |
| "the first task is understanding users goal and educational background, interests, career aspirations and other necessary information. " | |
| "Interact with the user until you have sufficient information, or the user offers a message ending with 'Exit'." | |
| "Please gently guide the user,ask questions one by one." | |
| "based on the critic feedback, if needed, further ask user." | |
| "summarizing the user's information but do not do recommandations.", | |
| llm_config={"config_list": config_list}, | |
| is_termination_msg=lambda msg: "EXIT" in msg["content"], # terminate | |
| human_input_mode="NEVER", # never ask for human input | |
| ) | |
| angent_recommander =ConversableAgent( | |
| name = "angent_recommander", | |
| system_message="you are dedicated Learning Path Advisor." | |
| "based on the user's information from 'agent_survey', critic and user, plan a learning path" | |
| "your task is to help user navigate through the vast ocean of courses and specializations, finding the perfect path that aligns with user's background." | |
| "the learning path should include necessary information of course such as course title, providers and thoughtful reasons for the recommendation" | |
| "Then, ask the critic's opinion. and try to improve based on the opinion of critics" | |
| "Rule 1. The total number of courses should be less than 4", | |
| llm_config={"config_list": config_list}, | |
| is_termination_msg=lambda msg: "EXIT" in msg["content"], | |
| human_input_mode="NEVER", # never ask for human input | |
| ) | |
| userproxy = UserProxyAgent( | |
| name="userproxy", | |
| system_message ="a human user", | |
| human_input_mode="NEVER", | |
| is_termination_msg=_is_termination_msg, | |
| max_consecutive_auto_reply=5, | |
| # code_execution_config=False, | |
| code_execution_config={ | |
| #"last_n_messages": 2, | |
| "work_dir": "path_advisor", | |
| "use_docker": False, # set to True or image name like "python:3" to use docker | |
| }, | |
| ) | |
| critic = AssistantAgent( | |
| name="Critic", | |
| system_message="Critic. Double check leanring path, reasons, from other agents and provide feedback. you should Reflect at least these questions" | |
| "Q1: Whether the recommended course meets the user's interests or objective?" | |
| "Q2: Do learning paths lead to higher motivation, or could they possibly lead to an overload of choices that paralyze some learners?" | |
| "Q3: Is the content provided in-depth enough to foster a comprehensive understanding?" | |
| "Q4: Is there a logical progression in the curriculum that builds on previous knowledge?" | |
| "if you think the plan should imporved further, give the feedback to 'angent_recommander' and ask it to improve the learning path" | |
| "if you think the plan is good enough, then ask the user if he would like to try the learning path?", | |
| llm_config={"config_list": config_list}, | |
| ) | |
| Learning_Path_summary = ConversableAgent( | |
| name = "Learning_Path_summary", | |
| system_message="You only followed by an approved leanring path plan by user." | |
| "Act as helpful and kind Learning Path Advisor, summarize the previous approved learning path for user, including course title, providers and thoughtful reasons for the recommendation" | |
| "The tone should be informative, friendly, and supportive." | |
| "And highlight the keypoints or keywords in orange" | |
| "Your task is to provide a detailed summary of an approved leanring path plan to user. This overview is designed to give user clarity on the structure, objectives, and resources. the key elements are below:" | |
| "Learning Path Overview: Goal Alignment; What were the initial goals set at the start of this learning path? How do these objectives align with user's current background or personal development needs?" | |
| "Curriculum Structure: provide a breakdown of the main topics and learning stages included in this path. What are the key outcomes expected at each stage, and how do they contribute to the overall goal?" | |
| "Certification and Completion: Upon completing the learning path, what certificates or qualifications will be awarded? How do these credentials support further professional advancement or learning?" | |
| "Future Learning Opportunities: What subsequent learning opportunities or advanced topics are recommended after completing this path?Are there any additional skills or areas of knowledge that suggest exploring to enhance professional growth?" | |
| "Conclusion and give encouragement" | |
| , | |
| llm_config={"config_list": config_list}, | |
| is_termination_msg=lambda msg: "EXIT" in msg["content"], | |
| human_input_mode="NEVER", # never ask for human input | |
| ) | |
| #group chat with critic | |
| groupchat = autogen.GroupChat(agents=[userproxy, angent_survey, angent_recommander,critic,Learning_Path_summary], messages=[], max_round=20) | |
| manager = autogen.GroupChatManager(groupchat=groupchat, llm_config={"config_list": config_list}) | |
| # assistant.register_reply([Agent, None], update_agent_history) | |
| # userproxy.register_reply([Agent, None], update_agent_history) | |
| return userproxy, angent_survey, angent_recommander,critic,Learning_Path_summary, manager | |
| def chat_to_oai_message(chat_history): | |
| """Convert chat history to OpenAI message format.""" | |
| messages = [] | |
| if LOG_LEVEL == "DEBUG": | |
| print(f"chat_to_oai_message: {chat_history}") | |
| for msg in chat_history: | |
| messages.append( | |
| { | |
| "content": msg[0].split()[0] if msg[0].startswith("exitcode") else msg[0], | |
| "role": "user", | |
| } | |
| ) | |
| messages.append({"content": msg[1], "role": "assistant"}) | |
| return messages | |
| def oai_message_to_chat(oai_messages, sender): | |
| """Convert OpenAI message format to chat history.""" | |
| chat_history = [] | |
| messages = oai_messages[sender] | |
| if LOG_LEVEL == "DEBUG": | |
| print(f"oai_message_to_chat: {messages}") | |
| for i in range(0, len(messages), 2): | |
| chat_history.append( | |
| [ | |
| messages[i]["content"], | |
| messages[i + 1]["content"] if i + 1 < len(messages) else "", | |
| ] | |
| ) | |
| return chat_history | |
| def agent_history_to_chat(agent_history): | |
| """Convert agent history to chat history.""" | |
| chat_history = [] | |
| for i in range(0, len(agent_history), 2): | |
| chat_history.append( | |
| [ | |
| agent_history[i], | |
| agent_history[i + 1] if i + 1 < len(agent_history) else None, | |
| ] | |
| ) | |
| return chat_history | |
| def initiate_chat(config_list, user_message, chat_history): | |
| if LOG_LEVEL == "DEBUG": | |
| print(f"chat_history_init: {chat_history}") | |
| # agent_history = flatten_chain(chat_history) | |
| if len(config_list[0].get("api_key", "")) < 2: | |
| chat_history.append( | |
| [ | |
| user_message, | |
| "Hi, nice to meet you!", | |
| ] | |
| ) | |
| return chat_history | |
| else: | |
| llm_config = { | |
| # "seed": 42, | |
| "timeout": TIMEOUT, | |
| "config_list": config_list, | |
| } | |
| manager.llm_config.update(llm_config) | |
| manager.client = OpenAIWrapper(**manager.llm_config) | |
| manager.reset() | |
| oai_messages = chat_to_oai_message(chat_history) | |
| manager._oai_system_message_origin = manager._oai_system_message.copy() | |
| manager._oai_system_message += oai_messages | |
| try: | |
| userproxy.initiate_chat(manager, message=user_message) | |
| messages = userproxy.chat_messages | |
| chat_history += oai_message_to_chat(messages, manager) | |
| # agent_history = flatten_chain(chat_history) | |
| except Exception as e: | |
| # agent_history += [user_message, str(e)] | |
| # chat_history[:] = agent_history_to_chat(agent_history) | |
| chat_history.append([user_message, str(e)]) | |
| manager._oai_system_message = manager._oai_system_message_origin.copy() | |
| if LOG_LEVEL == "DEBUG": | |
| print(f"chat_history: {chat_history}") | |
| # print(f"agent_history: {agent_history}") | |
| return chat_history | |
| def chatbot_reply_thread(input_text, chat_history, config_list): | |
| """Chat with the agent through terminal.""" | |
| thread = thread_with_trace(target=initiate_chat, args=(config_list, input_text, chat_history)) | |
| thread.start() | |
| try: | |
| messages = thread.join(timeout=TIMEOUT) | |
| if thread.is_alive(): | |
| thread.kill() | |
| thread.join() | |
| messages = [ | |
| input_text, | |
| "Timeout Error: Please check your API keys and try again later.", | |
| ] | |
| except Exception as e: | |
| messages = [ | |
| [ | |
| input_text, | |
| str(e) if len(str(e)) > 0 else "Invalid Request to OpenAI, please check your API keys.", | |
| ] | |
| ] | |
| return messages | |
| def chatbot_reply_plain(input_text, chat_history, config_list): | |
| """Chat with the agent through terminal.""" | |
| try: | |
| messages = initiate_chat(config_list, input_text, chat_history) | |
| except Exception as e: | |
| messages = [ | |
| [ | |
| input_text, | |
| str(e) if len(str(e)) > 0 else "Invalid Request to OpenAI, please check your API keys.", | |
| ] | |
| ] | |
| return messages | |
| def chatbot_reply(input_text, chat_history, config_list): | |
| """Chat with the agent through terminal.""" | |
| return chatbot_reply_thread(input_text, chat_history, config_list) | |
| def get_description_text(): | |
| return """ | |
| # Hello! 👋 My name is <span style="color:orange;">PathFinder</span>, | |
| ## your dedicated Learning Path Advisor here. | |
| Welcome aboard! | |
| I am here to help you navigate through the vast ocean of courses and specializations, finding the perfect path that aligns with your career goals and educational interests. | |
| Whether you are looking to advance in your current field, pivot to a new industry, or simply explore new areas of knowledge, I'm here to guide you every step of the way! | |
| """ | |
| def update_config(): | |
| config_list = autogen.config_list_from_models( | |
| model_list=[os.environ.get("MODEL", "gpt-4")], | |
| ) | |
| if not config_list: | |
| selected_model = "gpt-4" | |
| selected_key = "sk-ifbaI7viN2UnK634A92a07A9679046A392B907Df26AeCf8d" | |
| selected_url = "https://aihubmix.com/v1" | |
| config_list = [ | |
| { | |
| "api_key": selected_key, | |
| "base_url": selected_url, | |
| #"api_type": "azure", | |
| #"api_version": "2023-07-01-preview", | |
| "model": selected_model, | |
| } | |
| ] | |
| return config_list | |
| def set_params(model, oai_key, aoai_key, aoai_base): | |
| os.environ["MODEL"] = model | |
| os.environ["OPENAI_API_KEY"] = oai_key | |
| os.environ["AZURE_OPENAI_API_KEY"] = aoai_key | |
| os.environ["AZURE_OPENAI_API_BASE"] = aoai_base | |
| def respond(message, chat_history, model, oai_key, aoai_key, aoai_base): | |
| set_params(model, oai_key, aoai_key, aoai_base) | |
| config_list = update_config() | |
| chat_history[:] = chatbot_reply(message, chat_history, config_list) | |
| if LOG_LEVEL == "DEBUG": | |
| print(f"return chat_history: {chat_history}") | |
| return "" | |
| config_list= update_config() | |
| userproxy, angent_survey, angent_recommander,critic,Learning_Path_summary, manager = initialize_agents(config_list) | |
| description = gr.Markdown(get_description_text()) | |
| with gr.Row() as params: | |
| txt_model = gr.Dropdown( | |
| label="Model", | |
| choices=[ | |
| "gpt-4", | |
| "gpt-3.5-turbo", | |
| ], | |
| allow_custom_value=True, | |
| value="gpt-4", | |
| container=True, | |
| ) | |
| txt_oai_key = gr.Textbox( | |
| label="OpenAI API Key", | |
| placeholder="Enter OpenAI API Key", | |
| max_lines=1, | |
| show_label=True, | |
| container=True, | |
| type="password", | |
| ) | |
| txt_aoai_key = gr.Textbox( | |
| label="Azure OpenAI API Key", | |
| placeholder="Enter Azure OpenAI API Key", | |
| max_lines=1, | |
| show_label=True, | |
| container=True, | |
| type="password", | |
| ) | |
| txt_aoai_base_url = gr.Textbox( | |
| label="Base url", | |
| placeholder="Enter Base Url", | |
| max_lines=1, | |
| show_label=True, | |
| container=True, | |
| type="password", | |
| ) | |
| chatbot = gr.Chatbot( | |
| [], | |
| elem_id="chatbot", | |
| bubble_full_width=False, | |
| avatar_images=( | |
| "user.png", | |
| (os.path.join(os.path.dirname(__file__), "advisor.png")), | |
| ), | |
| render=False, | |
| height=600, | |
| ) | |
| txt_input = gr.Textbox( | |
| scale=4, | |
| show_label=False, | |
| placeholder="Enter text and press enter", | |
| container=False, | |
| render=False, | |
| autofocus=True, | |
| ) | |
| chatiface = myChatInterface( | |
| respond, | |
| chatbot=chatbot, | |
| textbox=txt_input, | |
| additional_inputs=[ | |
| txt_model, | |
| txt_oai_key, | |
| txt_aoai_key, | |
| txt_aoai_base_url, | |
| ], | |
| examples=[ | |
| [" I am interested in data science but do not know where to start."], | |
| [" I'm a software developer and I want to learn about artificial intelligence. I have some experience with Python."], | |
| [" I have a background in literature and I'm interested in exploring more about the philosophical aspects of humanity. I would like to understand how philosophical theories have influenced human behavior and society."], | |
| ], | |
| ) | |
| if __name__ == "__main__": | |
| demo.launch(share=True, server_name="0.0.0.0") | |