ADK 智能体的 Apigee AI 网关¶
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 的治理集成到你的智能体工作流程中。
from google.adk.agents import LlmAgent
from google.adk.models.apigee_llm import ApigeeLlm
# 实例化 ApigeeLlm 包装器
model = ApigeeLlm(
# 指定到你的模型的 Apigee 路由。有关更多信息，请查看 ApigeeLlm 文档 (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"},
# 可选：如果代理需要额外的 OAuth 作用域，请传递 google-auth 凭据
# credentials=my_credentials
)
# 将配置的模型包装器传递给你的 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重试非流式请求，但仅当构造函数传入了retry_options=types.HttpRetryOptions(...)时才会启用。默认情况下每个请求只尝试一次，流式请求永不重试。 - 规范化：将响应和流式数据块解析为 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. 初始化客户端
client = CompletionsHTTPClient(
base_url="https://your-apigee-proxy-url.com/v1",
headers={"Authorization": "Bearer YOUR_API_KEY"}
)
# 2. 构造一个最小请求
request = LlmRequest(
model="gpt-4o", # 替换为你的目标模型 ID
contents=[types.Content(role="user", parts=[types.Part.from_text(text="Hello!")])]
)
# 3. 执行一次非流式生成
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())
