第16章：资源感知优化 Resource-Aware Optimization

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

例如，考虑一个为金融分析师分析大型数据集的智能体。如果分析师需要立即获得初步报告，智能体可能会使用一个更快、更经济的模型来快速总结关键趋势。然而，如果分析师需要一个高度准确的预测来进行关键投资决策，并且有更大的预算和更多的时间，智能体将分配更多资源以使用一个功能更强大、速度较慢但更精确的预测模型。在这一领域的一个关键策略是回退机制，当首选模型由于过载或限制而不可用时，它充当一种保障措施。为了确保系统的优雅降级，系统会自动切换到默认或更经济的模型，从而保持服务的连续性而不是完全失败。

实际应用与使用场景

实际使用场景包括：

成本优化的LLM使用：智能体根据预算约束决定是否使用大型、昂贵的LLM处理复杂任务，或使用较小、更经济的模型处理简单查询。
时延敏感操作：在实时系统中，智能体选择更快但可能不够全面的推理路径，以确保及时响应。
能源效率：对于部署在边缘设备或有限电力设备上的智能体，优化其处理以节省电池寿命。
服务可靠性的回退机制：当首选模型不可用时，智能体自动切换到备用模型，确保服务连续性并实现优雅降级。
数据使用管理：智能体选择摘要数据检索而不是下载完整数据集，以节省带宽或存储空间。
自适应任务分配：在多智能体系统中，智能体根据其当前的计算负载或可用时间自我分配任务。

实践代码示例

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

例如，考虑一个由分层智能体构建的旅行规划器。高层规划涉及理解用户的复杂需求，将其分解为多步骤的行程，并做出逻辑决策。这部分将由一个复杂且功能更强大的LLM（如 Gemini Pro）管理。这个“规划器”智能体需要深入理解上下文并具备推理能力。

然而，一旦计划确定，计划中的各个任务（例如查询航班价格、检查酒店可用性或查找餐馆评论）本质上是简单的、重复的网络查询。这些“工具功能调用”可以由一个更快、更经济的模型（如 Gemini Flash）来执行。可以很容易地理解为什么可以使用经济型模型来处理这些简单的网络搜索，而复杂的规划阶段需要更高级模型的智能，以确保制定出连贯且逻辑合理的旅行计划。

谷歌的 ADK 通过其多智能体架构支持这种方法，使应用程序具有模块化和可扩展性。不同的智能体可以处理专门的任务。模型的灵活性使得可以直接使用各种 Gemini 模型，包括 Gemini Pro 和 Gemini Flash，或者通过 LiteLLM 集成其他模型。ADK 的编排功能支持动态、由 LLM 驱动的路由，以实现自适应行为。内置的评估功能允许系统性地评估智能体性能，这些评估结果可用于系统优化（详见评估与监控章节）。

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

## 概念性的类似 Python 的结构，非可运行代码
from google.adk.agents import Agent
## from google.adk.models.lite_llm import LiteLlm  # 如果使用 ADK 默认智能体不直接支持的模型

## 使用更昂贵的 Gemini Pro 2.5 的智能体
gemini_pro_agent = Agent(
    name="GeminiProAgent",
    model="gemini-2.5-pro",  # 如果实际模型名称不同，这里是占位符
    description="一个用于复杂查询的高能力智能体。",
    instruction="你是一个擅长解决复杂问题的专家助手。"
)

## 使用较便宜的 Gemini Flash 2.5 的智能体
gemini_flash_agent = Agent(
    name="GeminiFlashAgent",
    model="gemini-2.5-flash",  # 如果实际模型名称不同，这里是占位符
    description="一个快速高效的智能体，用于简单查询。",
    instruction="你是一个快速响应简单问题的助手。"
)

一个路由智能体可以基于简单的指标（例如查询长度）来分配查询，较短的查询分配给成本较低的模型，而较长的查询则分配给能力更强的模型。然而，更复杂的路由智能体可以利用 LLM 或 ML 模型来分析查询的细微差别和复杂性。这个 LLM 路由器可以确定最适合的下游语言模型。例如，请求事实性回忆的查询会被路由到 Flash 模型，而需要深度分析的复杂查询会被路由到 Pro 模型。

优化技术可以进一步提高 LLM 路由器的有效性。提示调优涉及设计提示以引导路由器 LLM 做出更好的路由决策。通过对包含查询及其最佳模型选择的数据集进行微调，可以提高 LLM 路由器的准确性和效率。这种动态路由能力在响应质量与成本效益之间实现了平衡。

## 概念性的类似 Python 的结构，非可运行代码
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 = "根据复杂性将用户查询路由到适当的 LLM 智能体。"

    async def _run_async_impl(self, context: InvocationContext) -> AsyncGenerator[Event, None]:
        user_query = context.current_message.text  # 假设输入为文本
        query_length = len(user_query.split())  # 简单指标：单词数量

        if query_length < 20:  # 简单与复杂的示例阈值
            print(f"将短查询（长度：{query_length}）路由到 Gemini Flash 智能体")
            # 在实际的 ADK 设置中，你可以使用 'transfer_to_agent' 或直接调用
            # 为演示目的，我们将模拟调用并生成其响应
            response = await gemini_flash_agent.run_async(context.current_message)
            yield Event(author=self.name, content=f"Flash 智能体处理结果: {response}")
        else:
            print(f"将长查询（长度：{query_length}）路由到 Gemini Pro 智能体")
            response = await gemini_pro_agent.run_async(context.current_message)
            yield Event(author=self.name, content=f"Pro 智能体处理结果: {response}")

语言模型的批评智能体（Critique Agent）负责评估模型的响应，并提供多功能的反馈。对于自我纠正，它识别错误或不一致之处，促使回答智能体优化输出以提高质量。同时，它系统性地评估响应以进行性能监控，追踪准确性和相关性等指标，这些指标可用于优化。

此外，其反馈还可用于强化学习或微调；例如，持续识别Flash模型响应不足的情况，可以优化路由智能体的逻辑。尽管批评智能体并不直接管理预算，但通过识别次优的路由选择（如将简单查询分配给Pro模型或将复杂查询分配给Flash模型导致结果不佳），它间接地促进了预算管理。这些信息有助于调整资源分配并节约成本。

批评智能体可以配置为仅审查回答智能体生成的文本，或同时审查原始查询和生成的文本，从而全面评估响应与初始问题的对齐程度。

CRITIC_SYSTEM_PROMPT = """
你是**批评智能体**，是我们协作研究助手系统的质量保证部门。
你的主要职责是**仔细审查并挑战**来自研究智能体的信息，确保其**准确性、完整性和无偏性**。

你的任务包括：
* **评估研究结果**的事实正确性、全面性以及潜在倾向性。
* **识别任何遗漏的数据**或推理中的不一致之处。
* **提出关键问题**以优化或扩展当前的理解。
* **提供建设性建议**以改进或探索不同的角度。
* **验证最终输出的全面性**和均衡性。

所有批评必须具有建设性。你的目标是强化研究，而不是否定它。
清晰地组织你的反馈，突出需要修订的具体点。
你的总体目标是确保最终的研究成果达到最高质量标准。
"""

批评智能体基于预定义的系统提示操作，该提示明确了其角色、职责和反馈方法。一个设计良好的提示必须清楚地定义其作为评估者的功能，明确批评的重点领域，并强调提供建设性反馈而非简单否定。提示还应鼓励识别优点和缺点，并指导智能体如何组织和呈现其反馈。

OpenAI 实践代码

该系统采用资源感知优化策略高效处理用户查询。它首先将每个查询分类为三类之一，以确定最适合且最具成本效益的处理路径。这种方法避免了在简单请求上浪费计算资源，同时确保复杂查询得到必要的关注。这三类包括：

simple：针对可以直接回答的简单问题，无需复杂推理或外部数据。
reasoning：针对需要逻辑推导或多步骤思维过程的查询，这类查询会被路由至更强大的模型。
internet_search：针对需要当前信息的问题，这类查询会自动触发Google搜索以提供最新答案。

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

## 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_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(
        "请在 .env 文件中设置 OPENAI_API_KEY、GOOGLE_CUSTOM_SEARCH_API_KEY 和 GOOGLE_CSE_ID。"
    )

client = OpenAI(api_key=OPENAI_API_KEY)

## --- 步骤 1: 分类用户输入 ---
def classify_prompt(prompt: str) -> dict:
    system_message = {
        "role": "system",
        "content": (
            "你是一个分类器，分析用户输入并仅返回以下三类之一：\n\n"
            "- simple\n"
            "- reasoning\n"
            "- internet_search\n\n"
            "规则：\n"
            "- 如果是直接的事实性问题，不需要推理或涉及当前事件，请使用 'simple'。\n"
            "- 如果是逻辑、数学或多步骤推理问题，请使用 'reasoning'。\n"
            "- 如果输入涉及当前事件、最新数据或超出你的训练数据范围，请使用 'internet_search'。\n\n"
            "仅以以下 JSON 格式响应：\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)

## --- 步骤 2: Google 搜索 ---
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)}

## --- 步骤 3: 生成响应 ---
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"
        # 将每个搜索结果字典转换为可读字符串
        if search_results:
            search_context = "\n".join(
                [
                    f"标题: {item.get('title')}\n摘要: {item.get('snippet')}\n链接: {item.get('link')}"
                    for item in search_results
                ]
            )
        else:
            search_context = "未找到搜索结果。"
        full_prompt = f"""使用以下网页结果回答用户查询: {search_context} 查询: {prompt}"""

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

## --- 步骤 4: 综合处理 ---
def handle_prompt(prompt: str) -> dict:
    classification_result = classify_prompt(prompt)
    # 删除或注释掉下一行以避免重复打印
    # print("\n🔍 分类结果:", 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 代码实现了一个用于处理用户问题的提示路由系统。它首先从 .env 文件加载 OpenAI 和 Google Custom Search 的必要 API 密钥。核心功能在于将用户的提示分类为三种类别：简单问题、推理问题或互联网搜索问题。一个专用函数使用 OpenAI 模型完成分类步骤。如果提示需要当前信息，则通过 Google Custom Search API 执行 Google 搜索。另一个函数根据分类选择适当的 OpenAI 模型生成最终响应。对于互联网搜索查询，搜索结果会作为上下文提供给模型。主函数 handle_prompt 协调这一工作流，调用分类和搜索（如果需要）功能，然后生成响应。它返回分类结果、使用的模型以及生成的答案。该系统能够高效地将不同类型的查询引导至优化的方法以获得更好的响应。

实操代码示例（OpenRouter）

OpenRouter 提供了一个统一的接口，通过单一 API 端点访问数百种 AI 模型。它支持自动故障转移和成本优化，并可通过您偏好的 SDK 或框架轻松集成。

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>",  # 可选。用于 openrouter.ai 排名的站点 URL。
    "X-Title": "<YOUR_SITE_NAME>",      # 可选。用于 openrouter.ai 排名的站点标题。
  },
  data=json.dumps({
    "model": "openai/gpt-4o",  # 可选
    "messages": [
      {
        "role": "user",
        "content": "What is the meaning of life?"
      }
    ]
  })
)

此代码片段使用 requests 库与 OpenRouter API 进行交互。它向聊天完成端点发送一个包含用户消息的 POST 请求。请求包括带有 API 密钥的授权头以及可选的站点信息。目标是从指定的语言模型（此处为 "openai/gpt-4o"）获取响应。

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

自动模型选择： 此功能将请求路由到从可用模型中精选的优化模型。选择基于用户提示的具体内容。最终处理请求的模型标识符会在响应的元数据中返回。

{
  "model": "openrouter/auto",
  ... // 其他参数
}

顺序模型回退机制（Sequential Model Fallback）：

该机制通过允许用户指定模型的层级列表来提供操作冗余。系统首先尝试使用序列中指定的主模型处理请求。如果由于服务不可用、速率限制或内容过滤等错误条件导致主模型无法响应，系统会自动将请求重新路由到序列中下一个指定的模型。此过程会持续进行，直到列表中的某个模型成功执行请求或列表耗尽为止。操作的最终成本以及响应中返回的模型标识符将对应于成功完成计算的模型。

{
  "models": ["anthropic/claude-3.5-sonnet", "gryphe/mythomax-l2-13b"],
  ... // 其他参数
}

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

图 1: OpenRouter 网站 (https://openrouter.ai/)

超越动态模型切换：AI智能体资源优化的多样化技术

在开发能够在现实世界约束下高效运行的AI智能体系统时，资源感知优化至关重要。以下是一些额外的优化技术：

动态模型切换（Dynamic Model Switching）

这是一个关键技术，涉及根据任务的复杂性和可用计算资源战略性地选择大型语言模型（LLM）。面对简单查询时，可以部署轻量化、成本低廉的 LLM；而复杂的多方面问题则需要使用更复杂且资源密集型的模型。

自适应工具使用与选择（Adaptive Tool Use & Selection）

确保智能体能够从一系列工具中智能选择最适合且高效的工具来完成每个具体子任务，同时仔细考虑 API 使用成本、延迟和执行时间等因素。动态工具选择通过优化外部 API 和服务的使用来提升系统整体效率。

上下文剪枝与摘要（Contextual Pruning & Summarization）

在管理智能体处理的信息量时发挥重要作用，通过智能摘要和选择性保留交互历史中最相关的信息，战略性地减少提示 token 数量，从而降低推理成本，避免不必要的计算开销。

资源需求预测（Proactive Resource Prediction）

通过预测未来的工作负载和系统需求来预估资源需求，使得能够主动分配和管理资源，确保系统响应能力并防止出现瓶颈。

成本敏感的探索（Cost-Sensitive Exploration）

在多智能体系统中，将优化考虑扩展到通信成本以及传统的计算成本，影响智能体协作和信息共享的策略，目标是尽量减少整体资源开销。

节能部署（Energy-Efficient Deployment）

专为资源限制严格的环境量身定制，旨在减少AI智能体系统的能源消耗，延长运行时间并降低整体运行成本。

并行化与分布式计算意识（Parallelization & Distributed Computing Awareness）

利用分布式资源来增强智能体的处理能力和吞吐量，通过将计算任务分配到多个机器或处理器上，实现更高效、更快速的任务完成。

学习型资源分配策略引入了一种学习机制，使AI智能体能够根据反馈和性能指标不断调整和优化其资源分配策略，从而通过持续改进提高效率。

优雅降级与回退机制确保AI智能体系统即使在资源限制严重的情况下，也能继续运行，尽管性能可能有所降低，但能够优雅地降级并回退到替代策略，以维持操作并提供基本功能。

概览

定义（What）

资源感知优化解决了在智能系统中管理计算、时间和财务资源消耗的挑战。基于大型语言模型（LLM）的应用通常成本高昂且运行缓慢，为每项任务选择最佳模型或工具通常效率低下。这导致了系统输出质量与所需资源之间的基本权衡。如果没有动态管理策略，系统无法适应任务复杂性的变化，也无法在预算和性能限制内运行。

设计意图（Why）

标准化解决方案是构建一个能够根据任务智能监控和分配资源的智能体系统。该模式通常采用“路由智能体”首先对传入请求的复杂性进行分类。然后将请求转发给最合适的LLM或工具——对于简单查询使用快速、低成本的模型，而对于复杂推理使用更强大的模型。“批判智能体”可以进一步优化流程，通过评估响应质量并提供反馈来改进路由逻辑。这种动态的多智能体方法确保系统高效运行，在响应质量与成本效益之间取得平衡。

使用原则（Rule of Thumb）

在以下情况下使用该模式：API调用或计算能力受严格财务预算限制、构建对响应时间敏感的应用程序、在资源受限的硬件（如电池寿命有限的边缘设备）上部署智能体、程序化地平衡响应质量与运行成本之间的权衡，以及管理复杂的多步骤工作流，其中不同任务具有不同的资源需求。

图解（Visual Summary）

图2：资源感知优化设计模式

关键要点

资源感知优化至关重要：AI智能体能够动态管理计算、时间和财务资源。关于模型使用和执行路径的决策基于实时约束和目标进行。
多智能体架构实现可扩展性：Google的ADK提供了一个多智能体框架，支持模块化设计。不同智能体（回答、路由、批判）处理特定任务。
动态的LLM驱动路由：路由智能体根据查询复杂性和预算将查询定向到语言模型（简单任务使用Gemini Flash，复杂任务使用Gemini Pro），优化了成本和性能。
批判智能体功能：专门的批判智能体提供反馈，用于自我纠正、性能监控以及优化路由逻辑，从而增强系统的有效性。
通过反馈和灵活性实现优化：批判评估能力和模型集成的灵活性促进了系统的自适应和自我改进行为。
其他资源感知优化方法：包括自适应工具使用与选择、上下文剪枝与摘要、主动资源预测、多智能体系统中的成本敏感探索、节能部署、并行化与分布式计算意识、学习型资源分配策略、优雅降级与回退机制，以及关键任务优先级。

结论

资源感知优化对于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

第16章：资源感知优化

第16章：资源感知优化 Resource-Aware Optimization

实际应用与使用场景

实践代码示例

OpenAI 实践代码

实操代码示例（OpenRouter）

超越动态模型切换：AI智能体资源优化的多样化技术

概览

定义（What）

设计意图（Why）

使用原则（Rule of Thumb）

图解（Visual Summary）

关键要点

结论

参考文献

results matching ""

No results matching ""

第16章：资源感知优化 Resource-Aware Optimization

实际应用与使用场景

实践代码示例

OpenAI 实践代码

实操代码示例（OpenRouter）

超越动态模型切换：AI智能体资源优化的多样化技术

概览

定义（What）

设计意图（Why）

使用原则（Rule of Thumb）

图解 （Visual Summary）

关键要点

结论

参考文献

results matching ""

No results matching ""

图解（Visual Summary）