第16章：Resource-Aware Optimization

Resource-Aware Optimization 使智能 Agent 能够在运行过程中动态监控和管理计算、时间和财务资源。这与主要关注行动排序的简单规划不同。Resource-Aware Optimization 要求 Agent 就行动执行做出决策，以便在指定的资源预算内实现目标，或优化效率。这涉及在更准确但更昂贵的模型与更快、成本更低的模型之间做出选择，或者决定是否分配额外的计算资源以获得更精细的响应，还是返回一个更快速但细节较少的答案。

例如，考虑一个负责为金融分析师分析大型数据集的 Agent。如果分析师需要立即获得初步报告，Agent 可能会使用更快、更实惠的模型来快速总结关键趋势。然而，如果分析师需要一个高度准确的预测来做出关键的投资决策，并且拥有更大的预算和更多的时间，Agent 就会分配更多资源来使用强大的、速度较慢但更精确的预测模型。这一类策略中的一个关键策略是 fallback 机制，当首选模型因过载或受限而无法使用时，它充当安全网。为了确保优雅降级（graceful degradation），系统会自动切换到默认或更实惠的模型，保持服务连续性，而不是完全失败。

实际应用与用例

实际用例包括：

Cost-Optimized LLM Usage：Agent 根据预算约束，决定是对复杂任务使用大型、昂贵的 LLM，还是对简单查询使用更小、更实惠的模型。
Latency-Sensitive Operations：在实时系统中，Agent 选择更快但可能不够全面的推理路径，以确保及时响应。
Energy Efficiency：对于部署在边缘设备或电源受限环境中的 Agent，优化其处理以节省电池寿命。
Fallback for service reliability：当主要选择的模型不可用时，Agent 自动切换到备用模型，确保服务连续性和优雅降级。
Data Usage Management：Agent 选择汇总数据检索，而不是下载完整数据集，以节省带宽或存储空间。
Adaptive Task Allocation：在多 Agent 系统中，Agent 根据当前的计算负载或可用时间自行分配任务。

动手代码示例

一个用于回答用户问题的智能系统可以评估每个问题的难度。对于简单查询，它使用经济高效的的语言模型，如 Gemini Flash。对于复杂的询问，则考虑使用更强大但更昂贵的语言模型（如 Gemini Pro）。使用更强大模型的决策还取决于资源的可用性，特别是预算和时间限制。该系统动态选择适当的模型。

例如，考虑一个使用分层 Agent 构建的旅行规划器。涉及理解用户复杂请求、将其分解为多步骤行程并做出逻辑决策的高级规划，将由像 Gemini Pro 这样复杂且更强大的 LLM 来管理。这是负责理解和推理能力的"规划者"Agent。

然而，一旦计划确定，该计划内的各个任务，如查找航班价格、查看酒店可用性，或查找餐厅评论，本质上是简单的、重复性的网络查询。这些"tool function calls"可以由像 Gemini Flash 这样更快、更实惠的模型来执行。更容易理解为什么经济实惠的模型可以用于这些简单的网络搜索，而复杂的规划阶段需要更先进模型的更高智能，以确保行程计划连贯且合理。

Google 的 ADK 通过其多 Agent 架构支持这种方法，该架构允许模块化和可扩展的应用程序。不同的 Agent 可以处理专门的任务。Model flexibility 使得可以直接使用各种 Gemini 模型，包括 Gemini Pro 和 Gemini Flash，或通过 LiteLLM 集成其他模型。ADK 的编排功能支持动态的、LLM 驱动的 routing，以实现自适应行为。内置的评估功能允许对 Agent 性能进行系统评估，可用于系统优化（参见 Evaluation and Monitoring 章节）。

接下来，将定义两个设置相同但使用不同模型和成本的 Agent。

python

# Conceptual Python-like structure, not runnable code
from google.adk.agents import Agent
# from google.adk.models.lite_llm import LiteLlm # If using models
# not directly supported by ADK's default Agent

# Agent using the more expensive Gemini Pro 2.5
gemini_pro_agent = Agent(
    name="GeminiProAgent",
    model="gemini-2.5-pro", # Placeholder for actual model name if different
    description="A highly capable agent for complex queries.",
    instruction="You are an expert assistant for complex problem-solving."
)

# Agent using the less expensive Gemini Flash 2.5
gemini_flash_agent = Agent(
    name="GeminiFlashAgent",
    model="gemini-2.5-flash", # Placeholder for actual model name if different
    description="A fast and efficient agent for simple queries.",
    instruction="You are a quick assistant for straightforward questions."
)

Router Agent 可以基于简单的指标（如查询长度）来引导查询，较短的查询去往较便宜的模型，较长的查询去往能力更强的模型。然而，更复杂的 Router Agent 可以利用 LLM 或 ML 模型来分析查询的细微差别和复杂性。这个 LLM router 可以确定哪个下游语言模型最适合。例如，请求事实回忆的查询被路由到 flash 模型，而需要深度分析的复杂查询则被路由到 pro 模型。

优化技术可以进一步提高 LLM router 的有效性。Prompt tuning 涉及精心设计 prompt 来引导 router LLM，以获得更好的路由决策。在查询数据集及其最佳模型选择上 fine-tuning LLM router，可以提高其准确性和效率。这种动态路由能力平衡了响应质量与成本效益。

python

# Conceptual Python-like structure, not runnable code
from google.adk.agents import Agent, BaseAgent
from google.adk.events import Event
from google.adk.agents.invocation_context import InvocationContext
import asyncio

class QueryRouterAgent(BaseAgent):
    name: str = "QueryRouter"
    description: str = "Routes user queries to the appropriate LLM agent based on complexity."

    async def _run_async_impl(self, context: InvocationContext) -> AsyncGenerator[Event, None]:
        user_query = context.current_message.text # Assuming text input
        query_length = len(user_query.split()) # Simple metric: number of words

        if query_length < 20: # Example threshold for simplicity vs. complexity
            print(f"Routing to Gemini Flash Agent for short query (length: {query_length})")
            # In a real ADK setup, you would 'transfer_to_agent' or directly invoke
            # For demonstration, we'll simulate a call and yield its response
            response = await gemini_flash_agent.run_async(context.current_message)
            yield Event(author=self.name, content=f"Flash Agent processed: {response}")
        else:
            print(f"Routing to Gemini Pro Agent for long query (length: {query_length})")
            response = await gemini_pro_agent.run_async(context.current_message)
            yield Event(author=self.name, content=f"Pro Agent processed: {response}")

Critique Agent 评估来自语言模型的响应，提供的反馈具有多种功能。对于 Self-Correction，它识别错误或不一致之处，提示回答 Agent 改进其输出以提高质量。它还系统地评估响应以进行性能监控，跟踪准确性和相关性等指标，这些指标用于优化。此外，其反馈可以为强化学习或 fine-tuning 提供信号；例如，持续识别不充分的 Flash 模型响应，可以改进 router Agent 的逻辑。虽然不直接管理预算，但 Critique Agent 通过识别次优的路由选择（例如，将简单查询导向 Pro 模型，或将复杂查询导向 Flash 模型，导致结果不佳）来间接促进预算管理。这为改进资源配置和节约成本提供了调整依据。

Critique Agent 可以配置为仅审查回答 Agent 生成的文本，或同时审查原始查询和生成的文本，从而对响应与初始问题的对齐情况进行全面评估。

text

CRITIC_SYSTEM_PROMPT = """
You are the **Critic Agent**, serving as the quality assurance arm of
our collaborative research assistant system. Your primary function is
to **meticulously review and challenge** information from the
Researcher Agent, guaranteeing **accuracy, completeness, and unbiased
presentation**.

Your duties encompass:
* **Assessing research findings** for factual correctness,
thoroughness, and potential leanings.
* **Identifying any missing data** or inconsistencies in reasoning.
* **Raising critical questions** that could refine or expand the
current understanding.
* **Offering constructive suggestions** for enhancement or exploring
different angles.
* **Validating that the final output is comprehensive** and balanced.

All criticism must be constructive. Your goal is to fortify the
research, not invalidate it. Structure your feedback clearly, drawing
attention to specific points for revision. Your overarching aim is to
ensure the final research product meets the highest possible quality
standards.
"""

Critic Agent 基于预定义的 system prompt 运行，该 prompt 概述了其角色、职责和反馈方法。为此 Agent 设计的良好 prompt 必须清楚地确立其作为评估者的功能。它应该指定重点关注领域，并强调提供建设性反馈而不是简单的否定。prompt 还应鼓励识别优点和缺点，并且必须引导 Agent 如何构建和呈现其反馈。

使用 OpenAI 的动手代码

该系统使用 resource-aware optimization 策略来高效处理用户查询。它首先将每个查询分类为三个类别之一，以确定最合适且最具成本效益的处理路径。这种方法避免了对简单请求浪费计算资源，同时确保复杂查询得到必要的关注。这三个类别是：

simple：用于可以直接回答而无需复杂推理或外部数据的简单问题。
reasoning：用于需要逻辑推演或多步骤思考过程的查询，这些查询被路由到更强大的模型。
internet_search：用于需要最新信息的问题，会自动触发 Google Search 以提供最新的答案。

代码基于 MIT 许可证，可在 Github 上获取： (https://github.com/mahtabsyed/21-Agentic-Patterns/blob/main/16_Resource_Aware_Opt_LLM_Reflection_v2.ipynb)

python

# MIT License
# Copyright (c) 2025 Mahtab Syed
# https://www.linkedin.com/in/mahtabsyed/

import os
import requests
import json
from dotenv import load_dotenv
from openai import OpenAI

# Load environment variables
load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
GOOGLE_CUSTOM_SEARCH_API_KEY = os.getenv("GOOGLE_CUSTOM_SEARCH_API_KEY")
GOOGLE_CSE_ID = os.getenv("GOOGLE_CSE_ID")

if not OPENAI_API_KEY or not GOOGLE_CUSTOM_SEARCH_API_KEY or not GOOGLE_CSE_ID:
    raise ValueError(
        "Please set OPENAI_API_KEY, GOOGLE_CUSTOM_SEARCH_API_KEY, and GOOGLE_CSE_ID in your .env file."
    )

client = OpenAI(api_key=OPENAI_API_KEY)

# --- Step 1: Classify the Prompt ---
def classify_prompt(prompt: str) -> dict:
    system_message = {
        "role": "system",
        "content": (
            "You are a classifier that analyzes user prompts and returns one of three categories ONLY:\n\n"
            "- simple\n"
            "- reasoning\n"
            "- internet_search\n\n"
            "Rules:\n"
            "- Use 'simple' for direct factual questions that need no reasoning or current events.\n"
            "- Use 'reasoning' for logic, math, or multi-step inference questions.\n"
            "- Use 'internet_search' if the prompt refers to current events, recent data, or things not in your training data.\n\n"
            "Respond ONLY with JSON like:\n"
            '{"classification": "simple"}'
        ),
    }
    user_message = {"role": "user", "content": prompt}
    response = client.chat.completions.create(
        model="gpt-4o", messages=[system_message, user_message],
        temperature=1
    )
    reply = response.choices[0].message.content
    return json.loads(reply)

# --- Step 2: Google Search ---
def google_search(query: str, num_results=1) -> list:
    url = "https://www.googleapis.com/customsearch/v1"
    params = {
        "key": GOOGLE_CUSTOM_SEARCH_API_KEY,
        "cx": GOOGLE_CSE_ID,
        "q": query,
        "num": num_results,
    }
    try:
        response = requests.get(url, params=params)
        response.raise_for_status()
        results = response.json()
        if "items" in results and results["items"]:
            return [
                {
                    "title": item.get("title"),
                    "snippet": item.get("snippet"),
                    "link": item.get("link"),
                }
                for item in results["items"]
            ]
        else:
            return []
    except requests.exceptions.RequestException as e:
        return {"error": str(e)}

# --- Step 3: Generate Response ---
def generate_response(prompt: str, classification: str, search_results=None) -> str:
    if classification == "simple":
        model = "gpt-4o-mini"
        full_prompt = prompt
    elif classification == "reasoning":
        model = "o4-mini"
        full_prompt = prompt
    elif classification == "internet_search":
        model = "gpt-4o"
        # Convert each search result dict to a readable string
        if search_results:
            search_context = "\n".join(
                [
                    f"Title: {item.get('title')}\nSnippet: {item.get('snippet')}\nLink: {item.get('link')}"
                    for item in search_results
                ]
            )
        else:
            search_context = "No search results found."
        full_prompt = f"""Use the following web results to answer the user query:
{search_context}

Query: {prompt}"""

    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": full_prompt}],
        temperature=1,
    )
    return response.choices[0].message.content, model

# --- Step 4: Combined Router ---
def handle_prompt(prompt: str) -> dict:
    classification_result = classify_prompt(prompt)
    # Remove or comment out the next line to avoid duplicate printing
    # print("\n Classification Result:", classification_result)
    classification = classification_result["classification"]
    search_results = None
    if classification == "internet_search":
        search_results = google_search(prompt)
        # print("\n Search Results:", search_results)
    answer, model = generate_response(prompt, classification, search_results)
    return {"classification": classification, "response": answer, "model": model}

test_prompt = "What is the capital of Australia?"
# test_prompt = "Explain the impact of quantum computing on cryptography."
# test_prompt = "When does the Australian Open 2026 start, give me full date?"

result = handle_prompt(test_prompt)
print(" Classification:", result["classification"])
print(" Model Used:", result["model"])
print(" Response:\n", result["response"])

此 Python 代码实现了一个 prompt routing 系统来回答用户问题。它首先加载来自 .env 文件的 OpenAI 和 Google Custom Search 所需的 API 密钥。核心功能在于将用户的 prompt 分类为三个类别：simple、reasoning 或 internet search。一个专用函数利用 OpenAI 模型来完成此分类步骤。如果 prompt 需要最新信息，则使用 Google Custom Search API 执行 Google 搜索。另一个函数随后生成最终响应，根据分类选择适当的 OpenAI 模型。对于 internet search 查询，搜索结果作为上下文提供给模型。主 handle_prompt 函数编排此工作流，在生成响应之前调用分类和搜索（如果需要）函数。它返回分类、所使用的模型以及生成的答案。该系统高效地将不同类型的查询引导至优化的方法，以获得更好的响应。

动手代码示例（OpenRouter）

OpenRouter 通过单一的 API endpoint 提供数百个 AI 模型的统一接口。它提供自动 failover 和 cost-optimization，并可通过你喜欢的 SDK 或框架轻松集成。

python

import requests
import json

response = requests.post(
    url="https://openrouter.ai/api/v1/chat/completions",
    headers={
        "Authorization": "Bearer <OPENROUTER_API_KEY>",
        "HTTP-Referer": "<YOUR_SITE_URL>", # Optional. Site URL for rankings on openrouter.ai.
        "X-Title": "<YOUR_SITE_NAME>", # Optional. Site title for rankings on openrouter.ai.
    },
    data=json.dumps({
        "model": "openai/gpt-4o", # Optional
        "messages": [
            {
                "role": "user",
                "content": "What is the meaning of life?"
            }
        ]
    })
)

此代码片段使用 requests 库与 OpenRouter API 交互。它向 chat completion endpoint 发送一个 POST 请求，并附带一条用户消息。该请求包含带有 API 密钥的授权 headers 和可选的站点信息。目标是获取来自指定语言模型的响应，在本例中为 "openai/gpt-4o"。

OpenRouter 提供了两种不同的方法来路由和确定用于处理给定请求的计算模型。

Automated Model Selection：此功能将请求路由到从精选的可用模型集中选择的优化模型。选择是基于用户 prompt 的具体内容做出的。最终处理请求的模型的标识符会在响应的元数据中返回。

json

{
  "model": "openrouter/auto",
  ... // Other params
}

Sequential Model Fallback：此机制通过允许用户指定分层模型列表来提供操作冗余。系统将首先尝试使用序列中指定的主模型来处理请求。如果此主模型由于任何错误条件（例如服务不可用、rate-limiting 或内容过滤）而无法响应，系统将自动将请求重新路由到序列中的下一个指定模型。此过程持续进行，直到列表中的某个模型成功执行请求，或者列表已耗尽。操作的最终成本和响应中返回的模型标识符将对应于成功完成计算的模型。

json

{
  "models": ["anthropic/claude-3.5-sonnet", "gryphe/mythomax-l2-13b"],
  ... // Other params
}

OpenRouter 提供了一个详细的排行榜（https://openrouter.ai/rankings），根据累积 token 产量对可用的 AI 模型进行排名。它还提供来自不同提供商（ChatGPT、Gemini、Claude）的最新模型（见图1）。

Fig. 1: OpenRouter Web site (https://openrouter.ai/)

超越动态模型切换：Agent Resource Optimizations 的谱系

Resource-aware optimization 在开发能够在现实世界约束下高效且有效地运行的智能 Agent 系统中至关重要。让我们看看一些额外的技术：

Dynamic Model Switching 是一项关键技术，涉及根据手头任务的复杂性和可用计算资源来策略性地选择大型语言模型。当面对简单查询时，可以部署轻量级、经济高效的 LLM，而复杂、多方面的问题则需要利用更复杂且资源密集型的模型。

Adaptive Tool Use & Selection 确保 Agent 能够智能地从一套工具中进行选择，为每个特定的子任务选择最合适和最高效的工具，并仔细考虑 API 使用成本、延迟和执行时间等因素。这种动态工具选择通过优化外部 API 和服务的使用，提高了整体系统效率。

Contextual Pruning & Summarization 在管理 Agent 处理的信息量方面起着至关重要的作用，通过智能地总结和选择性地仅保留交互历史中最相关的信息，策略性地最小化 prompt token 数量并降低 inference 成本，防止不必要的计算开销。

Proactive Resource Prediction 涉及通过预测未来工作负载和系统需求来预测资源需求，这使得能够主动分配和管理资源，确保系统响应能力并防止瓶颈。

Cost-Sensitive Exploration 在多 Agent 系统中将优化考虑扩展到涵盖通信成本以及传统计算成本，影响 Agent 用于协作和共享信息的策略，旨在最大限度地减少总体资源支出。

Energy-Efficient Deployment 专为具有严格资源约束的环境而设计，旨在最大限度地减少智能 Agent 系统的能源足迹，延长运行时间并降低总体运行成本。

Parallelization & Distributed Computing Awareness 利用分布式资源来增强 Agent 的处理能力和吞吐量，将计算工作负载分布在多台机器或处理器上，以实现更高的效率和更快的任务完成速度。

Learned Resource Allocation Policies 引入了学习机制，使 Agent 能够根据反馈和性能指标随时间适应和优化其资源分配策略，通过持续改进来提高效率。

Graceful Degradation and Fallback Mechanisms 确保智能 Agent 系统即使在资源约束严重的情况下也能继续运行（尽管可能能力有所降低），优雅地降低性能并 fallback 到替代策略，以维持操作并提供基本功能。

概览

What：Resource-Aware Optimization 解决了在智能系统中管理计算、时间和财务资源消耗的难题。基于 LLM 的应用程序可能既昂贵又缓慢，而为每个任务选择最佳模型或工具通常是低效的。这在系统输出的质量与产生该输出所需的资源之间创造了一种基本的权衡。

如果没有动态管理策略，系统就无法适应不同的任务复杂性，也无法在预算和性能约束内运行。

Why：标准化的解决方案是构建一个能够基于手头任务智能监控和分配资源的 agentic 系统。此模式通常采用"Router Agent"来首先分类传入请求的复杂性。然后将请求转发给最合适的 LLM 或工具——对简单查询使用快速、廉价的模型，对复杂推理使用更强大的模型。"Critique Agent"可以通过评估响应质量、提供反馈以随时间改进路由逻辑，进一步优化此过程。这种动态的、多 Agent 的方法确保系统高效运行，平衡响应质量与成本效益。

Rule of thumb：当在严格的 API 调用或计算能力财务预算下运行时；构建对快速响应时间至关重要的延迟敏感型应用程序时；在资源受限的硬件（如电池寿命有限的边缘设备）上部署 Agent 时；以编程方式平衡响应质量与运营成本之间的权衡时；以及管理不同任务具有不同资源需求的复杂、多步骤工作流时，请使用此模式。

视觉摘要

关键要点

Resource-Aware Optimization 至关重要：智能 Agent 可以动态管理计算、时间和财务资源。关于模型使用和执行路径的决策是基于实时约束和目标做出的。
多 Agent 架构的可扩展性：Google 的 ADK 提供了一个多 Agent 框架，支持模块化设计。不同的 Agent（回答、路由、评判）处理特定任务。
动态的、LLM 驱动的 Routing：Router Agent 根据查询复杂性和预算将查询引导至语言模型（简单查询用 Gemini Flash，复杂查询用 Gemini Pro）。这优化了成本和性能。
Critique Agent 功能：专用的 Critique Agent 为 Self-Correction、性能监控和改进路由逻辑提供反馈，增强了系统的有效性。
通过反馈和灵活性进行优化：用于评判的评估能力和模型集成灵活性有助于自适应和自改进的 system behavior。
其他 Resource-Aware Optimizations：其他方法包括 Adaptive Tool Use & Selection、Contextual Pruning & Summarization、Proactive Resource Prediction、Cost-Sensitive Exploration in Multi-Agent Systems、Energy-Efficient Deployment、Parallelization & Distributed Computing Awareness、Learned Resource Allocation Policies、Graceful Degradation and Fallback Mechanisms，以及 Prioritization of Critical Tasks。

结论

Resource-aware optimization 对于开发智能 Agent 至关重要，使其能够在现实世界的约束下高效运行。通过管理计算、时间和财务资源，Agent 可以实现最佳性能和成本效益。诸如动态模型切换、自适应工具使用和 contextual pruning 等技术对于实现这些效率至关重要。高级策略（包括 learned resource allocation policies 和 graceful degradation）增强了 Agent 在不同条件下的适应性和韧性。将这些优化原则整合到 Agent 设计中，是构建可扩展、稳健和可持续的 AI 系统的基础。

参考文献

Google's Agent Development Kit (ADK): https://google.github.io/adk-docs/
Gemini Flash 2.5 & Gemini 2.5 Pro: https://aistudio.google.com/
OpenRouter: https://openrouter.ai/docs/quickstart