-
Notifications
You must be signed in to change notification settings - Fork 8.2k
Description
There're a series of issues/bug on function_call not found in group chat (#252 #152 )
The reason is because the function_call is implemented in two steps in autogen agents
- propose a function_call
- run function_call
And usually the two steps are completed by different agents. This brings problem in group chat, where group chat manager doesn't have information on which agent has the correct function_map to run a function_call. If group chat "guesses" the executor agent incorrectly, a function_call not found error will be thrown
There're three suggested options to fix it
- Having an individual executor agent to hold all function_call in its
function_map, and reroute allfunction_callmessages to it. - Having all agents which have
function_definitionholds the correspondingfunction_mapas well, and reroute thefunction_callback to the propose-agent. - Having the next agent which has the key in
function_mapto run the function. And reroute thefunction_callto that agent
For the first option, we can accommodate the code from @ilaffey2 and combine the admin and executor into the same agent so as to keep backward compatibility
from autogen import GroupChat, ConversableAgent, UserProxyAgent
from dataclasses import dataclass
@dataclass
class ExecutorGroupchat(GroupChat):
def select_speaker(
self, last_speaker: ConversableAgent, selector: ConversableAgent
):
"""Select the next speaker."""
try:
message = self.messages[-1]
if "function_call" in message:
return self.admin
except Exception as e:
print(e)
pass
selector.update_system_message(self.select_speaker_msg())
final, name = selector.generate_oai_reply(
self.messages
+ [
{
"role": "system",
"content": f"Read the above conversation. Then select the next role from {self.agent_names} to play. Only return the role.",
}
]
)
if not final:
# i = self._random.randint(0, len(self._agent_names) - 1) # randomly pick an id
return self.next_agent(last_speaker)
try:
return self.agent_by_name(name)
except ValueError:
return self.next_agent(last_speaker)For the second option, the group manager just rewire function_call back to its proposal agent
from autogen import GroupChat, ConversableAgent, UserProxyAgent
from dataclasses import dataclass
@dataclass
class ExecutorGroupchat(GroupChat):
def select_speaker(
self, last_speaker: ConversableAgent, selector: ConversableAgent
):
"""Select the next speaker."""
try:
message = self.messages[-1]
if "function_call" in message:
return self.last_speaker
except Exception
# rest of code goes hereFor the third option, the group manager just sends function_call back to the next agent that has the correct key in its function_map
from autogen import GroupChat, ConversableAgent, UserProxyAgent
from dataclasses import dataclass
@dataclass
class ExecutorGroupchat(GroupChat):
def select_speaker(
self, last_speaker: ConversableAgent, selector: ConversableAgent
):
"""Select the next speaker."""
try:
message = self.messages[-1]
if "function_call" in message:
# for agent in self.agents:
# if message['function_call']['name'] in agent.function_map:
# return agent
except Exception as e:
print(e)
pass
# ...According to @sonichi suggestion, we can use a flag to toggle among the three options in order to accommodate the maximum flexibility.