aphilippov/python-server-api
0
1import asyncio2import copy3import functools4import inspect5import json6import logging7import re8from collections import defaultdict9from typing import Any, Awaitable, Callable, Dict, List, Literal, Optional, Tuple, Type, TypeVar, Union10 11from .. import OpenAIWrapper12from ..code_utils import DEFAULT_MODEL, UNKNOWN, content_str, execute_code, extract_code, infer_lang13from ..function_utils import get_function_schema, load_basemodels_if_needed, serialize_to_str14from .agent import Agent15from .._pydantic import model_dump16 17try:18 from termcolor import colored19except ImportError:20 21 def colored(x, *args, **kwargs):22 return x23 24 25__all__ = ("ConversableAgent",)26 27logger = logging.getLogger(__name__)28 29F = TypeVar("F", bound=Callable[..., Any])30 31 32class ConversableAgent(Agent):33 """(In preview) A class for generic conversable agents which can be configured as assistant or user proxy.34 35 After receiving each message, the agent will send a reply to the sender unless the msg is a termination msg.36 For example, AssistantAgent and UserProxyAgent are subclasses of this class,37 configured with different default settings.38 39 To modify auto reply, override `generate_reply` method.40 To disable/enable human response in every turn, set `human_input_mode` to "NEVER" or "ALWAYS".41 To modify the way to get human input, override `get_human_input` method.42 To modify the way to execute code blocks, single code block, or function call, override `execute_code_blocks`,43 `run_code`, and `execute_function` methods respectively.44 To customize the initial message when a conversation starts, override `generate_init_message` method.45 """46 47 DEFAULT_CONFIG = {} # An empty configuration48 MAX_CONSECUTIVE_AUTO_REPLY = 100 # maximum number of consecutive auto replies (subject to future change)49 50 llm_config: Union[Dict, Literal[False]]51 52 def __init__(53 self,54 name: str,55 system_message: Optional[Union[str, List]] = "You are a helpful AI Assistant.",56 is_termination_msg: Optional[Callable[[Dict], bool]] = None,57 max_consecutive_auto_reply: Optional[int] = None,58 human_input_mode: Optional[str] = "TERMINATE",59 function_map: Optional[Dict[str, Callable]] = None,60 code_execution_config: Optional[Union[Dict, Literal[False]]] = None,61 llm_config: Optional[Union[Dict, Literal[False]]] = None,62 default_auto_reply: Optional[Union[str, Dict, None]] = "",63 description: Optional[str] = None,64 ):65 """66 Args:67 name (str): name of the agent.68 system_message (str or list): system message for the ChatCompletion inference.69 is_termination_msg (function): a function that takes a message in the form of a dictionary70 and returns a boolean value indicating if this received message is a termination message.71 The dict can contain the following keys: "content", "role", "name", "function_call".72 max_consecutive_auto_reply (int): the maximum number of consecutive auto replies.73 default to None (no limit provided, class attribute MAX_CONSECUTIVE_AUTO_REPLY will be used as the limit in this case).74 When set to 0, no auto reply will be generated.75 human_input_mode (str): whether to ask for human inputs every time a message is received.76 Possible values are "ALWAYS", "TERMINATE", "NEVER".77 (1) When "ALWAYS", the agent prompts for human input every time a message is received.78 Under this mode, the conversation stops when the human input is "exit",79 or when is_termination_msg is True and there is no human input.80 (2) When "TERMINATE", the agent only prompts for human input only when a termination message is received or81 the number of auto reply reaches the max_consecutive_auto_reply.82 (3) When "NEVER", the agent will never prompt for human input. Under this mode, the conversation stops83 when the number of auto reply reaches the max_consecutive_auto_reply or when is_termination_msg is True.84 function_map (dict[str, callable]): Mapping function names (passed to openai) to callable functions, also used for tool calls.85 code_execution_config (dict or False): config for the code execution.86 To disable code execution, set to False. Otherwise, set to a dictionary with the following keys:87 - work_dir (Optional, str): The working directory for the code execution.88 If None, a default working directory will be used.89 The default working directory is the "extensions" directory under90 "path_to_autogen".91 - use_docker (Optional, list, str or bool): The docker image to use for code execution.92 If a list or a str of image name(s) is provided, the code will be executed in a docker container93 with the first image successfully pulled.94 If None, False or empty, the code will be executed in the current environment.95 Default is True when the docker python package is installed.96 When set to True, a default list will be used.97 We strongly recommend using docker for code execution.98 - timeout (Optional, int): The maximum execution time in seconds.99 - last_n_messages (Experimental, Optional, int or str): The number of messages to look back for code execution. Default to 1. If set to 'auto', it will scan backwards through all messages arriving since the agent last spoke (typically this is the last time execution was attempted).100 llm_config (dict or False): llm inference configuration.101 Please refer to [OpenAIWrapper.create](/docs/reference/oai/client#create)102 for available options.103 To disable llm-based auto reply, set to False.104 default_auto_reply (str or dict or None): default auto reply when no code execution or llm-based reply is generated.105 description (str): a short description of the agent. This description is used by other agents106 (e.g. the GroupChatManager) to decide when to call upon this agent. (Default: system_message)107 """108 super().__init__(name)109 # a dictionary of conversations, default value is list110 self._oai_messages = defaultdict(list)111 self._oai_system_message = [{"content": system_message, "role": "system"}]112 self.description = description if description is not None else system_message113 self._is_termination_msg = (114 is_termination_msg115 if is_termination_msg is not None116 else (lambda x: content_str(x.get("content")) == "TERMINATE")117 )118 119 if llm_config is False:120 self.llm_config = False121 self.client = None122 else:123 self.llm_config = self.DEFAULT_CONFIG.copy()124 if isinstance(llm_config, dict):125 self.llm_config.update(llm_config)126 self.client = OpenAIWrapper(**self.llm_config)127 128 self._code_execution_config: Union[Dict, Literal[False]] = (129 {} if code_execution_config is None else code_execution_config130 )131 self.human_input_mode = human_input_mode132 self._max_consecutive_auto_reply = (133 max_consecutive_auto_reply if max_consecutive_auto_reply is not None else self.MAX_CONSECUTIVE_AUTO_REPLY134 )135 self._consecutive_auto_reply_counter = defaultdict(int)136 self._max_consecutive_auto_reply_dict = defaultdict(self.max_consecutive_auto_reply)137 self._function_map = (138 {}139 if function_map is None140 else {name: callable for name, callable in function_map.items() if self._assert_valid_name(name)}141 )142 self._default_auto_reply = default_auto_reply143 self._reply_func_list = []144 self.reply_at_receive = defaultdict(bool)145 self.register_reply([Agent, None], ConversableAgent.generate_oai_reply)146 self.register_reply([Agent, None], ConversableAgent.a_generate_oai_reply)147 self.register_reply([Agent, None], ConversableAgent.generate_code_execution_reply)148 self.register_reply([Agent, None], ConversableAgent.generate_tool_calls_reply)149 self.register_reply([Agent, None], ConversableAgent.a_generate_tool_calls_reply)150 self.register_reply([Agent, None], ConversableAgent.generate_function_call_reply)151 self.register_reply([Agent, None], ConversableAgent.a_generate_function_call_reply)152 self.register_reply([Agent, None], ConversableAgent.check_termination_and_human_reply)153 self.register_reply([Agent, None], ConversableAgent.a_check_termination_and_human_reply)154 155 # Registered hooks are kept in lists, indexed by hookable method, to be called in their order of registration.156 # New hookable methods should be added to this list as required to support new agent capabilities.157 self.hook_lists = {self.process_last_message: []} # This is currently the only hookable method.158 159 def register_reply(160 self,161 trigger: Union[Type[Agent], str, Agent, Callable[[Agent], bool], List],162 reply_func: Callable,163 position: int = 0,164 config: Optional[Any] = None,165 reset_config: Optional[Callable] = None,166 ):167 """Register a reply function.168 169 The reply function will be called when the trigger matches the sender.170 The function registered later will be checked earlier by default.171 To change the order, set the position to a positive integer.172 173 Args:174 trigger (Agent class, str, Agent instance, callable, or list): the trigger.175 - If a class is provided, the reply function will be called when the sender is an instance of the class.176 - If a string is provided, the reply function will be called when the sender's name matches the string.177 - If an agent instance is provided, the reply function will be called when the sender is the agent instance.178 - If a callable is provided, the reply function will be called when the callable returns True.179 - If a list is provided, the reply function will be called when any of the triggers in the list is activated.180 - If None is provided, the reply function will be called only when the sender is None.181 Note: Be sure to register `None` as a trigger if you would like to trigger an auto-reply function with non-empty messages and `sender=None`.182 reply_func (Callable): the reply function.183 The function takes a recipient agent, a list of messages, a sender agent and a config as input and returns a reply message.184 ```python185 def reply_func(186 recipient: ConversableAgent,187 messages: Optional[List[Dict]] = None,188 sender: Optional[Agent] = None,189 config: Optional[Any] = None,190 ) -> Tuple[bool, Union[str, Dict, None]]:191 ```192 position (int): the position of the reply function in the reply function list.193 The function registered later will be checked earlier by default.194 To change the order, set the position to a positive integer.195 config (Any): the config to be passed to the reply function.196 When an agent is reset, the config will be reset to the original value.197 reset_config (Callable): the function to reset the config.198 The function returns None. Signature: ```def reset_config(config: Any)```199 """200 if not isinstance(trigger, (type, str, Agent, Callable, list)):201 raise ValueError("trigger must be a class, a string, an agent, a callable or a list.")202 self._reply_func_list.insert(203 position,204 {205 "trigger": trigger,206 "reply_func": reply_func,207 "config": copy.copy(config),208 "init_config": config,209 "reset_config": reset_config,210 },211 )212 213 @property214 def system_message(self) -> Union[str, List]:215 """Return the system message."""216 return self._oai_system_message[0]["content"]217 218 def update_system_message(self, system_message: Union[str, List]):219 """Update the system message.220 221 Args:222 system_message (str or List): system message for the ChatCompletion inference.223 """224 self._oai_system_message[0]["content"] = system_message225 226 def update_max_consecutive_auto_reply(self, value: int, sender: Optional[Agent] = None):227 """Update the maximum number of consecutive auto replies.228 229 Args:230 value (int): the maximum number of consecutive auto replies.231 sender (Agent): when the sender is provided, only update the max_consecutive_auto_reply for that sender.232 """233 if sender is None:234 self._max_consecutive_auto_reply = value235 for k in self._max_consecutive_auto_reply_dict:236 self._max_consecutive_auto_reply_dict[k] = value237 else:238 self._max_consecutive_auto_reply_dict[sender] = value239 240 def max_consecutive_auto_reply(self, sender: Optional[Agent] = None) -> int:241 """The maximum number of consecutive auto replies."""242 return self._max_consecutive_auto_reply if sender is None else self._max_consecutive_auto_reply_dict[sender]243 244 @property245 def chat_messages(self) -> Dict[Agent, List[Dict]]:246 """A dictionary of conversations from agent to list of messages."""247 return self._oai_messages248 249 def last_message(self, agent: Optional[Agent] = None) -> Optional[Dict]:250 """The last message exchanged with the agent.251 252 Args:253 agent (Agent): The agent in the conversation.254 If None and more than one agent's conversations are found, an error will be raised.255 If None and only one conversation is found, the last message of the only conversation will be returned.256 257 Returns:258 The last message exchanged with the agent.259 """260 if agent is None:261 n_conversations = len(self._oai_messages)262 if n_conversations == 0:263 return None264 if n_conversations == 1:265 for conversation in self._oai_messages.values():266 return conversation[-1]267 raise ValueError("More than one conversation is found. Please specify the sender to get the last message.")268 if agent not in self._oai_messages.keys():269 raise KeyError(270 f"The agent '{agent.name}' is not present in any conversation. No history available for this agent."271 )272 return self._oai_messages[agent][-1]273 274 @property275 def use_docker(self) -> Union[bool, str, None]:276 """Bool value of whether to use docker to execute the code,277 or str value of the docker image name to use, or None when code execution is disabled.278 """279 return None if self._code_execution_config is False else self._code_execution_config.get("use_docker")280 281 @staticmethod282 def _message_to_dict(message: Union[Dict, str]) -> Dict:283 """Convert a message to a dictionary.284 285 The message can be a string or a dictionary. The string will be put in the "content" field of the new dictionary.286 """287 if isinstance(message, str):288 return {"content": message}289 elif isinstance(message, dict):290 return message291 else:292 return dict(message)293 294 @staticmethod295 def _normalize_name(name):296 """297 LLMs sometimes ask functions while ignoring their own format requirements, this function should be used to replace invalid characters with "_".298 299 Prefer _assert_valid_name for validating user configuration or input300 """301 return re.sub(r"[^a-zA-Z0-9_-]", "_", name)[:64]302 303 @staticmethod304 def _assert_valid_name(name):305 """306 Ensure that configured names are valid, raises ValueError if not.307 308 For munging LLM responses use _normalize_name to ensure LLM specified names don't break the API.309 """310 if not re.match(r"^[a-zA-Z0-9_-]+$", name):311 raise ValueError(f"Invalid name: {name}. Only letters, numbers, '_' and '-' are allowed.")312 if len(name) > 64:313 raise ValueError(f"Invalid name: {name}. Name must be less than 64 characters.")314 return name315 316 def _append_oai_message(self, message: Union[Dict, str], role, conversation_id: Agent) -> bool:317 """Append a message to the ChatCompletion conversation.318 319 If the message received is a string, it will be put in the "content" field of the new dictionary.320 If the message received is a dictionary but does not have any of the three fields "content", "function_call", or "tool_calls",321 this message is not a valid ChatCompletion message.322 If only "function_call" or "tool_calls" is provided, "content" will be set to None if not provided, and the role of the message will be forced "assistant".323 324 Args:325 message (dict or str): message to be appended to the ChatCompletion conversation.326 role (str): role of the message, can be "assistant" or "function".327 conversation_id (Agent): id of the conversation, should be the recipient or sender.328 329 Returns:330 bool: whether the message is appended to the ChatCompletion conversation.331 """332 message = self._message_to_dict(message)333 # create oai message to be appended to the oai conversation that can be passed to oai directly.334 oai_message = {335 k: message[k]336 for k in ("content", "function_call", "tool_calls", "tool_responses", "tool_call_id", "name", "context")337 if k in message and message[k] is not None338 }339 if "content" not in oai_message:340 if "function_call" in oai_message or "tool_calls" in oai_message:341 oai_message["content"] = None # if only function_call is provided, content will be set to None.342 else:343 return False344 345 if message.get("role") in ["function", "tool"]:346 oai_message["role"] = message.get("role")347 else:348 oai_message["role"] = role349 350 if oai_message.get("function_call", False) or oai_message.get("tool_calls", False):351 oai_message["role"] = "assistant" # only messages with role 'assistant' can have a function call.352 self._oai_messages[conversation_id].append(oai_message)353 return True354 355 def send(356 self,357 message: Union[Dict, str],358 recipient: Agent,359 request_reply: Optional[bool] = None,360 silent: Optional[bool] = False,361 ):362 """Send a message to another agent.363 364 Args:365 message (dict or str): message to be sent.366 The message could contain the following fields:367 - content (str or List): Required, the content of the message. (Can be None)368 - function_call (str): the name of the function to be called.369 - name (str): the name of the function to be called.370 - role (str): the role of the message, any role that is not "function"371 will be modified to "assistant".372 - context (dict): the context of the message, which will be passed to373 [OpenAIWrapper.create](../oai/client#create).374 For example, one agent can send a message A as:375 ```python376 {377 "content": lambda context: context["use_tool_msg"],378 "context": {379 "use_tool_msg": "Use tool X if they are relevant."380 }381 }382 ```383 Next time, one agent can send a message B with a different "use_tool_msg".384 Then the content of message A will be refreshed to the new "use_tool_msg".385 So effectively, this provides a way for an agent to send a "link" and modify386 the content of the "link" later.387 recipient (Agent): the recipient of the message.388 request_reply (bool or None): whether to request a reply from the recipient.389 silent (bool or None): (Experimental) whether to print the message sent.390 391 Raises:392 ValueError: if the message can't be converted into a valid ChatCompletion message.393 """394 # When the agent composes and sends the message, the role of the message is "assistant"395 # unless it's "function".396 valid = self._append_oai_message(message, "assistant", recipient)397 if valid:398 recipient.receive(message, self, request_reply, silent)399 else:400 raise ValueError(401 "Message can't be converted into a valid ChatCompletion message. Either content or function_call must be provided."402 )403 404 async def a_send(405 self,406 message: Union[Dict, str],407 recipient: Agent,408 request_reply: Optional[bool] = None,409 silent: Optional[bool] = False,410 ):411 """(async) Send a message to another agent.412 413 Args:414 message (dict or str): message to be sent.415 The message could contain the following fields:416 - content (str or List): Required, the content of the message. (Can be None)417 - function_call (str): the name of the function to be called.418 - name (str): the name of the function to be called.419 - role (str): the role of the message, any role that is not "function"420 will be modified to "assistant".421 - context (dict): the context of the message, which will be passed to422 [OpenAIWrapper.create](../oai/client#create).423 For example, one agent can send a message A as:424 ```python425 {426 "content": lambda context: context["use_tool_msg"],427 "context": {428 "use_tool_msg": "Use tool X if they are relevant."429 }430 }431 ```432 Next time, one agent can send a message B with a different "use_tool_msg".433 Then the content of message A will be refreshed to the new "use_tool_msg".434 So effectively, this provides a way for an agent to send a "link" and modify435 the content of the "link" later.436 recipient (Agent): the recipient of the message.437 request_reply (bool or None): whether to request a reply from the recipient.438 silent (bool or None): (Experimental) whether to print the message sent.439 440 Raises:441 ValueError: if the message can't be converted into a valid ChatCompletion message.442 """443 # When the agent composes and sends the message, the role of the message is "assistant"444 # unless it's "function".445 valid = self._append_oai_message(message, "assistant", recipient)446 if valid:447 await recipient.a_receive(message, self, request_reply, silent)448 else:449 raise ValueError(450 "Message can't be converted into a valid ChatCompletion message. Either content or function_call must be provided."451 )452 453 def _print_received_message(self, message: Union[Dict, str], sender: Agent):454 # print the message received455 print(colored(sender.name, "yellow"), "(to", f"{self.name}):\n", flush=True)456 message = self._message_to_dict(message)457 458 if message.get("tool_responses"): # Handle tool multi-call responses459 for tool_response in message["tool_responses"]:460 self._print_received_message(tool_response, sender)461 if message.get("role") == "tool":462 return # If role is tool, then content is just a concatenation of all tool_responses463 464 if message.get("role") in ["function", "tool"]:465 func_print = f"***** Response from calling {message['role']} \"{message['name']}\" *****"466 print(colored(func_print, "green"), flush=True)467 print(message["content"], flush=True)468 print(colored("*" * len(func_print), "green"), flush=True)469 else:470 content = message.get("content")471 if content is not None:472 if "context" in message:473 content = OpenAIWrapper.instantiate(474 content,475 message["context"],476 self.llm_config and self.llm_config.get("allow_format_str_template", False),477 )478 print(content_str(content), flush=True)479 if "function_call" in message and message["function_call"]:480 function_call = dict(message["function_call"])481 func_print = (482 f"***** Suggested function Call: {function_call.get('name', '(No function name found)')} *****"483 )484 print(colored(func_print, "green"), flush=True)485 print(486 "Arguments: \n",487 function_call.get("arguments", "(No arguments found)"),488 flush=True,489 sep="",490 )491 print(colored("*" * len(func_print), "green"), flush=True)492 if "tool_calls" in message and message["tool_calls"]:493 for tool_call in message["tool_calls"]:494 id = tool_call.get("id", "(No id found)")495 function_call = dict(tool_call.get("function", {}))496 func_print = f"***** Suggested tool Call ({id}): {function_call.get('name', '(No function name found)')} *****"497 print(colored(func_print, "green"), flush=True)498 print(499 "Arguments: \n",500 function_call.get("arguments", "(No arguments found)"),501 flush=True,502 sep="",503 )504 print(colored("*" * len(func_print), "green"), flush=True)505 506 print("\n", "-" * 80, flush=True, sep="")507 508 def _process_received_message(self, message: Union[Dict, str], sender: Agent, silent: bool):509 # When the agent receives a message, the role of the message is "user". (If 'role' exists and is 'function', it will remain unchanged.)510 valid = self._append_oai_message(message, "user", sender)511 if not valid:512 raise ValueError(513 "Received message can't be converted into a valid ChatCompletion message. Either content or function_call must be provided."514 )515 if not silent:516 self._print_received_message(message, sender)517 518 def receive(519 self,520 message: Union[Dict, str],521 sender: Agent,522 request_reply: Optional[bool] = None,523 silent: Optional[bool] = False,524 ):525 """Receive a message from another agent.526 527 Once a message is received, this function sends a reply to the sender or stop.528 The reply can be generated automatically or entered manually by a human.529 530 Args:531 message (dict or str): message from the sender. If the type is dict, it may contain the following reserved fields (either content or function_call need to be provided).532 1. "content": content of the message, can be None.533 2. "function_call": a dictionary containing the function name and arguments. (deprecated in favor of "tool_calls")534 3. "tool_calls": a list of dictionaries containing the function name and arguments.535 4. "role": role of the message, can be "assistant", "user", "function", "tool".536 This field is only needed to distinguish between "function" or "assistant"/"user".537 5. "name": In most cases, this field is not needed. When the role is "function", this field is needed to indicate the function name.538 6. "context" (dict): the context of the message, which will be passed to539 [OpenAIWrapper.create](../oai/client#create).540 sender: sender of an Agent instance.541 request_reply (bool or None): whether a reply is requested from the sender.542 If None, the value is determined by `self.reply_at_receive[sender]`.543 silent (bool or None): (Experimental) whether to print the message received.544 545 Raises:546 ValueError: if the message can't be converted into a valid ChatCompletion message.547 """548 self._process_received_message(message, sender, silent)549 if request_reply is False or request_reply is None and self.reply_at_receive[sender] is False:550 return551 reply = self.generate_reply(messages=self.chat_messages[sender], sender=sender)552 if reply is not None:553 self.send(reply, sender, silent=silent)554 555 async def a_receive(556 self,557 message: Union[Dict, str],558 sender: Agent,559 request_reply: Optional[bool] = None,560 silent: Optional[bool] = False,561 ):562 """(async) Receive a message from another agent.563 564 Once a message is received, this function sends a reply to the sender or stop.565 The reply can be generated automatically or entered manually by a human.566 567 Args:568 message (dict or str): message from the sender. If the type is dict, it may contain the following reserved fields (either content or function_call need to be provided).569 1. "content": content of the message, can be None.570 2. "function_call": a dictionary containing the function name and arguments. (deprecated in favor of "tool_calls")571 3. "tool_calls": a list of dictionaries containing the function name and arguments.572 4. "role": role of the message, can be "assistant", "user", "function".573 This field is only needed to distinguish between "function" or "assistant"/"user".574 5. "name": In most cases, this field is not needed. When the role is "function", this field is needed to indicate the function name.575 6. "context" (dict): the context of the message, which will be passed to576 [OpenAIWrapper.create](../oai/client#create).577 sender: sender of an Agent instance.578 request_reply (bool or None): whether a reply is requested from the sender.579 If None, the value is determined by `self.reply_at_receive[sender]`.580 silent (bool or None): (Experimental) whether to print the message received.581 582 Raises:583 ValueError: if the message can't be converted into a valid ChatCompletion message.584 """585 self._process_received_message(message, sender, silent)586 if request_reply is False or request_reply is None and self.reply_at_receive[sender] is False:587 return588 reply = await self.a_generate_reply(sender=sender)589 if reply is not None:590 await self.a_send(reply, sender, silent=silent)591 592 def _prepare_chat(self, recipient, clear_history):593 self.reset_consecutive_auto_reply_counter(recipient)594 recipient.reset_consecutive_auto_reply_counter(self)595 self.reply_at_receive[recipient] = recipient.reply_at_receive[self] = True596 if clear_history:597 self.clear_history(recipient)598 recipient.clear_history(self)599 600 def initiate_chat(601 self,602 recipient: "ConversableAgent",603 clear_history: Optional[bool] = True,604 silent: Optional[bool] = False,605 **context,606 ):607 """Initiate a chat with the recipient agent.608 609 Reset the consecutive auto reply counter.610 If `clear_history` is True, the chat history with the recipient agent will be cleared.611 `generate_init_message` is called to generate the initial message for the agent.612 613 Args:614 recipient: the recipient agent.615 clear_history (bool): whether to clear the chat history with the agent.616 silent (bool or None): (Experimental) whether to print the messages for this conversation.617 **context: any context information.618 "message" needs to be provided if the `generate_init_message` method is not overridden.619 """620 self._prepare_chat(recipient, clear_history)621 self.send(self.generate_init_message(**context), recipient, silent=silent)622 623 async def a_initiate_chat(624 self,625 recipient: "ConversableAgent",626 clear_history: Optional[bool] = True,627 silent: Optional[bool] = False,628 **context,629 ):630 """(async) Initiate a chat with the recipient agent.631 632 Reset the consecutive auto reply counter.633 If `clear_history` is True, the chat history with the recipient agent will be cleared.634 `generate_init_message` is called to generate the initial message for the agent.635 636 Args:637 recipient: the recipient agent.638 clear_history (bool): whether to clear the chat history with the agent.639 silent (bool or None): (Experimental) whether to print the messages for this conversation.640 **context: any context information.641 "message" needs to be provided if the `generate_init_message` method is not overridden.642 """643 self._prepare_chat(recipient, clear_history)644 await self.a_send(self.generate_init_message(**context), recipient, silent=silent)645 646 def reset(self):647 """Reset the agent."""648 self.clear_history()649 self.reset_consecutive_auto_reply_counter()650 self.stop_reply_at_receive()651 for reply_func_tuple in self._reply_func_list:652 if reply_func_tuple["reset_config"] is not None:653 reply_func_tuple["reset_config"](reply_func_tuple["config"])654 else:655 reply_func_tuple["config"] = copy.copy(reply_func_tuple["init_config"])656 657 def stop_reply_at_receive(self, sender: Optional[Agent] = None):658 """Reset the reply_at_receive of the sender."""659 if sender is None:660 self.reply_at_receive.clear()661 else:662 self.reply_at_receive[sender] = False663 664 def reset_consecutive_auto_reply_counter(self, sender: Optional[Agent] = None):665 """Reset the consecutive_auto_reply_counter of the sender."""666 if sender is None:667 self._consecutive_auto_reply_counter.clear()668 else:669 self._consecutive_auto_reply_counter[sender] = 0670 671 def clear_history(self, agent: Optional[Agent] = None):672 """Clear the chat history of the agent.673 674 Args:675 agent: the agent with whom the chat history to clear. If None, clear the chat history with all agents.676 """677 if agent is None:678 self._oai_messages.clear()679 else:680 self._oai_messages[agent].clear()681 682 def generate_oai_reply(683 self,684 messages: Optional[List[Dict]] = None,685 sender: Optional[Agent] = None,686 config: Optional[OpenAIWrapper] = None,687 ) -> Tuple[bool, Union[str, Dict, None]]:688 """Generate a reply using autogen.oai."""689 client = self.client if config is None else config690 if client is None:691 return False, None692 if messages is None:693 messages = self._oai_messages[sender]694 695 # unroll tool_responses696 all_messages = []697 for message in messages:698 tool_responses = message.get("tool_responses", [])699 if tool_responses:700 all_messages += tool_responses701 # tool role on the parent message means the content is just concatenation of all of the tool_responses702 if message.get("role") != "tool":703 all_messages.append({key: message[key] for key in message if key != "tool_responses"})704 else:705 all_messages.append(message)706 707 # TODO: #1143 handle token limit exceeded error708 response = client.create(709 context=messages[-1].pop("context", None), messages=self._oai_system_message + all_messages710 )711 712 extracted_response = client.extract_text_or_completion_object(response)[0]713 714 # ensure function and tool calls will be accepted when sent back to the LLM715 if not isinstance(extracted_response, str):716 extracted_response = model_dump(extracted_response)717 if isinstance(extracted_response, dict):718 if extracted_response.get("function_call"):719 extracted_response["function_call"]["name"] = self._normalize_name(720 extracted_response["function_call"]["name"]721 )722 for tool_call in extracted_response.get("tool_calls") or []:723 tool_call["function"]["name"] = self._normalize_name(tool_call["function"]["name"])724 return True, extracted_response725 726 async def a_generate_oai_reply(727 self,728 messages: Optional[List[Dict]] = None,729 sender: Optional[Agent] = None,730 config: Optional[Any] = None,731 ) -> Tuple[bool, Union[str, Dict, None]]:732 """Generate a reply using autogen.oai asynchronously."""733 return await asyncio.get_event_loop().run_in_executor(734 None, functools.partial(self.generate_oai_reply, messages=messages, sender=sender, config=config)735 )736 737 def generate_code_execution_reply(738 self,739 messages: Optional[List[Dict]] = None,740 sender: Optional[Agent] = None,741 config: Optional[Union[Dict, Literal[False]]] = None,742 ):743 """Generate a reply using code execution."""744 code_execution_config = config if config is not None else self._code_execution_config745 if code_execution_config is False:746 return False, None747 if messages is None:748 messages = self._oai_messages[sender]749 last_n_messages = code_execution_config.pop("last_n_messages", 1)750 751 messages_to_scan = last_n_messages752 if last_n_messages == "auto":753 # Find when the agent last spoke754 messages_to_scan = 0755 for i in range(len(messages)):756 message = messages[-(i + 1)]757 if "role" not in message:758 break759 elif message["role"] != "user":760 break761 else:762 messages_to_scan += 1763 764 # iterate through the last n messages in reverse765 # if code blocks are found, execute the code blocks and return the output766 # if no code blocks are found, continue767 for i in range(min(len(messages), messages_to_scan)):768 message = messages[-(i + 1)]769 if not message["content"]:770 continue771 code_blocks = extract_code(message["content"])772 if len(code_blocks) == 1 and code_blocks[0][0] == UNKNOWN:773 continue774 775 # found code blocks, execute code and push "last_n_messages" back776 exitcode, logs = self.execute_code_blocks(code_blocks)777 code_execution_config["last_n_messages"] = last_n_messages778 exitcode2str = "execution succeeded" if exitcode == 0 else "execution failed"779 return True, f"exitcode: {exitcode} ({exitcode2str})\nCode output: {logs}"780 781 # no code blocks are found, push last_n_messages back and return.782 code_execution_config["last_n_messages"] = last_n_messages783 784 return False, None785 786 def generate_function_call_reply(787 self,788 messages: Optional[List[Dict]] = None,789 sender: Optional[Agent] = None,790 config: Optional[Any] = None,791 ) -> Tuple[bool, Union[Dict, None]]:792 """793 Generate a reply using function call.794 795 "function_call" replaced by "tool_calls" as of [OpenAI API v1.1.0](https://github.com/openai/openai-python/releases/tag/v1.1.0)796 See https://platform.openai.com/docs/api-reference/chat/create#chat-create-functions797 """798 if config is None:799 config = self800 if messages is None:801 messages = self._oai_messages[sender]802 message = messages[-1]803 if "function_call" in message and message["function_call"]:804 func_call = message["function_call"]805 func = self._function_map.get(func_call.get("name", None), None)806 if asyncio.coroutines.iscoroutinefunction(func):807 return False, None808 809 _, func_return = self.execute_function(message["function_call"])810 return True, func_return811 return False, None812 813 async def a_generate_function_call_reply(814 self,815 messages: Optional[List[Dict]] = None,816 sender: Optional[Agent] = None,817 config: Optional[Any] = None,818 ) -> Tuple[bool, Union[Dict, None]]:819 """820 Generate a reply using async function call.821 822 "function_call" replaced by "tool_calls" as of [OpenAI API v1.1.0](https://github.com/openai/openai-python/releases/tag/v1.1.0)823 See https://platform.openai.com/docs/api-reference/chat/create#chat-create-functions824 """825 if config is None:826 config = self827 if messages is None:828 messages = self._oai_messages[sender]829 message = messages[-1]830 if "function_call" in message:831 func_call = message["function_call"]832 func_name = func_call.get("name", "")833 func = self._function_map.get(func_name, None)834 if func and asyncio.coroutines.iscoroutinefunction(func):835 _, func_return = await self.a_execute_function(func_call)836 return True, func_return837 838 return False, None839 840 def _str_for_tool_response(self, tool_response):841 func_name = tool_response.get("name", "")842 func_id = tool_response.get("tool_call_id", "")843 response = tool_response.get("content", "")844 return f"Tool call: {func_name}\nId: {func_id}\n{response}"845 846 def generate_tool_calls_reply(847 self,848 messages: Optional[List[Dict]] = None,849 sender: Optional[Agent] = None,850 config: Optional[Any] = None,851 ) -> Tuple[bool, Union[Dict, None]]:852 """Generate a reply using tool call."""853 if config is None:854 config = self855 if messages is None:856 messages = self._oai_messages[sender]857 message = messages[-1]858 tool_returns = []859 for tool_call in message.get("tool_calls", []):860 id = tool_call["id"]861 function_call = tool_call.get("function", {})862 func = self._function_map.get(function_call.get("name", None), None)863 if asyncio.coroutines.iscoroutinefunction(func):864 continue865 _, func_return = self.execute_function(function_call)866 tool_returns.append(867 {868 "tool_call_id": id,869 "role": "tool",870 "name": func_return.get("name", ""),871 "content": func_return.get("content", ""),872 }873 )874 if tool_returns:875 return True, {876 "role": "tool",877 "tool_responses": tool_returns,878 "content": "\n\n".join([self._str_for_tool_response(tool_return) for tool_return in tool_returns]),879 }880 return False, None881 882 async def _a_execute_tool_call(self, tool_call):883 id = tool_call["id"]884 function_call = tool_call.get("function", {})885 _, func_return = await self.a_execute_function(function_call)886 return {887 "tool_call_id": id,888 "role": "tool",889 "name": func_return.get("name", ""),890 "content": func_return.get("content", ""),891 }892 893 async def a_generate_tool_calls_reply(894 self,895 messages: Optional[List[Dict]] = None,896 sender: Optional[Agent] = None,897 config: Optional[Any] = None,898 ) -> Tuple[bool, Union[Dict, None]]:899 """Generate a reply using async function call."""900 if config is None:901 config = self902 if messages is None:903 messages = self._oai_messages[sender]904 message = messages[-1]905 async_tool_calls = []906 for tool_call in message.get("tool_calls", []):907 func = self._function_map.get(tool_call.get("function", {}).get("name", None), None)908 if func and asyncio.coroutines.iscoroutinefunction(func):909 async_tool_calls.append(self._a_execute_tool_call(tool_call))910 if async_tool_calls:911 tool_returns = await asyncio.gather(*async_tool_calls)912 return True, {913 "role": "tool",914 "tool_responses": tool_returns,915 "content": "\n\n".join([self._str_for_tool_response(tool_return) for tool_return in tool_returns]),916 }917 918 return False, None919 920 def check_termination_and_human_reply(921 self,922 messages: Optional[List[Dict]] = None,923 sender: Optional[Agent] = None,924 config: Optional[Any] = None,925 ) -> Tuple[bool, Union[str, None]]:926 """Check if the conversation should be terminated, and if human reply is provided.927 928 This method checks for conditions that require the conversation to be terminated, such as reaching929 a maximum number of consecutive auto-replies or encountering a termination message. Additionally,930 it prompts for and processes human input based on the configured human input mode, which can be931 'ALWAYS', 'NEVER', or 'TERMINATE'. The method also manages the consecutive auto-reply counter932 for the conversation and prints relevant messages based on the human input received.933 934 Args:935 - messages (Optional[List[Dict]]): A list of message dictionaries, representing the conversation history.936 - sender (Optional[Agent]): The agent object representing the sender of the message.937 - config (Optional[Any]): Configuration object, defaults to the current instance if not provided.938 939 Returns:940 - Tuple[bool, Union[str, Dict, None]]: A tuple containing a boolean indicating if the conversation941 should be terminated, and a human reply which can be a string, a dictionary, or None.942 """943 # Function implementation...944 945 if config is None:946 config = self947 if messages is None:948 messages = self._oai_messages[sender]949 message = messages[-1]950 reply = ""951 no_human_input_msg = ""952 if self.human_input_mode == "ALWAYS":953 reply = self.get_human_input(954 f"Provide feedback to {sender.name}. Press enter to skip and use auto-reply, or type 'exit' to end the conversation: "955 )956 no_human_input_msg = "NO HUMAN INPUT RECEIVED." if not reply else ""957 # if the human input is empty, and the message is a termination message, then we will terminate the conversation958 reply = reply if reply or not self._is_termination_msg(message) else "exit"959 else:960 if self._consecutive_auto_reply_counter[sender] >= self._max_consecutive_auto_reply_dict[sender]:961 if self.human_input_mode == "NEVER":962 reply = "exit"963 else:964 # self.human_input_mode == "TERMINATE":965 terminate = self._is_termination_msg(message)966 reply = self.get_human_input(967 f"Please give feedback to {sender.name}. Press enter or type 'exit' to stop the conversation: "968 if terminate969 else f"Please give feedback to {sender.name}. Press enter to skip and use auto-reply, or type 'exit' to stop the conversation: "970 )971 no_human_input_msg = "NO HUMAN INPUT RECEIVED." if not reply else ""972 # if the human input is empty, and the message is a termination message, then we will terminate the conversation973 reply = reply if reply or not terminate else "exit"974 elif self._is_termination_msg(message):975 if self.human_input_mode == "NEVER":976 reply = "exit"977 else:978 # self.human_input_mode == "TERMINATE":979 reply = self.get_human_input(980 f"Please give feedback to {sender.name}. Press enter or type 'exit' to stop the conversation: "981 )982 no_human_input_msg = "NO HUMAN INPUT RECEIVED." if not reply else ""983 # if the human input is empty, and the message is a termination message, then we will terminate the conversation984 reply = reply or "exit"985 986 # print the no_human_input_msg987 if no_human_input_msg:988 print(colored(f"\n>>>>>>>> {no_human_input_msg}", "red"), flush=True)989 990 # stop the conversation991 if reply == "exit":992 # reset the consecutive_auto_reply_counter993 self._consecutive_auto_reply_counter[sender] = 0994 return True, None995 996 # send the human reply997 if reply or self._max_consecutive_auto_reply_dict[sender] == 0:998 # reset the consecutive_auto_reply_counter999 self._consecutive_auto_reply_counter[sender] = 01000 # User provided a custom response, return function and tool failures indicating user interruption1001 tool_returns = []1002 if message.get("function_call", False):1003 tool_returns.append(1004 {1005 "role": "function",1006 "name": message["function_call"].get("name", ""),1007 "content": "USER INTERRUPTED",1008 }1009 )1010 1011 if message.get("tool_calls", False):1012 tool_returns.extend(1013 [1014 {"role": "tool", "tool_call_id": tool_call.get("id", ""), "content": "USER INTERRUPTED"}1015 for tool_call in message["tool_calls"]1016 ]1017 )1018 1019 response = {"role": "user", "content": reply}1020 if tool_returns:1021 response["tool_responses"] = tool_returns1022 1023 return True, response1024 1025 # increment the consecutive_auto_reply_counter1026 self._consecutive_auto_reply_counter[sender] += 11027 if self.human_input_mode != "NEVER":1028 print(colored("\n>>>>>>>> USING AUTO REPLY...", "red"), flush=True)1029 1030 return False, None1031 1032 async def a_check_termination_and_human_reply(1033 self,1034 messages: Optional[List[Dict]] = None,1035 sender: Optional[Agent] = None,1036 config: Optional[Any] = None,1037 ) -> Tuple[bool, Union[str, None]]:1038 """(async) Check if the conversation should be terminated, and if human reply is provided.1039 1040 This method checks for conditions that require the conversation to be terminated, such as reaching1041 a maximum number of consecutive auto-replies or encountering a termination message. Additionally,1042 it prompts for and processes human input based on the configured human input mode, which can be1043 'ALWAYS', 'NEVER', or 'TERMINATE'. The method also manages the consecutive auto-reply counter1044 for the conversation and prints relevant messages based on the human input received.1045 1046 Args:1047 - messages (Optional[List[Dict]]): A list of message dictionaries, representing the conversation history.1048 - sender (Optional[Agent]): The agent object representing the sender of the message.1049 - config (Optional[Any]): Configuration object, defaults to the current instance if not provided.1050 1051 Returns:1052 - Tuple[bool, Union[str, Dict, None]]: A tuple containing a boolean indicating if the conversation1053 should be terminated, and a human reply which can be a string, a dictionary, or None.1054 """1055 if config is None:1056 config = self1057 if messages is None:1058 messages = self._oai_messages[sender]1059 message = messages[-1]1060 reply = ""1061 no_human_input_msg = ""1062 if self.human_input_mode == "ALWAYS":1063 reply = await self.a_get_human_input(1064 f"Provide feedback to {sender.name}. Press enter to skip and use auto-reply, or type 'exit' to end the conversation: "1065 )1066 no_human_input_msg = "NO HUMAN INPUT RECEIVED." if not reply else ""1067 # if the human input is empty, and the message is a termination message, then we will terminate the conversation1068 reply = reply if reply or not self._is_termination_msg(message) else "exit"1069 else:1070 if self._consecutive_auto_reply_counter[sender] >= self._max_consecutive_auto_reply_dict[sender]:1071 if self.human_input_mode == "NEVER":1072 reply = "exit"1073 else:1074 # self.human_input_mode == "TERMINATE":1075 terminate = self._is_termination_msg(message)1076 reply = await self.a_get_human_input(1077 f"Please give feedback to {sender.name}. Press enter or type 'exit' to stop the conversation: "1078 if terminate1079 else f"Please give feedback to {sender.name}. Press enter to skip and use auto-reply, or type 'exit' to stop the conversation: "1080 )1081 no_human_input_msg = "NO HUMAN INPUT RECEIVED." if not reply else ""1082 # if the human input is empty, and the message is a termination message, then we will terminate the conversation1083 reply = reply if reply or not terminate else "exit"1084 elif self._is_termination_msg(message):1085 if self.human_input_mode == "NEVER":1086 reply = "exit"1087 else:1088 # self.human_input_mode == "TERMINATE":1089 reply = await self.a_get_human_input(1090 f"Please give feedback to {sender.name}. Press enter or type 'exit' to stop the conversation: "1091 )1092 no_human_input_msg = "NO HUMAN INPUT RECEIVED." if not reply else ""1093 # if the human input is empty, and the message is a termination message, then we will terminate the conversation1094 reply = reply or "exit"1095 1096 # print the no_human_input_msg1097 if no_human_input_msg:1098 print(colored(f"\n>>>>>>>> {no_human_input_msg}", "red"), flush=True)1099 1100 # stop the conversation1101 if reply == "exit":1102 # reset the consecutive_auto_reply_counter1103 self._consecutive_auto_reply_counter[sender] = 01104 return True, None1105 1106 # send the human reply1107 if reply or self._max_consecutive_auto_reply_dict[sender] == 0:1108 # User provided a custom response, return function and tool results indicating user interruption1109 # reset the consecutive_auto_reply_counter1110 self._consecutive_auto_reply_counter[sender] = 01111 tool_returns = []1112 if message.get("function_call", False):1113 tool_returns.append(1114 {1115 "role": "function",1116 "name": message["function_call"].get("name", ""),1117 "content": "USER INTERRUPTED",1118 }1119 )1120 1121 if message.get("tool_calls", False):1122 tool_returns.extend(1123 [1124 {"role": "tool", "tool_call_id": tool_call.get("id", ""), "content": "USER INTERRUPTED"}1125 for tool_call in message["tool_calls"]1126 ]1127 )1128 1129 response = {"role": "user", "content": reply}1130 if tool_returns:1131 response["tool_responses"] = tool_returns1132 1133 return True, response1134 1135 # increment the consecutive_auto_reply_counter1136 self._consecutive_auto_reply_counter[sender] += 11137 if self.human_input_mode != "NEVER":1138 print(colored("\n>>>>>>>> USING AUTO REPLY...", "red"), flush=True)1139 1140 return False, None1141 1142 def generate_reply(1143 self,1144 messages: Optional[List[Dict]] = None,1145 sender: Optional[Agent] = None,1146 exclude: Optional[List[Callable]] = None,1147 ) -> Union[str, Dict, None]:1148 """Reply based on the conversation history and the sender.1149 1150 Either messages or sender must be provided.1151 Register a reply_func with `None` as one trigger for it to be activated when `messages` is non-empty and `sender` is `None`.1152 Use registered auto reply functions to generate replies.1153 By default, the following functions are checked in order:1154 1. check_termination_and_human_reply1155 2. generate_function_call_reply (deprecated in favor of tool_calls)1156 3. generate_tool_calls_reply1157 4. generate_code_execution_reply1158 5. generate_oai_reply1159 Every function returns a tuple (final, reply).1160 When a function returns final=False, the next function will be checked.1161 So by default, termination and human reply will be checked first.1162 If not terminating and human reply is skipped, execute function or code and return the result.1163 AI replies are generated only when no code execution is performed.1164 1165 Args:1166 messages: a list of messages in the conversation history.1167 default_reply (str or dict): default reply.1168 sender: sender of an Agent instance.1169 exclude: a list of functions to exclude.1170 1171 Returns:1172 str or dict or None: reply. None if no reply is generated.1173 """1174 if all((messages is None, sender is None)):1175 error_msg = f"Either {messages=} or {sender=} must be provided."1176 logger.error(error_msg)1177 raise AssertionError(error_msg)1178 1179 if messages is None:1180 messages = self._oai_messages[sender]1181 1182 # Call the hookable method that gives registered hooks a chance to process the last message.1183 # Message modifications do not affect the incoming messages or self._oai_messages.1184 messages = self.process_last_message(messages)1185 1186 for reply_func_tuple in self._reply_func_list:1187 reply_func = reply_func_tuple["reply_func"]1188 if exclude and reply_func in exclude:1189 continue1190 if asyncio.coroutines.iscoroutinefunction(reply_func):1191 continue1192 if self._match_trigger(reply_func_tuple["trigger"], sender):1193 final, reply = reply_func(self, messages=messages, sender=sender, config=reply_func_tuple["config"])1194 if final:1195 return reply1196 return self._default_auto_reply1197 1198 async def a_generate_reply(1199 self,1200 messages: Optional[List[Dict]] = None,