ADK 智能体的 Apigee AI 网关¶

Supported in ADKPython v1.18.0Java v0.4.0

Apigee 提供了强大的 AI Gateway，改变了你管理和治理生成式 AI 模型流量的方式。通过将你的 AI 模型端点（如 Agent Platform 或 Gemini API）暴露在 Apigee 代理之后，你可以立即获得企业级能力：

模型安全： 实施安全策略，如 Model Armor 以进行威胁防护。
流量治理： 执行速率限制和令牌限制以管理成本并防止滥用。
性能： 使用语义缓存和高级模型路由提高响应时间和效率。
监控与可见性： 获得对所有 AI 请求的细粒度监控、分析和审计。

The ApigeeLlm wrapper is designed for use with Agent Platform and the Gemini API (generateContent). We are continually expanding support for other models and interfaces. For OpenAI compatible models, including self-hosted or other providers, use the CompletionsHTTPClient to route traffic through your Apigee proxy.

实现示例¶

通过实例化 ApigeeLlm 包装器对象并将其传递给 LlmAgent 或其他智能体类型，将 Apigee 的治理集成到你的智能体工作流程中。

PythonJava

from google.adk.agents import LlmAgent
from google.adk.models.apigee_llm import ApigeeLlm

# 实例化 ApigeeLlm 包装器
model = ApigeeLlm(
    # Specify the Apigee route to your model. For more info, check out the ApigeeLlm documentation (https://github.com/google/adk-python/tree/main/contributing/samples/models/hello_world_apigeellm).
    model="apigee/gemini-flash-latest",
    # 已部署的 Apigee 代理的代理 URL,包括基本路径
    proxy_url=f"https://{APIGEE_PROXY_URL}",
    # 传递必要的身份验证/授权标头(如 API 密钥)
    custom_headers={"foo": "bar"}
)

# 将配置的模型包装器传递给你的 LlmAgent
agent = LlmAgent(
    model=model,
    name="my_governed_agent",
    instruction="你是一个由 Gemini 提供支持并由 Apigee 管理的得力助手。",
    # ... 其他智能体参数
)

import com.google.adk.agents.LlmAgent;
import com.google.adk.models.ApigeeLlm;
import com.google.common.collect.ImmutableMap;

ApigeeLlm apigeeLlm =
        ApigeeLlm.builder()
            .modelName("apigee/gemini-flash-latest") // 指定到你的模型的 Apigee 路由。有关更多信息,请查看 ApigeeLlm 文档
            .proxyUrl(APIGEE_PROXY_URL) // 已部署的 Apigee 代理的代理 URL,包括基本路径
            .customHeaders(ImmutableMap.of("foo", "bar")) // 传递必要的身份验证/授权标头(如 API 密钥)
            .build();
LlmAgent agent =
    LlmAgent.builder()
        .model(apigeeLlm)
        .name("my_governed_agent")
        .description("my_governed_agent")
        .instruction("你是一个由 Gemini 提供支持并由 Apigee 管理的得力助手。")
        // 接下来将添加工具
        .build();

使用此配置后，你的智能体发出的每个 API 调用都将首先通过 Apigee 路由，在那里执行所有必要的策略（安全、速率限制、日志记录），然后请求才会被安全地转发到底层 AI 模型端点。有关使用 Apigee 代理的完整代码示例，请参阅 Hello World Apigee LLM。

OpenAI 兼容性¶

CompletionsHTTPClient 是一个通用 HTTP 客户端，设计用于兼容 OpenAI API 格式。它允许你通过代理（如 Apigee）路由请求，这些代理期望标准的 OpenAI 兼容 /chat/completions 端点，而非原生 Gemini 或 Vertex AI 协议。此客户端处理：

Payload construction: Converts LlmRequest objects into the format required by OpenAI-compatible APIs.
Response handling: Manages streaming and non-streaming responses from the proxy.
Reliability: Uses tenacity to retry non-streaming requests, but only when you pass retry_options=types.HttpRetryOptions(...) to the constructor. By default each request is attempted once, and streaming requests are never retried.
Normalization: Parses responses and streaming chunks into the standard format expected by the rest of the ADK framework.

实现示例¶

import asyncio
from google.adk.models.apigee_llm import CompletionsHTTPClient
from google.adk.models.llm_request import LlmRequest
from google.genai import types

async def test_client():
    # 1. Initialize the client
    client = CompletionsHTTPClient(
        base_url="https://your-apigee-proxy-url.com/v1",
        headers={"Authorization": "Bearer YOUR_API_KEY"}
    )

    # 2. Construct a minimal request
    request = LlmRequest(
        model="gpt-4o",  # Replace with your target model ID
        contents=[types.Content(role="user", parts=[types.Part.from_text(text="Hello!")])]
    )

    # 3. Execute a non-streaming generation
    async for response in client.generate_content_async(request, stream=False):
        if response.content and response.content.parts:
            print(f"Response: {response.content.parts[0].text}")

if __name__ == "__main__":
    asyncio.run(test_client())