一文掌握ADK-Python：LLM代理Token用量统计与优化实战指南

【免费下载链接】adk-python 一款开源、代码优先的Python工具包，用于构建、评估和部署灵活可控的复杂 AI agents 项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python

你是否还在为AI代理的Token消耗失控而烦恼？作为开源、代码优先的Python工具包，ADK-Python（Google ADK）提供了灵活可控的LLM代理构建框架，其中Token用量统计功能更是帮助开发者精确掌控成本的关键。本文将从实现原理、核心代码到实际应用，全面解析ADK-Python的Token监控机制，读完你将能够：

• 理解LLM代理Token统计的底层实现逻辑
• 掌握ADK中Token用量数据的采集与分析方法
• 学会通过缓存优化减少30%以上的Token消耗
• 构建自定义Token监控仪表盘

Token统计功能架构解析

ADK-Python的Token用量统计系统采用事件驱动架构，通过监听LLM交互过程中的元数据事件实现全链路追踪。核心实现位于contributing/samples/cache_analysis/utils.py文件，主要包含数据采集、计算分析和结果展示三大模块。

核心数据结构

Token统计系统使用标准化的字典结构记录各类Token消耗：

token_usage = {
    "prompt_token_count": 0,          # 输入提示词Token数
    "candidates_token_count": 0,      # 候选回复Token数
    "cached_content_token_count": 0,  # 缓存命中节省的Token数
    "total_token_count": 0            # 总消耗Token数
}

这种结构化设计支持多维度分析，既可以追踪单次请求的Token使用，也能聚合计算批量任务的整体消耗。

工作流程

Token统计功能的工作流程可分为三个阶段：

实时性是该设计的关键优势，系统通过异步迭代器模式处理LLM响应流，确保每个Token消耗事件都能被即时捕获：

async for event in runner.run_async(...):
    if event.usage_metadata:
        # 实时累加Prompt Token计数
        if hasattr(event.usage_metadata, "prompt_token_count"):
            token_usage["prompt_token_count"] += event.usage_metadata.prompt_token_count
        # 处理其他类型Token计数...

核心实现代码解析

异步Token采集函数

contributing/samples/cache_analysis/utils.py中的call_agent_async函数实现了异步Token数据采集的核心逻辑。该函数通过迭代LLM响应流，同时完成内容构建和Token统计两项任务：

async def call_agent_async(
    runner: InMemoryRunner, user_id: str, session_id: str, prompt: str
) -> Dict[str, Any]:
    """Call agent asynchronously and return response with token usage."""
    response_parts = []
    token_usage = {
        "prompt_token_count": 0,
        "candidates_token_count": 0,
        "cached_content_token_count": 0,
        "total_token_count": 0,
    }

    async for event in runner.run_async(...):
        # 收集响应内容
        if event.content and event.content.parts:
            for part in event.content.parts:
                if hasattr(part, "text") and part.text:
                    response_parts.append(part.text)
        
        # 收集Token使用信息
        if event.usage_metadata:
            if hasattr(event.usage_metadata, "prompt_token_count"):
                token_usage["prompt_token_count"] += event.usage_metadata.prompt_token_count
            # 其他Token类型的处理逻辑...

    return {"response_text": "".join(response_parts), "token_usage": token_usage}

这种设计巧妙地将内容构建和Token统计合并在一个异步循环中，既保证了数据的完整性，又避免了额外的性能开销。

批量分析工具

为满足大规模实验需求，ADK提供了run_experiment_batch函数，支持对多轮对话进行批量Token统计分析。该函数能自动计算关键指标，如总Token消耗、缓存命中率等：

def run_experiment_batch(...):
    # 批量处理提示词列表
    for i, prompt in enumerate(prompts, 1):
        agent_response = await call_agent_async(...)
        # 记录单次请求结果
        results.append({
            "prompt_number": i,
            "prompt": prompt,
            "response_length": len(agent_response["response_text"]),
            "success": True,
            "token_usage": agent_response["token_usage"]
        })
    
    # 计算汇总统计指标
    total_prompt_tokens = sum(
        r.get("token_usage", {}).get("prompt_token_count", 0) for r in results
    )
    total_cached_tokens = sum(
        r.get("token_usage", {}).get("cached_content_token_count", 0) for r in results
    )
    cache_hit_ratio = (total_cached_tokens / total_prompt_tokens) * 100 if total_prompt_tokens > 0 else 0

缓存优化与Token节省策略

ADK-Python的Token统计系统与缓存机制深度集成，通过src/google/adk/models/gemini_context_cache_manager.py实现智能内容缓存，显著降低重复请求的Token消耗。

缓存命中率计算

缓存效果通过缓存命中率指标量化评估：

cache_hit_ratio = (total_cached_tokens / total_prompt_tokens) * 100

其中cached_content_token_count记录了通过缓存节省的Token数量。实际测试显示，在客服对话、代码审查等重复场景中，缓存可减少30-60%的Token消耗：

📊 BATCH CACHE STATISTICS:
   Cache Hit Ratio: 42.3% (1,245 / 2,945 tokens)
   Cache Utilization: 65.0% (13/20 requests)
   Avg Cached Tokens/Request: 62

智能缓存策略

系统采用多级缓存策略，根据内容类型和使用频率动态调整缓存优先级：

1. 高频查询缓存：对重复出现的标准问题（如产品说明、API文档）进行永久缓存
2. 会话上下文缓存：临时缓存对话历史，有效期设置为会话时长
3. 部分响应缓存：对函数调用结果等结构化数据进行针对性缓存

这种分层缓存架构在src/google/adk/models/gemini_context_cache_manager.py中实现，通过handle_context_caching方法动态决定缓存策略：

def handle_context_caching(self, llm_request: LlmRequest) -> Optional[CacheMetadata]:
    # 根据请求内容判断是否使用缓存
    if self._is_cache_valid(llm_request):
        cache_fingerprint = self._generate_cache_fingerprint(llm_request, cache_contents_count)
        # 应用缓存并更新统计
        self._apply_cache_to_request(llm_request, cache_name, cache_contents_count)
        return self._create_cache_metadata(...)
    return None

实战应用指南

基础使用方法

要在自定义代理中集成Token统计功能，只需三步即可实现：

1. 初始化统计字典：创建Token计数器并初始化为0
2. 注册事件处理器：监听LLM响应流中的元数据事件
3. 生成统计报告：调用分析函数生成可视化报告

以下是一个最小化实现示例：

from google.adk.runners import InMemoryRunner
from contributing.samples.cache_analysis.utils import call_agent_async

# 1. 创建运行器和代理实例
runner = InMemoryRunner.from_agent(your_agent)

# 2. 调用带Token统计的异步函数
result = await call_agent_async(
    runner=runner,
    user_id="user_123",
    session_id="session_456",
    prompt="你的问题..."
)

# 3. 提取Token统计结果
token_stats = result["token_usage"]
print(f"总消耗Token: {token_stats['total_token_count']}")
print(f"缓存节省Token: {token_stats['cached_content_token_count']}")

高级应用：构建自定义监控仪表盘

结合ADK的事件系统和Web框架，可以构建实时Token监控仪表盘。以下是实现方案：

1. 数据采集层：使用call_agent_async函数收集原始Token数据
2. 存储层：将统计结果写入时序数据库（如Prometheus）
3. 展示层：使用Grafana或Streamlit构建可视化界面

示例代码片段：

# 扩展Token统计函数，添加数据导出功能
def export_token_metrics(token_usage, user_id, session_id):
    # 写入Prometheus指标
    token_metrics.labels(
        user=user_id,
        session=session_id,
        type="prompt"
    ).set(token_usage["prompt_token_count"])
    # 其他类型指标...

性能优化最佳实践

Token节省技巧

基于ADK的Token统计数据，我们总结出以下高效节省策略：

优化策略	实施方法	预期效果
提示词压缩	使用text-compression工具预处理长文本	减少20-30%的Prompt Token
结构化输出	采用JSON Schema限制响应格式	降低15-25%的候选Token
缓存复用	对重复查询启用上下文缓存	最高节省60%的总Token
增量更新	只发送变更部分而非完整上下文	减少40-50%的历史Token

监控与告警设置

建议设置三级Token消耗告警阈值：

1. 警告阈值：单轮请求超过500 Token时发送通知
2. 限流阈值：连续3次超过1000 Token时自动启用压缩
3. 紧急阈值：单日累计超过10万Token时暂停非关键任务

这些阈值可根据实际业务需求在src/google/adk/models/llm_request.py中通过set_token_budget方法配置：

llm_request.set_token_budget({
    "per_request": 1000,
    "daily_total": 100000,
    "alert_callback": your_alert_function
})

总结与展望

ADK-Python的Token用量统计功能为LLM代理的成本控制提供了全方位解决方案，其核心价值体现在：

1. 成本可视化：将抽象的Token消耗转化为直观的量化指标
2. 性能优化：通过缓存机制显著降低重复请求的Token消耗
3. 预算控制：基于实时统计实现精细化的Token预算管理

随着LLM应用的普及，Token成本已成为企业采用AI技术的主要障碍之一。ADK-Python的Token统计与优化系统通过技术创新，帮助开发者在保持AI能力的同时，将Token消耗降低30-60%，为大规模部署LLM代理提供了经济可行的解决方案。

未来，该系统将进一步增强预测能力，通过机器学习模型提前预测Token消耗趋势，并自动调整请求策略以避免预算超支。同时，社区正在开发的插件系统将支持与各类成本管理平台集成，实现从Token统计到预算优化的全流程自动化。

要开始使用Token统计功能，建议从contributing/samples/cache_analysis/目录中的示例入手，该目录包含完整的实验脚本和可视化工具。如有定制需求，可参考src/google/adk/models/中的核心实现进行扩展开发。

【免费下载链接】adk-python 一款开源、代码优先的Python工具包，用于构建、评估和部署灵活可控的复杂 AI agents 项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python