第 16 章：资源感知优化

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

# 概念性 Python 结构，非可运行代码

from google.adk.agents import Agent
# from google.adk.models.lite_llm import LiteLlm # 如需调用 ADK 默认 Agent 不支持的模型

# 使用昂贵 Gemini Pro 2.5 的 Agent
gemini_pro_agent = Agent(
    name="GeminiProAgent",
    model="gemini-2.5-pro", # 实际模型名如有不同请替换
    description="复杂查询的高能力 Agent。",
    instruction="你是复杂问题解决的专家助手。"
)

# 使用经济型 Gemini Flash 2.5 的 Agent
gemini_flash_agent = Agent(
    name="GeminiFlashAgent",
    model="gemini-2.5-flash", # 实际模型名如有不同请替换
    description="简单问题的快速高效 Agent。",
    instruction="你是简单问题的快速助手。"
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# 概念性 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 Agent。"

    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 Agent")
              response = await gemini_flash_agent.run_async(context.current_message)
              yield Event(author=self.name, content=f"Flash Agent 处理结果：{response}")
         else:
              print(f"长问题（长度：{query_length}）路由到 Gemini Pro Agent")
              response = await gemini_pro_agent.run_async(context.current_message)
              yield Event(author=self.name, content=f"Pro Agent 处理结果：{response}")

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132

# 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"Title: {item.get('title')}\nSnippet: {item.get('snippet')}\nLink: {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)
    classification = classification_result["classification"]

    search_results = None
    if classification == "internet_search":
         search_results = google_search(prompt)

    answer, model = generate_response(prompt, classification, search_results)
    return {"classification": classification, "response": answer, "model": model}
test_prompt = "澳大利亚的首都是什么？"
# test_prompt = "解释量子计算对密码学的影响。"
# test_prompt = "澳网 2026 什么时候开始，请给出完整日期？"

result = handle_prompt(test_prompt)
print("🔍 分类：", result["classification"])
print("🧠 使用模型：", result["model"])
print("🧠 响应：\n", result["response"])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

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>", # 可选，站点 URL 用于 openrouter.ai 排名
    "X-Title": "<YOUR_SITE_NAME>", # 可选，站点名称用于 openrouter.ai 排名
 },
 data=json.dumps({
    "model": "openai/gpt-4o", # 可选
    "messages": [
      {
         "role": "user",
         "content": "生命的意义是什么？"
      }
    ]
 })
)