链载Ai

标题: 我把 Agent 的 Token 消耗降了 60%：ADK 多 Agent 架构实践（含关键代码） [打印本页]

---

作者: 链载Ai    时间: 昨天 21:53
标题: 我把 Agent 的 Token 消耗降了 60%：ADK 多 Agent 架构实践（含关键代码）

在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 处理单一职责，上下文短、交互简单。

• 路由清晰，减少无效澄清与工具试探。

为什么能省：

1. 上下文更短：Sub-Agent 只携带本职所需的少量上下文与历史信息。

2. 工具候选更少：每个 Sub-Agent 仅 1 个 Tool，减少工具选择带来的额外 token。

3. 调度更“干脆”：主 Agent 明确路由，避免在一个 Agent 里多轮试错。

4. 工具描述更精简：职责切分后，工具描述能够更加的精简，可以删除大量的冗长描述。

三、关键代码实现

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

数据库访问：

defexecute_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) ifnotvalidation_result["valid"]:   returnf"Query validation failed:{validation_result['error']}"
 try:   withget_database_connection(config.database_path)asconn:      cursor = conn.cursor()      cursor.execute(query)
     # Get column names      columns = [description[0]fordescriptionincursor.description]ifcursor.descriptionelse[]
     # Fetch results with limit      results = cursor.fetchmany(config.max_results)
     ifnotresults:       return"Query executed successfully but returned no results."
     # Format results     returnformat_results(results, columns, config.max_results)
 exceptExceptionase:   returnf"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

结果格式化：

defformat_results(resultsist[Any], columnsist[str], max_results:int) ->str: """Format query results for display.""" ifnotresults:   return"No results found."
 # Convert results to list of lists for tabulate  formatted_rows = [] forrowinresults:   ifhasattr(row,'_fields'): # sqlite3.Row object      formatted_rows.append([str(value)ifvalueisnotNoneelse'NULL'forvalueinrow])   else:      formatted_rows.append([str(value)ifvalueisnotNoneelse'NULL'forvalueinrow])
 # 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 iflen(results) >= max_results:    result +=f"\n\n⚠️ Results limited to{max_results}rows. Use LIMIT clause for more control."
 returnresult

3）agent.py

主 Agent 实现：

importosfromtypingimportOptionalfromdatetimeimportdatetimefromgoogle.adk.agentsimportAgentfromgoogle.adk.agents.callback_contextimportCallbackContextfromgoogle.adk.models.lite_llmimportLiteLlmfromgoogle.adk.tools.agent_toolimportAgentToolfromgoogle.genaiimporttypesfrom.configimportget_sqlite_config, get_model_name, get_DeepSeek_api_keyfrom.promptsimportget_sqlite_agent_instructionsfrom.utilsimportcreate_sample_database, get_database_connectionfrom.sub_agentsimport(  list_tables,  table_info,  query_execution,  sample_data)from.monitored_llmimportcreate_monitored_llm

#创建多Agent协调的SQLiteAgentsqlite_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 提示词：

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

5）sub_agents/query_execution/agent.py

智能查询子 Agent：

"""Queryexecutionagent-specializedagentforexecutingSQLqueries."""fromgoogle.adk.agentsimportAgentfromgoogle.genaiimporttypesfrom...configimportget_model_namefrom...toolsimportexecute_query_tool,get_table_info_toolfrom...monitored_llmimportcreate_monitored_llmfrom.promptimportQUERY_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：

"""Queryexecutionagent-specializedagentforexecutingSQLqueries."""fromgoogle.adk.agentsimportAgentfromgoogle.genaiimporttypesfrom...configimportget_model_namefrom...toolsimportexecute_query_tool,get_table_info_toolfrom...monitored_llmimportcreate_monitored_llmfrom.promptimportQUERY_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 提示词：

"""romptsforthequeryexecutionagent."""QUERY_EXECUTION_AGENT_INSTR="""执行SQL查询专家。直接执行查询并返回结果给用户。**使用execute_query工具：**-将用户问题直接转换为SQL查询-例如："最贵的订单"→"SELECT*FROMordersORDERBYtotal_amountDESCLIMIT1"-例如："最新用户"→"SELECT*FROMusersORDERBYcreated_atDESCLIMIT1"**输出格式：**```sql[实际SQL]```[查询结果表格]统计：X条记录**规则：**-仅执行SELECT查询-直接回答，不解释过程"""

8）monitored_llm.py

使用 LiteLLM 和 deepseek 交互：

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

4. 实现总结

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

ADK 架构落地的建议：

• 先确定设计原则：单 Agent 多 Toolvs主 + 多 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%。这次改造对工程复杂度的增加可控，重构后的代码结构更加合理，而且对成本和吞吐的改善非常可观。

---

欢迎光临 链载Ai (https://www.lianzai.com/)

Powered by Discuz! X3.5