如何用 wrap_tools_with_headroom 给 AutoGen 代理接入 Headroom 工具输出压缩-开发者社区

如何用 wrap_tools_with_headroom 给 AutoGen 代理接入 Headroom 工具输出压缩

【免费下载链接】headroomCompress tool outputs, logs, files, and RAG chunks before they reach the LLM. 20% fewer tokens for coding agents, 60-95% fewer tokens for JSON, same answers. Library, proxy, MCP server.项目地址: https://gitcode.com/GitHub_Trending/head/headroom

如果你的 AutoGen 代理（autogen-agentchat>=0.7）挂载了返回大 JSON 数组、数据库结果或冗长日志的工具，这些输出会原样进入模型上下文，直接推高 token 消耗。Headroom 的headroom.integrations.autogen模块提供了wrap_tools_with_headroom，它把每个FunctionTool的返回值在字符串化进入model_context之前压缩掉，并保留原工具的名称、描述和参数 schema，是 drop-in 替换，不需要改工具函数本身。

适用前提：AutoGen >=0.7、Python 3.10+，工具以FunctionTool形式挂载在AssistantAgent上。完整文档见 AutoGen 集成文档。

安装

pip install headroom-ai autogen-agentchat

Headroom 的[all]extra 不覆盖框架适配器，AutoGen 集成依赖autogen-agentchat，缺它时导入会抛出ImportError并提示执行上面的安装命令（见集成源码中的_check_autogen_available）。

用 wrap_tools_with_headroom 包装工具并挂到代理

最短主路径是：把现有FunctionTool列表整体传给wrap_tools_with_headroom，拿到包装后的列表再交给AssistantAgent：

from autogen_agentchat.agents import AssistantAgent from autogen_core.tools import FunctionTool from headroom.integrations.autogen import wrap_tools_with_headroom def search_database(query: str) -> str: """Search the database and return results.""" return json.dumps({"results": [...], "total": 1000}) tool = FunctionTool(search_database, description="Search the database") wrapped = wrap_tools_with_headroom([tool]) agent = AssistantAgent( name="researcher", model_client=model_client, tools=wrapped, )

其中model_client是你代理原有的 AutoGen 模型客户端，文档示例中直接沿用，创建方式以你现有 AutoGen 代码为准。search_database中的返回体{"results": [...], "total": 1000}是文档示例，替换为你的真实查询逻辑。

包装后每个工具的行为变化（来自文档 How it works 一节）：

调用原始函数；
把返回值字符串化后判断是否超过min_chars_to_compress（默认 1000 字符）；
超过则调用 Headroom 的compress_tool_result()压缩；
记录指标并返回压缩后的字符串。

低于阈值的短输出不做压缩、原样返回；压缩过程本身出错时也会回退为原始输出，不会中断工具调用。

调整压缩阈值（可选）

默认阈值 1000 字符。如果工具输出普遍较短但仍想压缩，可以调低：

wrapped = wrap_tools_with_headroom( [search_tool, log_tool], min_chars_to_compress=500, # Default: 1000 )

这里的search_tool、log_tool是文档中使用的占位名称，替换为你自己的FunctionTool实例。

验证压缩是否生效

方法一：检查工具级压缩指标

不指定metrics_collector时，所有包装工具共用一个全局收集器，会话结束后读取：

from headroom.integrations.autogen import get_tool_metrics metrics = get_tool_metrics() print(metrics.get_summary())

文档示例输出（数值为文档示例，不是固定预期）：

# { # 'total_invocations': 25, # 'total_compressions': 18, # 'total_chars_saved': 450000, # 'average_compression_ratio': 0.35, # 'by_tool': { # 'search_database': {'invocations': 15, 'compressions': 12, 'chars_saved': 320000}, # } # }

判断方式：被压缩过的大输出工具应出现在by_tool中且compressions大于 0。多个会话或多次评估之间用reset_tool_metrics()清空全局指标，避免跨会话累计：

from headroom.integrations.autogen import reset_tool_metrics reset_tool_metrics()

指标收集器只保留最近 1000 条调用记录，超出的旧记录会被丢弃。

如果不想用全局收集器，可以传入独立的ToolMetricsCollector，让同一批工具共享一个实例：

from headroom.integrations.autogen import ToolMetricsCollector, wrap_tools_with_headroom collector = ToolMetricsCollector() wrapped = wrap_tools_with_headroom( [search_tool], metrics_collector=collector, ) print(collector.get_summary())

方法二：直接执行单个包装工具核对输出

集成测试展示的验证方式是对包装后的工具直接跑一次run_json，核对返回内容与指标：

from unittest.mock import patch from autogen_core import CancellationToken from headroom.integrations.autogen import HeadroomToolWrapper, ToolMetricsCollector collector = ToolMetricsCollector() wrapper = HeadroomToolWrapper(tool, min_chars_to_compress=1000, metrics_collector=collector) wrapped_tool = wrapper.wrapped_tool result = asyncio.run(wrapped_tool.run_json({"query": "test"}, CancellationToken())) print(result) print(collector.get_summary())

测试中确认的行为可以直接作为核对依据：

短输出（如返回"ok"）：结果原样返回，get_summary()["total_compressions"]为 0；
大输出且压缩成功：run_json返回的是压缩后的字符串，total_compressions增加 1；
包装保留元数据：wrapper.name、wrapper.description与wrapped_tool.name都和原工具一致，保证 LLM 侧看到的工具定义不变。

单独包装某个工具（可选分支）

wrap_tools_with_headroom适合批量处理；如果只想给个别工具压缩，用HeadroomToolWrapper逐个包装：

from headroom.integrations.autogen import HeadroomToolWrapper wrapper = HeadroomToolWrapper( search_tool, min_chars_to_compress=500, ) # Get the wrapped FunctionTool compressed_tool = wrapper.as_function_tool() agent = AssistantAgent( name="researcher", model_client=model_client, tools=[compressed_tool], )

search_tool同样替换为你自己的FunctionTool。

Async 工具与常见误区

AutoGen 工具原生支持 async，wrapper 对同步和异步工具函数都透明处理，压缩行为一致，无需为 async 工具单独处理：

async def async_search(query: str) -> str: """Async database search.""" results = await db.search(query) return json.dumps(results) tool = FunctionTool(async_search, description="Async search") wrapped = wrap_tools_with_headroom([tool])

一个容易走偏的替代方案：AssistantAgent的tool_call_summary_formatter参数看起来也是工具输出的钩子，但它只控制工具循环结束后的最终汇总消息，不触碰真正写入model_context的原始FunctionExecutionResult，也就是 LLM 实际读到的内容。文档明确说明包装函数才是唯一的干净拦截点，不要用tool_call_summary_formatter替代wrap_tools_with_headroom。