使用 LangChain 智能体

准备工作

本教程假定您已阅读并遵循以下说明：

开发 LangChain 智能体：将 agent 作为 LangchainAgent 的实例进行开发。
用户身份验证，以用户身份进行身份验证，以便查询智能体。
导入并初始化 SDK：初始化客户端，以获取已部署的实例（如有需要）。

获取代理的实例

如需查询 LangchainAgent，您需要先创建新实例或获取现有实例。

如需获取与特定资源 ID 对应的 LangchainAgent，请执行以下操作：

Vertex AI SDK for Python

运行以下代码：

import vertexai

client = vertexai.Client(  # For service interactions via client.agent_engines
    project="PROJECT_ID",
    location="LOCATION",
)

agent = client.agent_engines.get(name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID")

print(agent)

其中

PROJECT_ID 是您在其中开发和部署代理的 Google Cloud 项目 ID。
LOCATION 是支持的区域之一。
RESOURCE_ID 是所部署智能体的 ID，以 reasoningEngine 资源的形式表示。

Python 请求库

运行以下代码：

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests

def get_identity_token():
    credentials, _ = google_auth.default()
    auth_request = google_requests.Request()
    credentials.refresh(auth_request)
    return credentials.token

response = requests.get(
f"https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
    headers={
        "Content-Type": "application/json; charset=utf-8",
        "Authorization": f"Bearer {get_identity_token()}",
    },
)

REST API

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID

使用 Vertex AI SDK for Python 时，agent 对象对应于包含以下内容的 AgentEngine 类：

包含已部署代理相关信息的 agent.api_resource。您还可以调用 agent.operation_schemas() 以返回代理支持的操作列表。如需了解详情，请参阅支持的操作。
一种允许同步服务交互的 agent.api_client
一种允许异步服务交互的 agent.async_api_client

本部分的其余内容假设您有一个名为 agent 的 AgentEngine 实例。

支持的操作

支持以下操作：

query：用于同步获取对查询的响应。
stream_query：用于对查询的响应进行流式传输。

query 和 stream_query 方法都支持相同类型的参数：

input：要发送给智能体的消息。
config：查询上下文的配置（如适用）。

查询智能体

命令：

agent.query(input="What is the exchange rate from US dollars to SEK today?")

它相当于以下内容（完整形式）：

agent.query(input={
    "input": [ # The input is represented as a list of messages (each message as a dict)
        {
            # The role (e.g. "system", "user", "assistant", "tool")
            "role": "user",
            # The type (e.g. "text", "tool_use", "image_url", "media")
            "type": "text",
            # The rest of the message (this varies based on the type)
            "text": "What is the exchange rate from US dollars to Swedish currency?",
        },
    ]
})

角色用于帮助模型在回答时区分不同类型的消息。如果输入中省略了 role，则默认为 "user"。

角色	说明
`system`	用于告知聊天模型如何表现并提供其他背景信息。并非所有聊天模型提供方都支持此参数。
`user`	表示用户与模型互动时的输入，通常以文本或其他互动输入的形式呈现。
`assistant`	表示来自模型的回答，可以包含文本或调用工具的请求。
`tool`	一种消息，用于在检索到外部数据或完成处理后，将工具调用的结果传递回模型。

此外，该消息的 type 部分决定了应如何解读消息的其余部分（请参阅处理多模态内容）。

使用多模态内容查询智能体

我们将使用以下智能体来演示如何将多模态输入传递给智能体，该智能体会将输入转发给模型，而不会使用任何工具：

agent = agent_engines.LangchainAgent(
    model="gemini-2.0-flash",
    runnable_builder=lambda model, **kwargs: model,
)

多模态消息通过指定 type 及相应数据的内容块来表示。一般来说，对于多模态内容，您需要将 type 指定为 "media"，将 file_uri 指向 Cloud Storage URI，并指定用于解读文件的 mime_type。

图片

agent.query(input={"input": [
    {"type": "text", "text": "Describe the attached media in 5 words!"},
    {"type": "media", "mime_type": "image/jpeg", "file_uri": "gs://cloud-samples-data/generative-ai/image/cricket.jpeg"},
]})

视频

agent.query(input={"input": [
    {"type": "text", "text": "Describe the attached media in 5 words!"},
    {"type": "media", "mime_type": "video/mp4", "file_uri": "gs://cloud-samples-data/generative-ai/video/pixel8.mp4"},
]})

音频

agent.query(input={"input": [
    {"type": "text", "text": "Describe the attached media in 5 words!"},
    {"type": "media", "mime_type": "audio/mp3", "file_uri": "gs://cloud-samples-data/generative-ai/audio/pixel.mp3"},
]})

如需查看 Gemini 支持的 MIME 类型列表，请参阅相应类型的文档：

使用可运行的配置查询智能体

查询智能体时，您还可以为智能体指定 config（遵循 RunnableConfig 的架构）。以下是两种常见场景：

默认配置参数：
- run_id/run_name：运行作业的标识符。
- tags/metadata：运行作业的分类器（当通过 OpenTelemetry 进行跟踪时）。
自定义配置参数（通过 configurable）：
- session_id：运行作业所处的会话（请参阅存储对话记录）。
- thread_id：运行作业所处的线程（请参阅存储检查点）。

例如：

import uuid

run_id = uuid.uuid4()  # Generate an ID for tracking the run later.

response = agent.query(
    input="What is the exchange rate from US dollars to Swedish currency?",
    config={  # Specify the RunnableConfig here.
        "run_id": run_id                               # Optional.
        "tags": ["config-tag"],                        # Optional.
        "metadata": {"config-key": "config-value"},    # Optional.
        "configurable": {"session_id": "SESSION_ID"}   # Optional.
    },
)

print(response)