揭开生成式人工智能代理的神秘面纱

Rifx.Online
Generative AI , Chatbots , Autonomous Systems
19 Jan, 2025

从单一交互到复杂的多代理系统

概述

在生成式人工智能代理的热潮中迷失了吗？你并不孤单。这篇文章穿透噪音，提供了对代理的清晰定义及其工作原理。我们分解了关键组件，包括“工具”的重要角色，并提供了从单次交互到复杂的多代理系统的构建和部署的实用见解。我们还探讨了多代理架构如何在企业环境中实施，并与微服务进行类比。未来的文章将深入探讨代理与运营（AgentOps）以及如何为企业规模的多代理系统构建平台。

什么是代理？

“代理就是一个提示，它指示基础模型与特定工具进行交互”

生成式 AI 代理通过精心设计的提示来协调基础模型 (FM) 与外部工具之间的交互。这些提示指示 FM 在何时和如何使用这些工具。

每个“工具”本质上是一组功能声明（或者我们称之为“声明”）。这些声明包括：

功能名称： 工具的标识符。
描述： 工具目的的全面解释，解决的问题，以及在何种情况下可以使用它。
参数： 输入参数的列表，包括它们的含义、类型和预期值的描述。
输出（可选）： 预期输出格式和内容的描述。

为了规范这些声明，市场上通常使用基于 JSON 的 OpenAPI 格式。这种标准化格式允许清晰、机器可读的 API 描述，从而促进与生成式 AI 模型的无缝集成。以下是使用 OpenAPI 格式在工具列表中检索股票价格的功能声明示例：

{
  "tools": [
    {
      "functionDeclarations": [
        {
          "name": "get_stock_price",
          "description": "Fetch the current stock price of a given company.",
          "parameters": {
            "type": "object",
            "properties": {
              "ticker": {
                "type": "string",
                "description": "Stock ticker symbol (e.g., AAPL, MSFT)."
              }
            },
            "required": ["ticker"]
          },
          "returns": {
              "type": "number",
              "description": "The current stock price."
          }
        }
      ]
    }
  ]
}

可以使用 Python SDK（例如 Vertex AI 中可用的 SDK）以编程方式创建相同的功能声明。这允许动态创建和管理工具规范：

from vertexai.generative_models import (
    FunctionDeclaration,
    GenerationConfig,
    GenerativeModel,
    Part,
    Tool,
)

## 创建功能声明
get_stock_price = FunctionDeclaration(
    name="get_stock_price",
    description="Fetch the current stock price of a given company",
    parameters={
        "type": "object",
        "properties": {
            "ticker": {
                "type": "string",
                "description": "Stock ticker symbol for a company",
            }
        },
        "required": ["ticker"]
    },
    returns={ # 添加返回字段！
        "type": "number",
        "description": "The current stock price."
    }
)
## ... 提供更多功能声明

## 将可用功能定义为工具
company_insights_tool = Tool(
    function_declarations=[
        get_stock_price,
        # ... 其他功能声明
    ],
)

为了实际展示这些概念，我们将在以下示例中使用 Vertex AI SDK 和 Gemini 的功能调用能力，基于这个代码库（我们强烈建议您进行探索）。这种方法为理解代理如何在更低层次上工作提供了坚实的基础。一旦掌握这些基本知识，您将能够很好地使用更高级的代理框架，如 LangChain。

到目前为止，我们专注于使用 JSON 定义工具的结构，并演示如何使用 Vertex AI SDK 以编程方式创建这些定义。这些工具定义最终被转换为文本，并附加到指令提示中。这使得模型能够推理出哪个工具（如果有的话）是满足用户请求所必需的，以及使用哪些参数。

以下是一个示例，演示这些元素——工具、模型和指令——如何结合在一起：

## 选择 LLM，配置并提供可用的工具
gemini_model = GenerativeModel(
    "gemini-2.0-flash-exp",
    generation_config=GenerationConfig(temperature=0),
    tools=[company_insights_tool],
)

## 为 LLM 准备指令
instruction = """
    给出简洁的高层次摘要。 
    仅使用您从
    API 响应中学到的信息。
    """

agent = gemini_model.start_chat()

下一步是开始向模型发送新输入：

## 为 LLM 准备您的查询/问题
query = "Google 当前的股票价格是多少？"

## 将指令和查询发送给 LLM
prompt = instruction + query
response = agent.send_message(prompt)

您认为响应会是什么样的？模型会调用一个真实的函数吗？

如果 company_insights_tool 正确地定义（包括具有 ticker 参数和返回字段的 get_stock_price 函数），Gemini 应该识别出它有一个能够回答问题的工具。它很可能会生成一个结构化请求，以调用 get_stock_price 函数，参数为 ticker=”GOOG”（或“GOOGL”，具体取决于您如何处理 Google 的两类股票）。

重要的一点是，Gemini 不会直接执行外部代码或实时调用股票价格 API。相反，它会为您（开发人员）生成一个结构化请求。通过运行以下代码：

## LLM 检查可用的工具声明
## LLM 返回最适用的函数和参数
function_call = response.candidates[0].content.parts[0].function_call

响应可能看起来像这样（简化）：

name: "get_stock_price" 
args { 
   fields 
     { key: "ticker" 
       value {string_value: "GOOG"}        
  } 
}

那么，您如何实际获取股票价格呢？ 这就是您的代码发挥作用的地方：

因此，用户（或更可能是您的代码）负责根据模型的响应触发正确的代码。为此，我们需要为每个工具声明实现单独的 Python 函数。以下是 get_stock_price 函数的示例：

## 为每个声明实现一个 Python 函数
def get_stock_price_from_api(content):
   url = f"https://www.alphavantage.co/query?function=GLOBAL_QUOTE"
         f"&symbol={content['ticker']}&apikey={API_KEY}"
   api_request = requests.get(url)
   return api_request.text

## ... 其他函数实现

为了简化函数调用的触发，我们建议创建一个函数处理程序（即 Python 字典），将功能声明中的函数名称与实际代码函数链接起来：

## 将功能声明与特定 Python 函数链接
function_handler = {
   "get_stock_price": get_stock_price_from_api,
   # ...,
}

通过实现 Python 函数并能够从模型的响应中获取函数名称和参数，下一步是执行相应的函数：

## LLM 检查可用的工具声明
## LLM 返回最适用的函数和参数
function_call = response.candidates[0].content.parts[0].function_call

function_name = function_call.name
params = {key: value for key, value in function_call.args.items()}

## 调用相应的 Python 函数（或 API）
function_api_response = function_handler[function_name](params)[:20000]

API 调用的输出看起来像这样：

{
   "Global Quote": { 
      "01. symbol": "GOOG", 
      "02. open": "179.7500",     
      "03. high": "180.4450", 
      "04. low": "176.0300", 
      "05. price": "177.3500", 
      "06. volume": "17925763", 
      "07. latest trading day": "2024-11-14", 
      "08. previous close": "180.4900", 
      "09. change": "-3.1400", 
      "10. change percent": "-1.7397%" 
   }
}

然后，结果可以发送回模型进行最终处理和响应生成：

## 将函数的返回值发送给 LLM 以生成最终答案
final_response = agent.send_message(
    Part.from_function_response(
           name=function_name,
           response={"content": function_api_response},
    ),
)

通过将函数的响应传递给 LLM，我们得到最终的响应：

Google 的 (GOOG) 股票价格目前为 \$177.35，较昨天的 \$180.49 收盘价下跌了 1.74%。

这就是最终用户期望的答案。

以下是该过程的五个步骤摘要：

用户提问： 用户向模型提问（“Google 当前的股票价格是多少？”）。
模型请求函数调用： 模型生成请求，指明要调用哪个函数（“get_stock_price”）以及使用什么参数值（“GOOG”）。
开发者执行函数： 您的代码从响应中提取函数调用信息，并使用参数值运行相应的 Python 函数（“get_stock_price_from_api”）。该函数从外部 API 检索实际的股票价格数据。
函数结果发送回模型以生成最终响应： 从外部 API 调用（股票价格数据）返回的结果发送回 LLM。

我们所描述的过程是代理的基础。 特别是，我们概述的工作流程代表了单轮代理的核心逻辑：一个输入触发一个函数调用并产生响应。这种模块化设计非常适合云部署和扩展。通过将此逻辑容器化并部署在 Google Cloud Run 等服务上，您可以创建一个健壮的、无服务器的代理，通过 API 访问，无论是在您的 VPC 内部还是对外部世界。

移动到多轮代理

虽然单轮模型提供了一个关键的基础，但大多数现实世界的生成AI应用需要更复杂的交互。用户很少能通过一个问题和答案得到他们所需的内容。本节探讨多轮代理，它们能够保持上下文，处理后续问题，并协调多个函数调用以实现更复杂的目标。

为了说明这个概念，我们将使用一个灵感来自于这个代码库的例子。我们的目标是创建一个生成AI代理，能够回答特定区域内关于电影和电影院的问题。与单轮代理一样，我们首先需要定义代理可以使用的函数。为了简单起见，我们将直接在代码中提供函数签名和描述：

def find_movies(description: str, location: str = “”): “””根据任何描述、类型、标题词等查找当前在电影院上映的电影标题。参数： description：任何类型的描述，包括类别或类型、标题词、属性等。 location：城市和州，例如：旧金山，加州或邮政编码，例如：95616 “””

def find_movies(description: str, location: str = ""): 
    """find movie titles currently playing in theaters based on any description, genre, title words, etc. Args: description: Any kind of description including category or genre, title words, attributes, etc. location: The city and state, e.g. San Francisco, CA or a zip code e.g. 95616 """ 
    ...
    return ["Barbie", "Oppenheimer"] 

def find_theaters(location: str, movie: str = ""): 
    """Find theaters based on location and optionally movie title which are is currently playing in theaters. Args: location: The city and state, e.g. San Francisco, CA or a zip code e.g. 95616 movie: Any movie title """ 
    ...
    return ["Googleplex 16", "Android Theatre"] 

def get_showtimes(location: str, movie: str, theater: str, date: str): 
    """ Find the start times for movies playing in a specific theater. Args: location: The city and state, e.g. San Francisco, CA or a zip code e.g. 95616 movie: Any movie title theater: Name of the theater date: Date for requested showtime """ 
    ...
    return ["10:00", "11:00"]

下一步是运行函数识别、执行（使用函数处理程序）和响应生成的循环，直到模型获得足够的信息以完全回应用户的请求。为了实现多轮交互，Gemini支持自动函数调用，这可以通过以下代码自动完成：

chat = model.start_chat(enable_automatic_function_calling=True) 
response = chat.send_message( 
"Which comedy movies are shown tonight in Mountain view and at what time?") 

for content in chat.history: 
   print(content.role, "->", [type(part).to_dict(part) for part in content.parts]) 
   print("-" * 80)

以下交互演示了当代码与用户查询“今晚在山景城放映哪些喜剧电影，时间是什么？”一起执行时模型的行为：

user -> [{'text': 'Which comedy movies are shown tonight in Mountain view and at what time?’}] 
-------------------------------------------------------------------------------- 
model -> [{'function_call': {'name': 'find_movies', 'args': {'location': 'Mountain View, CA', 'description': 'comedy'}}}] 
-------------------------------------------------------------------------------- 
user -> [{'function_response': {'name': 'find_movies', 'response': {'result': ['Barbie', 'Oppenheimer']}}}] 
-------------------------------------------------------------------------------- 
model -> [{'function_call': {'name': 'find_theaters', 'args': {'movie': 'Barbie', 'location': 'Mountain View, CA'}}}] 
-------------------------------------------------------------------------------- 
user -> [{'function_response': {'name': 'find_theaters', 'response': {'result': ['Googleplex 16', 'Android Theatre']}}}] 
-------------------------------------------------------------------------------- model -> [{'function_call': {'name': 'get_showtimes', 'args': {'date': 'tonight', 'location': 'Mountain View, CA', 'theater': 'Googleplex 16', 'movie': 'Barbie'}}}] 
-------------------------------------------------------------------------------- 
user -> [{'function_response': {'name': 'get_showtimes', 'response': {'result': ['10:00', '11:00']}}}] 
-------------------------------------------------------------------------------- 
model -> [{'text': '喜剧电影《芭比》今晚在Googleplex 16放映，时间是10:00和11:00。\n’}] 
--------------------------------------------------------------------------------

这个交互演示了模型在每一轮中使用完整的对话历史来确定仍然需要哪些信息，使用哪个工具，以及如何构建其响应。过去交互的记录对于多轮对话至关重要，这被称为短期记忆。此外，除了对话历史，存储操作指标（例如执行时间、延迟、内存）也很重要，以便进行进一步的实验和优化。

以下是图中多轮代理执行过程的7步总结：

新查询： 用户通过提供新的查询或问题来启动交互。
函数识别： 基础模型（FM）及可用工具和指令分析查询，并确定是否需要函数调用。如果需要，它会识别适当的函数名称和所需参数。
函数调用准备： FM生成一个结构化的函数调用请求，指定函数名称和要传递的参数。
函数调用执行： 此步骤由开发者的代码执行，而不是FM本身（但是Gemini支持自动函数调用，使实现更容易）。代码接收函数调用请求，执行相应的函数（例如，进行API调用），并检索结果。
中间响应： 函数执行的结果（由函数检索的数据）作为中间响应发送回FM。
上下文更新（对话历史）： FM用中间响应更新其对话历史（在图中表示为“短期记忆”）。FM随后使用此更新的上下文来决定是否需要进一步的函数调用，或者是否已经收集了足够的信息以生成最终响应。如果需要更多信息，过程将回到第2步。
最终响应： 一旦FM确定它拥有所有必要的信息（或达到最大步骤数以避免无限循环），它生成最终响应给用户，结合从所有函数调用中收集的信息。

这种多轮交互真实地展示了生成AI代理如何处理单个用户请求。然而，现实世界的使用通常涉及随时间的重复交互。想象一下，一个用户在一周内使用代理查找电影放映时间，然后在下一周返回进行类似请求。如果代理保留了过去交互的长期记忆，它可以提供更个性化的推荐，可能会建议用户之前感兴趣的电影或电影院。这个长期记忆存储每次交互的短期对话历史的摘要或完整记录。

短期和长期记忆在实现有效的多轮代理交互中发挥着至关重要的作用。以下是每种记忆的细分及实施选项：

短期记忆（对话历史）： 存储单个用户会话中的正在进行的对话。这包括用户的查询、模型的函数调用及这些函数调用的响应。此上下文对于模型理解后续问题并在交互中保持连贯性至关重要。 实施选项：

日志（小文本日志）： 对于简单应用程序和短对话，将交互历史存储为纯文本日志可能足够。这易于实现，但对于长对话或高流量可能效率不高。
云存储/数据库（大非文本日志）： 对于更复杂的应用程序（例如，利用多模态模型能力使用图像或音频作为输入），云存储服务或数据库是更好的选择。这允许更结构化的存储和高效检索对话历史。
API会话（客户端内存）： 对话历史也可以在客户端（例如，在网页浏览器或移动应用中）使用API会话进行管理。这减少了服务器端存储需求，但可能对可存储的数据量有一定限制。
以上所有的组合： 可以使用混合方法，根据应用程序的具体需求结合不同的存储机制。

长期记忆： 存储跨多个会话的用户过去交互的信息。这允许代理学习用户偏好，提供个性化推荐，并随着时间的推移提供更高效的服务。 实施选项：

向量数据库（用于RAG - 检索增强生成）： 向量数据库特别适合于代理应用中的长期记忆。它们将数据存储为向量嵌入，捕捉数据的语义含义。这使得高效的相似性搜索成为可能，允许代理根据当前用户查询从过去的交互中检索相关信息。这通常用于检索增强生成（RAG）管道。
元数据存储/图（会话ID，其他元数据）： 元数据存储，例如图数据库或键值存储，可以用于存储有关用户会话的信息，例如会话ID、时间戳和其他相关元数据。这可以用于组织和检索过去的对话历史和交互关系。
云存储/数据库（实际日志）： 过去对话的完整日志可以存储在云存储或数据库中。这提供了所有交互的完整记录，但可能需要更多的存储空间和更复杂的检索机制。
以上所有的组合： 类似于短期记忆，可以使用多种存储机制的组合来优化性能和存储效率。例如，摘要信息可以存储在向量数据库中以便快速检索，而完整日志可以存储在更便宜的云存储中以便审计或更详细的分析。

代理调用代理：多代理系统的力量

虽然单个代理可以处理复杂任务，但某些问题需要协调的努力。多代理系统通过使多个代理协同工作来解决这一问题，每个代理专注于特定的子任务。这种协作方法使复杂问题可以分解为更小、更易管理的部分，从而导致更高效和更稳健的解决方案。

多代理系统中的一个关键概念是将代理视为工具：就像一个代理可以使用外部 API 或函数一样，它也可以使用其他代理来执行特定的子任务。这种“代理作为工具”的范式允许创建分层系统，其中一个代理协调其他代理的工作。

以下是最常见的多代理模式的描述：

路由代理（逐个处理）： 在此模式中，一个中央“路由”代理接收初始请求，然后逐个将其委派给其他代理。路由器充当协调者，确定哪个代理最适合处理任务的每个部分。在一个代理完成其子任务后，结果被传回路由器，路由器随后决定下一步。这对于可以分解为顺序步骤的任务非常有用，其中一个步骤的输出会影响下一个步骤。
并行（一个对多个）： 在这里，一个代理同时将子任务分配给多个代理。当子任务是独立的并且可以并行执行时，这种方法非常有效，可以显著减少整体处理时间。一旦所有代理完成其工作，它们的结果通常由最初分配任务的代理进行汇总。
顺序（预定义顺序）： 此模式涉及代理之间的信息流的预定义。一个代理的输出直接作为输入传递给下一个代理，形成固定顺序。这适用于具有明确定义的线性工作流程的任务。
循环流（预定义顺序）： 类似于顺序模式，但信息流形成一个循环。序列中最后一个代理的输出被传回第一个代理，形成一个周期。这对于迭代过程非常有用，代理可以根据循环中其他代理的反馈来完善其输出。
动态（全对全）： 在这种更复杂的模式中，任何代理都可以与任何其他代理进行通信。没有中央协调者或预定义的流程。代理可以动态地交换信息并相互协商以实现共同目标。这种模式更灵活，但管理起来也更复杂，需要复杂的通信和协调机制。

总结一下，可以这样理解：

在路由模式中，路由代理将其他代理作为专门工具，按需逐个调用它们。
在并行模式中，初始代理同时使用多个代理作为工具，以加快过程。
在顺序和循环流模式中，代理作为工具在预定义的管道或循环中使用。
在动态模式中，代理像一个团队一样互动，每个代理根据情况充当其他代理的用户和工具。

这些模式提供了不同的方式来构建多代理交互，使开发人员能够根据其应用程序的具体需求选择最合适的方法。

在建立了多代理协作的概念后，下一个逻辑问题是在企业环境中如何将其付诸实践。上面的图表展示了一种基于微服务方法的企业级架构。这种方法将每个代理视为独立服务，类似于微服务如何将大型应用程序分解为更小、可独立部署的组件。这种类比非常强大，因为它使我们能够利用现有的微服务最佳实践：每个业务单元可以根据其特定需求开发和部署自己的代理，作为独立的微服务。这种去中心化的方法允许更大的灵活性和更快的开发周期，因为团队可以独立工作，而不会影响系统的其他部分。正如微服务通过 API 进行通信，多代理系统中的代理通过交换消息进行通信，通常采用 JSON 等结构化格式。为了确保互操作性并避免冗余开发，中央工具注册表提供对共享工具的访问，而代理模板目录提供可重用的代码和最佳实践。这种方法促进了协作，加速了开发，并在组织内促进了一致性。我们将在即将发布的文章中更深入地探讨这种架构。