agentchat.conversable_agent

ConversableAgent​

class ConversableAgent(LLMAgent)

（预览中）一个用于通用对话代理的类，可以配置为助手或用户代理。

在收到每条消息后，代理将向发送者发送回复，除非该消息是终止消息。 例如，AssistantAgent和UserProxyAgent是该类的子类， 配置了不同的默认设置。

要修改自动回复，请重写generate_reply方法。 要在每一轮禁用/启用人机响应，将human_input_mode设置为“NEVER”或“ALWAYS”。 要修改获取人工输入的方式，请重写get_human_input方法。 要修改执行代码块、单个代码块或函数调用的方式，请分别重写execute_code_blocks、 run_code和execute_function方法。

DEFAULT_CONFIG​

False 或 dict，llm 推理的默认配置

MAX_CONSECUTIVE_AUTO_REPLY​

最大连续自动回复次数（未来可能会有变动）

__init__​

def __init__(name: str,
             system_message: Optional[Union[
                 str, List]] = "You are a helpful AI Assistant.",
             is_termination_msg: Optional[Callable[[Dict], bool]] = None,
             max_consecutive_auto_reply: Optional[int] = None,
             human_input_mode: Literal["ALWAYS", "NEVER",
                                       "TERMINATE"] = "TERMINATE",
             function_map: Optional[Dict[str, Callable]] = None,
             code_execution_config: Union[Dict, Literal[False]] = False,
             llm_config: Optional[Union[Dict, Literal[False]]] = None,
             default_auto_reply: Union[str, Dict] = "",
             description: Optional[str] = None,
             chat_messages: Optional[Dict[Agent, List[Dict]]] = None,
             silent: Optional[bool] = None)

参数:

• name str - 代理的名称。
• system_message str or list - ChatCompletion 推理的系统消息。
• is_termination_msg 函数 - 一个接收字典形式的消息并返回布尔值的函数，用于指示接收到的消息是否为终止消息。 该字典可以包含以下键："content"（内容）, "role"（角色）, "name"（名称）, "function_call"（函数调用）。
• max_consecutive_auto_reply int - 最大连续自动回复次数。 默认为 None（未提供限制，此时将使用类属性 MAX_CONSECUTIVE_AUTO_REPLY 作为限制）。 当设置为 0 时，不会生成自动回复。
• human_input_mode str - 是否在每次收到消息时请求人工输入。 可能的值为 "ALWAYS", "TERMINATE", "NEVER"。 (1) 当值为 "ALWAYS" 时，代理每次收到消息时都会提示人工输入。 在此模式下，当人工输入为 "exit" 时，对话结束， 或者当 is_termination_msg 为 True 且没有人工输入时。 (2) 当值为 "TERMINATE" 时，代理仅在收到终止消息或 自动回复次数达到 max_consecutive_auto_reply 时提示人工输入。 (3) 当值为 "NEVER" 时，代理永远不会提示人工输入。在此模式下，对话结束 当自动回复次数达到 max_consecutive_auto_reply 或 is_termination_msg 为 True 时。
• function_map dict[str, callable] - 将函数名称（传递给openai）映射到可调用函数，也用于工具调用。
• code_execution_config dict or False - config for the code execution. To disable code execution, set to False. Otherwise, set to a dictionary with the following keys:
  • work_dir（可选，str）：代码执行的工作目录。 如果为None，将使用默认的工作目录。 默认工作目录是“path_to_autogen”下的“extensions”目录。
  • use_docker (可选, list, str 或 bool): 用于代码执行的docker镜像。 默认值为 True，意味着代码将在docker容器中执行。将使用默认的镜像列表。 如果提供了一个镜像名称的列表或字符串，代码将在成功拉取到的第一个镜像的docker容器中执行。 如果为 False，代码将在当前环境中执行。 我们强烈建议使用docker来执行代码。
  • timeout（可选，int）：最大执行时间（以秒为单位）。
  • last_n_messages（实验性，int 或 str）：需要回顾的消息数量以执行代码。如果设置为 'auto'，它将扫描自代理上次发言以来收到的所有消息，这通常是上次尝试执行的时间。（默认值：auto）
• llm_config 字典或False或None - llm推理配置。 请参考OpenAIWrapper.create 以获取可用选项。 当使用OpenAI或Azure OpenAI端点时，请在llm_config或llm_config中每个config_list的配置中指定一个非空的'model'。 要禁用基于llm的自动回复，请设置为False。 当设置为None时，将使用self.DEFAULT_CONFIG，默认值为False。
• default_auto_reply str or dict - 当没有代码执行或基于LLM的回复生成时的默认自动回复。
• description str - 代理的简短描述。此描述由其他代理（例如 GroupChatManager）用来决定何时调用此代理。（默认：system_message）
• chat_messages dict or None - 该代理与其他代理之前的聊天记录。 可以通过提供聊天历史记录来给予代理记忆。这将允许代理继续之前的对话。默认为空的聊天记录。
• silent 布尔值或空 - (实验性) 是否打印发送的消息。如果为空，将使用每个函数中的silent值。

名称​

@property
def name() -> str

获取代理的名称。

描述​

@property
def description() -> str

获取代理的描述。

描述​

@description.setter
def description(description: str)

设置代理的描述。

代码执行器​

@property
def code_executor() -> Optional[CodeExecutor]

此代理使用的代码执行器。如果代码执行被禁用，则返回 None。

register_reply​

def register_reply(trigger: Union[Type[Agent], str, Agent,
                                  Callable[[Agent], bool], List],
                   reply_func: Callable,
                   position: int = 0,
                   config: Optional[Any] = None,
                   reset_config: Optional[Callable] = None,
                   *,
                   ignore_async_in_sync_chat: bool = False,
                   remove_other_reply_funcs: bool = False)

注册一个回复函数。

当触发器与发送者匹配时，将调用回复函数。 默认情况下，稍后注册的函数会优先检查。 要更改顺序，请将位置设置为正整数。

同步和异步回复函数都可以被注册。同步回复函数将会从同步和异步聊天中触发。然而，异步回复函数只会从异步聊天中触发（通过ConversableAgent.a_initiate_chat发起）。如果注册了一个async回复函数并且使用同步函数初始化了一个聊天，ignore_async_in_sync_chat将决定行为如下：如果ignore_async_in_sync_chat设置为False（默认值），将会引发异常，如果ignore_async_in_sync_chat设置为True，回复函数将会被忽略。

参数:

• trigger Agent 类、字符串、Agent 实例、可调用对象或列表 - 触发器。 如果提供了一个类，当发送者是该类的实例时，将调用回复函数。 如果提供了一个字符串，当发送者的名称与该字符串匹配时，将调用回复函数。 如果提供了一个 agent 实例，当发送者是该 agent 实例时，将调用回复函数。 如果提供了一个可调用对象，当该可调用对象返回 True 时，将调用回复函数。 如果提供了一个列表，当列表中的任何触发器被激活时，将调用回复函数。 如果提供了 None，则仅在发送者为 None 时调用回复函数。

• 注意 - 如果你想在收到非空消息时触发自动回复功能，并且sender=None，请确保将None注册为触发器。

• reply_func Callable - 回复函数。 该函数接收一个接收者代理、消息列表、发送者代理和配置作为输入，并返回一个回复消息。

    ```python
    def reply_func(
        recipient: ConversableAgent,
        messages: Optional[List[Dict]] = None,
        sender: Optional[Agent] = None,
        config: Optional[Any] = None,
    ) -> Tuple[bool, Union[str, Dict, None]]:
    ```
  

• position int - 回复函数在回复函数列表中的位置。 默认情况下，后注册的函数会先被检查。 要改变顺序，请将位置设置为正整数。

• config Any - 传递给回复函数的配置。 当代理被重置时，配置将重置为原始值。

• reset_config 可调用 - 用于重置配置的函数。 该函数返回 None。签名：def reset_config(config: Any)

• ignore_async_in_sync_chat bool - 是否在同步对话中忽略异步回复函数。如果为False，当异步回复函数已注册且对话使用同步函数初始化时，将抛出异常。

• remove_other_reply_funcs bool - 在注册此回复函数时是否移除其他回复函数。

replace_reply_func​

def replace_reply_func(old_reply_func: Callable, new_reply_func: Callable)

用一个新的函数替换已注册的回复函数。

参数:

• old_reply_func 可调用的 - 要被替换的旧回复函数。
• new_reply_func 可调用对象 - 用于替换旧回复函数的新回复函数。

register_nested_chats​

def register_nested_chats(chat_queue: List[Dict[str, Any]],
                          trigger: Union[Type[Agent], str, Agent,
                                         Callable[[Agent], bool], List],
                          reply_func_from_nested_chats: Union[
                              str, Callable] = "summary_from_nested_chats",
                          position: int = 2,
                          use_async: Union[bool, None] = None,
                          **kwargs) -> None

注册一个嵌套的聊天回复函数。

参数:

• chat_queue 列表 - 要初始化的聊天对象列表。如果使用了 use_async，那么 chat_queue 中的所有消息都必须有一个与之关联的 chat-id。
• trigger Agent类、字符串、Agent实例、可调用对象或列表 - 详情请参阅register_reply。
• reply_func_from_nested_chats Callable, str - the reply function for the nested chat. The function takes a chat_queue for nested chat, recipient agent, a list of messages, a sender agent and a config as input and returns a reply message. Default to "summary_from_nested_chats", which corresponds to a built-in reply function that get summary from the nested chat_queue.

  def reply_func_from_nested_chats(
      chat_queue: List[Dict],
      recipient: ConversableAgent,
      messages: Optional[List[Dict]] = None,
      sender: Optional[Agent] = None,
      config: Optional[Any] = None,
  ) -> Tuple[bool, Union[str, Dict, None]]:
  
• position int - 详情请参考 register_reply。默认值为 2。这意味着我们首先检查终止条件和人类回复，然后检查已注册的嵌套聊天回复。
• use_async - 内部使用a_initiate_chats来启动嵌套聊天。如果原始聊天是通过a_initiate_chats启动的，您可以将其设置为true，以便嵌套聊天不会同步运行。
• kwargs - 详情请参考 register_reply.

系统消息​

@property
def system_message() -> str

返回系统消息。

update_system_message​

def update_system_message(system_message: str) -> None

更新系统消息。

参数:

• system_message str - ChatCompletion推理的系统消息。

update_max_consecutive_auto_reply​

def update_max_consecutive_auto_reply(value: int,
                                      sender: Optional[Agent] = None)

更新最大连续自动回复次数。

参数:

• value int - 最大连续自动回复次数。
• sender Agent - 当提供了发送者时，仅更新该发送者的最大连续自动回复次数。

max_consecutive_auto_reply​

def max_consecutive_auto_reply(sender: Optional[Agent] = None) -> int

最大连续自动回复次数。

聊天消息​

@property
def chat_messages() -> Dict[Agent, List[Dict]]

从代理到消息列表的对话字典。

chat_messages_for_summary​

def chat_messages_for_summary(agent: Agent) -> List[Dict]

作为对话需要总结的消息列表。

last_message​

def last_message(agent: Optional[Agent] = None) -> Optional[Dict]

与代理交换的最后一条消息。

参数:

• agent Agent - 对话中的代理。 如果为 None 并且找到多个代理的对话，将会引发错误。 如果为 None 并且只找到一个对话，将返回该对话的最后一条消息。

返回:

与代理的最后一条消息。

使用docker​

@property
def use_docker() -> Union[bool, str, None]

布尔值，表示是否使用docker来执行代码，或字符串值，表示要使用的docker镜像名称，或None当代码执行被禁用时。

发送​

def send(message: Union[Dict, str],
         recipient: Agent,
         request_reply: Optional[bool] = None,
         silent: Optional[bool] = False)

向另一个代理发送消息。

参数:

• message dict or str - message to be sent. The message could contain the following fields:
  • content (str or List): 必需，消息的内容。（可以为None）
  • function_call (str): 要调用的函数名称。
  • name (str): 要调用的函数的名称。
  • role (str): 消息的角色，任何不是"function"的角色将被修改为"assistant"。
  • context (dict): 消息的上下文，将会传递给 OpenAIWrapper.create。 例如，一个代理可以发送消息 A 如下：

{
    "content": lambda context: context["use_tool_msg"],
    "context": {
        "use_tool_msg": "Use tool X if they are relevant."
    }
}

下次，一个代理可以发送带有不同“use_tool_msg”的消息B。 然后消息A的内容将被刷新为新的“use_tool_msg”。 因此，这有效地提供了一种方式，让代理可以发送一个“链接”并在之后修改“链接”的内容。

• recipient Agent - 消息的接收者。
• request_reply 布尔值或空值 - 是否请求接收者回复。
• silent bool or None - （实验性）是否打印发送的消息。

引发:

• ValueError - 如果消息无法转换为有效的ChatCompletion消息。

a_send​

async def a_send(message: Union[Dict, str],
                 recipient: Agent,
                 request_reply: Optional[bool] = None,
                 silent: Optional[bool] = False)

(异步) 发送消息给另一个代理。

参数:

• message dict or str - message to be sent. The message could contain the following fields:
  • content (str or List): 必需，消息的内容。（可以为None）
  • function_call (str): 要调用的函数名称。
  • name (str): 要调用的函数的名称。
  • role (str): 消息的角色，任何不是"function"的角色将被修改为"assistant"。
  • context (dict): 消息的上下文，将会传递给 OpenAIWrapper.create。 例如，一个代理可以发送消息 A 如下：

{
    "content": lambda context: context["use_tool_msg"],
    "context": {
        "use_tool_msg": "Use tool X if they are relevant."
    }
}

下一次，一个代理可以发送一条带有不同“use_tool_msg”的消息B。 然后，消息A的内容将刷新为新的“use_tool_msg”。 因此，这实际上为代理提供了一种发送“链接”并稍后修改“链接”内容的方式。

• recipient Agent - 消息的接收者。
• request_reply 布尔值或空值 - 是否请求接收者回复。
• silent 布尔值或空值 - （实验性）是否打印发送的消息。

引发:

• ValueError - 如果消息无法转换为有效的ChatCompletion消息。

接收​

def receive(message: Union[Dict, str],
            sender: Agent,
            request_reply: Optional[bool] = None,
            silent: Optional[bool] = False)

从另一个代理接收消息。

一旦接收到消息，此函数会向发送者发送回复或停止。 回复可以自动生成或由人工手动输入。

参数:

• 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).
  1. "content": 消息的内容，可以为空。
  2. "function_call": 包含函数名称和参数的字典。（已弃用，推荐使用"tool_calls"）
  3. "tool_calls"：包含函数名称和参数的字典列表。
  4. "role": 消息的角色，可以是 "assistant"、"user"、"function"、"tool"。 此字段仅用于区分 "function" 或 "assistant"/"user"。
  5. "name": 在大多数情况下，不需要此字段。当角色为 "function" 时，需要此字段来指示函数名称。
  6. "context" (dict): 消息的上下文，将传递给 OpenAIWrapper.create。
• sender - Agent 实例的发送者。
• request_reply 布尔值或None - 是否从发送者那里请求回复。 如果为None，该值由self.reply_at_receive[sender]决定。
• silent bool 或 None - (实验性) 是否打印接收到的消息。

引发:

• ValueError - 如果消息无法转换为有效的ChatCompletion消息。

a_receive​

async def a_receive(message: Union[Dict, str],
                    sender: Agent,
                    request_reply: Optional[bool] = None,
                    silent: Optional[bool] = False)

（异步）从另一个代理接收消息。

一旦接收到消息，此函数会向发送者发送回复或停止。 回复可以自动生成或由人工手动输入。

参数:

• 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).
  1. "content": 消息的内容，可以为空。
  2. "function_call": 包含函数名称和参数的字典。（已弃用，推荐使用"tool_calls"）
  3. "tool_calls"：包含函数名称和参数的字典列表。
  4. "role": 消息的角色，可以是 "assistant"、"user"、"function"。 该字段仅用于区分 "function" 或 "assistant"/"user"。
  5. "name": 在大多数情况下，不需要此字段。当角色为 "function" 时，需要此字段来指示函数名称。
  6. "context" (dict): 消息的上下文，将传递给 OpenAIWrapper.create。
• sender - Agent 实例的发送者。
• request_reply 布尔值或None - 是否从发送者那里请求回复。 如果为None，该值由self.reply_at_receive[sender]决定。
• silent bool 或 None - (实验性) 是否打印接收到的消息。

引发:

• ValueError - 如果消息无法转换为有效的ChatCompletion消息。

initiate_chat​

def initiate_chat(recipient: "ConversableAgent",
                  clear_history: bool = True,
                  silent: Optional[bool] = False,
                  cache: Optional[AbstractCache] = None,
                  max_turns: Optional[int] = None,
                  summary_method: Optional[Union[
                      str, Callable]] = DEFAULT_SUMMARY_METHOD,
                  summary_args: Optional[dict] = {},
                  message: Optional[Union[Dict, str, Callable]] = None,
                  **kwargs) -> ChatResult

启动与接收代理的聊天。

重置连续自动回复计数器。 如果 clear_history 为 True，与接收方代理的聊天历史将被清除。

参数:

• recipient - 接收方代理。

• clear_history bool - 是否清除与代理的聊天历史记录。默认为True。

• silent 布尔值或空值 - （实验性）是否打印此对话的消息。默认值为False。

• cache AbstractCache or None - 用于此对话的缓存客户端。默认为None。

• max_turns int or None - 两个代理之间聊天的最大轮次。一轮意味着一次完整的对话往返。请注意，这与 max_consecutive_auto_reply 不同，后者是连续自动回复的最大次数；它也与 GroupChat 中的 max_rounds 不同，后者是群聊会话中的最大轮次。 如果 max_turns 设置为 None，聊天将继续，直到满足终止条件。默认值为 None。

• summary_method str or callable - 一种从聊天中获取摘要的方法。默认是 DEFAULT_SUMMARY_METHOD，即 "last_msg"。

  支持的字符串为 "last_msg" 和 "reflection_with_llm"：

  • 当设置为“last_msg”时，它将返回对话的最后一条消息作为摘要。
  • 当设置为"reflection_with_llm"时，它将返回使用llm客户端提取的摘要。 llm_config必须在接收者或发送者中设置。

  一个可调用的 summary_method 应该将聊天中的接收者和发送者代理作为输入，并返回一个摘要字符串。例如，

  def my_summary_method(
      sender: ConversableAgent,
      recipient: ConversableAgent,
      summary_args: dict,
  ):
      return recipient.last_message(sender)["content"]
  

• summary_args dict - 一个传递给 summary_method 的参数字典。 一个示例键是 "summary_prompt"，其值是一个用于提示基于LLM的代理（发送者或接收者代理）在 summary_method 为 "reflection_with_llm" 时，反思对话并提取摘要的文本字符串。 默认的 summary_prompt 是 DEFAULT_SUMMARY_PROMPT，即 "总结对话的要点。不要添加任何介绍性短语。如果未妥善处理预期请求，请指出。" 另一个可用的键是 "summary_role"，即发送给负责摘要的代理的消息角色。默认是 "system"。

• message str, dict or Callable - 发送给接收者的初始消息。需要提供。否则，将调用 input() 来获取初始消息。

  • 如果提供的是字符串或字典，它将作为初始消息。 generate_init_message 被调用来根据此字符串和上下文生成代理的初始消息。 如果是字典，它可能包含以下保留字段（需要提供内容或工具调用）。

  1. "content": 消息的内容，可以为None。

  2. "function_call": 一个包含函数名称和参数的字典。（已弃用，推荐使用"tool_calls"）

  3. "tool_calls": 包含函数名称和参数的字典列表。

  4. "role": 消息的角色，可以是 "assistant"、"user"、"function"。 此字段仅用于区分 "function" 或 "assistant"/"user"。

  5. "name": 在大多数情况下，不需要该字段。当角色为"function"时，需要该字段来指示函数名称。

  6. "context" (dict): 消息的上下文，将会传递给 OpenAIWrapper.create。

  • 如果提供了一个可调用对象，它将被调用来获取初始消息，形式为字符串或字典。 如果返回的类型是字典，它可能包含上述提到的保留字段。

  可调用消息的示例（返回字符串）：

  def my_message(sender: ConversableAgent, recipient: ConversableAgent, context: dict) -> Union[str, Dict]:
      carryover = context.get("carryover", "")
      if isinstance(message, list):
          carryover = carryover[-1]
      final_msg = "Write a blogpost." + "\nContext: \n" + carryover
      return final_msg
  

  可调用消息的示例（返回一个字典）：

  def my_message(sender: ConversableAgent, recipient: ConversableAgent, context: dict) -> Union[str, Dict]:
      final_msg = {}
      carryover = context.get("carryover", "")
      if isinstance(message, list):
          carryover = carryover[-1]
      final_msg["content"] = "Write a blogpost." + "\nContext: \n" + carryover
      final_msg["context"] = {"prefix": "Today I feel"}
      return final_msg
  

• **kwargs - 任何附加信息。它包含以下保留字段：

  • "carryover": 一个字符串或字符串列表，用于指定要传递给本次聊天信息的延续信息。 如果提供了该信息，我们将在生成初始聊天消息时，通过在消息内容后附加一个"context: "字符串和延续内容，与"message"内容结合使用。
  • "verbose": 一个布尔值，用于指定是否打印聊天中的消息和剩余部分。默认值为False。

引发:

• RuntimeError - 如果在同步聊天中注册了任何异步回复函数且未被忽略。

返回:

• ChatResult - 一个 ChatResult 对象。

a_initiate_chat​

async def a_initiate_chat(recipient: "ConversableAgent",
                          clear_history: bool = True,
                          silent: Optional[bool] = False,
                          cache: Optional[AbstractCache] = None,
                          max_turns: Optional[int] = None,
                          summary_method: Optional[Union[
                              str, Callable]] = DEFAULT_SUMMARY_METHOD,
                          summary_args: Optional[dict] = {},
                          message: Optional[Union[str, Callable]] = None,
                          **kwargs) -> ChatResult

(异步) 启动与接收代理的聊天。

重置连续自动回复计数器。 如果 clear_history 为 True，则与接收方代理的聊天记录将被清除。 a_generate_init_message 被调用来生成代理的初始消息。

参数：请参考initiate_chat。

返回:

• ChatResult - 一个 ChatResult 对象。

initiate_chats​

def initiate_chats(chat_queue: List[Dict[str, Any]]) -> List[ChatResult]

（实验性）与多个代理发起聊天。

参数:

• chat_queue List[Dict] - 包含聊天信息的一个字典列表。 每个字典应包含initiate_chat的输入参数。

• Returns - 返回与chat_queue中已完成聊天对应的ChatResult对象列表。

获取聊天结果​

def get_chat_results(
        chat_index: Optional[int] = None
) -> Union[List[ChatResult], ChatResult]

总结自特定代理的已完成聊天记录。

重置​

def reset()

重置代理。

stop_reply_at_receive​

def stop_reply_at_receive(sender: Optional[Agent] = None)

重置发送者的reply_at_receive。

reset_consecutive_auto_reply_counter​

def reset_consecutive_auto_reply_counter(sender: Optional[Agent] = None)

重置发送者的 consecutive_auto_reply_counter。

clear_history​

def clear_history(recipient: Optional[Agent] = None,
                  nr_messages_to_preserve: Optional[int] = None)

清除代理的聊天历史记录。

参数:

• recipient - 要清除聊天记录的代理。如果为 None，则清除与所有代理的聊天记录。
• nr_messages_to_preserve - 要在聊天历史中保留的最新消息的数量。

generate_oai_reply​

def generate_oai_reply(
    messages: Optional[List[Dict]] = None,
    sender: Optional[Agent] = None,
    config: Optional[OpenAIWrapper] = None
) -> Tuple[bool, Union[str, Dict, None]]

使用autogen.oai生成回复。

a_generate_oai_reply​

async def a_generate_oai_reply(
        messages: Optional[List[Dict]] = None,
        sender: Optional[Agent] = None,
        config: Optional[Any] = None) -> Tuple[bool, Union[str, Dict, None]]

使用autogen.oai异步生成回复。

generate_code_execution_reply​

def generate_code_execution_reply(
        messages: Optional[List[Dict]] = None,
        sender: Optional[Agent] = None,
        config: Optional[Union[Dict, Literal[False]]] = None)

使用代码执行生成回复。

generate_function_call_reply​

def generate_function_call_reply(
        messages: Optional[List[Dict]] = None,
        sender: Optional[Agent] = None,
        config: Optional[Any] = None) -> Tuple[bool, Union[Dict, None]]

使用函数调用生成回复。

"function_call" 自 OpenAI API v1.1.0 起替换为 "tool_calls" 参见 https://platform.openai.com/docs/api-reference/chat/create#chat-create-functions

a_generate_function_call_reply​

async def a_generate_function_call_reply(
        messages: Optional[List[Dict]] = None,
        sender: Optional[Agent] = None,
        config: Optional[Any] = None) -> Tuple[bool, Union[Dict, None]]

使用异步函数调用生成回复。

自OpenAI API v1.1.0起，"function_call"已被替换为"tool_calls" 参见https://platform.openai.com/docs/api-reference/chat/create#chat-create-functions

generate_tool_calls_reply​

def generate_tool_calls_reply(
        messages: Optional[List[Dict]] = None,
        sender: Optional[Agent] = None,
        config: Optional[Any] = None) -> Tuple[bool, Union[Dict, None]]

使用工具调用生成回复。

一个生成工具调用的回复​

async def a_generate_tool_calls_reply(
        messages: Optional[List[Dict]] = None,
        sender: Optional[Agent] = None,
        config: Optional[Any] = None) -> Tuple[bool, Union[Dict, None]]

使用异步函数调用生成回复。

check_termination_and_human_reply​

def check_termination_and_human_reply(
        messages: Optional[List[Dict]] = None,
        sender: Optional[Agent] = None,
        config: Optional[Any] = None) -> Tuple[bool, Union[str, None]]

检查对话是否应终止，以及是否提供了人类回复。

该方法检查是否需要终止对话的条件，例如达到连续自动回复的最大次数或遇到终止消息。此外，根据配置的人工输入模式（可以是'ALWAYS'、'NEVER'或'TERMINATE'）提示并处理人工输入。该方法还管理对话的连续自动回复计数器，并根据接收到的人工输入打印相关消息。

参数:

• messages（可选[List[Dict]]）：消息字典列表，表示对话历史记录。
• sender (Optional[Agent]): 表示消息发送者的代理对象。
• config (可选[Any]): 配置对象，如果未提供，默认为当前实例。

返回:

• Tuple[bool, Union[str, Dict, None]]: 一个包含布尔值的元组，表示对话是否应终止，以及人类回复，回复可以是字符串、字典或None。

a_check_termination_and_human_reply​

async def a_check_termination_and_human_reply(
        messages: Optional[List[Dict]] = None,
        sender: Optional[Agent] = None,
        config: Optional[Any] = None) -> Tuple[bool, Union[str, None]]

(异步) 检查对话是否应终止，以及是否提供了人类回复。

该方法检查是否需要终止对话的条件，例如达到连续自动回复的最大次数或遇到终止消息。此外，它会根据配置的人工输入模式（可以是'ALWAYS'、'NEVER'或'TERMINATE'）提示并处理人工输入。该方法还管理对话的连续自动回复计数器，并根据接收到的人工输入打印相关消息。

参数:

• messages（可选[List[Dict]]）：消息字典列表，表示对话历史记录。
• sender (Optional[Agent]): 表示消息发送者的代理对象。
• config (Optional[Any]): 配置对象，如果未提供则默认为当前实例。

返回:

• Tuple[bool, Union[str, Dict, None]]: 一个包含布尔值的元组，指示对话是否应终止，以及可以是一个字符串、字典或None的人类回复。

generate_reply​

def generate_reply(messages: Optional[List[Dict[str, Any]]] = None,
                   sender: Optional["Agent"] = None,
                   **kwargs: Any) -> Union[str, Dict, None]

根据对话历史和发件人进行回复。

必须提供消息或发送者。 注册一个回复函数，将None作为其中一个触发器，以便在messages非空且sender为None时激活它。 使用注册的自动回复函数来生成回复。 默认情况下，按顺序检查以下函数：

1. check_termination_and_human_reply
2. generate_function_call_reply (已弃用，推荐使用 tool_calls)
3. generate_tool_calls_reply
4. generate_code_execution_reply
5. generate_oai_reply 每个函数返回一个元组 (final, reply)。 当函数返回 final=False 时，将检查下一个函数。 因此默认情况下，将首先检查终止和人类回复。 如果未终止且人类回复被跳过，则执行函数或代码并返回结果。 只有在未执行代码时才会生成AI回复。

参数:

• messages - 对话历史中的消息列表。

• sender - 一个Agent实例的发送者。

  额外的关键字参数：

• exclude List[Callable] - 要排除的回复函数列表。

返回:

字符串 或 字典 或 无：回复。如果没有生成回复则为无。

a_generate_reply​

async def a_generate_reply(messages: Optional[List[Dict[str, Any]]] = None,
                           sender: Optional["Agent"] = None,
                           **kwargs: Any) -> Union[str, Dict[str, Any], None]

（异步）根据对话历史和发送者进行回复。

必须提供消息或发送者。 注册一个回复函数，将None作为其中一个触发器，以便在messages非空且sender为None时激活它。 使用注册的自动回复函数来生成回复。 默认情况下，按顺序检查以下函数：

1. check_termination_and_human_reply
2. generate_function_call_reply
3. generate_tool_calls_reply
4. generate_code_execution_reply
5. generate_oai_reply 每个函数返回一个元组 (final, reply)。 当函数返回 final=False 时，将检查下一个函数。 因此默认情况下，将首先检查终止和人类回复。 如果未终止且人类回复被跳过，则执行函数或代码并返回结果。 只有在未执行代码时才会生成AI回复。

参数:

• messages - 对话历史中的消息列表。

• sender - 一个Agent实例的发送者。

  额外的关键字参数：

• exclude List[Callable] - 要排除的回复函数列表。

返回:

str 或 dict 或 None: 回复。如果没有生成回复，则为 None。

获取人工输入​

def get_human_input(prompt: str) -> str

获取人类输入。

重写此方法以自定义获取人工输入的方式。

参数:

• prompt str - 人类输入的提示。

返回:

• str - 人类输入。

a_get_human_input​

async def a_get_human_input(prompt: str) -> str

(异步) 获取用户输入。

重写此方法以自定义获取人工输入的方式。

参数:

• prompt str - 人类输入的提示。

返回:

• str - 用户输入。

run_code​

def run_code(code, **kwargs)

运行代码并返回结果。

重写此函数以修改运行代码的方式。

参数:

• code str - 要执行的代码。
• **kwargs - 其他关键字参数。

返回:

一个包含 (exitcode, logs, image) 的元组。

• exitcode int - 代码执行的退出码。
• logs str - 代码执行的日志。
• image str 或 None - 用于代码执行的docker镜像。

执行代码块​

def execute_code_blocks(code_blocks)

执行代码块并返回结果。

execute_function​

def execute_function(func_call,
                     verbose: bool = False) -> Tuple[bool, Dict[str, str]]

执行一个函数调用并返回结果。

重写此函数以修改执行函数和工具调用的方式。

参数:

• func_call - 一个从openai消息中提取的字典，位于"function_call"或"tool_calls"处，包含键"name"和"arguments"。

返回:

一个元组，包含 (is_exec_success, result_dict)。

• is_exec_success boolean - 判断执行是否成功。

• result_dict - 一个包含键 "name", "role", 和 "content" 的字典。"role" 的值为 "function"。

  "function_call" 自 OpenAI API v1.1.0 起已弃用 参见 https://platform.openai.com/docs/api-reference/chat/create#chat-create-function_call

a_execute_function​

async def a_execute_function(func_call)

执行一个异步函数调用并返回结果。

重写此函数以修改异步函数和工具的执行方式。

参数:

• func_call - 从openai消息中提取的字典，位于键"function_call"或"tool_calls"下，包含键"name"和"arguments"。

返回:

一个元组，包含 (is_exec_success, result_dict)。

• is_exec_success boolean - 判断执行是否成功。

• result_dict - 一个包含键“name”、“role”和“content”的字典。“role”的值为“function”。

  从OpenAI API v1.1.0开始，已弃用“function_call” 参见https://platform.openai.com/docs/api-reference/chat/create#chat-create-function_call

generate_init_message​

def generate_init_message(message: Union[Dict, str, None],
                          **kwargs) -> Union[str, Dict]

为代理生成初始消息。 如果消息为None，将调用input()来获取初始消息。

参数:

• message str 或 None - 要处理的消息。
• **kwargs - 任何附加信息。它有以下保留字段：
• "carryover" - 一个字符串或字符串列表，用于指定要传递给此聊天的继承信息。它可以是字符串或字符串列表。如果提供了，我们将在生成初始聊天消息时将此继承信息与"message"内容结合。

返回:

str 或 dict：处理后的消息。

a_generate_init_message​

async def a_generate_init_message(message: Union[Dict, str, None],
                                  **kwargs) -> Union[str, Dict]

为代理生成初始消息。 如果消息为None，将调用input()来获取初始消息。

参数:

请参考generate_init_message了解参数的描述。

返回:

str 或 dict：处理后的消息。

register_function​

def register_function(function_map: Dict[str, Union[Callable, None]])

将函数注册到代理。

参数:

• function_map - 一个将函数名映射到函数的字典。如果function_map[name]为None，该函数将从function_map中移除。

更新函数签名​

def update_function_signature(func_sig: Union[str, Dict], is_remove: None)

更新LLM配置中的function_signature以进行function_call。

参数:

• func_sig str or dict - 要更新/移除到模型的函数的描述/名称。参见：https://platform.openai.com/docs/api-reference/chat/create#chat/create-functions

• is_remove - 是否从名为 'func_sig' 的 llm_config 中移除该函数

  自 OpenAI API v1.1.0 起已弃用 参见 https://platform.openai.com/docs/api-reference/chat/create#chat-create-function_call

update_tool_signature​

def update_tool_signature(tool_sig: Union[str, Dict], is_remove: None)

更新LLM配置中的tool_signature以进行tool_call。

参数:

• tool_sig str or dict - 要更新或从模型中移除的工具的描述/名称。参见：https://platform.openai.com/docs/api-reference/chat/create#chat-create-tools
• is_remove - 是否从名为'tool_sig'的llm_config中移除该工具

can_execute_function​

def can_execute_function(name: Union[List[str], str]) -> bool

代理是否可以执行该函数。

function_map​

@property
def function_map() -> Dict[str, Callable]

返回函数映射。

注册LLM

def register_for_llm(
        *,
        name: Optional[str] = None,
        description: Optional[str] = None,
        api_style: Literal["function", "tool"] = "tool") -> Callable[[F], F]

用于注册供代理使用的函数的装饰器工厂。

它的返回值用于装饰一个要注册到agent的函数。该函数使用类型提示来指定参数和返回类型。函数名称用作函数的默认名称，但可以提供自定义名称。函数描述用于在agent的配置中描述该函数。

参数:

name (optional(str)): 函数的名称。如果为None，将使用函数名称（默认值：None）。 description (optional(str)): 函数的描述（默认值：None）。对于初始装饰器是必需的， 但后续的装饰器可以省略它。

• api_style - (literal): 函数调用的API风格。 对于Azure OpenAI API，请使用版本2023-12-01-preview或更高版本。 "function"风格将被弃用。对于早期版本，如果"tool"不起作用，请使用"function"。 详情请参阅Azure OpenAI 文档。

返回:

用于注册一个函数供代理使用的装饰器。

示例:

```
@user_proxy.register_for_execution()
@agent2.register_for_llm()
@agent1.register_for_llm(description="This is a very useful function")
def my_function(a: Annotated[str, "description of a parameter"] = "a", b: int, c=3.14) -> str:
     return a + str(b * c)
```

对于2023-12-01-preview之前的Azure OpenAI版本，如果"tool"不起作用，请将api_style设置为"function"： @agent2.register_for_llm(api_style="function") def my_function(a: Annotated[str, "a参数的描述"] = "a", b: int, c=3.14) -> str: return a + str(b * c)

register_for_execution​

def register_for_execution(name: Optional[str] = None) -> Callable[[F], F]

用于注册一个由代理执行的函数的装饰器工厂。

它的返回值用于装饰一个要注册到代理的函数。

参数:

名称 (optional(str)): 函数的名称。如果为 None，将使用函数名（默认值：None）。

返回:

用于注册一个函数供代理使用的装饰器。

示例:

```
@user_proxy.register_for_execution()
@agent2.register_for_llm()
@agent1.register_for_llm(description="This is a very useful function")
def my_function(a: Annotated[str, "description of a parameter"] = "a", b: int, c=3.14):
     return a + str(b * c)
```

register_model_client​

def register_model_client(model_client_cls: ModelClient, **kwargs)

注册一个模型客户端。

参数:

• model_client_cls - 一个遵循Client接口的自定义客户端类
• **kwargs - 用于初始化自定义客户端类的kwargs

register_hook​

def register_hook(hookable_method: str, hook: Callable)

注册一个挂钩，以便被可挂钩方法调用，从而为代理添加功能。 已注册的挂钩保存在列表中（每个可挂钩方法一个），并按注册顺序调用。

参数:

• hookable_method - 由ConversableAgent实现的可挂钩方法名称。
• hook - 由 AgentCapability 子类实现的方法。

process_all_messages_before_reply​

def process_all_messages_before_reply(messages: List[Dict]) -> List[Dict]

调用任何已注册的能力钩子来处理所有消息，可能会修改这些消息。

a_process_all_messages_before_reply​

async def a_process_all_messages_before_reply(
        messages: List[Dict]) -> List[Dict]

调用任何已注册的能力钩子来处理所有消息，可能会修改这些消息。

处理最后接收到的消息​

def process_last_received_message(messages: List[Dict]) -> List[Dict]

调用任何已注册的功能钩子来使用并可能修改最后一条消息的文本，只要最后一条消息不是函数调用或退出命令。

a_process_last_received_message​

async def a_process_last_received_message(messages: List[Dict]) -> List[Dict]

调用任何已注册的功能钩子以使用并可能修改上一条消息的文本，只要上一条消息不是函数调用或退出命令。

print_usage_summary​

def print_usage_summary(
        mode: Union[str, List[str]] = ["actual", "total"]) -> None

打印使用摘要。

get_actual_usage​

def get_actual_usage() -> Union[None, Dict[str, int]]

获取实际使用摘要。

获取总使用量​

def get_total_usage() -> Union[None, Dict[str, int]]

获取总使用情况摘要。

register_function​

def register_function(f: Callable[..., Any],
                      *,
                      caller: ConversableAgent,
                      executor: ConversableAgent,
                      name: Optional[str] = None,
                      description: str) -> None

注册一个由代理提出并由执行者执行的函数。

这个函数可以代替函数装饰器@ConversationAgent.register_for_llm和 @ConversationAgent.register_for_execution。

参数:

• f - 要注册的函数。
• caller - 调用该函数的代理，通常是 ConversableAgent 的一个实例。
• executor - 执行该函数的代理，通常是UserProxy的实例。
• name - 函数的名称。如果为None，将使用函数名称（默认值：None）。
• description - 函数的描述。该描述由LLM用于解码是否调用该函数。确保描述正确地描述了函数的功能，否则在需要时可能不会被LLM调用。