Spaces:
Runtime error
Runtime error
File size: 5,133 Bytes
b628e9d |
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 121 122 123 124 125 126 127 |
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]]
|