akadmin/headroom
1
0
Fork 0
mirror of https://github.com/headroomlabs-ai/headroom.git synced 2026-08-27 14:17:10 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

Remove hardcoded source hint system from ContentRouter

Browse source 
ContentRouter now routes purely based on content analysis instead of
relying on hardcoded tool name mappings. This makes the router work
with any MCP tool regardless of naming convention.

Changes:
- Remove generate_source_hint() function and _strategy_from_hint() method
- Remove source_hint parameter from compress() method
- Remove _get_tool_source_hint() from IntelligentContextManager
- Update tests to remove source hint test cases
- Update docs to document content detection approach
This commit is contained in:
chopratejas 2026-01-23 00:28:22 -08:00
parent dcae408dcd
commit 0b3e4d2586
6 changed files with 35 additions and 361 deletions
Download patch file Download diff file Expand all files Collapse all files

30
docs/transforms.md
View file

@ -516,15 +516,14 @@ router = ContentRouter(config)
### Example
```python
from headroom.transforms import ContentRouter, generate_source_hint
from headroom.transforms import ContentRouter
router = ContentRouter()
# With source hint for high-confidence routing
hint = generate_source_hint(tool_name="grep", file_path="src/auth.py")
result = router.compress(content, source_hint=hint)
# Router auto-detects content type and routes to optimal compressor
result = router.compress(content)
print(result.strategy) # CompressionStrategy.SEARCH or CODE_AWARE
print(result.strategy_used) # CompressionStrategy.CODE_AWARE, SMART_CRUSHER, etc.
print(result.routing_log) # List of routing decisions
```
@ -540,22 +539,17 @@ print(result.routing_log) # List of routing decisions
| LLMLINGUA | Any (max compression) | LLMLinguaCompressor |
| PASSTHROUGH | Small content | None |
### Source Hints
### Content Detection
Use source hints for accurate routing:
The router automatically detects content types by analyzing the content itself:
```python
from headroom.transforms import generate_source_hint
- **Source code**: Detected by syntax patterns, indentation, keywords
- **JSON arrays**: Detected by JSON structure with array elements
- **Search results**: Detected by `file:line:` patterns
- **Log output**: Detected by timestamp and log level patterns
- **Plain text**: Fallback for prose content
# From tool invocation
hint = generate_source_hint(tool_name="Read", file_path="main.py")
# From file extension
hint = generate_source_hint(file_path="components/Button.tsx")
# From explicit tool
hint = generate_source_hint(tool_name="Grep") # Routes to SEARCH
```
No manual hints required - the router inspects content directly.
---

2
headroom/transforms/__init__.py
View file

@ -57,7 +57,6 @@ from .content_router import (
ContentRouter,
ContentRouterConfig,
RouterCompressionResult,
generate_source_hint,
)
__all__ = [
@ -101,7 +100,6 @@ __all__ = [
"ContentRouterConfig",
"RouterCompressionResult",
"CompressionStrategy",
"generate_source_hint",
# Other transforms
"CacheAligner",
"RollingWindow",

164
headroom/transforms/content_router.py
View file

@ -360,51 +360,6 @@ def _extract_json_block(lines: list[str], start: int) -> tuple[str | None, int]:
return None, start
def generate_source_hint(tool_name: str, tool_input: dict[str, Any]) -> str:
"""Generate a source hint from tool metadata.
This enables higher-confidence routing decisions.
Args:
tool_name: Name of the tool that produced the output.
tool_input: Input parameters to the tool.
Returns:
Source hint string (e.g., "file:auth.py", "tool:grep").
"""
# File read operations
if tool_name in ("Read", "read_file", "cat", "ReadFile"):
file_path = tool_input.get("file_path", tool_input.get("path", ""))
if file_path:
return f"file:{file_path}"
# Search operations
if tool_name in ("Grep", "grep", "ripgrep", "rg", "search", "Search"):
return "tool:grep"
# Glob operations
if tool_name in ("Glob", "glob", "find"):
return "tool:glob"
# Build/test operations
if tool_name == "Bash":
command = str(tool_input.get("command", ""))
if any(cmd in command for cmd in ["pytest", "npm test", "cargo test", "go test"]):
return "tool:pytest"
if any(cmd in command for cmd in ["npm run build", "cargo build", "make"]):
return "tool:build"
if "git diff" in command:
return "tool:git-diff"
if "git log" in command:
return "tool:git-log"
# Web fetch
if tool_name in ("WebFetch", "fetch", "curl", "WebSearch"):
return "tool:web"
return ""
class ContentRouter(Transform):
"""Intelligent router that selects optimal compression strategy.
@ -462,15 +417,12 @@ class ContentRouter(Transform):
def compress(
self,
content: str,
source_hint: str | None = None,
context: str = "",
) -> RouterCompressionResult:
"""Compress content using optimal strategy.
"""Compress content using optimal strategy based on content detection.
Args:
content: Content to compress.
source_hint: Optional hint about content source.
Examples: "file:auth.py", "tool:grep", "tool:pytest"
context: Optional context for relevance-aware compression.
Returns:
@ -484,87 +436,31 @@ class ContentRouter(Transform):
routing_log=[],
)
# Determine strategy
strategy = self._determine_strategy(content, source_hint)
# Determine strategy from content analysis
strategy = self._determine_strategy(content)
if strategy == CompressionStrategy.MIXED:
return self._compress_mixed(content, context)
else:
return self._compress_pure(content, strategy, context)
def _determine_strategy(
self,
content: str,
source_hint: str | None,
) -> CompressionStrategy:
"""Determine the compression strategy.
def _determine_strategy(self, content: str) -> CompressionStrategy:
"""Determine the compression strategy from content analysis.
Args:
content: Content to analyze.
source_hint: Optional source hint.
Returns:
Selected compression strategy.
"""
# 1. Source hint takes priority
if source_hint:
strategy = self._strategy_from_hint(source_hint)
if strategy:
return strategy
# 2. Check for mixed content
# 1. Check for mixed content
if is_mixed_content(content):
return CompressionStrategy.MIXED
# 3. Detect content type
# 2. Detect content type from content itself
detection = detect_content_type(content)
return self._strategy_from_detection(detection)
def _strategy_from_hint(self, hint: str) -> CompressionStrategy | None:
"""Get strategy from source hint.
Args:
hint: Source hint string.
Returns:
Strategy if determinable, None otherwise.
"""
hint_lower = hint.lower()
# File hints
if hint_lower.startswith("file:"):
file_path = hint_lower[5:]
if file_path.endswith((".py", ".pyw")):
return CompressionStrategy.CODE_AWARE
if file_path.endswith((".js", ".jsx", ".ts", ".tsx", ".mjs")):
return CompressionStrategy.CODE_AWARE
if file_path.endswith((".go", ".rs", ".java", ".c", ".cpp", ".h", ".hpp")):
return CompressionStrategy.CODE_AWARE
if file_path.endswith(".json"):
return CompressionStrategy.SMART_CRUSHER
if file_path.endswith((".md", ".txt", ".rst")):
return CompressionStrategy.TEXT
if file_path.endswith((".log", ".out")):
return CompressionStrategy.LOG
# Tool hints
if hint_lower.startswith("tool:"):
tool = hint_lower[5:]
if tool in ("grep", "rg", "ripgrep", "ag", "search"):
return CompressionStrategy.SEARCH
if tool in ("pytest", "jest", "cargo-test", "go-test", "npm-test"):
return CompressionStrategy.LOG
if tool in ("build", "make", "cargo-build", "npm-build"):
return CompressionStrategy.LOG
if tool in ("git-diff", "diff"):
return CompressionStrategy.DIFF
# Direct strategy hints (used by _process_content_blocks for tool_result)
if hint_lower == "json_array":
return CompressionStrategy.SMART_CRUSHER
return None
def _strategy_from_detection(self, detection: Any) -> CompressionStrategy:
"""Get strategy from content detection result.
@ -897,14 +793,13 @@ class ContentRouter(Transform):
Args:
messages: Messages to transform.
tokenizer: Tokenizer for counting.
**kwargs: Additional arguments (context, source_hints).
**kwargs: Additional arguments (context).
Returns:
TransformResult with routed and compressed messages.
"""
tokens_before = sum(tokenizer.count_text(str(m.get("content", ""))) for m in messages)
context = kwargs.get("context", "")
source_hints = kwargs.get("source_hints", {}) # message_id -> hint
transformed_messages: list[dict[str, Any]] = []
transforms_applied: list[str] = []
@ -945,9 +840,6 @@ class ContentRouter(Transform):
transformed_messages.append(message)
continue
# Get source hint if available
source_hint = source_hints.get(i) or source_hints.get(str(i))
# Detect content type for protection decisions
detection = detect_content_type(content)
is_code = detection.content_type == ContentType.SOURCE_CODE
@ -969,8 +861,8 @@ class ContentRouter(Transform):
transforms_applied.append("router:protected:analysis_context")
continue
# Route and compress
result = self.compress(content, source_hint=source_hint, context=context)
# Route and compress based on content detection
result = self.compress(content, context=context)
if result.compression_ratio < 0.9:
transformed_messages.append({**message, "content": result.compressed})
@ -1013,8 +905,6 @@ class ContentRouter(Transform):
Returns:
Transformed message with compressed content blocks.
"""
import json
new_blocks = []
any_compressed = False
@ -1031,33 +921,7 @@ class ContentRouter(Transform):
# Only process string content
if isinstance(tool_content, str) and len(tool_content) > 500:
# Try to detect if it's JSON array data (SmartCrusher target)
try:
parsed = json.loads(tool_content)
if isinstance(parsed, list) and len(parsed) > 10:
# Route to SmartCrusher for arrays
result = self.compress(
tool_content,
source_hint="json_array",
context=context,
)
if result.compression_ratio < 0.9:
new_blocks.append(
{
**block,
"content": result.compressed,
}
)
transforms_applied.append(
f"router:tool_result:{result.strategy_used.value}"
)
any_compressed = True
continue
except (json.JSONDecodeError, TypeError):
# Not JSON, try general compression
pass
# Try general compression for large non-JSON content
# Compress using content detection (will auto-detect JSON arrays, etc.)
result = self.compress(tool_content, context=context)
if result.compression_ratio < 0.9:
new_blocks.append({**block, "content": result.compressed})
@ -1140,15 +1004,13 @@ class ContentRouter(Transform):
def route_and_compress(
content: str,
source_hint: str | None = None,
context: str = "",
) -> str:
"""Convenience function for one-off routing and compression.
Args:
content: Content to compress.
source_hint: Optional source hint.
context: Optional context.
context: Optional context for relevance-aware compression.
Returns:
Compressed content.
@ -1157,5 +1019,5 @@ def route_and_compress(
>>> compressed = route_and_compress(mixed_content)
"""
router = ContentRouter()
result = router.compress(content, source_hint=source_hint, context=context)
result = router.compress(content, context=context)
return result.compressed

52
headroom/transforms/intelligent_context.py
View file

@ -444,16 +444,8 @@ class IntelligentContextManager(Transform):
# Focus on tool messages (highest compression potential)
if role == "tool" and isinstance(content, str) and len(content) > 100:
try:
# Get source hint from tool call context
tool_call_id = msg.get("tool_call_id", "")
source_hint = self._get_tool_source_hint(messages, tool_call_id)
# Compress using ContentRouter
result = router.compress(
content,
source_hint=source_hint,
context="", # No specific context in compress-first mode
)
# Compress using ContentRouter (auto-detects content type)
result = router.compress(content)
# Check if compression was effective
if result.compression_ratio < 0.9: # At least 10% savings
@ -500,46 +492,6 @@ class IntelligentContextManager(Transform):
return compressed_messages, transforms_applied, total_tokens_saved
def _get_tool_source_hint(self, messages: list[dict[str, Any]], tool_call_id: str) -> str:
"""Extract source hint from the tool call that produced this result.
Args:
messages: List of all messages.
tool_call_id: The ID of the tool call to find.
Returns:
Source hint string (e.g., "tool:grep", "file:main.py").
"""
if not tool_call_id:
return ""
# Find the assistant message with this tool call
for msg in messages:
if msg.get("role") == "assistant" and msg.get("tool_calls"):
for tc in msg.get("tool_calls", []):
if tc.get("id") == tool_call_id:
func = tc.get("function", {})
tool_name = func.get("name", "")
# Import here to avoid circular imports
try:
import json
from .content_router import generate_source_hint
args_str = func.get("arguments", "{}")
try:
args = (
json.loads(args_str) if isinstance(args_str, str) else args_str
)
except json.JSONDecodeError:
args = {}
return generate_source_hint(tool_name, args)
except ImportError:
return f"tool:{tool_name}" if tool_name else ""
return ""
def _compress_content_blocks(
self,
content_blocks: list[Any],

105
tests/test_transforms/test_content_router.py
View file

@ -4,7 +4,6 @@ Comprehensive tests covering:
- ContentRouterConfig: Configuration validation and defaults
- ContentRouter: Core routing functionality
- Strategy detection: Code, JSON, search, logs, text
- Source hint parsing: File paths, tool names
- Mixed content handling: Split, route, reassemble
- Transform interface: apply(), should_apply() methods
"""
@ -18,7 +17,6 @@ from headroom.transforms.content_router import (
ContentRouterConfig,
RouterCompressionResult,
RoutingDecision,
generate_source_hint,
)
# =============================================================================
@ -279,51 +277,6 @@ class TestRouterCompressionResult:
assert result.savings_percentage == 0.0
# =============================================================================
# TestSourceHintGeneration
# =============================================================================
class TestSourceHintGeneration:
"""Tests for generate_source_hint function."""
def test_file_path_hint_with_file_path(self):
"""File path generates file: hint."""
hint = generate_source_hint("Read", {"file_path": "/src/auth.py"})
assert hint == "file:/src/auth.py"
def test_file_path_hint_with_path(self):
"""File path with 'path' key also works."""
hint = generate_source_hint("read_file", {"path": "/src/auth.py"})
assert hint == "file:/src/auth.py"
def test_grep_tool_hint(self):
"""Grep tool generates tool:grep hint."""
hint = generate_source_hint("Grep", {"pattern": "def.*"})
assert hint == "tool:grep"
def test_glob_tool_hint(self):
"""Glob tool generates tool:glob hint."""
hint = generate_source_hint("Glob", {"pattern": "*.py"})
assert hint == "tool:glob"
def test_bash_test_command(self):
"""Bash with test command generates tool:pytest hint."""
hint = generate_source_hint("Bash", {"command": "pytest tests/"})
assert hint == "tool:pytest"
def test_bash_build_command(self):
"""Bash with build command generates tool:build hint."""
hint = generate_source_hint("Bash", {"command": "npm run build"})
assert "tool:" in hint
def test_unknown_tool(self):
"""Unknown tool returns generic tool hint."""
hint = generate_source_hint("CustomTool", {"arg": "value"})
# Should return some hint, not crash
assert hint is not None
# =============================================================================
# TestStrategyDetection
# =============================================================================
@ -335,54 +288,32 @@ class TestStrategyDetection:
def test_detect_python_code(self, router):
"""Python code is detected."""
code = generate_python_code(5)
strategy = router._determine_strategy(code, None)
strategy = router._determine_strategy(code)
# Should be either CODE_AWARE or fallback
assert strategy in CompressionStrategy
def test_detect_json_content(self, router):
"""JSON content is detected."""
json_data = generate_json_data(20)
strategy = router._determine_strategy(json_data, None)
strategy = router._determine_strategy(json_data)
assert strategy in CompressionStrategy
def test_detect_search_results(self, router):
"""Search/grep results are detected."""
search_results = generate_search_results(10)
strategy = router._determine_strategy(search_results, None)
strategy = router._determine_strategy(search_results)
assert strategy in CompressionStrategy
def test_detect_log_output(self, router):
"""Build/test logs are detected."""
logs = generate_log_output(30)
strategy = router._determine_strategy(logs, None)
strategy = router._determine_strategy(logs)
assert strategy in CompressionStrategy
def test_detect_plain_text(self, router):
"""Plain text detection."""
text = "This is just plain text without any special formatting."
strategy = router._determine_strategy(text, None)
assert strategy in CompressionStrategy
# =============================================================================
# TestSourceHintRouting
# =============================================================================
class TestSourceHintRouting:
"""Tests for source hint-based routing."""
def test_file_hint_routes_appropriately(self, router):
"""file:*.py hint influences routing."""
content = generate_python_code(3)
strategy = router._determine_strategy(content, "file:/src/auth.py")
# Should return a valid strategy
assert strategy in CompressionStrategy
def test_grep_hint_routes_appropriately(self, router):
"""tool:grep hint influences routing."""
content = generate_search_results(5)
strategy = router._determine_strategy(content, "tool:grep")
strategy = router._determine_strategy(text)
assert strategy in CompressionStrategy
@ -430,14 +361,6 @@ class TestContentRouter:
assert result.original == content
assert result.strategy_used is not None
def test_compress_with_source_hint(self, router):
"""Source hint influences routing decision."""
content = generate_python_code(5)
result = router.compress(content, source_hint="file:/src/auth.py")
# Should return a valid result
assert isinstance(result, RouterCompressionResult)
def test_name_property(self, router):
"""Router has correct name."""
assert router.name == "content_router"
@ -515,7 +438,7 @@ class TestCompressorDisabling:
code = generate_python_code(10)
# Should not crash
result = router.compress(code, source_hint="file:/src/auth.py")
result = router.compress(code)
assert result is not None
def test_config_accepts_disable_search_compression(self):
@ -528,7 +451,7 @@ class TestCompressorDisabling:
search_results = generate_search_results(10)
# Should not crash
result = router.compress(search_results, source_hint="tool:grep")
result = router.compress(search_results)
assert result is not None
def test_config_accepts_disable_log_compression(self):
@ -541,7 +464,7 @@ class TestCompressorDisabling:
logs = generate_log_output(30)
# Should not crash
result = router.compress(logs, source_hint="tool:pytest")
result = router.compress(logs)
assert result is not None
@ -570,18 +493,6 @@ class TestEdgeCases:
result = router.compress(content)
assert result is not None
def test_null_source_hint(self, router):
"""Null source hint doesn't crash."""
content = generate_python_code(5)
result = router.compress(content, source_hint=None)
assert result is not None
def test_empty_source_hint(self, router):
"""Empty source hint doesn't crash."""
content = generate_python_code(5)
result = router.compress(content, source_hint="")
assert result is not None
# =============================================================================
# TestRoutingLog

43
tests/test_transforms/test_intelligent_context.py
View file

@ -1017,49 +1017,6 @@ class TestCompressFirstStrategy:
router2 = manager._get_content_router()
assert router is router2
def test_source_hint_extraction(self):
"""Source hints should be extracted from tool calls."""
manager = IntelligentContextManager()
messages = [
{
"role": "assistant",
"tool_calls": [
{
"id": "call_1",
"function": {
"name": "Read",
"arguments": '{"file_path": "/src/main.py"}',
},
}
],
},
{
"role": "assistant",
"tool_calls": [
{
"id": "call_2",
"function": {
"name": "Grep",
"arguments": '{"pattern": "def"}',
},
}
],
},
]
# Test file read hint
hint1 = manager._get_tool_source_hint(messages, "call_1")
assert "file:" in hint1 or hint1 == "" # May not have content_router import
# Test grep hint
hint2 = manager._get_tool_source_hint(messages, "call_2")
assert "grep" in hint2.lower() or hint2 == ""
# Test unknown tool call
hint3 = manager._get_tool_source_hint(messages, "unknown")
assert hint3 == ""
class TestCompressFirstWithContentBlocks:
"""Tests for COMPRESS_FIRST with Anthropic-style content blocks."""