嘿，大家好！这里是一个专注于AI智能体的频道~

在a16z预测2025年的发展中，Agent占据了很重要的一个环节。昨天中金也发研报，表示看好Agent的发展趋势。包括langchain发布的24年总结报告，Agent是持续保持增长的。

所以25年，Agent或许值得期待？今天给家人们完全梳理一下Agent 智能体的系统，作为Agent入门指南（超长超长）！

这篇文章会帮你理清头绪，明确地告诉你智能体到底是什么，以及它们是怎么工作的。我们会拆解智能体的关键组成部分，包括“工具”这个重要角色。

同时，也会分享一些实用的经验，教你如何构建和部署智能体，从简单的单次交互，到复杂的多智能体系统。

我们还会探讨如何在企业环境中应用多智能体架构，并将其与微服务进行对比。在后续的文章中，我们会更深入地研究智能体和运营（AgentOps），以及如何搭建一个企业级多智能体系统的平台。

什么是Agent智能体？

简单来说，“智能体就是一个prompt，它指示基础模型去和特定的工具互动。”

生成式 AI 智能体通过一个设计好的prompt，协调基础模型（大模型）与外部工具之间的互动。这个prompt告诉 LLM 何时以及如何使用这些工具。

每一个“工具”本质上都包含了一组函数规范（或者我们叫它“声明”）。这些声明包括：

• 函数名称：工具的唯一标识符。
• 描述：详细解释工具的用途、能解决什么问题，以及在什么情况下可以使用它。
• 参数：列出工具需要的输入参数，并解释每个参数的含义、类型和期望值。
• 输出（可选）：描述工具返回结果的格式和内容。

为了让这些声明标准化，市面上通常使用基于 JSON 的 OpenAPI 格式。这种标准化格式可以清晰、机器可读地描述 API，方便与生成式 AI 模型无缝集成。举个例子，下面是用 OpenAPI 格式声明一个可以获取股票价格的工具：

{
  "tools": [
    {
      "functionDeclarations": [
        {
          "name": "get_stock_price",
          "description": "获取指定公司的当前股票价格。",
          "parameters": {
            "type": "object",
            "properties": {
              "ticker": {
                "type": "string",
                "description": "股票代码（如，AAPL, MSFT）。"
              }
            },
            "required": ["ticker"]
          },
          "returns": {
              "type": "number",
              "description": "当前股票价格。"
          }
        }
      ]
    }
  ]
}

我们也可以用 Python SDK 以编程方式创建相同的函数声明，比如 Vertex AI 提供的 SDK。这样做可以更灵活地创建和管理工具规范：

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

# 创建函数声明
get_stock_price = FunctionDeclaration(
    name="get_stock_price",
    description="获取指定公司的当前股票价格",
    parameters={
        "type": "object",
        "properties": {
            "ticker": {
                "type": "string",
                "description": "公司的股票代码",
            }
        },
        "required": ["ticker"]
    },
    returns={ # 添加 returns 字段！
        "type": "number",
        "description": "当前股票价格。"
    }
)
# ... 可以添加更多的函数声明

# 定义一个工具，包含可以被 FM 使用的函数列表
company_insights_tool = Tool(
    function_declarations=[
        get_stock_price,
        # ... 其他函数声明
    ],
)

为了更直观地展示这些概念，我们将在下面的例子中使用 Vertex AI SDK 和 Gemini 的函数调用功能。这些例子基于这个代码仓库，推荐去看看（https://github.com/GoogleCloudPlatform/generative-ai/blob/main/gemini/function-calling/use_case_company_news_and_insights.ipynb）。这种方法能让你从更底层的角度理解智能体的工作原理。一旦你掌握了这些基础知识，就可以更轻松地使用 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 参数和 returns 字段的 get_stock_price 函数，就像前面的例子那样），Gemini 应该能识别出它有一个可以回答这个问题的工具。它很可能会生成一个结构化的请求，调用 get_stock_price 函数，并将 ticker 参数设为 "GOOG" （或者 "GOOGL"，取决于你如何处理谷歌的两类股票）。

这里要注意的是，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: "Google"}        
  } 
}

那么，你该如何真正获取股票价格呢？这就轮到你的代码出场了：

也就是说，用户（更常见的情况是你的代码）需要负责根据模型的响应触发正确的代码。为了做到这一点，我们需要为每一个工具声明都实现一个单独的 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 后，我们就能得到最终的答案：

谷歌 (GOOG) 的股价目前为 177.35 美元，相比昨天的 180.49 美元收盘价下跌了 1.74%。

这才是最终用户期望得到的答案。

简单总结一下，这个过程包含五个步骤：

1. 用户提问：用户向模型提问（“谷歌的当前股票价格是多少？”）。
2. 模型请求函数调用：模型生成一个请求，指出应该调用哪个函数（“get_stock_price”），以及应该使用哪个参数值（“谷歌”）。
3. 开发者执行函数：你的代码从响应中提取函数调用信息，并用对应的参数值运行相应的 Python 函数（“get_stock_price_from_api”）。这个函数会从外部 API 获取真实的股票价格数据。
4. 将函数结果发回模型：从外部 API 得到的结果（股票价格数据）被发送回 LLM。
5. 模型生成最终响应：模型使用得到的数据，生成最终的、人类可以理解的响应。

我们刚才描述的流程是智能体的基础。具体来说，我们展示的流程是单轮智能体的核心逻辑：一个输入触发一个函数调用，并产生一个响应。这种模块化设计非常适合云部署和扩展。你可以把这个逻辑容器化，然后部署在 Google Cloud Run 等服务上。这样，你就可以创建一个稳定、无服务器的智能体，它可以通过 API 访问，并且可以在你的 VPC 内或者更广阔的网络中使用。

迈向多轮智能体

虽然单轮模型奠定了基础，但大多数现实世界的生成式 AI 应用需要更复杂的交互。用户往往不能通过一次问答就得到所有需要的信息。所以接下来，我们来探索多轮智能体。这种智能体可以记住上下文，处理后续问题，并且协调多个函数调用来实现更复杂的目标。

为了说明这个概念，我们用一个受这个代码仓库启发的例子（https://github.com/google-gemini/cookbook/blob/main/quickstarts/Function_calling.ipynb）。我们的目标是创建一个可以回答特定区域电影和影院相关问题的生成式 AI 智能体。和单轮智能体一样，我们首先需要定义智能体可以使用的函数。为了简单起见，我们直接在代码中提供函数签名和描述：

def find_movies(description: str, location: str = ""):
    """根据任何描述（类型、标题词等）查找当前正在影院上映的电影。参数：description：任何类型的描述，包括类别或类型、标题词、属性等。location：城市和州，例如旧金山，CA，或者邮政编码，例如 95616"""
    ...
    return ["Barbie", "Oppenheimer"]

def find_theaters(location: str, movie: str = ""):
    """查找特定地点，以及可选的正在上映的电影的影院。参数：location：城市和州，例如旧金山，CA，或者邮政编码，例如 95616。movie：任何电影标题"""
    ...
    return ["Googleplex 16", "Android Theatre"]

def get_showtimes(location: str, movie: str, theater: str, date: str):
    """查找特定影院上映电影的开始时间。参数：location：城市和州，例如旧金山，CA，或者邮政编码，例如 95616。movie：任何电影标题。theater：影院名称。date：请求放映时间的日期"""
    ...
    return ["10:00", "11:00"]

下一步是在循环中运行函数识别、执行（使用函数处理器）和响应生成，直到模型拥有足够的信息来完全响应用户的请求。为了实现多轮交互，Gemini 支持自动函数调用，我们可以使用以下代码自动完成：

chat = model.start_chat(enable_automatic_function_calling=True)
response = chat.send_message(
"今晚在山景城上映哪些喜剧电影？什么时间？")

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

下面的交互展示了，当代码使用用户查询“今晚在山景城上映哪些喜剧电影？什么时间？”执行时，模型的行为：

user -> [{'text': '今晚在山景城上映哪些喜剧电影？什么时间？'}]
--------------------------------------------------------------------------------
model -> [{'function_call': {'name': 'find_movies', 'args': {'location': '山景城, CA', 'description': '喜剧'}}}]
--------------------------------------------------------------------------------
user -> [{'function_response': {'name': 'find_movies', 'response': {'result': ['Barbie', 'Oppenheimer']}}}]
--------------------------------------------------------------------------------
model -> [{'function_call': {'name': 'find_theaters', 'args': {'movie': 'Barbie', 'location': '山景城, CA'}}}]
--------------------------------------------------------------------------------
user -> [{'function_response': {'name': 'find_theaters', 'response': {'result': ['Googleplex 16', 'Android Theatre']}}}]
--------------------------------------------------------------------------------
model -> [{'function_call': {'name': 'get_showtimes', 'args': {'date': '今晚', 'location': '山景城, CA', 'theater': 'Googleplex 16', 'movie': 'Barbie'}}}]
--------------------------------------------------------------------------------
user -> [{'function_response': {'name': 'get_showtimes', 'response': {'result': ['10:00', '11:00']}}}]
--------------------------------------------------------------------------------
model -> [{'text': '喜剧电影《Barbie》今晚在 Googleplex 16 的放映时间是 10:00 和 11:00。'}]
--------------------------------------------------------------------------------

这个交互表明，模型在每一轮中都会使用完整的对话历史，以此来判断还需要哪些信息，应该使用哪个工具，以及如何组织自己的响应。这种记录过去的交互信息的方式，对多轮对话来说至关重要，我们称之为短期记忆。此外，除了对话历史记录，存储一些操作指标也很重要，例如每个模型交互的执行时间、延迟和内存等，方便我们进一步实验和优化。

下面是图中描述的多轮智能体执行过程的 7 步骤摘要：

1. 新查询：用户发起新的交互，提出新的查询或问题。
2. 函数识别：基础模型 (FM) 会分析查询，结合可用的工具和指令，判断是否需要进行函数调用。如果需要，它会识别出合适的函数名称和需要的参数。
3. 准备函数调用：FM 会生成一个结构化的函数调用请求，指定函数名称和要传递的参数。
4. 执行函数调用：这一步由开发者的代码执行，而不是 FM 本身（但是 Gemini 支持自动函数调用，方便实现）。代码会接收函数调用请求，执行对应的函数（例如，调用 API），并获取结果。
5. 中间响应：函数执行的结果（函数获取的数据）会被发送回 FM，作为中间响应。
6. 更新上下文（对话历史）：FM 会使用中间响应更新其对话历史（在图中被称为“短期记忆”）。然后，FM 会使用更新后的上下文，来决定是否需要进一步的函数调用，或者是否已经收集到足够的信息来生成最终响应。如果需要更多信息，过程会循环回到步骤 2。
7. 最终响应：一旦 FM 认为它已经掌握所有必要的信息（或者达到了为避免无限循环而设定的最大步骤数），它会生成最终的响应，发送给用户。这个响应会整合从所有函数调用中获取的信息。

这个多轮交互比较真实地展示了，生成式 AI 智能体是如何处理单个用户请求的。然而，在现实世界中，我们经常会多次重复使用智能体。想象一下，用户这周用智能体查找电影放映时间，下周又回来提出类似的要求。如果智能体能够记住过去交互的长期记忆，它就可以提供更有针对性的推荐，比如推荐用户之前感兴趣的电影或者影院。这种长期记忆会存储每次交互的短期对话历史的摘要或完整记录。

短期和长期记忆都在实现有效的多轮智能体交互中起着至关重要的作用。下面是它们各自的说明和实现方式：

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

• 日志（小文本日志）：对于对话比较短的简单应用，可以将交互历史记录存储为纯文本日志。这种方式实现简单，但是对于长对话或高流量，效率可能会降低。
• 云存储/数据库（大型非文本日志）：对于更复杂的应用（例如，利用多模态模型处理图像或者音频输入），云存储服务或者数据库会是更好的选择。这种方式能更结构化地存储，并高效地检索对话历史。
• API 会话（客户端存储）：对话历史也可以在客户端（例如，网页浏览器或者手机应用）使用 API 会话管理。这种方式能减少服务器端的存储压力，但存储的数据量可能会有限制。
• 以上所有方式的组合：可以混合使用不同的存储方式，根据应用的特定需求来选择合适的方案。

长期记忆：存储用户在多个会话中的历史交互信息。这使得智能体能够学习用户的偏好，提供个性化的推荐，并在长期使用中提供更高效的服务。实现选项：

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

智能体调用智能体：多智能体系统的力量

虽然单个智能体可以处理复杂的任务，但有些问题需要协同努力才能解决。多智能体系统通过让多个智能体一起工作来解决这个问题，每个智能体负责一个特定的子任务。这种协作方法允许将复杂问题分解为更小的、更易于管理的部分，从而实现更高效、更强大的解决方案。

多智能体系统的一个关键概念是把智能体当作工具：就像一个智能体可以使用外部 API 或者函数一样，它也可以使用其他的智能体来执行特定的子任务。这种“智能体即工具”的模式允许我们创建分层系统，由一个智能体来协调其他智能体的工作。

下面是一些最常见的多智能体模式：

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

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

• 在 路由器 模式中，路由器智能体把其他的智能体当作专门的工具来使用，根据需要逐一调用它们。
• 在 并行 模式中，初始智能体同时使用多个智能体作为工具，来加速整个过程。
• 在 顺序 和 循环流 模式中，智能体在一个预定义的流程或循环中使用，像工具一样。
• 在 动态 模式中，智能体的互动更像是一个团队，每个智能体都同时充当其他智能体的用户和工具，具体情况而定。

这些模式提供了不同的方式来构建多智能体交互，开发者可以根据应用的特定需求，选择最合适的方式。

在了解了多智能体协作的概念后，下一个问题自然是如何在企业环境中实践。上图展示了一个基于微服务方法的企业级架构。这种方法把每个智能体都视为一个独立的微服务，就像微服务把大型应用分解成可以独立部署的组件一样。这个类比非常有用，因为它允许我们利用现有的微服务最佳实践：每个业务部门可以独立开发和部署自己的智能体，作为独立的微服务，并根据自己的特定需求进行定制。这种分散式的方法可以提高灵活性，并加快开发周期，因为团队可以独立工作，而不会影响系统的其他部分。就像微服务通过 API 进行通信一样，多智能体系统中的智能体也通过交换消息进行通信，消息通常使用 JSON 等结构化格式。为了确保互操作性并避免重复开发，一个中央工具注册表会提供对共享工具的访问，而一个智能体模板目录会提供可重用的代码和最佳实践。这种方法可以促进协作，加快开发速度，并提高整个组织的一致性。我们会在后续的文章中，更深入地探讨这个架构。

结论

这篇文章的关键要点包括：

• 生成式 AI 模型依赖智能体通过定义良好的函数声明，与外部工具进行交互。
• 智能体本质上是一个提示，它指示基础模型与特定工具进行交互。
• 多轮智能体通过动态调用各种函数，并在整个交互过程中维护上下文，来处理复杂的用户请求。
• 多智能体系统使智能体能够通过委派子任务和协调彼此的工作，来协作解决复杂的问题。

好了，这就是我今天想分享的内容。如果你对构建AI智能体感兴趣，别忘了点赞、关注噢~