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 请求的细粒度监控、分析和审计。

ApigeeLLM 包装器设计用于 Agent Platform 和 Gemini API (generateContent)。我们正在持续扩展对其他模型和接口的支持。对于兼容 OpenAI 的模型，包括自托管或其他提供者，请使用 CompletionsHTTPClient 通过你的 Apigee 代理路由流量。

实现示例¶

通过实例化 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 协议。此客户端处理：

负载构建：将 LlmRequest 对象转换为 OpenAI 兼容 API 所需的格式。
响应处理：管理来自代理的流式和非流式响应。
可靠性：使用 tenacity 进行内置重试逻辑。
规范化：将响应和流式分块解析为 ADK 框架其余部分期望的标准格式。

实现示例¶

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):
        print(f"Response: {response.text}")

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