Spaces:
Runtime error
Runtime error
| import autogen | |
| from autogen.agentchat.contrib.text_analyzer_agent import TextAnalyzerAgent | |
| from autogensop.gradio_chat_manager import GroupChat, GroupChatManager | |
| class GradioAutogenSop(autogen.ConversableAgent): | |
| def __init__( | |
| self, | |
| target="", | |
| states=[], | |
| agents=[], | |
| user_name="", | |
| llm_config={}, | |
| max_user_input=100, | |
| **kwargs, | |
| ) -> None: | |
| super().__init__( | |
| name="State manager", | |
| human_input_mode="NEVER", | |
| llm_config=llm_config, | |
| system_message="You are a state manager", | |
| **kwargs, | |
| ) | |
| self.target = target | |
| self.states = states | |
| self.agents = agents | |
| self.user_name = user_name | |
| self.llm_config = llm_config | |
| self.max_user_input = max_user_input | |
| self.groupchat = autogen.GroupChat(agents, messages=[]) | |
| self.user = self.groupchat.agent_by_name(self.user_name) | |
| self.manager = GroupChatManager(self.groupchat, llm_config=self.llm_config) | |
| self.messages = [] | |
| self.state_groupchat = None | |
| def _state_start_condition(self): | |
| return "\n".join([f"{key}: {item['start_condition']}" for key, item in self.states.items()]) | |
| def _state_task(self): | |
| return "\n".join([f"{key}: {item['task']}" for key, item in self.states.items()]) | |
| def judge_target_reached(self): | |
| # 判断self.target是否满足 | |
| rule = f"""Your judgment condition is {self.target} If the judgment condition is reached, only return: EXIT, else only return: CONTINUE. """ | |
| prompt = self.messages + [{ | |
| "role": "system", | |
| "content": rule, | |
| }] | |
| final, res = self.generate_oai_reply(prompt) | |
| print('***************target_reached:' + res) | |
| if final: | |
| if 'EXIT' in res: | |
| return True | |
| else: | |
| return False | |
| else: | |
| print('judge_target_reached 错误') | |
| exit() | |
| def select_state(self): | |
| # 判断进入的state | |
| states_rule = f"""Your ultimate goal is {self.target} The optional states are as follows: | |
| {self._state_start_condition()}. | |
| Read the above conversation. Then select the next state from {[key for key in self.states]}. Only return the state.""" | |
| prompt = self.messages[-4:] + [{ | |
| "role": "system", | |
| "content": states_rule, | |
| }] | |
| print('【判断进入的state】') | |
| print(prompt) | |
| final, name = self.generate_oai_reply(prompt) | |
| while name not in list(self.states.keys()): | |
| prompt += [{ | |
| "role": "system", | |
| "content": name + "is wrong answer", | |
| }] | |
| final, name = self.generate_oai_reply(prompt) | |
| print('************进入：' + name + '阶段************\n') | |
| if final: | |
| return name | |
| else: | |
| print('select_state 错误') | |
| exit() | |
| def get_response(self, history): | |
| while True: | |
| print('【选择发言人】') | |
| speaker = self.manager.groupchat.select_speaker(self.user, self.manager) | |
| print('************选择发言的是：' + speaker.name + '**************\n') | |
| if speaker.name == self.user.name: | |
| break | |
| # 如果是可学习的agent则学习 | |
| if hasattr(speaker, 'learn_from_user_feedback'): | |
| speaker.learn_from_user_feedback() | |
| reply = speaker.generate_reply(messages=self.messages, sender=self.manager) | |
| self.manager.broadcast(reply, speaker) | |
| message = self.manager.chat_messages[self.user][-1] | |
| message["name"] = speaker.name | |
| self.messages.append(message) | |
| self.manager.groupchat.messages = self.messages | |
| history += [[None, " **" + speaker.name + ":** " + reply]] | |
| yield history | |
| yield history | |
| def ready_response(self, msg, history): | |
| self.user.send(message=msg, recipient=self.manager, request_reply=False) | |
| message = self.manager.chat_messages[self.user][-1] | |
| message["name"] = self.user.name | |
| self.messages.append(message) | |
| state_name = self.select_state() | |
| # 根据state更改参与者的sys_msg，将阶段目标拼接到sys_msg后面 | |
| for agent_name, sys_msg in self.states[state_name]['sys_msg'].items(): | |
| agent = self.groupchat.agent_by_name(agent_name) | |
| if sys_msg not in agent.system_message: | |
| agent.update_system_message(agent.system_message + '此阶段你的目标是：' + sys_msg) | |
| # 创建group chat，继承已有对话 | |
| self.state_groupchat = GroupChat( | |
| [self.groupchat.agent_by_name(agent_name) for agent_name in | |
| self.states[state_name]['participate_agent_names']], | |
| messages=self.messages, | |
| max_round=4 | |
| ) | |
| self.manager.groupchat = self.state_groupchat | |
| self.manager.broadcast(msg, self.user) | |
| return "", history + [[msg, None]] | |