Headroom Agno 集成：为 Agno Agent 框架实现自动上下文 Token 优化

Headroom 为 Agno（前身 Phidata）Agent 框架提供了一方集成模块 headroom.integrations.agno，通过在 LLM 调用边界自动包裹任意 Agno 模型，使用户消息、工具调用、工具结果和系统提示在每次 API 请求前被压缩，并内置可观测性钩子。本文基于仓库 wiki/agno.md 的集成指南，结合 headroom/integrations/agno/ 目录下的源码实现逐节展开，读完后可掌握在 Agno 项目中落地 Headroom 的全部集成模式、Provider 自动检测机制、配置参考与故障排查方法。

安装

先安装带 agno extra 的 Headroom，再安装 Agno 本体：

pip install "headroom-ai[agno]"
pip install agno

在 pyproject.toml 中，agno extra 的定义为：

# Agno agent framework integration
agno = [
    "agno>=1.0.0",
]

即集成声明自 Agno 1.0.0 起兼容。Agno 属于可选依赖：未安装 agno 时导入 Headroom 其他模块不会报错——model.py 顶部用 try/except ImportError 设置 AGNO_AVAILABLE 标志，只有真正实例化 HeadroomAgnoModel 时才抛出 ImportError 并提示执行 pip install agno。

快速开始

一行代码包裹模型，其余 Agno 代码无需改动：

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from headroom.integrations.agno import HeadroomAgnoModel

# Wrap your model
model = HeadroomAgnoModel(OpenAIChat(id="gpt-4o"))

# Create agent as usual
agent = Agent(model=model)

# Use exactly like before
response = agent.run("What's the capital of France?")

# Check savings
print(f"Tokens saved: {model.total_tokens_saved}")
print(model.get_savings_summary())
# {'total_requests': 1, 'total_tokens_saved': 245, 'average_savings_percent': 12.3}

HeadroomAgnoModel 直接继承 agno.models.base.Model（见 model.py），因此对 Agno 而言它是"一等公民"：可以像普通模型一样传给 Agent(model=...)，Agno 的工具循环、多轮对话、结构化输出等机制全部由基类处理。

从源码结构看，包装器在构造期完成了几件关键初始化：

• __post_init__ 中校验 wrapped_model 非空，将自身 id 设为 headroom:{被包模型 id}（便于在日志中识别包装模型），并从被包模型复制 name/provider 以维持框架自省兼容；
• 转发 thinking、reasoning_effort、supports_native_structured_outputs 等能力属性（_forward_capability_attributes()），这是 Agno 推理检测能正确工作的关键；
• headroom_mode 参数已标记废弃，官方推荐的配置方式是 HeadroomConfig(default_mode=...)。

集成模式

模式 1：基础模型包装

最通用的方式——用 HeadroomAgnoModel 包裹任意 Agno 模型：

from agno.models.openai import OpenAIChat
from agno.models.anthropic import Claude
from agno.models.google import Gemini
from headroom.integrations.agno import HeadroomAgnoModel

# Works with any Agno model
openai_model = HeadroomAgnoModel(OpenAIChat(id="gpt-4o"))
claude_model = HeadroomAgnoModel(Claude(id="claude-3-5-sonnet-20241022"))
gemini_model = HeadroomAgnoModel(Gemini(id="gemini-2.0-flash"))

# Each automatically uses the correct provider for accurate token counting

这一模式的意义在于：Headroom 会自动检测底层 provider 并选用对应分词器做精确的 token 计数（检测机制详见下文"Provider 自动检测"一节）。

模式 2：带可观测性钩子的 Agent

在不改变模型行为的前提下，用钩子跟踪请求模式：

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from headroom.integrations.agno import (
    HeadroomAgnoModel,
    HeadroomPreHook,
    HeadroomPostHook,
)

# Model wrapper for optimization
model = HeadroomAgnoModel(OpenAIChat(id="gpt-4o"))

# Hooks for observability
pre_hook = HeadroomPreHook()
post_hook = HeadroomPostHook(token_alert_threshold=10000)

agent = Agent(
    model=model,
    pre_hooks=[pre_hook],
    post_hooks=[post_hook],
)

# Run agent
response = agent.run("Analyze this large dataset...")

# Check metrics from model
print(f"Tokens saved: {model.total_tokens_saved}")

# Check observability from hooks
print(f"Post-hook summary: {post_hook.get_summary()}")
print(f"Alerts triggered: {post_hook.alerts}")

钩子提供对 Agent 行为的可观测性，token_alert_threshold 允许在 token 用量超过阈值时触发告警。

模式 3：钩子对工厂函数

用 create_headroom_hooks() 一次性创建匹配的 pre/post 钩子对：

from headroom.integrations.agno import create_headroom_hooks

pre_hook, post_hook = create_headroom_hooks(
    token_alert_threshold=5000,
    log_level="DEBUG",
)

agent = Agent(
    model=model,
    pre_hooks=[pre_hook],
    post_hooks=[post_hook],
)

该工厂函数完整签名为 create_headroom_hooks(config=None, mode=HeadroomMode.OPTIMIZE, model="gpt-4o", log_level="INFO", token_alert_threshold=None)，返回 (HeadroomPreHook, HeadroomPostHook) 元组（见 hooks.py）。

模式 4：自定义配置

需要细粒度控制时传入 HeadroomConfig 实例：

from headroom import HeadroomConfig, HeadroomMode
from headroom.integrations.agno import HeadroomAgnoModel

config = HeadroomConfig(
    default_mode=HeadroomMode.OPTIMIZE,
    # Add other configuration options as needed
)

model = HeadroomAgnoModel(
    wrapped_model=OpenAIChat(id="gpt-4o"),
    headroom_config=config,
)

从源码看，headroom_config 为 None 时包装器会自动回退到默认 HeadroomConfig()；配置最终交由 TransformPipeline 参与变换决策（model.py 的 pipeline 属性）。

模式 5：独立消息优化

不包裹模型，也可直接用 optimize_messages() 压缩消息列表：

from headroom.integrations.agno import optimize_messages

messages = [
    {"role": "system", "content": "You are a helpful assistant."},
    {"role": "user", "content": "Analyze this large JSON: " + large_json},
]

optimized_messages, metrics = optimize_messages(messages, model="gpt-4o")

print(f"Tokens saved: {metrics['tokens_saved']}")
print(f"Transforms applied: {metrics['transforms_applied']}")

该函数完整签名为 optimize_messages(messages, config=None, mode=HeadroomMode.OPTIMIZE, model="gpt-4o")（见 model.py），返回 (optimized_messages, metrics_dict) 元组；metrics_dict 包含 tokens_before、tokens_after、tokens_saved、savings_percent、transforms_applied 字段。注意其内部固定使用 OpenAIProvider 分词器，不做 provider 检测——若关心跨 provider 的精确统计，推荐用模型包装方式。

模式 6：异步操作

面向高吞吐场景的完整异步支持：

import asyncio
from headroom.integrations.agno import HeadroomAgnoModel


async def process_async():
    model = HeadroomAgnoModel(OpenAIChat(id="gpt-4o"))

    # Async response
    response = await model.aresponse(messages)

    # Async streaming
    async for chunk in model.aresponse_stream(messages):
        print(chunk, end="", flush=True)

    print(f"\nTokens saved: {model.total_tokens_saved}")


asyncio.run(process_async())

源码中 ainvoke/ainvoke_stream 通过 loop.run_in_executor 把 CPU 密集的压缩过程放到线程池执行，避免阻塞事件循环；被包模型若无原生异步流式方法，会自动降级为在 executor 中包装的同步流（model.py）。

真实场景示例

示例 1：工具密集型 Agent

工具输出是 Headroom 收益最大的场景：

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.duckduckgo import DuckDuckGoTools
from headroom.integrations.agno import HeadroomAgnoModel

# Wrap model for optimization
model = HeadroomAgnoModel(OpenAIChat(id="gpt-4o"))

# Agent with search tools
agent = Agent(
    model=model,
    tools=[DuckDuckGoTools()],
    show_tool_calls=True,
)

# Tool outputs get compressed automatically
response = agent.run("Research the latest AI developments and summarize")

# Impact: Tool outputs (often 10K+ tokens) compressed by 70-90%
print(f"Tokens saved: {model.total_tokens_saved}")
print(model.get_savings_summary())

工具结果之所以能在后续请求中被自动压缩，调用链是这样的：HeadroomAgnoModel.response() 不直接压缩，而是把控制权交给 Agno 基类 Model.response() 的工具循环；循环中每次 API 调用都会回调被覆写的 invoke()，而 invoke() 对"包含新工具结果的完整消息历史"执行 _optimize_messages() 后再委托给被包模型（model.py）。即工具结果一旦进入对话历史，就会在下一个请求边界被压缩。

示例 2：多模型路由

from agno.models.openai import OpenAIChat
from agno.models.anthropic import Claude
from headroom.integrations.agno import HeadroomAgnoModel

# Different models for different tasks
fast_model = HeadroomAgnoModel(OpenAIChat(id="gpt-4o-mini"))
powerful_model = HeadroomAgnoModel(Claude(id="claude-3-5-sonnet-20241022"))

# Use fast model for simple tasks
simple_agent = Agent(model=fast_model)

# Use powerful model for complex reasoning
complex_agent = Agent(model=powerful_model)

# Each tracks its own metrics
print(f"Fast model saved: {fast_model.total_tokens_saved}")
print(f"Powerful model saved: {powerful_model.total_tokens_saved}")

每个包装器实例持有独立的指标状态，多模型可以并存并分别统计。

示例 3：生产环境监控

from agno.agent import Agent
from headroom.integrations.agno import (
    HeadroomAgnoModel,
    create_headroom_hooks,
)

model = HeadroomAgnoModel(OpenAIChat(id="gpt-4o"))
pre_hook, post_hook = create_headroom_hooks(
    token_alert_threshold=50000,  # Alert on large requests
    log_level="WARNING",
)

agent = Agent(
    model=model,
    pre_hooks=[pre_hook],
    post_hooks=[post_hook],
)

# Run multiple requests
for query in user_queries:
    response = agent.run(query)

# Check for alerts
if post_hook.alerts:
    print(f"WARNING: {len(post_hook.alerts)} requests exceeded threshold")
    for alert in post_hook.alerts:
        print(f"  - {alert}")

# Summary stats
summary = post_hook.get_summary()
print(f"Total requests: {summary['total_requests']}")
print(f"Average tokens: {summary['average_tokens']}")

HeadroomPostHook 会在每次响应后从 run_output.metrics 提取 input_tokens/output_tokens/total_tokens，当 total_tokens 超过阈值时记录告警文本并打印 WARNING 日志（hooks.py）。它最多保留 1000 条请求记录，所有指标更新都由 threading.Lock 保护，适合多线程服务。

示例 4：新会话重置指标

model = HeadroomAgnoModel(OpenAIChat(id="gpt-4o"))

# Session 1
agent.run("First conversation...")
print(f"Session 1 savings: {model.get_savings_summary()}")

# Reset for new session
model.reset()

# Session 2 - metrics start fresh
agent.run("Second conversation...")
print(f"Session 2 savings: {model.get_savings_summary()}")

reset() 线程安全，会清空 metrics_history 与累计节省计数器，适用于按用户会话或测试运行周期重置统计。

Provider 自动检测

HeadroomAgnoModel 会从被包模型自动检测 provider：

Provider	Agno 模型	自动检测
OpenAI	OpenAIChat、OpenAILike	是
Anthropic	Claude、AwsBedrock	是
Google	Gemini、VertexAI	是
Cohere	Cohere、CohereChat	是
Groq	Groq	是（OpenAI 兼容）
Mistral	Mistral	是（OpenAI 兼容）
Together	Together	是（OpenAI 兼容）
Ollama	Ollama	是（OpenAI 兼容）

如需禁用自动检测：

model = HeadroomAgnoModel(
    wrapped_model=some_model,
    auto_detect_provider=False,  # Falls back to OpenAI tokenizer
)

检测逻辑实现在 providers.py，采用四级渐进策略：

1. 类名匹配：查 _AGNO_MODEL_PROVIDERS 静态映射表（providers.py），如 OpenAIChat → OpenAIProvider、Claude/AwsBedrock/BedrockClaude → AnthropicProvider、Gemini/VertexAI → GoogleProvider、Cohere → CohereProvider。表中还覆盖 LiteLLM、Fireworks、DeepSeek、xAI/Grok、Perplexity、OpenRouter、HuggingFace 等；
2. 模块路径检测：类名未命中时，从 __class__.__module__ 中查找 anthropic、google、cohere、openai 等关键字；
3. 模型 ID 推断：id/model 属性含 claude、gemini、gpt/o1/o3、command/cohere 时推断对应 provider；
4. 兜底：全部失败则打印 WARNING 日志（提示 token 计数可能不精确），回退到 OpenAIProvider。

get_model_name_from_agno() 按 id → model → model_name → model_id 顺序尝试提取模型名，无法提取时回退为 "gpt-4o" 并告警。模型名与 provider 的 get_context_limit(model) 上下文窗口值共同决定传给压缩管线的策略参数。

功能覆盖与已知限制

HeadroomAgnoModel 在 LLM 调用边界做优化，覆盖范围如下：

特性	是否优化	说明
用户/助手消息	是	完整消息历史被压缩
工具调用	是	工具调用参数被优化
工具结果	是	JSON 响应经 SmartCrusher 压缩 70-90%
系统提示	是	包含在消息优化内
流式响应	是	同步与异步均支持
多轮对话	是	完整历史可用于优化

该集成工作于模型层而非 Agent 层，部分 Agno 特性处于此边界之外：

Agno 特性	状态	说明
Agent Memory	部分	记忆内容进入消息时会被优化，但持久化记忆存储本身不被压缩。若在 agent memory 中存大量数据，建议存储前先做摘要。
Knowledge Bases	部分	KB 检索发生在消息到达模型之前。检索到的上下文会作为消息一部分被优化，但无法影响 KB 检索过程本身。
Agent Teams	不支持	每个 agent 的模型被独立包装，无跨 agent 优化或团队级协调。
Tool Definitions	不去重	工具 schema 随每次请求发送。后续版本可能去重重复的工具定义。
Structured Outputs	支持	response_model 正常工作，优化不影响输出解析。
Reasoning Models	支持	扩展思维（extended thinking）正常工作，但不压缩推理轨迹。

"Reasoning Models 支持"一项有一个重要实现细节：当 _optimize_messages() 检测到消息含扩展思维内容块（thinking/redacted_thinking 块，或 reasoning_content/redacted_reasoning_content 字段）时，会跳过整个压缩流程，让思维结构原样通过（Claude API 要求如此），并记录 transforms_applied 为 ["skipped:extended_thinking"]，token 计数退化为粗略估算（model.py）。

此外源码明确记录了一对互斥约束：Claude 原生扩展思维与 Agno 的 reasoning=True 框架流程不兼容——要么在模型上配置 thinking（不用 Agno reasoning），要么用 Agent 的 reasoning=True（不在模型上配 thinking），且后者建议传 reasoning_model=wrapped.underlying_model 帮助 Agno 正确检测模型类型。

最大化节省的最佳实践

1. 工具密集型 agent 收益最大——工具结果（JSON、日志、搜索结果）可压缩 70-90%；
2. 长对话自动处理——Headroom 原地压缩最新的工具输出与内容块（仅 live-zone 压缩），从不丢弃历史消息，缓存热区保持完整，无需配置上下文上限；
3. 在模型层而非 Agent 层包装——确保所有 LLM 调用都经过优化；
4. 用钩子做可观测性——跟踪 token 用量模式，发现优化机会。

路线图（未来改进）中在追踪的方向包括：memory 存储前压缩钩子、知识库层的检索上下文优化、重复工具 schema 去重、以及跨 agent 团队的共享上下文压缩。

配置参考

HeadroomAgnoModel

参数	类型	默认值	说明
wrapped_model	Any	必填	要包装的 Agno 模型
headroom_config	HeadroomConfig	None（回退默认配置）	自定义配置
headroom_mode	HeadroomMode	None	已废弃，请用 HeadroomConfig(default_mode=...)
auto_detect_provider	bool	True	为 token 计数自动检测 provider；为 False 时使用 OpenAI 分词器

属性：

• wrapped_model / underlying_model——访问底层 Agno 模型（后者专为框架类型自省设计）；
• total_tokens_saved——跨所有调用的累计节省 token 数；
• metrics_history——最近 100 条 OptimizationMetrics（线程安全副本）；
• pipeline——惰性初始化的 TransformPipeline（双重检查锁，线程安全）。

方法：

• response(messages, **kwargs)——同步响应并优化；
• response_stream(messages, **kwargs)——同步流式响应；
• aresponse(messages, **kwargs)——异步响应；
• aresponse_stream(messages, **kwargs)——异步流式；
• get_savings_summary()——返回统计 dict，实际实现包含 total_requests、total_tokens_saved、average_savings_percent、total_tokens_before、total_tokens_after 五个字段（model.py）；
• reset()——清空全部指标。

OptimizationMetrics 每条记录含：request_id、timestamp、tokens_before、tokens_after、tokens_saved、savings_percent、transforms_applied、model（model.py）。当压缩管线抛出异常时，包装器回退到原始消息（请求照常发出而不中断），并记录 transforms_applied 为 ["fallback:error"]——设计上以可用性优先于节省率。

HeadroomPreHook

参数	类型	默认值	说明
config	HeadroomConfig	None	配置（保留供未来使用）
mode	HeadroomMode	HeadroomMode.OPTIMIZE	保留供未来使用
model	str	"gpt-4o"	用于估算的模型名

预钩子只做请求跟踪（记录 request_id 与时间戳，保留最近 100 条），原样返回输入。源码注释明确说明：Agno 的 pre_hooks 只接收用户输入字符串而非完整消息历史，因此真正的优化应在模型层通过 HeadroomAgnoModel 完成。

HeadroomPostHook

参数	类型	默认值	说明
log_level	str	"INFO"	日志级别（支持 DEBUG/INFO/WARNING）
token_alert_threshold	int	None	响应 total_tokens 超过该值时触发告警

属性：

• total_requests——已跟踪的请求数（上限 1000）；
• alerts——告警消息列表。

方法：

• get_summary()——返回含 total_requests、total_tokens、average_tokens、alerts 的 dict；
• reset()——清空历史与告警。

create_headroom_hooks()

参数	类型	默认值	说明
config	HeadroomConfig	None	传给 pre-hook 的配置
mode	HeadroomMode	HeadroomMode.OPTIMIZE	传给 pre-hook 的模式
model	str	"gpt-4o"	pre-hook 的模型名
log_level	str	"INFO"	post-hook 的日志级别
token_alert_threshold	int	None	post-hook 的告警阈值

返回值：tuple[HeadroomPreHook, HeadroomPostHook]。

导入参考

# Main integration
from headroom.integrations.agno import HeadroomAgnoModel

# Hooks
from headroom.integrations.agno import HeadroomPreHook
from headroom.integrations.agno import HeadroomPostHook
from headroom.integrations.agno import create_headroom_hooks

# Utilities
from headroom.integrations.agno import optimize_messages
from headroom.integrations.agno import agno_available
from headroom.integrations.agno import get_headroom_provider
from headroom.integrations.agno import get_model_name_from_agno

# Or import everything from parent
from headroom.integrations import (
    HeadroomAgnoModel,
    HeadroomPreHook,
    HeadroomPostHook,
    create_headroom_hooks,
)

父包 headroom/integrations/init.py 在 Agno 已安装时还会以别名形式重导出 AgnoOptimizationMetrics、get_agno_provider、optimize_agno_messages 等符号。

故障排查

检查 Agno 是否可用

from headroom.integrations.agno import agno_available

if agno_available():
    from headroom.integrations.agno import HeadroomAgnoModel
else:
    print("Install agno: pip install agno")

Provider 检测问题

自动检测失败时，可用工具函数排查检测结果：

from headroom.integrations.agno import get_headroom_provider, get_model_name_from_agno

model = OpenAIChat(id="gpt-4o")
provider = get_headroom_provider(model)
model_name = get_model_name_from_agno(model)

print(f"Detected provider: {type(provider).__name__}")
print(f"Model name: {model_name}")

若日志中出现 WARNING ... defaulting to OpenAIProvider，说明四级检测策略全部未命中，需检查模型类名是否在映射表内，或显式改用 auto_detect_provider=False。

指标不更新

注意检查的是正确的对象：

# Model metrics (optimization)
print(model.total_tokens_saved)  # Actual savings

# Hook metrics (observability)
print(post_hook.get_summary())  # Request tracking

注意：钩子只跟踪请求数与输出 token 用量，不跟踪 token 节省。节省统计必须从模型包装器读取——这正是 hooks.py 中 HookMetrics（保留字段恒为 0）与 OptimizationMetrics 在设计上的分工。

总结：包装器如何挂入 Agno 运行时

通读 headroom/integrations/agno/model.py 后，HeadroomAgnoModel 的设计可以概括为"委托型装饰器"：

• 它自己不做工具循环、响应解析和流式分块——response()/response_stream()/aresponse()/aresponse_stream() 在确认消息是 Message 对象后全部委托给 Agno 基类（Agno 内部 _log_messages() 要求 Message 对象，故有专门的 _ensure_message_objects() 做 dict→Message 转换）；
• 它只覆写 invoke/ainvoke/invoke_stream/ainvoke_stream 四个抽象方法——这是"每次 API 调用必经"的唯一入口，_optimize_messages() 在此处插入：转 OpenAI 格式 → 检测 thinking 块 → TransformPipeline.apply() → 转回 Agno Message → 委托给被包模型；
• 消息转换特别保留 Claude 扩展思维结构、reasoning_content、provider_data，并把 ChoiceDeltaToolCall 等 provider SDK 对象规范化为 dict（该处理修复了 issue #1312 中 'ChoiceDeltaToolCall' object has no attribute 'get' 的报错）；
• 所有指标写入（历史、累计节省）均由 threading.Lock 保护，历史保留最近 100 条。

该集成由专用测试覆盖：tests/test_integrations/agno/test_model.py（1200 余行，覆盖包装器行为、provider 检测与 optimize_messages()）和 tests/test_integrations/agno/test_hooks.py。测试套件在 Agno 未安装时整体 skipif，不影响基础测试流程。若希望扩展该集成，贡献流程见 CONTRIBUTING.md。