我要投稿

我把 Agent 的 Token 消耗降了 60%：ADK 多 Agent 架构实践（含关键代码）

发布日期：2025-08-02 16:24:16 浏览次数： 1513

作者：弓长先生的杂货铺

微信搜一搜，关注“弓长先生的杂货铺”

在Agent：从零到一的实践指南（含关键代码）" data-itemshowtype="0" linktype="text" data-linktype="2">用 LangChain 打造你的第一个 AI Agent：从零到一的实践指南（含关键代码）这篇文章中，我用 LangChain实现了一个 AI Agent，结构是单主 Agent + 多个 Tool（如：DatabaseInfoTool、TableSchemaTool、QueryExecutorTool、SampleDataTool等）。后来我用 Google ADK把这个 Agent 重写了一遍，依旧沿用了“一个主 Agent 带多个 Tool”的思路。

在随后的分析中从ReAct流程看LangChain和Google ADK的Agent设计理念差异，我发现：在 ADK 框架中，单 Agent 多 Tool 的架构 Token 消耗偏高。于是我提出一个设计思路：ADK 更适合“主 Agent + 多个 Sub-Agent”的结构——每个 Sub-Agent 只包含一个 Tool，职责单一、边界清晰。

一、优化效果：立竿见影

我按这个思路把 ADK 架构改成了1 个主 Agent 负责调度，多个 Sub-Agent 负责执行，且每个 Sub-Agent 只绑定 1 个 Tool，严格遵守“单一职责”原则。

主要做了以下两方面的调整：

第一步（架构从单 Agent 多 Tool → 主 + 多 Sub-Agent）：Token 立减约 35%。
第二步（在此基础上优化提示词 + 引入缓存）：总降幅达到约 60%。

由于ADK 框架使用了 LLM 的原生能力（如工具筛选、Function Call等），所以在 Agent 的代码实现上确实更加的简洁。调整完成后，项目总代码量大约为 1800 行，比 LangChain 版本的体量小很多。

说明：上述降幅是在同一任务下统计的，统计口径一致（包括 prompt、function call等Agent和LLM之间交互的所有 token）。

二、针对 ADK 的架构演进

之前：单 Agent 多 Tool

统一持有多个 Tool，路由调度集中在主 Agent 内部。
上下文容易“越滚越大”，无形增加 Token。
工具候选多时，模型在选择与澄清上会产生大量Token消耗，也容易造成额外的交互。

现在：主 Agent + 多 Sub-Agent（每个 Sub-Agent 仅 1 个 Tool）

主 Agent 只负责意图识别和任务分发。
Sub-Agent 处理单一职责，上下文短、交互简单。
路由清晰，减少无效澄清与工具试探。

为什么能省：

上下文更短：Sub-Agent 只携带本职所需的少量上下文与历史信息。
工具候选更少：每个 Sub-Agent 仅 1 个 Tool，减少工具选择带来的额外 token。
调度更“干脆”：主 Agent 明确路由，避免在一个 Agent 里多轮试错。
工具描述更精简：职责切分后，工具描述能够更加的精简，可以删除大量的冗长描述。

三、关键代码实现

1. 概述

以下代码是基于SQLite多智能体项目的实际实现，详细介绍如何使用Google ADK框架开发AI Agent。该项目展示了一个完整的多智能体协作架构，包含主智能体协调、子智能体分工、工具集成和安全性设计。

2. 项目结构

sqlite-multi-agent/├── sqlite_agent/           # 主要智能体代码│   ├── agent.py           # 主智能体实现│   ├── config.py          # 配置管理│   ├── tools.py           # 工具函数│   ├── utils.py           # 工具类│   ├── prompts.py         # 提示词模板│   ├── monitored_llm.py   # LLM监控包装器（LLM必须，监控为可选）│   └── sub_agents/        # 子智能体│       ├── list_tables/   # 表列表智能体│       ├── table_info/    # 表信息智能体│       ├── query_execution/  # 查询执行智能体│       └── sample_data/   # 样本数据智能体├── examples/              # 示例代码├── tests/                 # 测试代码└── deployment/            # 部署脚本

ADK 架构的 Agent 实现起来比较简单，只需要实现工具函数、LLM 交互工具类、主 Agent 、Sub-Agent 就可以了。

3. 关键代码

1）tools.py

数据库访问：

def execute_query(query: str) -> str:    """Execute a SELECT query on the SQLite database. Use for data retrieval, filtering, aggregation, and analysis. Only SELECT queries are allowed."""    config = get_sqlite_config()
    # Validate query for safety    validation_result = validate_query(query)    if not validation_result["valid"]:        return f"Query validation failed: {validation_result['error']}"
    try:        with get_database_connection(config.database_path) as conn:            cursor = conn.cursor()            cursor.execute(query)
            # Get column names            columns = [description[0] for description in cursor.description] if cursor.description else []
            # Fetch results with limit            results = cursor.fetchmany(config.max_results)
            if not results:                return "Query executed successfully but returned no results."
            # Format results            return format_results(results, columns, config.max_results)
    except Exception as e:        return f"Error executing query: {str(e)}"

工具注册：

# Create ADK function toolslist_tables_tool = FunctionTool(list_tables)get_table_info_tool = FunctionTool(get_table_info)execute_query_tool = FunctionTool(execute_query)get_sample_data_tool = FunctionTool(get_sample_data)
# Export all toolssqlite_tools = [    list_tables_tool,    get_table_info_tool,    execute_query_tool,    get_sample_data_tool]

2）utils.py

结果格式化：

def format_results(results: List[Any], columns: List[str], max_results: int) -> str:    """Format query results for display."""    if not results:        return "No results found."
    # Convert results to list of lists for tabulate    formatted_rows = []    for row in results:        if hasattr(row, '_fields'):  # sqlite3.Row object            formatted_rows.append([str(value) if value is not None else 'NULL' for value in row])        else:            formatted_rows.append([str(value) if value is not None else 'NULL' for value in row])
    # Create table    table = tabulate(formatted_rows, headers=columns, tablefmt="grid")
    result = f"Query Results ({len(results)} rows):\n\n{table}"
    # Add warning if results were limited    if len(results) >= max_results:        result += f"\n\n⚠️ Results limited to {max_results} rows. Use LIMIT clause for more control."
    return result

3）agent.py

主 Agent 实现：

import osfrom typing import Optionalfrom datetime import datetimefrom google.adk.agents import Agentfrom google.adk.agents.callback_context import CallbackContextfrom google.adk.models.lite_llm import LiteLlmfrom google.adk.tools.agent_tool import AgentToolfrom google.genai import typesfrom .config import get_sqlite_config, get_model_name, get_DeepSeek_api_keyfrom .prompts import get_sqlite_agent_instructionsfrom .utils import create_sample_database, get_database_connectionfrom .sub_agents import (    list_tables,    table_info,    query_execution,    sample_data)from .monitored_llm import create_monitored_llm

# 创建多Agent协调的SQLite Agentsqlite_agent = Agent(    name="sqlite_agent",    model=create_monitored_llm(model=get_model_name()),    description="SQLite数据库助手，通过协调多个专门的子agent来完成数据库操作",    instruction=get_sqlite_agent_instructions,    sub_agents=[        list_tables,        table_info,        query_execution,        sample_data,    ],    before_agent_callback=setup_database,    generate_content_config=types.GenerateContentConfig(        safety_settings=[            types.SafetySetting(                category=types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,                threshold=types.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,            ),        ],        temperature=0.1,        top_p=0.8,        max_output_tokens=2048,    ),)

4）prompts.py

主 Agent 提示词：

def get_sqlite_agent_instructions(ctx) -> str:    """Get the main instructions for the SQLite agent."""    return f"""你是专业的SQLite数据库助手，通过协调专门的子agent来高效完成数据库操作。## 核心职责- 分析用户意图，精准选择最合适的agent- 确保查询结果准确、格式清晰- 主动识别潜在需求，提供相关建议## 可用agent选择指南### list_tables - 表发现**用途**：当用户询问"有哪些表"、"显示所有表"、"查看数据库结构"时使用**关键词**：表、列表、所有、结构、数据库### table_info - 表结构分析  **用途**：当用户询问"表结构"、"字段信息"、"列信息"、"schema"时使用**关键词**：结构、字段、列、schema、信息、详情### query_execution - 数据查询**用途**：当用户需要执行具体查询、分析数据、筛选记录时使用**关键词**：查询、查找、统计、分析、最贵、最新、数量、总和### sample_data - 样本数据**用途**：当用户想"看看数据"、"预览内容"、"示例数据"时使用**关键词**：看看、样本、示例、预览、数据## 智能选择策略1. **探索阶段**：用户首次接触数据库 → list_tables2. **结构了解**：用户想了解特定表 → table_info  3. **数据分析**：用户有具体查询需求 → query_execution4. **快速预览**：用户想快速了解数据 → sample_data## 响应要求- 直接调用agent，不解释选择过程- 确保结果对用户可见- 中文问题用中文回答- 表格数据用清晰格式展示当前数据库表：{', '.join(ctx.state.get('available_tables', []))}"""

5）sub_agents/query_execution/agent.py

智能查询子 Agent：

"""Query execution agent - specialized agent for executing SQL queries."""from google.adk.agents import Agentfrom google.genai import typesfrom ...config import get_model_namefrom ...tools import execute_query_tool, get_table_info_toolfrom ...monitored_llm import create_monitored_llmfrom .prompt import QUERY_EXECUTION_AGENT_INSTRquery_execution = Agent(    name="query_execution",    model=create_monitored_llm(model=get_model_name()),    description="专门负责执行SQLite数据库查询的agent，遵循单一职责原则",    instruction=QUERY_EXECUTION_AGENT_INSTR,    tools=[execute_query_tool],  # 仅执行查询，避免多余表结构查询    disallow_transfer_to_parent=True,    disallow_transfer_to_peers=True,    generate_content_config=types.GenerateContentConfig(        temperature=0.1,        top_p=0.8,        max_output_tokens=2048,    ),)

6）sub_agents/query_execution/agent.py

智能执行子 Agent：

"""Query execution agent - specialized agent for executing SQL queries."""from google.adk.agents import Agentfrom google.genai import typesfrom ...config import get_model_namefrom ...tools import execute_query_tool, get_table_info_toolfrom ...monitored_llm import create_monitored_llmfrom .prompt import QUERY_EXECUTION_AGENT_INSTRquery_execution = Agent(    name="query_execution",    model=create_monitored_llm(model=get_model_name()),    description="专门负责执行SQLite数据库查询的agent，遵循单一职责原则",    instruction=QUERY_EXECUTION_AGENT_INSTR,    tools=[execute_query_tool],  # 仅执行查询，避免多余表结构查询    disallow_transfer_to_parent=True,    disallow_transfer_to_peers=True,    generate_content_config=types.GenerateContentConfig(        temperature=0.1,        top_p=0.8,        max_output_tokens=2048,    ),)

7）sub_agents/query_execution/prompt.py

智能执行子 Agent 提示词：

"""Prompts for the query execution agent."""QUERY_EXECUTION_AGENT_INSTR = """执行SQL查询专家。直接执行查询并返回结果给用户。**使用execute_query工具：**- 将用户问题直接转换为SQL查询- 例如："最贵的订单" → "SELECT * FROM orders ORDER BY total_amount DESC LIMIT 1"- 例如："最新用户" → "SELECT * FROM users ORDER BY created_at DESC LIMIT 1"**输出格式：**```sql[实际SQL]```[查询结果表格]统计：X条记录**规则：**- 仅执行SELECT查询- 直接回答，不解释过程"""

8）monitored_llm.py

使用 LiteLLM 和 deepseek 交互：

def create_monitored_llm(model: str, **kwargs) -> LiteLlm:    """创建带监控功能的LLM实例"""    return LiteLlm(model=model, **kwargs)

4. 实现总结

和 LangChain 相比，ADK 更偏向使用 LLM 原生机制（工具选择、Function Call），代码更少，工程上更“轻”。

ADK 架构落地的建议：

先确定设计原则：单 Agent 多 Tool vs 主 + 多 Sub-Agent。
子 Agent 职责切到最小：每个 Sub-Agent 只干一件事。
主 Agent 只负责调度：主 Agent 只做意图识别与任务分发。
精简系统提示：优化 Sub-Agent 提示词，不要背不必要的上下文。
逐步优化 Token：深入分析 Token 的损耗点，逐步进行优化。

四、优化效果

下图是 ADK 自带的Web UI，通过“adk web”命令即可启动。这个Web工具提供了非常实用的Trace、Event、Session等跟踪工具，可以大幅提升调试的效率。

优化后，一次查询产生的三次 Agent 和 LLM 之间交互消耗的 Token 数，从5330降至了2107，降幅60%。

五、总结

把架构从“单 Agent 多 Tool”切到“主 Agent + 多个单一职责的 Sub-Agent”，再配合提示词精简，我的 Token 消耗最终降到了原来的约 40%。这次改造对工程复杂度的增加可控，重构后的代码结构更加合理，而且对成本和吞吐的改善非常可观。

下一篇，我应该会写 ADK 自带的评估机制，让你能够利用框架自身的能力编写自动化评估测试Case。

53AI，企业落地大模型首选服务商

产品：场景落地咨询+大模型应用平台+行业解决方案

承诺：免费POC验证，效果达标后再合作。零风险落地应用大模型，已交付160+中大型企业