建立自訂代理

使用 Agent Runtime 開發自訂代理時，您會以 Python 類別定義代理的邏輯和功能。這種以類別為基礎的方法，可讓您彈性整合任何與 Python 相容的程式庫或服務，同時讓 Agent Platform 處理代理執行個體的部署、擴充和管理作業。

下列步驟說明如何建立自訂範本，用於例項化可在 Agent Platform 部署的代理：

1. 基本範例
2. 選用：逐句顯示回覆
3. 選用：註冊自訂方法
4. 選用：提供類型註解
5. 選用：將追蹤記錄傳送至 Cloud Trace
6. 選用：使用環境變數
7. 選用：與 Secret Manager 整合
8. 選用：處理憑證
9. 選用：處理錯誤

基本範例

以下提供一個基本範例，這個 Python 類別是範本，用於例項化可部署在 Agent Platform 上的代理 (您可以為 CLASS_NAME 變數提供 MyAgent 等值)：

from typing import Callable, Sequence

class CLASS_NAME:
    def __init__(
        self,
        model: str,
        tools: Sequence[Callable],
        project: str,
        location: str,
    ):
        self.model_name = model
        self.tools = tools
        self.project = project
        self.location = location

    def set_up(self):
        import vertexai
        from langchain_google_vertexai import ChatVertexAI
        from langgraph.prebuilt import create_react_agent

        vertexai.init(project=self.project, location=self.location)

        model = ChatVertexAI(model_name=self.model_name)
        self.graph = create_react_agent(model, tools=self.tools)

    def query(self, **kwargs):
        return self.graph.invoke(**kwargs)

部署考量事項

編寫 Python 類別時，請留意下列三種方法：

1. __init__()：
   • 這個方法僅適用於代理設定參數。舉例來說，您可以使用這個方法，收集模型參數和安全屬性，做為使用者的輸入引數。您也可以使用這個方法收集專案 ID、區域、應用程式憑證和 API 金鑰等參數。
   • 建構函式傳回的物件必須可「pickle」，才能部署至 Agent Runtime。因此，您應該在 .set_up 方法中初始化服務用戶端，並建立與資料庫的連線，而不是在 __init__ 方法中。
   • 此為選用方法。如未指定，Agent Runtime 會使用該類別的預設 Python 建構函式。
2. set_up()：
   • 您必須使用這個方法定義代理初始化邏輯。舉例來說，您可以使用這個方法建立與資料庫或依附服務的連線、匯入依附套件，或預先計算用於提供查詢的資料。
   • 此為選用方法。如未指定，Agent Runtime 會假設代理不需要在處理使用者查詢前呼叫 .set_up 方法。
3. query()/stream_query()：
   • 使用 query()，將完整回覆做為單一結果傳回。
   • 使用 stream_query()，在回覆可用時分塊傳回，提供串流體驗。stream_query 方法必須傳回可疊代物件 (例如產生器)，才能啟用串流。
   • 如要同時支援單一回覆和串流互動，可以同時導入這兩種方法。
   • 您應為這個方法提供清楚的 docstring，定義其功能、記錄其屬性，並為其輸入提供類型註解。避免在 query 和 stream_query 方法中使用變數引數。

在本機例項化代理

您可以使用下列程式碼，建立代理的本機執行個體：

agent = CLASS_NAME(
    model=model,  # Required.
    tools=[get_exchange_rate],  # Optional.
    project="PROJECT_ID",
    location="LOCATION",
)
agent.set_up()

測試 query 方法

您可以傳送查詢至本機執行個體，測試代理程式：

response = agent.query(
    input="What is the exchange rate from US dollars to Swedish currency?"
)

print(response)

回應會是類似下列內容的字典：

{"input": "What is the exchange rate from US dollars to Swedish currency?",
 # ...
 "output": "For 1 US dollar you will get 10.7345 Swedish Krona."}

以非同步方式查詢

如要以非同步方式回應查詢，您可以定義傳回 Python 協同程式的方法 (例如 async_query)。舉例來說，下列範本擴充了基本範例，可非同步回應，並可在 Agent Platform 上部署：

class AsyncAgent(CLASS_NAME):

    async def async_query(self, **kwargs):
        from langchain.load.dump import dumpd

        for chunk in self.graph.ainvoke(**kwargs):
            yield dumpd(chunk)

agent = AsyncAgent(
    model=model,                # Required.
    tools=[get_exchange_rate],  # Optional.
    project="PROJECT_ID",
    location="LOCATION",
)
agent.set_up()

測試 async_query 方法

您可以呼叫 async_query 方法，在本機測試代理。範例如下：

response = await agent.async_query(
    input="What is the exchange rate from US dollars to Swedish Krona today?"
)
print(response)

回應會是類似下列內容的字典：

{"input": "What is the exchange rate from US dollars to Swedish currency?",
 # ...
 "output": "For 1 US dollar you will get 10.7345 Swedish Krona."}

逐句顯示回覆

如要串流查詢的回覆，您可以定義名為 stream_query 的方法，產生回覆。舉例來說，下列範本擴充了基本範例，可串流傳輸回覆，並可在 Agent Platform 上部署：

from typing import Iterable

class StreamingAgent(CLASS_NAME):

    def stream_query(self, **kwargs) -> Iterable:
        from langchain.load.dump import dumpd

        for chunk in self.graph.stream(**kwargs):
            yield dumpd(chunk)

agent = StreamingAgent(
    model=model,                # Required.
    tools=[get_exchange_rate],  # Optional.
    project="PROJECT_ID",
    location="LOCATION",
)
agent.set_up()

使用 Streaming API 時，請留意下列幾項重點：

• 逾時上限：串流回應的逾時上限為 15 分鐘。如果代理需要較長的處理時間，建議將工作拆成多個小任務。
• 串流模型和鏈結：LangChain 的 Runnable 介面支援串流，因此您不只能串流代理的回覆，還能串流模型和鏈結的回覆。
• LangChain 相容性：請注意，目前不支援 LangChain 的 astream_event 方法等非同步方法。
• 限制內容生成速度：如果遇到反壓問題 (產生器生成資料的速度快於消費者處理資料的速度)，應限制內容生成速度。這有助於防止緩衝區溢位，確保串流體驗順暢。

測試 stream_query 方法

您可以在本機測試串流查詢，方法是呼叫 stream_query 方法，並疊代結果。範例如下：

import pprint

for chunk in agent.stream_query(
    input="What is the exchange rate from US dollars to Swedish currency?"
):
    # Use pprint with depth=1 for a more concise, high-level view of the
    # streamed output.
    # To see the full content of the chunk, use:
    # print(chunk)
    pprint.pprint(chunk, depth=1)

這段程式碼會在回覆生成時，逐一輸出各個部分。輸出內容大致如下：

{'actions': [...], 'messages': [...]}
{'messages': [...], 'steps': [...]}
{'messages': [...],
 'output': 'The exchange rate from US dollars to Swedish currency is 1 USD to '
           '10.5751 SEK. \n'}

在這個範例中，每個區塊都包含不同的回覆資訊，例如代理採取的動作、交換的訊息和最終輸出內容。

以非同步方式逐句顯示回覆

如要以非同步方式串流回覆，您可以定義傳回非同步產生器的方法 (例如 async_stream_query)。舉例來說，下列範本擴充了基本範例，以非同步方式串流回覆，並可部署在 Agent Platform：

class AsyncStreamingAgent(CLASS_NAME):

    async def async_stream_query(self, **kwargs):
        from langchain.load.dump import dumpd

        for chunk in self.graph.astream(**kwargs):
            yield dumpd(chunk)

agent = AsyncStreamingAgent(
    model=model,                # Required.
    tools=[get_exchange_rate],  # Optional.
    project="PROJECT_ID",
    location="LOCATION",
)
agent.set_up()

測試 async_stream_query 方法

與測試串流查詢的程式碼類似，您可以呼叫 async_stream_query 方法並疊代結果，在本機測試代理。範例如下：

import pprint

async for chunk in agent.async_stream_query(
    input="What is the exchange rate from US dollars to Swedish currency?"
):
    # Use pprint with depth=1 for a more concise, high-level view of the
    # streamed output.
    # To see the full content of the chunk, use:
    # print(chunk)
    pprint.pprint(chunk, depth=1)

這段程式碼會在回覆生成時，逐一輸出各個區塊。輸出內容大致如下：

{'actions': [...], 'messages': [...]}
{'messages': [...], 'steps': [...]}
{'messages': [...],
 'output': 'The exchange rate from US dollars to Swedish currency is 1 USD to '
           '10.5751 SEK. \n'}

註冊自訂方法

根據預設，query 和 stream_query 方法會在已部署的代理中註冊為作業。您可以覆寫預設行為，並使用 register_operations 方法定義要註冊的一組作業。作業可以註冊為標準 (以空字串 "" 表示) 或串流 ("stream") 執行模式。

如要註冊多項作業，您可以定義名為 register_operations 的方法，列出代理部署後可供使用者使用的方法。在下列程式碼範例中，register_operations 方法會導致部署的代理將 query 和 get_state 註冊為同步執行的作業，並將 stream_query 和 get_state_history 註冊為串流回應的作業：

from typing import Iterable

class CustomAgent(StreamingAgent):

    def get_state(self) -> dict: # new synchronous method
        return self.graph.get_state(**kwargs)._asdict()

    def get_state_history(self) -> Iterable: # new streaming operation
        for state_snapshot in self.graph.get_state_history(**kwargs):
            yield state_snapshot._asdict()

    def register_operations(self):
        return {
            # The list of synchronous operations to be registered
            "": ["query", "get_state"],
            # The list of streaming operations to be registered
            "stream": ["stream_query", "get_state_history"],
        }

注意： 您不必同時為標準和串流作業註冊方法。舉例來說，如要僅支援標準作業，您可以在 register_operations 方法中傳回 {"": ["query", "get_state"]}，而不必為 "stream" 指定對應的方法清單。

您可以直接在本機代理執行個體上呼叫自訂方法，測試這些方法，就像測試 query 和 stream_query 方法一樣。

提供類型註解

您可以使用類型註解，指定代理方法預期的輸入和輸出類型。部署代理後，代理支援的作業輸入和輸出內容，僅支援可執行 JSON 序列化的類型。輸入和輸出的結構定義可以使用 TypedDict 或 Pydantic 模型加上註解。

在以下範例中，我們將輸入內容標註為 TypedDict，並使用 ._asdict() 方法，將 .get_state (NamedTuple) 的原始輸出內容轉換為可序列化的字典：

from typing import Any, Dict, TypedDict

# schemas.py
class RunnableConfig(TypedDict, total=False):
    metadata: Dict[str, Any]
    configurable: Dict[str, Any]

# agents.py
class AnnotatedAgent(CLASS_NAME):

    def get_state(self, config: RunnableConfig) -> dict:
        return self.graph.get_state(config=config)._asdict()

    def register_operations(self):
        return {"": ["query", "get_state"]}

將追蹤記錄傳送至 Cloud Trace

如要使用支援 OpenTelemetry 的檢測程式庫，將追蹤記錄傳送至 Cloud Trace，您可以在 .set_up 方法中匯入並初始化這些程式庫。如果是常見的代理架構，您或許可以搭配使用 Open Telemetry Google Cloud整合和檢測架構，例如 OpenInference 或 OpenLLMetry。

舉例來說，下列範本是基本範例的修改版本，可將追蹤記錄匯出至 Cloud Trace：

OpenInference

首先，請執行下列指令，使用 pip 安裝必要套件：

pip install openinference-instrumentation-langchain==0.1.34

接著，匯入並初始化檢測工具：

from typing import Callable, Sequence

class CLASS_NAME:
    def __init__(
        self,
        model: str,
        tools: Sequence[Callable],
        project: str,
        location: str,
    ):
        self.model_name = model
        self.tools = tools
        self.project = project
        self.location = location

    def set_up(self):
        # The additional code required for tracing instrumentation.
        from opentelemetry import trace
        from opentelemetry.exporter.cloud_trace import CloudTraceSpanExporter
        from opentelemetry.sdk.trace import TracerProvider
        from opentelemetry.sdk.trace.export import SimpleSpanProcessor
        from openinference.instrumentation.langchain import LangChainInstrumentor
        import google.cloud.trace_v2 as cloud_trace_v2
        import google.auth

        credentials, _ = google.auth.default()

        trace.set_tracer_provider(TracerProvider())
        cloud_trace_exporter = CloudTraceSpanExporter(
            project_id=self.project,
            client=cloud_trace_v2.TraceServiceClient(
                credentials=credentials.with_quota_project(self.project),
            ),
        )
        trace.get_tracer_provider().add_span_processor(
            SimpleSpanProcessor(cloud_trace_exporter)
        )
        LangChainInstrumentor().instrument()
        # end of additional code required

        import vertexai
        from langchain_google_vertexai import ChatVertexAI
        from langgraph.prebuilt import create_react_agent

        vertexai.init(project=self.project, location=self.location)

        model = ChatVertexAI(model_name=self.model_name)
        self.graph = create_react_agent(model, tools=self.tools)

    def query(self, **kwargs):
        return self.graph.invoke(**kwargs)

OpenLLMetry

首先，請執行下列指令，使用 pip 安裝必要套件：

pip install opentelemetry-instrumentation-langchain==0.38.10

接著，匯入並初始化檢測工具：

from typing import Callable, Sequence

class CLASS_NAME:
    def __init__(
        self,
        model: str,
        tools: Sequence[Callable],
        project: str,
        location: str,
    ):
        self.model_name = model
        self.tools = tools
        self.project = project
        self.location = location

    def set_up(self):
        # The additional code required for tracing instrumentation.
        from opentelemetry import trace
        from opentelemetry.exporter.cloud_trace import CloudTraceSpanExporter
        from opentelemetry.sdk.trace import TracerProvider
        from opentelemetry.sdk.trace.export import SimpleSpanProcessor
        from opentelemetry.instrumentation.langchain import LangchainInstrumentor
        import google.cloud.trace_v2 as cloud_trace_v2
        import google.auth

        credentials, _ = google.auth.default()

        trace.set_tracer_provider(TracerProvider())
        cloud_trace_exporter = CloudTraceSpanExporter(
            project_id=self.project,
            client=cloud_trace_v2.TraceServiceClient(
                credentials=credentials.with_quota_project(self.project),
            ),
        )
        trace.get_tracer_provider().add_span_processor(
            SimpleSpanProcessor(cloud_trace_exporter)
        )
        LangchainInstrumentor().instrument()
        # end of additional code required

        import vertexai
        from langchain_google_vertexai import ChatVertexAI
        from langgraph.prebuilt import create_react_agent

        vertexai.init(project=self.project, location=self.location)

        model = ChatVertexAI(model_name=self.model_name)
        self.graph = create_react_agent(model, tools=self.tools)

    def query(self, **kwargs):
        return self.graph.invoke(**kwargs)

使用環境變數

如要設定環境變數，請確保在開發期間可透過 os.environ 取得這些變數，並在部署代理時，按照「定義環境變數」一節的指示操作。

與 Secret Manager 整合

如要與 Secret Manager 整合，請按照下列步驟操作：

1. 執行下列指令，安裝用戶端程式庫：

   pip install google-cloud-secret-manager

2. 請按照「為已部署的代理程式授予角色」一節的操作說明，透過 Google Cloud 控制台將「Secret Manager 密鑰存取者」角色 (roles/secretmanager.secretAccessor) 授予服務帳戶。

3. 在 .set_up 方法中匯入並初始化用戶端，並在需要時取得對應的密鑰。舉例來說，下列範本是修改基本範例，使用儲存在 Secret Manager 的 ChatAnthropic API 金鑰：

from typing import Callable, Sequence

class CLASS_NAME:
    def __init__(
        self,
        model: str,
        tools: Sequence[Callable],
        project: str,
    ):
        self.model_name = model
        self.tools = tools
        self.project = project
        self.secret_id = secret_id # <- new

    def set_up(self):
        from google.cloud import secretmanager
        from langchain_anthropic import ChatAnthropic
        from langgraph.prebuilt import create_react_agent

        # Get the API Key from Secret Manager here.
        self.secret_manager_client = secretmanager.SecretManagerServiceClient()
        secret_version = self.secret_manager_client.access_secret_version(request={
            "name": "projects/PROJECT_ID/secrets/SECRET_ID/versions/SECRET_VERSION",
        })
        # Use the API Key from Secret Manager here.
        model = ChatAnthropic(
            model_name=self.model_name,
            model_kwargs={"api_key": secret_version.payload.data.decode()},  # <- new
        )
        self.graph = create_react_agent(model, tools=self.tools)

    def query(self, **kwargs):
        return self.graph.invoke(**kwargs)

處理憑證

部署代理時，可能需要處理不同類型的憑證：

1. 應用程式預設憑證 (ADC) 通常來自服務帳戶，
2. OAuth：通常來自使用者帳戶
3. 外部帳戶憑證的識別資訊提供者 (Workload Identity Federation)。

應用程式預設憑證

import google.auth

credentials, project = google.auth.default(
    scopes=["https://www.googleapis.com/auth/cloud-platform"]
)

在程式碼中的使用方式如下：

from typing import Callable, Sequence

class CLASS_NAME:
    def __init__(
        self,
        model: str = "meta/llama3-405b-instruct-maas",
        tools: Sequence[Callable],
        location: str,
        project: str,
    ):
        self.model_name = model
        self.tools = tools
        self.project = project
        self.endpoint = f"https://{location}-aiplatform.googleapis.com"
        self.base_url = f'{self.endpoint}/v1beta1/projects/{project}/locations/{location}/endpoints/openapi'

    def query(self, **kwargs):
        import google.auth
        from langchain_openai import ChatOpenAI
        from langgraph.prebuilt import create_react_agent

        # Note: the credential lives for 1 hour by default.
        # After expiration, it must be refreshed.
        creds, _ = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform"])
        creds.refresh(google.auth.transport.requests.Request())

        model = ChatOpenAI(
            model=self.model_name,
            base_url=self.base_url,
            api_key=creds.token,  # Use the token from the credentials here.
        )
        graph = create_react_agent(model, tools=self.tools)
        return graph.invoke(**kwargs)

詳情請參閱「應用程式預設憑證的運作方式」。

OAuth

使用者憑證通常是使用 OAuth 2.0 取得。

如果您有存取權杖 (例如來自 oauthlib)，可以建立 google.oauth2.credentials.Credentials 執行個體。此外，如果您取得更新權杖，也可以指定更新權杖和權杖 URI，讓憑證自動更新：

credentials = google.oauth2.credentials.Credentials(
    token="ACCESS_TOKEN",
    refresh_token="REFRESH_TOKEN",  # Optional
    token_uri="TOKEN_URI",          # E.g. "https://oauth2.googleapis.com/token"
    client_id="CLIENT_ID",          # Optional
    client_secret="CLIENT_SECRET"   # Optional
)

這裡的 TOKEN_URI、CLIENT_ID 和 CLIENT_SECRET 是根據「建立 OAuth 用戶端憑證」取得。

如果您沒有存取權杖，可以使用 google_auth_oauthlib.flow 執行 OAuth 2.0 授權授予流程，取得對應的 google.oauth2.credentials.Credentials 執行個體：

from google.cloud import secretmanager
from google_auth_oauthlib.flow import InstalledAppFlow
import json

# Get the client config from Secret Manager here.
secret_manager_client = secretmanager.SecretManagerServiceClient()
secret_version = client.access_secret_version(request={
    "name": "projects/PROJECT_ID/secrets/SECRET_ID/versions/SECRET_VERSION",
})
client_config = json.loads(secret_version.payload.data.decode())

# Create flow instance to manage the OAuth 2.0 Authorization Grant Flow steps.
flow = InstalledAppFlow.from_client_config(
    client_config,
    scopes=['https://www.googleapis.com/auth/cloud-platform'],
    state="OAUTH_FLOW_STATE"  # from flow.authorization_url(...)
)

# You can get the credentials from the flow object.
credentials: google.oauth2.credentials.Credentials = flow.credentials

# After obtaining the credentials, you can then authorize API requests on behalf
# of the given user or service account. For example, to authorize API requests
# to vertexai services, you'll specify it in vertexai.init(credentials=)
import vertexai

vertexai.init(
    project="PROJECT_ID",
    location="LOCATION",
    credentials=credentials, # specify the credentials here
)

詳情請參閱 google_auth_oauthlib.flow 模組的說明文件。

識別資訊提供者

如要使用電子郵件/密碼、電話號碼、社群媒體供應商 (例如 Google、Facebook 或 GitHub) 或自訂驗證機制驗證使用者，可以使用 Identity Platform 或 Firebase 驗證，或任何支援 OpenID Connect (OIDC) 的識別資訊提供者。

詳情請參閱「透過 OIDC 識別資訊提供者存取資源」。

處理錯誤

為確保 API 錯誤以結構化 JSON 格式傳回，建議您使用 try...except 區塊在代理程式程式碼中實作錯誤處理，並可將其抽象化為裝飾器。

雖然 Agent Platform 可以在內部處理各種狀態碼，但 Python 缺乏標準化方式，無法在所有例外狀況類型中，以相關聯的 HTTP 狀態碼表示錯誤。如果嘗試將所有可能的 Python 例外狀況對應至基礎服務中的 HTTP 狀態，會很複雜且難以維護。

更具擴充性的做法是在代理方法中明確擷取相關例外狀況，或使用 error_wrapper 等可重複使用的裝飾器。接著，您可以建立適當的狀態碼關聯 (例如將 code 和 error 屬性新增至自訂例外狀況，或特別處理標準例外狀況)，並將錯誤格式化為 JSON 字典，做為傳回值。這只需要對代理方法本身進行最少的程式碼變更，通常只需要新增裝飾器。

以下範例說明如何在代理中實作錯誤處理：

from functools import wraps
import json

def error_wrapper(func):
    @wraps(func)  # Preserve original function metadata
    def wrapper(*args, **kwargs):
        try:
            # Execute the original function with its arguments
            return func(*args, **kwargs)
        except Exception as err:
            error_code = getattr(err, 'code')
            error_message = getattr(err, 'error')

            # Construct the error response dictionary
            error_response = {
                "error": {
                    "code": error_code,
                    "message": f"'{func.__name__}': {error_message}"
                }
            }
            # Return the Python dictionary directly.
            return error_response

    return wrapper

# Example exception
class SessionNotFoundError(Exception):
    def __init__(self, session_id, message="Session not found"):
        self.code = 404
        self.error = f"{message}: {session_id}"
        super().__init__(self.error)

# Example Agent Class
class MyAgent:
    @error_wrapper
    def get_session(self, session_id: str):
        # Simulate the condition where the session isn't found
        raise SessionNotFoundError(session_id=session_id)


# Example Usage: Session Not Found
agent = MyAgent()
error_result = agent.get_session(session_id="nonexistent_session_123")
print(json.dumps(error_result, indent=2))

上述程式碼會產生下列輸出內容： json { "error": { "code": 404, "message": "Invocation error in 'get_session': Session not found: nonexistent_session_123" } }

後續步驟

指南

部署代理

瞭解如何根據開發需求，在 Agent Platform Runtime 部署代理。

指南

使用代理

使用代理與 Agent Platform 執行階段。

指南

評估代理

建立及部署基本代理，並使用 Gen AI Evaluation Service 評估代理

疑難排解

疑難排解代理建立問題

瞭解如何解決建立自訂代理時的常見錯誤。

資源

取得支援

尋找 Google Agent Platform 的資源和支援。