在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 里多轮试错。
工具描述更精简：职责切分后，工具描述能够更加的精简，可以删除大量的冗长描述。
三、关键代码实现
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/ # 部署脚本
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 safetyvalidation_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 namescolumns = [description[0] for description in cursor.description] if cursor.description else []# Fetch results with limitresults = cursor.fetchmany(config.max_results)if not results:return "Query executed successfully but returned no results."# Format resultsreturn 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]
结果格式化：
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 tabulateformatted_rows = []for row in results:if hasattr(row, '_fields'): # sqlite3.Row objectformatted_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 tabletable = tabulate(formatted_rows, headers=columns, tablefmt="grid")result = f"Query Results ({len(results)} rows):\n\n{table}"# Add warning if results were limitedif len(results) >= max_results:result += f"\n\n⚠️ Results limited to {max_results} rows. Use LIMIT clause for more control."return result
主 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, ),)
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%。这次改造对工程复杂度的增加可控，重构后的代码结构更加合理，而且对成本和吞吐的改善非常可观。
