Skip to main content

conversable_agent

可对话代理​

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)

参数：

  • name str - 代理的名称。
  • system_message str 或 list - 用于 ChatCompletion 推理的系统消息。
  • is_termination_msg function - 一个函数，接受一个以字典形式表示的消息，并返回一个布尔值，指示接收到的消息是否为终止消息。 字典可以包含以下键："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 - 代码执行的配置。 要禁用代码执行，请设置为 False。否则，设置为一个包含以下键的字典：
  • 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 dict or False or 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 - 代理以前与其他代理进行的聊天消息。 可以通过提供聊天历史记录来给代理提供记忆。这将允许代理恢复以前的对话。默认为空的聊天历史记录。

name​

@property
def name() -> str

获取代理的名称。

description​

@property
def description() -> str

获取代理的描述。

description​

@description.setter
def description(description: str)
@description.setter
def description(description: str)

description 是一个属性的 setter 方法，它接受一个 str 类型的参数 description。 设置代理的描述。

code_executor​

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

  • Note - 如果您希望使用非空消息和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 任意类型 - 要传递给回复函数的配置。当代理重置时，配置将重置为初始值。

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

  • ignore_async_in_sync_chat 布尔值 - 是否忽略同步聊天中的异步回复函数。如果为 False，并且已注册异步回复函数并且使用同步函数初始化了聊天，将会引发异常。

  • remove_other_reply_funcs 布尔值 - 是否在注册此回复函数时删除其他回复函数。

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,
                          **kwargs) -> None

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

参数：

  • chat_queue 列表 - 要初始化的聊天对象列表。
  • trigger Agent 类、字符串、Agent 实例、可调用对象或列表 - 详见 register_reply 的详细信息。
  • reply_func_from_nested_chats 可调用对象、字符串 - 嵌套聊天的回复函数。该函数接受嵌套聊天的 chat_queue、接收者代理、消息列表、发送者代理和配置作为输入，并返回一个回复消息。默认为 "summary_from_nested_chats"，对应一个内置的回复函数，从嵌套的 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 整数 - 详见 register_reply 的详细信息。默认为 2。这意味着我们首先检查终止和人工回复，然后检查已注册的嵌套聊天回复。
  • kwargs - 详见 register_reply 的详细信息。

system_message​

@property
def system_message() -> str

返回系统消息。

update_system_message​

def update_system_message(system_message: str) -> None

更新系统消息。

参数：

  • system_message 字符串 - 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。

max_consecutive_auto_reply​

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

连续自动回复的最大次数。

chat_messages​

@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 并且只找到一个对话，则返回唯一对话的最后一条消息。

返回值：

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

use_docker​

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

是否使用 Docker 执行代码的布尔值， 或者要使用的 Docker 镜像名称的字符串值，或者当禁用代码执行时为 None。

send​

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

向另一个代理发送消息。

参数：

  • message dict or str - 要发送的消息。 消息可以包含以下字段：
    • 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": "如果相关，使用工具 X。"
    }
}

下一次，一个代理可以发送消息 B，其中包含不同的 "use_tool_msg"。 然后，消息 A 的内容将刷新为新的 "use_tool_msg"。 因此，这实际上为代理发送一个 "链接" 并在以后修改 "链接" 的内容提供了一种方式。

  • recipient Agent - 消息的接收者。
  • request_reply bool or None - 是否要求接收者回复。
  • silent bool 或 None - (实验性) 是否打印发送的消息。

抛出：

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

a_send​

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

(异步) 向另一个 agent 发送消息。

参数：

  • message dict 或 str - 要发送的消息。 消息可以包含以下字段：
    • content (str 或 List): 必需，消息的内容。(可以为 None)
    • function_call (str): 要调用的函数的名称。
    • name (str): 要调用的函数的名称。
    • role (str): 消息的角色，任何不是 "function" 的角色都会被修改为 "assistant"。
    • context (dict): 消息的上下文，将传递给 OpenAIWrapper.create。 例如，一个 agent 可以发送消息 A：
{
    "content": lambda context: context["use_tool_msg"],
    "context": {
        "use_tool_msg": "如果相关，请使用工具 X。"
    }
}

下一次，一个 agent 可以发送消息 B，其中包含不同的 "use_tool_msg"。 然后，消息 A 的内容将被刷新为新的 "use_tool_msg"。 因此，这实际上提供了一种让 agent 发送一个 "链接" 并在以后修改 "链接" 内容的方法。

  • recipient Agent - 消息的接收者。
  • request_reply bool 或 None - 是否要求接收者回复。
  • silent bool 或 None - (实验性) 是否打印发送的消息。

抛出：

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

receive​

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

从另一个 agent 接收消息。

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

参数：

  • message dict 或 str - 来自发送者的消息。如果类型为 dict，则可能包含以下保留字段（需要提供 content 或 function_call 中的一个）。

    1. "content": 消息的内容，可以为 None。
    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函数用于创建一个Agent实例。

  • sender - Agent实例的发送者。

  • request_reply bool或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)

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

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

参数：

  • message dict或str - 来自发送者的消息。如果类型为dict，则可以包含以下保留字段（需要提供content或function_call之一）。
    1. "content"：消息的内容，可以为None。
    2. "function_call"：包含函数名和参数的字典（已弃用，推荐使用"tool_calls"）。
    3. "tool_calls"：包含函数名和参数的字典列表。
    4. "role"：消息的角色，可以是"assistant"、"user"或"function"。此字段仅用于区分"function"和"assistant"/"user"。
    5. "name"：在大多数情况下，不需要此字段。当角色为"function"时，需要此字段来指示函数名。
    6. "context"（字典）：消息的上下文，将传递给OpenAIWrapper.create函数。
  • sender - Agent实例的发送者。
  • request_reply bool或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

与接收者Agent开始一次聊天。

重置连续自动回复计数器。 如果clear_history为True，则会清除与接收者Agent的聊天历史记录。

参数：

  • recipient - 接收者Agent。

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

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

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

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

  • summary_method str 或 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，即 "Summarize takeaway from the conversation. Do not add any introductory phrases. If the intended request is NOT properly addressed, please point it out."。 另一个可用的键是 "summary_role"，它是发送给负责总结的代理的消息的角色。默认为 "system"。

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

    • 如果提供了字符串或字典，它将被用作初始消息。generate_init_message 将根据该字符串和上下文生成代理的初始消息。 如果是字典，则可能包含以下保留字段（content 或 tool_calls 需要提供其中之一）。

    1. "content"：消息的内容，可以为 None。
    2. "function_call"：包含函数名称和参数的字典。（已弃用，推荐使用 "tool_calls"）
    3. "tool_calls"：包含函数名称和参数的字典列表。
  1. "role": 消息的角色，可以是 "assistant"（助手）、"user"（用户）或 "function"（函数）。这个字段只用于区分 "function" 和 "assistant"/"user"。
  2. "name": 在大多数情况下，不需要这个字段。当角色是 "function" 时，需要使用这个字段来指定函数的名称。
  3. "context"（字典）：消息的上下文，将传递给 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 = "写一篇博客文章。" + "\n上下文：\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"] = "写一篇博客文章。" + "\n上下文：\n" + carryover
    final_msg["context"] = {"prefix": "今天我感觉"}
    return final_msg
  • **kwargs - 任何其他信息。它具有以下保留字段：
    • "carryover"：一个字符串或字符串列表，用于指定要传递给此聊天的保留信息。如果提供了此字段，我们将在生成 generate_init_message 中的初始聊天消息时将此保留信息（通过在消息内容后附加一个 "context: " 字符串和保留内容）与 "message" 内容结合起来。
    • "verbose"：一个布尔值，用于指定是否在聊天中打印消息和保留信息。默认为 False。

Raises:

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

Returns:

  • 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 的输入参数。

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

get_chat_results​

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

从特定代理的已完成聊天中获取摘要。

reset​

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]]

使用函数调用生成回复。

OpenAI API v1.1.0 中将 "function_call" 替换为 "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]]

使用工具调用生成回复。

a_generate_tool_calls_reply​

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 (Optional[List[Dict]]): 一个消息字典列表，表示对话历史记录。
  • sender (Optional[Agent]): 表示消息发送者的代理对象。
  • config (Optional[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'）提示并处理人类输入。该方法还管理对话的连续自动回复计数器，并根据接收到的人类输入打印相关消息。

generate_reply​

根据对话历史和发送者生成回复。

必须提供 messages 或 sender 中的一个。 注册一个 reply_func，其中一个触发器为 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] - 要排除的回复函数列表。

返回：

str 或 dict 或 None：回复。如果没有生成回复，则返回 None。

a_generate_reply​

(async) 根据对话历史和发送者生成回复。

必须提供 messages 或 sender 中的一个。 注册一个 reply_func，其中一个触发器为 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。

get_human_input​

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 镜像。

execute_code_blocks​

def execute_code_blocks(code_blocks)

执行代码块并返回结果。

execute_function​

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

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

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

参数：

  • func_call - 从 "function_call" 或 "tool_calls" 中提取的字典，包含键 "name" 和 "arguments"。

返回值：

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

a_execute_function​

async def a_execute_function(func_call)

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

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

参数：

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

返回值：

一个元组 (is_exec_success, result_dict)。

generate_init_message​

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

生成代理的初始消息。 如果 message 为 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]

生成代理的初始消息。 如果 message 为 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 中删除该函数。

update_function_signature​

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

更新 LLM 配置中 function_call 的函数签名。

参数：

update_tool_signature​

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

更新 LLM 配置中 tool_call 的工具签名。

参数：

can_execute_function​

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

判断代理是否能执行该函数。

function_map​

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

返回函数映射表。

register_for_llm​

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

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

返回值用于装饰要注册到代理的函数。函数使用类型提示来指定参数和返回类型。函数名将作为函数的默认名称，但也可以提供自定义名称。函数描述用于描述代理的配置中的函数。

参数：

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

  • api_style - (字面值)：函数调用的 API 风格。 对于 Azure OpenAI API，请使用 2023-12-01-preview 版本或更高版本。 如果 "tool" 不起作用，请使用 "function" 风格。对于较早版本，请参阅 Azure OpenAI 文档 了解详情。

返回值：

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

示例：

```
@user_proxy.register_for_execution()
@agent2.register_for_llm()
@agent1.register_for_llm(description="这是一个非常有用的函数")
def my_function(a: Annotated[str, "a 参数的描述"] = "a", b: int, c=3.14) -> str:
     return a + str(b * c)
```

对于 Azure OpenAI 版本在 2023-12-01-preview 之前的版本，如果 "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]

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

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

参数：

name (可选(str))：函数的名称。如果为 None，则使用函数名（默认值：None）。

返回值：

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

register_model_client​

def register_model_client(model_client_cls: ModelClient, **kwargs)

注册一个模型客户端。

参数：

  • model_client_cls - 一个遵循 Client 接口的自定义客户端类
  • **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]

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

process_last_received_message​

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

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

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

打印使用情况摘要。

get_actual_usage​

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

获取实际使用情况摘要。

get_total_usage​

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 调用。