Add seamless LangChain integration

- Add HeadroomChatModel wrapper with auto provider detection (OpenAI, Anthropic, Google)
- Add HeadroomChatMessageHistory for automatic conversation compression
- Add HeadroomDocumentCompressor for retriever integration
- Add wrap_tools_with_headroom() for agent tool output compression
- Add async support (ainvoke, astream)
- Add LangSmith integration for observability
- Restructure integrations package into nested langchain/ and mcp/ subpackages
- Fix Pydantic v2 deprecation warning
- Add comprehensive docs/langchain.md guide with real-world examples
- Update README with LangChain quickstart and framework integrations

Bump version to 0.2.3
This commit is contained in:
chopratejas 2026-01-14 15:22:26 -08:00
parent 905c229251
commit bb041047c8
26 changed files with 3806 additions and 66 deletions

View file

@ -27,45 +27,89 @@
## What It Does
Headroom is a **smart compression proxy** for LLM applications:
Headroom is a **smart compression layer** for LLM applications:
- **Compresses tool outputs** — 1000 search results → 15 items (keeps errors, anomalies, relevant items)
- **Enables provider caching** — Stabilizes prefixes so cache hits actually happen
- **Manages context windows** — Prevents token limit failures without breaking tool calls
- **Reversible compression** — LLM can retrieve original data if needed ([CCR architecture](docs/ccr.md))
**Zero code changes required** — point your existing tools at the proxy.
Works as a **proxy** (zero code changes) or **SDK** (fine-grained control).
---
## 30-Second Quickstart
### Option 1: Proxy (Zero Code Changes)
```bash
# Install
pip install "headroom-ai[proxy]"
# Start proxy
headroom proxy --port 8787
# Verify
curl http://localhost:8787/health
```
**Use with your tools:**
Point your tools at the proxy:
```bash
# Claude Code
ANTHROPIC_BASE_URL=http://localhost:8787 claude
# Cursor / Continue / any OpenAI client
# Any OpenAI-compatible client
OPENAI_BASE_URL=http://localhost:8787/v1 cursor
# Python scripts
export OPENAI_BASE_URL=http://localhost:8787/v1
python your_script.py
```
That's it. You're saving tokens.
### Option 2: LangChain Integration
```bash
pip install "headroom-ai[langchain]"
```
```python
from langchain_openai import ChatOpenAI
from headroom.integrations import HeadroomChatModel
# Wrap your model - that's it!
llm = HeadroomChatModel(ChatOpenAI(model="gpt-4o"))
# Use exactly like before
response = llm.invoke("Hello!")
```
See the full [LangChain Integration Guide](docs/langchain.md) for memory, retrievers, agents, and more.
---
## Framework Integrations
| Framework | Integration | Docs |
|-----------|-------------|------|
| **LangChain** | `HeadroomChatModel`, memory, retrievers, agents | [Guide](docs/langchain.md) |
| **MCP** | Tool output compression for Claude | [Guide](docs/ccr.md) |
| **Any OpenAI Client** | Proxy server | [Guide](docs/proxy.md) |
### LangChain Highlights
```python
from headroom.integrations import (
HeadroomChatModel, # Wrap any chat model
HeadroomChatMessageHistory, # Auto-compress conversation history
HeadroomDocumentCompressor, # Filter retrieved documents
wrap_tools_with_headroom, # Compress agent tool outputs
)
# Memory that auto-compresses when over 4K tokens
memory = ConversationBufferMemory(
chat_memory=HeadroomChatMessageHistory(base_history)
)
# Retriever that keeps only relevant docs
retriever = ContextualCompressionRetriever(
base_compressor=HeadroomDocumentCompressor(max_documents=10),
base_retriever=vectorstore.as_retriever(search_kwargs={"k": 50}),
)
# Agent tools with automatic output compression
tools = wrap_tools_with_headroom([search_tool, database_tool])
```
---
@ -82,13 +126,21 @@ curl http://localhost:8787/stats
}
```
Or in Python:
```python
print(llm.get_metrics())
# {'tokens_saved': 12500, 'savings_percent': 45.2}
```
---
## Installation
```bash
pip install "headroom-ai[proxy]" # Proxy server (recommended)
pip install headroom-ai # SDK only
pip install "headroom-ai[proxy]" # Proxy server
pip install "headroom-ai[langchain]" # LangChain integration
pip install "headroom-ai[code]" # AST-based code compression
pip install "headroom-ai[llmlingua]" # ML-based compression
pip install "headroom-ai[all]" # Everything
@ -106,10 +158,10 @@ pip install "headroom-ai[all]" # Everything
| **CacheAligner** | Stabilizes prefixes for provider caching | [Transforms](docs/transforms.md) |
| **RollingWindow** | Manages context limits without breaking tools | [Transforms](docs/transforms.md) |
| **CCR** | Reversible compression with automatic retrieval | [CCR Guide](docs/ccr.md) |
| **LangChain** | Memory, retrievers, agents, streaming | [LangChain](docs/langchain.md) |
| **Text Utilities** | Opt-in compression for search/logs | [Text Compression](docs/text-compression.md) |
| **LLMLingua-2** | ML-based 20x compression (opt-in) | [LLMLingua](docs/llmlingua.md) |
| **Code-Aware** | AST-based code compression (tree-sitter) | [Transforms](docs/transforms.md) |
| **ContentRouter** | Auto-routes content to optimal compressor | [Transforms](docs/transforms.md) |
---
@ -123,7 +175,7 @@ pip install "headroom-ai[all]" # Everything
| Cohere | Official API | - |
| Mistral | Official tokenizer | - |
**New models auto-supported** — Unknown models get sensible defaults based on naming patterns (e.g., `claude-opus-*` gets Opus pricing). Custom limits via `~/.headroom/models.json` or `HEADROOM_MODEL_LIMITS` env var.
**New models auto-supported** — Unknown models get sensible defaults based on naming patterns.
---
@ -134,6 +186,7 @@ pip install "headroom-ai[all]" # Everything
| Search results (1000 items) | 45,000 tokens | 4,500 tokens | 90% |
| Log analysis (500 entries) | 22,000 tokens | 3,300 tokens | 85% |
| Long conversation (50 turns) | 80,000 tokens | 32,000 tokens | 60% |
| Agent with tools (10 calls) | 100,000 tokens | 15,000 tokens | 85% |
Overhead: ~1-5ms per request.
@ -152,13 +205,13 @@ Overhead: ~1-5ms per request.
| Guide | Description |
|-------|-------------|
| [LangChain Integration](docs/langchain.md) | Full LangChain support |
| [SDK Guide](docs/sdk.md) | Wrap your client for fine-grained control |
| [Proxy Guide](docs/proxy.md) | Production deployment |
| [Configuration](docs/configuration.md) | All configuration options |
| [CCR Guide](docs/ccr.md) | Reversible compression architecture |
| [Metrics](docs/metrics.md) | Monitoring and observability |
| [Troubleshooting](docs/troubleshooting.md) | Common issues |
| [Architecture](docs/ARCHITECTURE.md) | How it works internally |
---
@ -168,6 +221,8 @@ See [`examples/`](examples/) for runnable code:
- `basic_usage.py` — Simple SDK usage
- `proxy_integration.py` — Using with different clients
- `langchain_agent.py` — LangChain ReAct agent with Headroom
- `rag_pipeline.py` — RAG with document compression
- `ccr_demo.py` — CCR architecture demonstration
---

View file

@ -10,6 +10,13 @@ Welcome to the Headroom documentation.
| [SDK Guide](sdk.md) | Python SDK usage |
| [Proxy Guide](proxy.md) | Proxy server deployment |
## Framework Integrations
| Framework | Description |
|-----------|-------------|
| [LangChain](langchain.md) | Chat models, memory, retrievers, agents, streaming |
| MCP | See [CCR Guide](ccr.md) for tool compression |
## Core Concepts
| Topic | Description |

622
docs/langchain.md Normal file
View file

@ -0,0 +1,622 @@
# LangChain Integration
Headroom provides seamless integration with LangChain, enabling automatic context optimization across all LangChain patterns: chat models, memory, retrievers, agents, and observability.
## Installation
```bash
pip install "headroom-ai[langchain]"
```
This installs Headroom with LangChain dependencies (`langchain-core`).
## Quick Start
### Wrap Any Chat Model (1 Line)
```python
from langchain_openai import ChatOpenAI
from headroom.integrations import HeadroomChatModel
# Wrap your model - that's it!
llm = HeadroomChatModel(ChatOpenAI(model="gpt-4o"))
# Use exactly like before
response = llm.invoke("Hello!")
```
Headroom automatically:
- Detects the provider (OpenAI, Anthropic, Google)
- Compresses tool outputs in conversation history
- Optimizes for provider caching
- Tracks token savings
### Check Your Savings
```python
# After some usage
print(llm.get_metrics())
# {'tokens_saved': 12500, 'savings_percent': 45.2, 'requests': 50}
```
---
## Integration Patterns
### 1. Chat Model Wrapper
The `HeadroomChatModel` wraps any LangChain `BaseChatModel`:
```python
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from headroom.integrations import HeadroomChatModel
# OpenAI
llm = HeadroomChatModel(ChatOpenAI(model="gpt-4o"))
# Anthropic (auto-detected)
llm = HeadroomChatModel(ChatAnthropic(model="claude-3-5-sonnet-20241022"))
# Custom configuration
from headroom import HeadroomConfig, HeadroomMode
config = HeadroomConfig(
default_mode=HeadroomMode.OPTIMIZE,
smart_crusher_target_ratio=0.3, # Target 70% compression
)
llm = HeadroomChatModel(
ChatOpenAI(model="gpt-4o"),
headroom_config=config,
)
```
#### Async Support
Full async support for `ainvoke` and `astream`:
```python
# Async invoke
response = await llm.ainvoke("Hello!")
# Async streaming
async for chunk in llm.astream("Tell me a story"):
print(chunk.content, end="", flush=True)
```
#### Tool Calling
Works seamlessly with LangChain tool calling:
```python
from langchain_core.tools import tool
@tool
def search(query: str) -> str:
"""Search the web."""
return {"results": [...]} # Large JSON response
llm_with_tools = llm.bind_tools([search])
response = llm_with_tools.invoke("Search for Python tutorials")
# Tool outputs are automatically compressed in subsequent turns
```
---
### 2. Memory Integration
`HeadroomChatMessageHistory` wraps any chat history with automatic compression:
```python
from langchain.memory import ConversationBufferMemory
from langchain_community.chat_message_histories import ChatMessageHistory
from headroom.integrations import HeadroomChatMessageHistory
# Wrap any history
base_history = ChatMessageHistory()
compressed_history = HeadroomChatMessageHistory(
base_history,
compress_threshold_tokens=4000, # Compress when over 4K tokens
keep_recent_turns=5, # Always keep last 5 turns
)
# Use with any memory class
memory = ConversationBufferMemory(chat_memory=compressed_history)
# Zero changes to your chain!
chain = ConversationChain(llm=llm, memory=memory)
```
**Why this matters**: Long conversations can blow up to 50K+ tokens. HeadroomChatMessageHistory automatically compresses older turns while preserving recent context.
```python
# Check compression stats
print(compressed_history.get_compression_stats())
# {'compression_count': 12, 'total_tokens_saved': 28000}
```
---
### 3. Retriever Integration
`HeadroomDocumentCompressor` filters retrieved documents by relevance:
```python
from langchain.retrievers import ContextualCompressionRetriever
from langchain_community.vectorstores import FAISS
from headroom.integrations import HeadroomDocumentCompressor
# Create vector store retriever (retrieve many for recall)
vectorstore = FAISS.from_documents(documents, embeddings)
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 50})
# Wrap with Headroom compression (keep best for precision)
compressor = HeadroomDocumentCompressor(
max_documents=10, # Keep top 10
min_relevance=0.3, # Minimum relevance score
prefer_diverse=True, # MMR-style diversity
)
retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=base_retriever,
)
# Retrieves 50 docs, returns best 10
docs = retriever.invoke("What is Python?")
```
**Why this matters**: Vector search often returns many marginally-relevant documents. HeadroomDocumentCompressor uses BM25-style scoring to keep only the most relevant ones, reducing context size while improving answer quality.
---
### 4. Agent Tool Wrapping
`wrap_tools_with_headroom` compresses tool outputs for agents:
```python
from langchain.agents import create_openai_tools_agent, AgentExecutor
from langchain_core.tools import tool
from headroom.integrations import wrap_tools_with_headroom
@tool
def search_database(query: str) -> str:
"""Search the database."""
# Returns 1000 results as JSON
return json.dumps({"results": [...], "total": 1000})
@tool
def fetch_logs(service: str) -> str:
"""Fetch service logs."""
# Returns 500 log entries
return json.dumps({"logs": [...]})
# Wrap tools with compression
tools = [search_database, fetch_logs]
wrapped_tools = wrap_tools_with_headroom(
tools,
min_chars_to_compress=1000, # Only compress large outputs
)
# Create agent with wrapped tools
agent = create_openai_tools_agent(llm, wrapped_tools, prompt)
executor = AgentExecutor(agent=agent, tools=wrapped_tools)
# Tool outputs are automatically compressed
result = executor.invoke({"input": "Find users who logged in yesterday"})
```
**Per-tool metrics:**
```python
from headroom.integrations import get_tool_metrics
metrics = get_tool_metrics()
print(metrics.get_summary())
# {
# 'total_invocations': 25,
# 'total_compressions': 18,
# 'total_chars_saved': 450000,
# 'by_tool': {
# 'search_database': {'invocations': 15, 'chars_saved': 320000},
# 'fetch_logs': {'invocations': 10, 'chars_saved': 130000},
# }
# }
```
---
### 5. Streaming Metrics
Track output tokens during streaming:
```python
from headroom.integrations import StreamingMetricsTracker
tracker = StreamingMetricsTracker(model="gpt-4o")
for chunk in llm.stream("Write a poem about coding"):
tracker.add_chunk(chunk)
print(chunk.content, end="", flush=True)
metrics = tracker.finish()
print(f"\nOutput tokens: {metrics.output_tokens}")
print(f"Duration: {metrics.duration_ms:.0f}ms")
```
**Context manager style:**
```python
from headroom.integrations import StreamingMetricsCallback
with StreamingMetricsCallback(model="gpt-4o") as tracker:
for chunk in llm.stream(messages):
tracker.add_chunk(chunk)
print(chunk.content, end="")
print(f"Metrics: {tracker.metrics}")
```
---
### 6. LangSmith Integration
Add Headroom metrics to LangSmith traces:
```python
from headroom.integrations import HeadroomLangSmithCallbackHandler
# Create callback handler
langsmith_handler = HeadroomLangSmithCallbackHandler()
# Use with your LLM
llm = HeadroomChatModel(
ChatOpenAI(model="gpt-4o"),
callbacks=[langsmith_handler],
)
# After calls, metrics appear in LangSmith traces:
# - headroom.tokens_before
# - headroom.tokens_after
# - headroom.tokens_saved
# - headroom.compression_ratio
```
---
## Real-World Examples
### Example 1: LangGraph ReAct Agent
The ReAct pattern is the most common agent architecture. Here's how to optimize it:
```python
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from headroom.integrations import HeadroomChatModel, wrap_tools_with_headroom
# Define tools that return large outputs
@tool
def search_web(query: str) -> str:
"""Search the web for information."""
# Simulating large search results
return json.dumps({
"results": [
{"title": f"Result {i}", "snippet": "..." * 100, "url": f"https://..."}
for i in range(100)
],
"total": 1000,
})
@tool
def query_database(sql: str) -> str:
"""Execute SQL query."""
return json.dumps({
"rows": [{"id": i, "data": "..." * 50} for i in range(500)],
"total": 500,
})
# Wrap model with Headroom
llm = HeadroomChatModel(ChatOpenAI(model="gpt-4o"))
# Wrap tools with compression
tools = wrap_tools_with_headroom([search_web, query_database])
# Create ReAct agent
agent = create_react_agent(llm, tools)
# Run - tool outputs are automatically compressed between iterations
result = agent.invoke({
"messages": [("user", "Find all users who signed up last week and their activity")]
})
# Check savings
print(f"Tokens saved: {llm.get_metrics()['tokens_saved']}")
```
**Without Headroom**: Each tool call adds 10-50K tokens to context.
**With Headroom**: Tool outputs compressed to 1-2K tokens, agent runs faster and cheaper.
---
### Example 2: RAG Pipeline with Document Filtering
```python
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.retrievers import ContextualCompressionRetriever
from headroom.integrations import HeadroomChatModel, HeadroomDocumentCompressor
# Setup vector store
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(documents, embeddings)
# High-recall retriever (get many candidates)
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 50})
# Headroom compressor for precision
compressor = HeadroomDocumentCompressor(
max_documents=5, # Keep only top 5
min_relevance=0.4, # Must be 40%+ relevant
prefer_diverse=True, # Avoid redundant docs
)
# Combine into compression retriever
retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=base_retriever,
)
# Wrap LLM
llm = HeadroomChatModel(ChatOpenAI(model="gpt-4o"))
# Create QA chain
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
retriever=retriever,
return_source_documents=True,
)
# Query - retrieves 50 docs, uses best 5
result = qa_chain.invoke({"query": "How do I configure authentication?"})
print(f"Answer: {result['result']}")
print(f"Sources: {len(result['source_documents'])} docs")
```
**Impact**:
- Without filtering: 50 docs × ~500 tokens = 25K context tokens
- With Headroom: 5 docs × ~500 tokens = 2.5K context tokens (90% reduction)
---
### Example 3: Conversational Agent with Memory
```python
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain.chains import ConversationChain
from headroom.integrations import HeadroomChatModel, HeadroomChatMessageHistory
# Wrap LLM
llm = HeadroomChatModel(ChatOpenAI(model="gpt-4o"))
# Wrap memory with auto-compression
base_history = ChatMessageHistory()
compressed_history = HeadroomChatMessageHistory(
base_history,
compress_threshold_tokens=8000, # Compress when over 8K
keep_recent_turns=10, # Always keep last 10 turns
)
memory = ConversationBufferMemory(
chat_memory=compressed_history,
return_messages=True,
)
# Create conversation chain
chain = ConversationChain(llm=llm, memory=memory)
# Long conversation - memory auto-compresses
for i in range(100):
response = chain.invoke({"input": f"Tell me about topic {i}"})
print(f"Turn {i}: {len(response['response'])} chars")
# Check memory stats
print(compressed_history.get_compression_stats())
# {'compression_count': 8, 'total_tokens_saved': 45000}
```
**Impact**: Without compression, 100-turn conversation = 100K+ tokens. With HeadroomChatMessageHistory, it stays under 8K tokens while preserving recent context.
---
### Example 4: Multi-Tool Research Agent
```python
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from headroom.integrations import (
HeadroomChatModel,
wrap_tools_with_headroom,
get_tool_metrics,
reset_tool_metrics,
)
@tool
def search_arxiv(query: str) -> str:
"""Search arXiv for papers."""
return json.dumps({"papers": [{"title": f"Paper {i}", "abstract": "..." * 200} for i in range(50)]})
@tool
def search_github(query: str) -> str:
"""Search GitHub repositories."""
return json.dumps({"repos": [{"name": f"repo-{i}", "description": "..." * 100, "stars": i * 100} for i in range(100)]})
@tool
def fetch_documentation(url: str) -> str:
"""Fetch documentation from URL."""
return "..." * 5000 # Large doc content
# Wrap everything
llm = HeadroomChatModel(ChatOpenAI(model="gpt-4o"))
tools = wrap_tools_with_headroom([search_arxiv, search_github, fetch_documentation])
prompt = ChatPromptTemplate.from_messages([
("system", "You are a research assistant. Use tools to gather information."),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
])
agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
# Reset metrics for this session
reset_tool_metrics()
# Run complex research task
result = executor.invoke({
"input": "Research the latest advances in LLM context compression and find relevant GitHub projects"
})
# Check per-tool metrics
metrics = get_tool_metrics().get_summary()
print(f"Total chars saved: {metrics['total_chars_saved']:,}")
print(f"Per-tool breakdown: {metrics['by_tool']}")
```
---
## Configuration Options
### HeadroomChatModel
```python
HeadroomChatModel(
wrapped_model, # Any LangChain BaseChatModel
headroom_config=HeadroomConfig(), # Headroom configuration
auto_detect_provider=True, # Auto-detect from wrapped model
)
```
### HeadroomChatMessageHistory
```python
HeadroomChatMessageHistory(
base_history, # Any BaseChatMessageHistory
compress_threshold_tokens=4000, # Token threshold for compression
keep_recent_turns=5, # Minimum turns to preserve
model="gpt-4o", # Model for token counting
)
```
### HeadroomDocumentCompressor
```python
HeadroomDocumentCompressor(
max_documents=10, # Maximum docs to return
min_relevance=0.0, # Minimum relevance score (0-1)
prefer_diverse=False, # Use MMR for diversity
)
```
### wrap_tools_with_headroom
```python
wrap_tools_with_headroom(
tools, # List of LangChain tools
min_chars_to_compress=1000, # Minimum output size
smart_crusher_config=None, # SmartCrusher configuration
)
```
---
## Import Reference
```python
from headroom.integrations import (
# Chat Model
HeadroomChatModel,
# Memory
HeadroomChatMessageHistory,
# Retrievers
HeadroomDocumentCompressor,
# Agents
HeadroomToolWrapper,
wrap_tools_with_headroom,
get_tool_metrics,
reset_tool_metrics,
# Streaming
StreamingMetricsTracker,
StreamingMetricsCallback,
track_streaming_response,
# LangSmith
HeadroomLangSmithCallbackHandler,
# Provider Detection
detect_provider,
get_headroom_provider,
)
# Or import from subpackage directly
from headroom.integrations.langchain import HeadroomChatModel
from headroom.integrations.langchain.memory import HeadroomChatMessageHistory
```
---
## Troubleshooting
### LangChain not detected
```python
from headroom.integrations import langchain_available
if not langchain_available():
print("Install with: pip install headroom-ai[langchain]")
```
### Provider detection failing
```python
# Force a specific provider
from headroom.providers import AnthropicProvider
llm = HeadroomChatModel(
ChatAnthropic(model="claude-3-5-sonnet-20241022"),
auto_detect_provider=False,
)
llm._provider = AnthropicProvider()
```
### Memory not compressing
Check that your message count exceeds the threshold:
```python
history = HeadroomChatMessageHistory(
base_history,
compress_threshold_tokens=1000, # Lower threshold
keep_recent_turns=2, # Fewer preserved turns
)
```
---
## Performance Tips
1. **Use tool wrapping for agents** - Agents with tools benefit most from compression
2. **Set appropriate thresholds** - Don't compress small conversations
3. **Enable diversity for RAG** - `prefer_diverse=True` improves answer quality
4. **Monitor with LangSmith** - Use the callback handler to track savings over time
5. **Batch similar requests** - Provider caching works better with stable prefixes

View file

@ -292,7 +292,8 @@ class CompressionStore:
tool_signature_hash=entry.tool_signature_hash,
)
# CRITICAL: Make a deep copy to return (entry could be modified/evicted after lock release)
# CRITICAL: Make a deep copy to return
# (entry could be modified/evicted after lock release)
# The entry contains mutable fields (search_queries list) that must be copied
result_entry = replace(entry, search_queries=list(entry.search_queries))

View file

@ -588,13 +588,19 @@ class NERDetector:
self._load_error: str | None = None
if not _SPACY_AVAILABLE:
self._load_error = "spaCy not installed. Install with: pip install spacy && python -m spacy download en_core_web_sm"
self._load_error = (
"spaCy not installed. Install with: "
"pip install spacy && python -m spacy download en_core_web_sm"
)
return
try:
self._nlp = spacy.load(config.spacy_model)
except OSError:
self._load_error = f"spaCy model '{config.spacy_model}' not found. Install with: python -m spacy download {config.spacy_model}"
self._load_error = (
f"spaCy model '{config.spacy_model}' not found. "
f"Install with: python -m spacy download {config.spacy_model}"
)
@property
def is_available(self) -> bool:
@ -704,7 +710,10 @@ class SemanticDetector:
self._load_error: str | None = None
if not _SENTENCE_TRANSFORMERS_AVAILABLE:
self._load_error = "sentence-transformers not installed. Install with: pip install sentence-transformers"
self._load_error = (
"sentence-transformers not installed. "
"Install with: pip install sentence-transformers"
)
return
try:

View file

@ -109,9 +109,10 @@ class CCRMCPServer:
Tool(
name=CCR_TOOL_NAME,
description=(
"Retrieve original uncompressed content that was compressed to save tokens. "
"Use this when you need more data than what's shown in compressed tool results. "
"The hash is provided in compression markers like [N items compressed... hash=abc123]."
"Retrieve original uncompressed content that was compressed "
"to save tokens. Use this when you need more data than what's "
"shown in compressed tool results. The hash is provided in "
"compression markers like [N items compressed... hash=abc123]."
),
inputSchema={
"type": "object",

View file

@ -1,18 +1,69 @@
"""Headroom integrations with popular LLM frameworks.
Available integrations:
- LangChain: HeadroomChatModel, HeadroomCallbackHandler, optimize_messages
- MCP: HeadroomMCPCompressor, compress_tool_result, HeadroomMCPClientWrapper
Install LangChain support: pip install headroom[langchain]
LangChain (pip install headroom[langchain]):
- HeadroomChatModel: Drop-in wrapper for any LangChain chat model
- HeadroomChatMessageHistory: Automatic conversation compression
- HeadroomDocumentCompressor: Relevance-based document filtering
- HeadroomToolWrapper: Tool output compression for agents
- StreamingMetricsTracker: Token counting during streaming
- HeadroomLangSmithCallbackHandler: LangSmith trace enrichment
MCP (Model Context Protocol):
- HeadroomMCPCompressor: Compress MCP tool results
- compress_tool_result: Simple function for tool compression
Example:
# LangChain integration
from headroom.integrations import HeadroomChatModel
# or explicitly:
from headroom.integrations.langchain import HeadroomChatModel
# MCP integration
from headroom.integrations import compress_tool_result
# or explicitly:
from headroom.integrations.mcp import compress_tool_result
"""
# Re-export from langchain subpackage for backwards compatibility
from .langchain import (
# Retrievers
CompressionMetrics,
# Core
HeadroomCallbackHandler,
# Memory
HeadroomChatMessageHistory,
HeadroomChatModel,
HeadroomDocumentCompressor,
# LangSmith
HeadroomLangSmithCallbackHandler,
HeadroomRunnable,
# Agents
HeadroomToolWrapper,
OptimizationMetrics,
# Streaming
StreamingMetrics,
StreamingMetricsCallback,
StreamingMetricsTracker,
ToolCompressionMetrics,
ToolMetricsCollector,
# Provider Detection
detect_provider,
get_headroom_provider,
get_model_name_from_langchain,
get_tool_metrics,
is_langsmith_available,
is_langsmith_tracing_enabled,
langchain_available,
optimize_messages,
reset_tool_metrics,
track_async_streaming_response,
track_streaming_response,
wrap_tools_with_headroom,
)
# Re-export from mcp subpackage for backwards compatibility
from .mcp import (
DEFAULT_MCP_PROFILES,
HeadroomMCPClientWrapper,
@ -25,11 +76,39 @@ from .mcp import (
)
__all__ = [
# LangChain
# LangChain Core
"HeadroomChatModel",
"HeadroomCallbackHandler",
"optimize_messages",
"HeadroomRunnable",
"OptimizationMetrics",
"optimize_messages",
"langchain_available",
# Provider Detection
"detect_provider",
"get_headroom_provider",
"get_model_name_from_langchain",
# Memory
"HeadroomChatMessageHistory",
# Retrievers
"HeadroomDocumentCompressor",
"CompressionMetrics",
# Agents
"HeadroomToolWrapper",
"ToolCompressionMetrics",
"ToolMetricsCollector",
"wrap_tools_with_headroom",
"get_tool_metrics",
"reset_tool_metrics",
# LangSmith
"HeadroomLangSmithCallbackHandler",
"is_langsmith_available",
"is_langsmith_tracing_enabled",
# Streaming
"StreamingMetricsTracker",
"StreamingMetricsCallback",
"StreamingMetrics",
"track_streaming_response",
"track_async_streaming_response",
# MCP
"HeadroomMCPCompressor",
"HeadroomMCPClientWrapper",

View file

@ -0,0 +1,106 @@
"""LangChain integration for Headroom.
This package provides seamless integration with LangChain, including:
- HeadroomChatModel: Drop-in wrapper for any LangChain chat model
- HeadroomChatMessageHistory: Automatic conversation compression
- HeadroomDocumentCompressor: Relevance-based document filtering
- HeadroomToolWrapper: Tool output compression for agents
- StreamingMetricsTracker: Token counting during streaming
- HeadroomLangSmithCallbackHandler: LangSmith trace enrichment
Example:
from langchain_openai import ChatOpenAI
from headroom.integrations.langchain import HeadroomChatModel
# Wrap any LangChain model
llm = HeadroomChatModel(ChatOpenAI(model="gpt-4o"))
# Use like normal - optimization happens automatically
response = llm.invoke("Hello!")
Install: pip install headroom[langchain]
"""
# Core chat model wrapper
# Agent tool wrapping
from .agents import (
HeadroomToolWrapper,
ToolCompressionMetrics,
ToolMetricsCollector,
get_tool_metrics,
reset_tool_metrics,
wrap_tools_with_headroom,
)
from .chat_model import (
HeadroomCallbackHandler,
HeadroomChatModel,
HeadroomRunnable,
OptimizationMetrics,
langchain_available,
optimize_messages,
)
# LangSmith integration
from .langsmith import (
HeadroomLangSmithCallbackHandler,
is_langsmith_available,
is_langsmith_tracing_enabled,
)
# Memory integration
from .memory import HeadroomChatMessageHistory
# Provider auto-detection
from .providers import (
detect_provider,
get_headroom_provider,
get_model_name_from_langchain,
)
# Retriever integration
from .retriever import CompressionMetrics, HeadroomDocumentCompressor
# Streaming metrics
from .streaming import (
StreamingMetrics,
StreamingMetricsCallback,
StreamingMetricsTracker,
track_async_streaming_response,
track_streaming_response,
)
__all__ = [
# Core
"HeadroomChatModel",
"HeadroomCallbackHandler",
"HeadroomRunnable",
"OptimizationMetrics",
"optimize_messages",
"langchain_available",
# Provider Detection
"detect_provider",
"get_headroom_provider",
"get_model_name_from_langchain",
# Memory
"HeadroomChatMessageHistory",
# Retrievers
"HeadroomDocumentCompressor",
"CompressionMetrics",
# Agents
"HeadroomToolWrapper",
"ToolCompressionMetrics",
"ToolMetricsCollector",
"wrap_tools_with_headroom",
"get_tool_metrics",
"reset_tool_metrics",
# LangSmith
"HeadroomLangSmithCallbackHandler",
"is_langsmith_available",
"is_langsmith_tracing_enabled",
# Streaming
"StreamingMetricsTracker",
"StreamingMetricsCallback",
"StreamingMetrics",
"track_streaming_response",
"track_async_streaming_response",
]

View file

@ -0,0 +1,326 @@
"""Agent tool integration for LangChain with output compression.
This module provides HeadroomToolWrapper and wrap_tools_with_headroom
for wrapping LangChain tools to automatically compress their outputs
and track per-tool compression metrics.
Example:
from langchain.agents import create_openai_tools_agent
from langchain.tools import Tool
from headroom.integrations import wrap_tools_with_headroom
# Define tools
tools = [
Tool(name="search", func=search_func, description="Search"),
Tool(name="database", func=db_func, description="Query DB"),
]
# Wrap with Headroom compression
wrapped_tools = wrap_tools_with_headroom(tools)
# Use in agent - outputs are automatically compressed
agent = create_openai_tools_agent(llm, wrapped_tools, prompt)
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any
# LangChain imports - these are optional dependencies
try:
from langchain_core.tools import BaseTool, StructuredTool, Tool
LANGCHAIN_AVAILABLE = True
except ImportError:
LANGCHAIN_AVAILABLE = False
BaseTool = object # type: ignore[misc,assignment]
StructuredTool = object # type: ignore[misc,assignment]
Tool = object # type: ignore[misc,assignment]
from headroom.integrations.mcp import compress_tool_result
logger = logging.getLogger(__name__)
def _check_langchain_available() -> None:
"""Raise ImportError if LangChain is not installed."""
if not LANGCHAIN_AVAILABLE:
raise ImportError(
"LangChain is required for this integration. "
"Install with: pip install headroom[langchain] "
"or: pip install langchain-core"
)
@dataclass
class ToolCompressionMetrics:
"""Metrics from a single tool compression."""
tool_name: str
timestamp: datetime
chars_before: int
chars_after: int
chars_saved: int
compression_ratio: float
was_compressed: bool
@dataclass
class ToolMetricsCollector:
"""Collects compression metrics across all tool invocations."""
metrics: list[ToolCompressionMetrics] = field(default_factory=list)
def add(self, metric: ToolCompressionMetrics) -> None:
"""Add a metric entry."""
self.metrics.append(metric)
# Keep only last 1000
if len(self.metrics) > 1000:
self.metrics = self.metrics[-1000:]
def get_summary(self) -> dict[str, Any]:
"""Get summary statistics."""
if not self.metrics:
return {
"total_invocations": 0,
"total_compressions": 0,
"total_chars_saved": 0,
}
compressed = [m for m in self.metrics if m.was_compressed]
return {
"total_invocations": len(self.metrics),
"total_compressions": len(compressed),
"total_chars_saved": sum(m.chars_saved for m in self.metrics),
"average_compression_ratio": (
sum(m.compression_ratio for m in compressed) / len(compressed) if compressed else 0
),
"by_tool": self._get_by_tool_stats(),
}
def _get_by_tool_stats(self) -> dict[str, dict[str, Any]]:
"""Get per-tool statistics."""
by_tool: dict[str, list[ToolCompressionMetrics]] = {}
for m in self.metrics:
if m.tool_name not in by_tool:
by_tool[m.tool_name] = []
by_tool[m.tool_name].append(m)
result = {}
for name, tool_metrics in by_tool.items():
compressed = [m for m in tool_metrics if m.was_compressed]
result[name] = {
"invocations": len(tool_metrics),
"compressions": len(compressed),
"chars_saved": sum(m.chars_saved for m in tool_metrics),
}
return result
# Global metrics collector
_global_metrics = ToolMetricsCollector()
def get_tool_metrics() -> ToolMetricsCollector:
"""Get the global tool metrics collector."""
return _global_metrics
def reset_tool_metrics() -> None:
"""Reset global tool metrics."""
global _global_metrics
_global_metrics = ToolMetricsCollector()
class HeadroomToolWrapper:
"""Wraps a LangChain tool to compress its output.
Applies SmartCrusher compression to tool outputs, particularly
useful for tools that return large JSON arrays (search results,
database queries, etc.).
Example:
from langchain.tools import Tool
from headroom.integrations import HeadroomToolWrapper
def search(query: str) -> str:
# Returns large JSON with 1000 results
return json.dumps({"results": [...1000 items...]})
search_tool = Tool(name="search", func=search, description="Search")
wrapped = HeadroomToolWrapper(search_tool)
# Use wrapped tool - output automatically compressed
result = wrapped("python tutorials")
Attributes:
tool: The wrapped LangChain tool
min_chars_to_compress: Minimum output size to trigger compression
metrics_collector: Collector for compression metrics
"""
def __init__(
self,
tool: BaseTool,
min_chars_to_compress: int = 1000,
metrics_collector: ToolMetricsCollector | None = None,
):
"""Initialize HeadroomToolWrapper.
Args:
tool: The LangChain BaseTool to wrap.
min_chars_to_compress: Minimum character count for output
before compression is applied. Default 1000.
metrics_collector: Collector for metrics. Uses global
collector if not specified.
"""
_check_langchain_available()
self.tool = tool
self.min_chars_to_compress = min_chars_to_compress
self._metrics = metrics_collector or _global_metrics
# Copy tool metadata
self.name = tool.name
self.description = tool.description
def __call__(self, *args: Any, **kwargs: Any) -> str:
"""Invoke the tool and compress output.
Args:
*args: Arguments to pass to the tool.
**kwargs: Keyword arguments to pass to the tool.
Returns:
Compressed tool output as string.
"""
# Invoke underlying tool
result = self.tool.invoke(*args, **kwargs)
# Convert to string if needed
if not isinstance(result, str):
result = str(result)
# Check if compression is needed
if len(result) < self.min_chars_to_compress:
self._record_metrics(result, result, was_compressed=False)
return result
# Try to compress
compressed = self._compress_output(result)
self._record_metrics(result, compressed, was_compressed=True)
return compressed
def invoke(self, *args: Any, **kwargs: Any) -> str:
"""Invoke the tool (alias for __call__)."""
return self(*args, **kwargs)
def _compress_output(self, output: str) -> str:
"""Apply compression to tool output.
Args:
output: Tool output string.
Returns:
Compressed output.
"""
try:
return compress_tool_result(
content=output,
tool_name=self.name,
)
except Exception as e:
logger.debug(f"Tool compression failed: {e}")
return output
def _record_metrics(self, original: str, compressed: str, was_compressed: bool) -> None:
"""Record compression metrics.
Args:
original: Original output.
compressed: Compressed output.
was_compressed: Whether compression was applied.
"""
chars_before = len(original)
chars_after = len(compressed)
chars_saved = chars_before - chars_after
metric = ToolCompressionMetrics(
tool_name=self.name,
timestamp=datetime.now(),
chars_before=chars_before,
chars_after=chars_after,
chars_saved=max(0, chars_saved),
compression_ratio=chars_after / chars_before if chars_before > 0 else 1.0,
was_compressed=was_compressed and chars_saved > 0,
)
self._metrics.add(metric)
if was_compressed and chars_saved > 0:
logger.info(
f"HeadroomToolWrapper[{self.name}]: {chars_before} -> {chars_after} chars "
f"({chars_saved} saved, {metric.compression_ratio:.1%} of original)"
)
def as_langchain_tool(self) -> StructuredTool:
"""Convert wrapper back to a LangChain tool.
Useful when you need to pass the wrapped tool to APIs
that expect a LangChain tool type.
Returns:
StructuredTool that wraps this wrapper.
"""
return StructuredTool.from_function(
func=self.__call__,
name=self.name,
description=self.description,
)
def wrap_tools_with_headroom(
tools: list[BaseTool],
min_chars_to_compress: int = 1000,
metrics_collector: ToolMetricsCollector | None = None,
) -> list[StructuredTool]:
"""Wrap multiple LangChain tools with Headroom compression.
Convenience function to wrap all tools in a list at once.
Args:
tools: List of LangChain tools to wrap.
min_chars_to_compress: Minimum output size for compression.
metrics_collector: Shared metrics collector for all tools.
Returns:
List of wrapped tools as StructuredTools.
Example:
from langchain.tools import Tool
from headroom.integrations import wrap_tools_with_headroom
tools = [search_tool, database_tool, api_tool]
wrapped = wrap_tools_with_headroom(tools)
# Use wrapped tools in agent
agent = create_openai_tools_agent(llm, wrapped, prompt)
"""
_check_langchain_available()
collector = metrics_collector or _global_metrics
wrapped = []
for tool in tools:
wrapper = HeadroomToolWrapper(
tool=tool,
min_chars_to_compress=min_chars_to_compress,
metrics_collector=collector,
)
wrapped.append(wrapper.as_langchain_tool())
return wrapped

View file

@ -27,9 +27,10 @@ Example:
from __future__ import annotations
import asyncio
import json
import logging
from collections.abc import Iterator, Sequence
from collections.abc import AsyncIterator, Iterator, Sequence
from dataclasses import dataclass
from datetime import datetime
from typing import Any
@ -48,13 +49,14 @@ try:
)
from langchain_core.outputs import ChatGeneration, ChatResult
from langchain_core.runnables import RunnableLambda
from pydantic import Field, PrivateAttr
from pydantic import ConfigDict, Field, PrivateAttr
LANGCHAIN_AVAILABLE = True
except ImportError:
LANGCHAIN_AVAILABLE = False
BaseChatModel = object
BaseCallbackHandler = object
ConfigDict = lambda **kwargs: {} # type: ignore[assignment,misc] # noqa: E731
Field = lambda **kwargs: None # type: ignore[assignment] # noqa: E731
PrivateAttr = lambda **kwargs: None # type: ignore[assignment] # noqa: E731
@ -62,10 +64,12 @@ from headroom import HeadroomConfig, HeadroomMode
from headroom.providers import OpenAIProvider
from headroom.transforms import TransformPipeline
from .providers import get_headroom_provider, get_model_name_from_langchain
logger = logging.getLogger(__name__)
def _check_langchain_available():
def _check_langchain_available() -> None:
"""Raise ImportError if LangChain is not installed."""
if not LANGCHAIN_AVAILABLE:
raise ImportError(
@ -133,6 +137,10 @@ class HeadroomChatModel(BaseChatModel):
wrapped_model: Any = Field(description="The wrapped LangChain chat model")
headroom_config: Any = Field(default=None, description="Headroom configuration")
mode: HeadroomMode = Field(default=HeadroomMode.OPTIMIZE, description="Headroom mode")
auto_detect_provider: bool = Field(
default=True,
description="Auto-detect provider from wrapped model (OpenAI, Anthropic, Google)",
)
# Private attributes (not serialized)
_metrics_history: list = PrivateAttr(default_factory=list)
@ -140,24 +148,27 @@ class HeadroomChatModel(BaseChatModel):
_pipeline: Any = PrivateAttr(default=None)
_provider: Any = PrivateAttr(default=None)
class Config:
"""Pydantic config for LangChain compatibility."""
arbitrary_types_allowed = True
# Pydantic v2 config for LangChain compatibility
model_config = ConfigDict(arbitrary_types_allowed=True)
def __init__(
self,
wrapped_model: BaseChatModel,
config: HeadroomConfig | None = None,
mode: HeadroomMode = HeadroomMode.OPTIMIZE,
**kwargs,
):
auto_detect_provider: bool = True,
**kwargs: Any,
) -> None:
"""Initialize HeadroomChatModel.
Args:
wrapped_model: Any LangChain BaseChatModel to wrap
config: HeadroomConfig for optimization settings
mode: HeadroomMode (AUDIT, OPTIMIZE, or SIMULATE)
auto_detect_provider: Auto-detect provider from wrapped model.
When True (default), automatically detects if the wrapped model
is OpenAI, Anthropic, Google, etc. and uses the appropriate
Headroom provider for accurate token counting.
**kwargs: Additional arguments passed to BaseChatModel
"""
_check_langchain_available()
@ -166,6 +177,7 @@ class HeadroomChatModel(BaseChatModel):
wrapped_model=wrapped_model,
headroom_config=config or HeadroomConfig(),
mode=mode,
auto_detect_provider=auto_detect_provider,
**kwargs,
)
self._metrics_history = []
@ -188,9 +200,17 @@ class HeadroomChatModel(BaseChatModel):
@property
def pipeline(self) -> TransformPipeline:
"""Lazily initialize TransformPipeline."""
"""Lazily initialize TransformPipeline.
When auto_detect_provider is True, automatically detects the provider
from the wrapped model's class path (e.g., ChatAnthropic -> AnthropicProvider).
"""
if self._pipeline is None:
self._provider = OpenAIProvider()
if self.auto_detect_provider:
self._provider = get_headroom_provider(self.wrapped_model)
logger.debug(f"Auto-detected provider: {self._provider.__class__.__name__}")
else:
self._provider = OpenAIProvider()
self._pipeline = TransformPipeline(
config=self.headroom_config,
provider=self._provider,
@ -290,10 +310,11 @@ class HeadroomChatModel(BaseChatModel):
# Convert to OpenAI format
openai_messages = self._convert_messages_to_openai(messages)
# Get model name and context limit
model = getattr(self.wrapped_model, "model_name", None)
if model is None:
model = getattr(self.wrapped_model, "model", "gpt-4o")
# Get model name from wrapped model
model = get_model_name_from_langchain(self.wrapped_model)
# Ensure pipeline is initialized (this also sets up provider)
_ = self.pipeline
# Get model context limit from provider
model_limit = self._provider.get_context_limit(model) if self._provider else 128000
@ -342,7 +363,7 @@ class HeadroomChatModel(BaseChatModel):
messages: list[BaseMessage],
stop: list[str] | None = None,
run_manager: Any = None,
**kwargs,
**kwargs: Any,
) -> ChatResult:
"""Generate response with Headroom optimization.
@ -371,14 +392,15 @@ class HeadroomChatModel(BaseChatModel):
messages: list[BaseMessage],
stop: list[str] | None = None,
run_manager: Any = None,
**kwargs,
**kwargs: Any,
) -> Iterator[ChatGeneration]:
"""Stream response with Headroom optimization."""
# Optimize messages
optimized_messages, metrics = self._optimize_messages(messages)
logger.info(
f"Headroom optimized (streaming): {metrics.tokens_before} -> {metrics.tokens_after} tokens"
f"Headroom optimized (streaming): {metrics.tokens_before} -> "
f"{metrics.tokens_after} tokens"
)
# Stream from wrapped model
@ -389,13 +411,78 @@ class HeadroomChatModel(BaseChatModel):
**kwargs,
)
def bind_tools(self, tools: Sequence[Any], **kwargs) -> HeadroomChatModel:
async def _agenerate(
self,
messages: list[BaseMessage],
stop: list[str] | None = None,
run_manager: Any = None,
**kwargs: Any,
) -> ChatResult:
"""Async generate response with Headroom optimization.
This enables `await model.ainvoke(messages)` to work correctly.
The optimization step runs in a thread executor since it's CPU-bound.
"""
# Run optimization in executor (CPU-bound)
loop = asyncio.get_event_loop()
optimized_messages, metrics = await loop.run_in_executor(
None, self._optimize_messages, messages
)
logger.info(
f"Headroom optimized (async): {metrics.tokens_before} -> {metrics.tokens_after} tokens "
f"({metrics.savings_percent:.1f}% saved)"
)
# Call wrapped model's async generate
result = await self.wrapped_model._agenerate(
optimized_messages,
stop=stop,
run_manager=run_manager,
**kwargs,
)
return result
async def _astream(
self,
messages: list[BaseMessage],
stop: list[str] | None = None,
run_manager: Any = None,
**kwargs: Any,
) -> AsyncIterator[ChatGeneration]:
"""Async stream response with Headroom optimization.
This enables `async for chunk in model.astream(messages)` to work correctly.
"""
# Run optimization in executor (CPU-bound)
loop = asyncio.get_event_loop()
optimized_messages, metrics = await loop.run_in_executor(
None, self._optimize_messages, messages
)
logger.info(
f"Headroom optimized (async streaming): {metrics.tokens_before} -> "
f"{metrics.tokens_after} tokens"
)
# Async stream from wrapped model
async for chunk in self.wrapped_model._astream(
optimized_messages,
stop=stop,
run_manager=run_manager,
**kwargs,
):
yield chunk
def bind_tools(self, tools: Sequence[Any], **kwargs: Any) -> HeadroomChatModel:
"""Bind tools to the wrapped model."""
new_wrapped = self.wrapped_model.bind_tools(tools, **kwargs)
return HeadroomChatModel(
wrapped_model=new_wrapped,
config=self.headroom_config,
mode=self.mode,
auto_detect_provider=self.auto_detect_provider,
)
def get_savings_summary(self) -> dict[str, Any]:
@ -494,7 +581,7 @@ class HeadroomCallbackHandler(BaseCallbackHandler):
self,
serialized: dict[str, Any],
prompts: list[str],
**kwargs,
**kwargs: Any,
) -> None:
"""Called when LLM starts processing."""
self._current_request = {
@ -511,7 +598,7 @@ class HeadroomCallbackHandler(BaseCallbackHandler):
self,
serialized: dict[str, Any],
messages: list[list[BaseMessage]],
**kwargs,
**kwargs: Any,
) -> None:
"""Called when chat model starts processing."""
# Estimate tokens from messages
@ -532,7 +619,10 @@ class HeadroomCallbackHandler(BaseCallbackHandler):
# Check token alert
if self.token_alert_threshold and estimated_tokens > self.token_alert_threshold:
alert = f"Token alert: {estimated_tokens} tokens exceeds threshold {self.token_alert_threshold}"
alert = (
f"Token alert: {estimated_tokens} tokens exceeds "
f"threshold {self.token_alert_threshold}"
)
self._alerts.append(alert)
logger.warning(alert)
@ -542,7 +632,7 @@ class HeadroomCallbackHandler(BaseCallbackHandler):
f"Chat model request: ~{estimated_tokens} input tokens",
)
def on_llm_end(self, response: Any, **kwargs) -> None:
def on_llm_end(self, response: Any, **kwargs: Any) -> None:
"""Called when LLM finishes processing."""
if self._current_request is None:
return
@ -579,7 +669,7 @@ class HeadroomCallbackHandler(BaseCallbackHandler):
self._current_request = None
def on_llm_error(self, error: Exception, **kwargs) -> None:
def on_llm_error(self, error: Exception, **kwargs: Any) -> None:
"""Called when LLM encounters an error."""
if self._current_request:
self._current_request["error"] = str(error)
@ -677,19 +767,19 @@ class HeadroomRunnable:
)
return self._pipeline
def __or__(self, other):
def __or__(self, other: Any) -> Any:
"""Support pipe operator for LCEL composition."""
from langchain_core.runnables import RunnableSequence
return RunnableSequence(first=self.as_runnable(), last=other)
def __ror__(self, other):
def __ror__(self, other: Any) -> Any:
"""Support reverse pipe operator."""
from langchain_core.runnables import RunnableSequence
return RunnableSequence(first=other, last=self.as_runnable())
def as_runnable(self):
def as_runnable(self) -> RunnableLambda:
"""Convert to LangChain Runnable."""
return RunnableLambda(self._optimize)

View file

@ -0,0 +1,324 @@
"""LangSmith integration for Headroom compression metrics.
This module provides HeadroomLangSmithCallbackHandler, a LangChain callback
handler that adds Headroom compression metrics to LangSmith traces.
When used with HeadroomChatModel, it automatically captures:
- Tokens before/after optimization
- Savings percentage
- Transforms applied
- Per-request compression details
Example:
import os
from langchain_openai import ChatOpenAI
from headroom.integrations import (
HeadroomChatModel,
HeadroomLangSmithCallbackHandler,
)
# Enable LangSmith tracing
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "..."
# Create handler
handler = HeadroomLangSmithCallbackHandler()
# Use with HeadroomChatModel
llm = HeadroomChatModel(
ChatOpenAI(model="gpt-4o"),
callbacks=[handler],
)
# Traces will include headroom.* metadata
response = llm.invoke("Hello!")
"""
from __future__ import annotations
import logging
import os
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any
from uuid import UUID
# LangChain imports - these are optional dependencies
try:
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.messages import BaseMessage
from langchain_core.outputs import LLMResult
LANGCHAIN_AVAILABLE = True
except ImportError:
LANGCHAIN_AVAILABLE = False
BaseCallbackHandler = object # type: ignore[misc,assignment]
LLMResult = object # type: ignore[misc,assignment]
# LangSmith imports - optional
try:
from langsmith import Client as LangSmithClient
LANGSMITH_AVAILABLE = True
except ImportError:
LANGSMITH_AVAILABLE = False
LangSmithClient = None # type: ignore[misc,assignment]
logger = logging.getLogger(__name__)
def _check_langchain_available() -> None:
"""Raise ImportError if LangChain is not installed."""
if not LANGCHAIN_AVAILABLE:
raise ImportError(
"LangChain is required for this integration. "
"Install with: pip install headroom[langchain] "
"or: pip install langchain-core"
)
@dataclass
class PendingMetrics:
"""Metrics pending attachment to a LangSmith run."""
tokens_before: int
tokens_after: int
tokens_saved: int
savings_percent: float
transforms_applied: list[str]
timestamp: datetime = field(default_factory=datetime.now)
class HeadroomLangSmithCallbackHandler(BaseCallbackHandler):
"""Callback handler that adds Headroom metrics to LangSmith traces.
Integrates with LangSmith to provide visibility into context
optimization within traces. Metrics appear as metadata with
the `headroom.` prefix.
Works automatically when:
1. LANGCHAIN_TRACING_V2=true is set
2. Used as a callback with HeadroomChatModel
3. LangSmith API key is configured
Example:
from headroom.integrations import (
HeadroomChatModel,
HeadroomLangSmithCallbackHandler,
)
handler = HeadroomLangSmithCallbackHandler()
llm = HeadroomChatModel(
ChatOpenAI(model="gpt-4o"),
callbacks=[handler],
)
response = llm.invoke("Hello!")
# LangSmith trace now includes:
# - headroom.tokens_before
# - headroom.tokens_after
# - headroom.tokens_saved
# - headroom.savings_percent
# - headroom.transforms_applied
Attributes:
langsmith_client: LangSmith client for updating runs.
pending_metrics: Metrics waiting to be attached to runs.
"""
def __init__(
self,
langsmith_client: Any = None,
auto_update_runs: bool = True,
):
"""Initialize HeadroomLangSmithCallbackHandler.
Args:
langsmith_client: LangSmith client instance. Auto-creates
one if not provided and LangSmith is available.
auto_update_runs: If True, automatically updates LangSmith
runs with Headroom metadata. Default True.
"""
_check_langchain_available()
self._client = langsmith_client
self._auto_update = auto_update_runs
self._pending_metrics: dict[str, PendingMetrics] = {}
self._run_metrics: dict[str, dict[str, Any]] = {}
# Initialize LangSmith client if available and not provided
if self._client is None and LANGSMITH_AVAILABLE and auto_update_runs:
try:
if os.environ.get("LANGCHAIN_API_KEY"):
self._client = LangSmithClient()
except Exception as e:
logger.debug(f"Could not initialize LangSmith client: {e}")
def set_headroom_metrics(
self,
run_id: str | UUID,
tokens_before: int,
tokens_after: int,
transforms_applied: list[str] | None = None,
) -> None:
"""Set Headroom metrics for a run.
Call this from HeadroomChatModel after optimization to attach
metrics to the current run.
Args:
run_id: The LangSmith run ID.
tokens_before: Token count before optimization.
tokens_after: Token count after optimization.
transforms_applied: List of transforms that were applied.
"""
run_id_str = str(run_id)
tokens_saved = tokens_before - tokens_after
savings_percent = (tokens_saved / tokens_before * 100) if tokens_before > 0 else 0.0
metrics = PendingMetrics(
tokens_before=tokens_before,
tokens_after=tokens_after,
tokens_saved=tokens_saved,
savings_percent=savings_percent,
transforms_applied=transforms_applied or [],
)
self._pending_metrics[run_id_str] = metrics
logger.debug(
f"Headroom metrics set for run {run_id_str}: "
f"{tokens_before} -> {tokens_after} tokens ({savings_percent:.1f}% saved)"
)
def on_chat_model_start(
self,
serialized: dict[str, Any],
messages: list[list[BaseMessage]],
*,
run_id: UUID,
**kwargs: Any,
) -> None:
"""Called when chat model starts.
Records the run ID for later metric attachment.
"""
run_id_str = str(run_id)
# Initialize empty metrics for this run
self._run_metrics[run_id_str] = {}
def on_llm_end(
self,
response: LLMResult,
*,
run_id: UUID,
**kwargs: Any,
) -> None:
"""Called when LLM completes.
Attaches pending Headroom metrics to the LangSmith run.
"""
run_id_str = str(run_id)
# Check for pending metrics
if run_id_str in self._pending_metrics:
metrics = self._pending_metrics.pop(run_id_str)
self._attach_metrics_to_run(run_id_str, metrics)
def _attach_metrics_to_run(self, run_id: str, metrics: PendingMetrics) -> None:
"""Attach Headroom metrics to a LangSmith run.
Args:
run_id: The run ID.
metrics: Metrics to attach.
"""
metadata = {
"headroom.tokens_before": metrics.tokens_before,
"headroom.tokens_after": metrics.tokens_after,
"headroom.tokens_saved": metrics.tokens_saved,
"headroom.savings_percent": round(metrics.savings_percent, 2),
"headroom.transforms_applied": metrics.transforms_applied,
"headroom.optimization_timestamp": metrics.timestamp.isoformat(),
}
# Store in run metrics
self._run_metrics[run_id] = metadata
# Update LangSmith run if client available
if self._client and self._auto_update:
try:
self._client.update_run(
run_id=run_id,
extra={"metadata": metadata},
)
logger.debug(f"Updated LangSmith run {run_id} with Headroom metrics")
except Exception as e:
logger.debug(f"Could not update LangSmith run: {e}")
def get_run_metrics(self, run_id: str | UUID) -> dict[str, Any]:
"""Get Headroom metrics for a specific run.
Args:
run_id: The run ID.
Returns:
Dictionary of headroom.* metrics for the run.
"""
return self._run_metrics.get(str(run_id), {})
def get_all_metrics(self) -> dict[str, dict[str, Any]]:
"""Get all recorded run metrics.
Returns:
Dictionary mapping run IDs to their metrics.
"""
return self._run_metrics.copy()
def get_summary(self) -> dict[str, Any]:
"""Get summary statistics across all runs.
Returns:
Summary with total runs, tokens saved, etc.
"""
if not self._run_metrics:
return {
"total_runs": 0,
"total_tokens_saved": 0,
"average_savings_percent": 0,
}
total_saved = sum(m.get("headroom.tokens_saved", 0) for m in self._run_metrics.values())
savings_percents = [
m.get("headroom.savings_percent", 0) for m in self._run_metrics.values()
]
return {
"total_runs": len(self._run_metrics),
"total_tokens_saved": total_saved,
"average_savings_percent": (
sum(savings_percents) / len(savings_percents) if savings_percents else 0
),
}
def reset(self) -> None:
"""Clear all recorded metrics."""
self._pending_metrics.clear()
self._run_metrics.clear()
def is_langsmith_available() -> bool:
"""Check if LangSmith is available and configured.
Returns:
True if LangSmith is installed and API key is set.
"""
return LANGSMITH_AVAILABLE and bool(os.environ.get("LANGCHAIN_API_KEY"))
def is_langsmith_tracing_enabled() -> bool:
"""Check if LangSmith tracing is enabled.
Returns:
True if LANGCHAIN_TRACING_V2 is set to "true".
"""
return os.environ.get("LANGCHAIN_TRACING_V2", "").lower() == "true"

View file

@ -0,0 +1,319 @@
"""Memory integration for LangChain with automatic compression.
This module provides HeadroomChatMessageHistory, a wrapper for any LangChain
chat message history that automatically compresses conversation history
when it exceeds a token threshold.
Example:
from langchain.memory import ConversationBufferMemory
from langchain_community.chat_message_histories import ChatMessageHistory
from headroom.integrations import HeadroomChatMessageHistory
# Wrap any chat message history
base_history = ChatMessageHistory()
compressed_history = HeadroomChatMessageHistory(base_history)
# Use with ConversationBufferMemory (zero code changes to chain)
memory = ConversationBufferMemory(chat_memory=compressed_history)
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from headroom.providers.base import Provider
# LangChain imports - these are optional dependencies
try:
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import (
AIMessage,
BaseMessage,
HumanMessage,
SystemMessage,
ToolMessage,
)
LANGCHAIN_AVAILABLE = True
except ImportError:
LANGCHAIN_AVAILABLE = False
BaseChatMessageHistory = object # type: ignore[misc,assignment]
from headroom import HeadroomConfig
from headroom.config import RollingWindowConfig
from headroom.providers import OpenAIProvider
from headroom.transforms import TransformPipeline
logger = logging.getLogger(__name__)
def _check_langchain_available() -> None:
"""Raise ImportError if LangChain is not installed."""
if not LANGCHAIN_AVAILABLE:
raise ImportError(
"LangChain is required for this integration. "
"Install with: pip install headroom[langchain] "
"or: pip install langchain-core"
)
class HeadroomChatMessageHistory(BaseChatMessageHistory):
"""Wraps any LangChain chat message history with automatic compression.
When conversation history exceeds the token threshold, automatically
applies RollingWindow compression to keep recent turns while fitting
within the limit.
This works with ANY memory type because it wraps at the storage layer:
- ConversationBufferMemory
- ConversationSummaryMemory
- ConversationBufferWindowMemory
- Redis, PostgreSQL, or any custom history
Example:
from langchain.memory import ConversationBufferMemory
from langchain_community.chat_message_histories import ChatMessageHistory
from headroom.integrations import HeadroomChatMessageHistory
# Wrap base history
base = ChatMessageHistory()
compressed = HeadroomChatMessageHistory(
base,
compress_threshold_tokens=4000,
keep_recent_turns=5,
)
# Use with any memory class
memory = ConversationBufferMemory(chat_memory=compressed)
# Messages are compressed automatically when accessed
chain = ConversationChain(llm=llm, memory=memory)
chain.invoke({"input": "Hello!"})
Attributes:
base_history: The underlying chat message history
compress_threshold_tokens: Token count that triggers compression
keep_recent_turns: Minimum recent turns to always preserve
model: Model name for token counting (default: "gpt-4o")
"""
def __init__(
self,
base_history: BaseChatMessageHistory,
compress_threshold_tokens: int = 4000,
keep_recent_turns: int = 5,
model: str = "gpt-4o",
provider: Provider | None = None,
):
"""Initialize HeadroomChatMessageHistory.
Args:
base_history: Any LangChain BaseChatMessageHistory to wrap
compress_threshold_tokens: Apply compression when history exceeds
this many tokens. Default 4000.
keep_recent_turns: Minimum number of recent user/assistant turns
to always preserve during compression. Default 5.
model: Model name for token counting. Default "gpt-4o".
provider: Headroom provider for token counting. Auto-uses
OpenAIProvider if not specified.
"""
_check_langchain_available()
self._base = base_history
self._threshold = compress_threshold_tokens
self._keep_recent_turns = keep_recent_turns
self._model = model
self._provider: Provider = provider or OpenAIProvider()
# Track compression stats
self._compression_count = 0
self._total_tokens_saved = 0
@property
def messages(self) -> list[BaseMessage]:
"""Get messages, applying compression if over threshold.
Returns:
List of messages, potentially compressed to fit within threshold.
"""
raw_messages = self._base.messages
if not raw_messages:
return []
# Count tokens
token_count = self._count_tokens(raw_messages)
if token_count <= self._threshold:
return list(raw_messages)
# Apply compression
compressed = self._apply_rolling_window(raw_messages)
tokens_after = self._count_tokens(compressed)
self._compression_count += 1
self._total_tokens_saved += token_count - tokens_after
logger.info(
f"HeadroomChatMessageHistory compressed: {token_count} -> {tokens_after} tokens "
f"({len(raw_messages)} -> {len(compressed)} messages)"
)
return compressed
def add_message(self, message: BaseMessage) -> None:
"""Add a message to the underlying history.
Args:
message: The message to add.
"""
self._base.add_message(message)
def add_user_message(self, message: str) -> None:
"""Add a user message to the history.
Args:
message: The user message content.
"""
self._base.add_user_message(message)
def add_ai_message(self, message: str) -> None:
"""Add an AI message to the history.
Args:
message: The AI message content.
"""
self._base.add_ai_message(message)
def clear(self) -> None:
"""Clear all messages from history."""
self._base.clear()
def _count_tokens(self, messages: list[BaseMessage]) -> int:
"""Count tokens in messages using provider's tokenizer.
Args:
messages: List of messages to count.
Returns:
Total token count.
"""
token_counter = self._provider.get_token_counter(self._model)
total = 0
for msg in messages:
content = msg.content if isinstance(msg.content, str) else str(msg.content)
total += token_counter.count_text(content)
return total
def _apply_rolling_window(self, messages: list[BaseMessage]) -> list[BaseMessage]:
"""Apply RollingWindow compression to messages.
Args:
messages: Messages to compress.
Returns:
Compressed messages fitting within threshold.
"""
# Convert to OpenAI format for Headroom transforms
openai_messages = self._convert_to_openai(messages)
# Use TransformPipeline which handles tokenizer setup
config = HeadroomConfig(
rolling_window=RollingWindowConfig(keep_last_turns=self._keep_recent_turns),
)
pipeline = TransformPipeline(config=config, provider=self._provider)
# Apply compression via pipeline
result = pipeline.apply(
messages=openai_messages,
model=self._model,
model_limit=self._threshold,
)
# Convert back to LangChain format
return self._convert_from_openai(result.messages)
def _convert_to_openai(self, messages: list[BaseMessage]) -> list[dict[str, Any]]:
"""Convert LangChain messages to OpenAI format.
Args:
messages: LangChain messages.
Returns:
OpenAI format messages.
"""
result = []
for msg in messages:
content = msg.content if isinstance(msg.content, str) else str(msg.content)
if isinstance(msg, SystemMessage):
result.append({"role": "system", "content": content})
elif isinstance(msg, HumanMessage):
result.append({"role": "user", "content": content})
elif isinstance(msg, AIMessage):
entry: dict[str, Any] = {"role": "assistant", "content": content}
if hasattr(msg, "tool_calls") and msg.tool_calls:
entry["tool_calls"] = msg.tool_calls
result.append(entry)
elif isinstance(msg, ToolMessage):
result.append(
{
"role": "tool",
"tool_call_id": getattr(msg, "tool_call_id", ""),
"content": content,
}
)
else:
# Generic fallback
result.append(
{
"role": getattr(msg, "type", "user"),
"content": content,
}
)
return result
def _convert_from_openai(self, messages: list[dict[str, Any]]) -> list[BaseMessage]:
"""Convert OpenAI format back to LangChain messages.
Args:
messages: OpenAI format messages.
Returns:
LangChain messages.
"""
result: list[BaseMessage] = []
for msg in messages:
role = msg.get("role", "user")
content = msg.get("content", "")
if role == "system":
result.append(SystemMessage(content=content))
elif role == "user":
result.append(HumanMessage(content=content))
elif role == "assistant":
tool_calls = msg.get("tool_calls", [])
result.append(AIMessage(content=content, tool_calls=tool_calls))
elif role == "tool":
result.append(
ToolMessage(
content=content,
tool_call_id=msg.get("tool_call_id", ""),
)
)
return result
def get_compression_stats(self) -> dict[str, Any]:
"""Get statistics about compression operations.
Returns:
Dictionary with compression_count, total_tokens_saved.
"""
return {
"compression_count": self._compression_count,
"total_tokens_saved": self._total_tokens_saved,
"threshold_tokens": self._threshold,
"keep_recent_turns": self._keep_recent_turns,
}

View file

@ -0,0 +1,200 @@
"""Provider detection for LangChain models.
This module provides automatic provider detection from LangChain chat models
without requiring explicit provider imports. It uses duck-typing based on
class paths to identify the appropriate Headroom provider.
Example:
from langchain_anthropic import ChatAnthropic
from headroom.integrations.langchain import get_headroom_provider
model = ChatAnthropic(model="claude-3-5-sonnet-20241022")
provider = get_headroom_provider(model) # Returns AnthropicProvider
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from headroom.providers.base import Provider
logger = logging.getLogger(__name__)
# Provider detection patterns
# Maps provider name to list of class path patterns to match
PROVIDER_PATTERNS: dict[str, list[str]] = {
"openai": [
"langchain_openai.ChatOpenAI",
"langchain_openai.chat_models.ChatOpenAI",
"langchain_community.chat_models.ChatOpenAI",
"langchain.chat_models.ChatOpenAI",
"ChatOpenAI",
],
"anthropic": [
"langchain_anthropic.ChatAnthropic",
"langchain_anthropic.chat_models.ChatAnthropic",
"langchain_community.chat_models.ChatAnthropic",
"langchain.chat_models.ChatAnthropic",
"ChatAnthropic",
],
"google": [
"langchain_google_genai.ChatGoogleGenerativeAI",
"langchain_google_genai.chat_models.ChatGoogleGenerativeAI",
"langchain_community.chat_models.ChatGoogleGenerativeAI",
"ChatGoogleGenerativeAI",
# Also match Vertex AI
"langchain_google_vertexai.ChatVertexAI",
"ChatVertexAI",
],
"cohere": [
"langchain_cohere.ChatCohere",
"langchain_community.chat_models.ChatCohere",
"ChatCohere",
],
"mistral": [
"langchain_mistralai.ChatMistralAI",
"langchain_community.chat_models.ChatMistralAI",
"ChatMistralAI",
],
}
# Model name patterns for fallback detection
MODEL_NAME_PATTERNS: dict[str, list[str]] = {
"anthropic": ["claude", "anthropic"],
"openai": ["gpt", "o1", "o3", "davinci", "turbo"],
"google": ["gemini", "palm", "bison"],
"cohere": ["command", "cohere"],
"mistral": ["mistral", "mixtral"],
}
def detect_provider(model: Any) -> str:
"""Detect provider name from a LangChain model using duck-typing.
Detection strategy:
1. Check class module and name against known patterns
2. Check model_name attribute against known model patterns
3. Fall back to "openai" as safe default
Args:
model: Any LangChain chat model instance
Returns:
Provider name string: "openai", "anthropic", "google", "cohere", "mistral"
Example:
>>> from langchain_anthropic import ChatAnthropic
>>> model = ChatAnthropic(model="claude-3-5-sonnet-20241022")
>>> detect_provider(model)
'anthropic'
"""
# Strategy 1: Check class path
class_module = getattr(model.__class__, "__module__", "")
class_name = model.__class__.__name__
class_path = f"{class_module}.{class_name}"
for provider_name, patterns in PROVIDER_PATTERNS.items():
for pattern in patterns:
if pattern in class_path or class_name == pattern.split(".")[-1]:
logger.debug(f"Detected provider '{provider_name}' from class path: {class_path}")
return provider_name
# Strategy 2: Check model_name attribute
model_name = _get_model_name(model)
if model_name:
model_name_lower = model_name.lower()
for provider_name, name_patterns in MODEL_NAME_PATTERNS.items():
for pattern in name_patterns:
if pattern in model_name_lower:
logger.debug(
f"Detected provider '{provider_name}' from model name: {model_name}"
)
return provider_name
# Strategy 3: Fall back to OpenAI (most common, safe default)
logger.debug(f"Could not detect provider for {class_path}, falling back to 'openai'")
return "openai"
def _get_model_name(model: Any) -> str | None:
"""Extract model name from a LangChain model.
Tries common attribute names used by different LangChain models.
"""
# Try common attribute names
for attr in ["model_name", "model", "model_id", "_model_name"]:
value = getattr(model, attr, None)
if isinstance(value, str):
return value
return None
def get_headroom_provider(model: Any) -> Provider:
"""Get appropriate Headroom Provider instance for a LangChain model.
This function automatically detects the provider from the model type
and returns a configured Headroom provider for accurate token counting
and context limit detection.
Args:
model: Any LangChain chat model instance
Returns:
Configured Headroom Provider instance
Example:
>>> from langchain_anthropic import ChatAnthropic
>>> model = ChatAnthropic(model="claude-3-5-sonnet-20241022")
>>> provider = get_headroom_provider(model)
>>> provider.name
'anthropic'
"""
# Import providers lazily to avoid circular imports
from headroom.providers import (
AnthropicProvider,
GoogleProvider,
OpenAIProvider,
)
provider_name = detect_provider(model)
if provider_name == "anthropic":
return AnthropicProvider()
elif provider_name == "google":
return GoogleProvider()
# Cohere and Mistral fall back to OpenAI-compatible for now
# TODO: Add dedicated providers when needed
# Default to OpenAI
return OpenAIProvider()
def get_model_name_from_langchain(model: Any) -> str:
"""Extract the model name string from a LangChain model.
Useful for getting the model identifier for token counting
and context limit lookup.
Args:
model: Any LangChain chat model instance
Returns:
Model name string (e.g., "gpt-4o", "claude-3-5-sonnet-20241022")
"""
name = _get_model_name(model)
if name:
return name
# Try to infer from class name
class_name = model.__class__.__name__
if "GPT" in class_name or "OpenAI" in class_name:
return "gpt-4o" # Safe default for OpenAI
elif "Anthropic" in class_name or "Claude" in class_name:
return "claude-3-5-sonnet-20241022" # Safe default for Anthropic
elif "Google" in class_name or "Gemini" in class_name:
return "gemini-1.5-pro" # Safe default for Google
return "gpt-4o" # Ultimate fallback

View file

@ -0,0 +1,371 @@
"""Retriever integration for LangChain with intelligent document compression.
This module provides HeadroomDocumentCompressor, a LangChain BaseDocumentCompressor
that reduces retrieved documents based on relevance scoring while preserving
the most important information.
Example:
from langchain.retrievers import ContextualCompressionRetriever
from langchain_community.vectorstores import Chroma
from headroom.integrations import HeadroomDocumentCompressor
# Create vector store retriever
vectorstore = Chroma.from_documents(documents, embeddings)
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 50})
# Wrap with Headroom compression
compressor = HeadroomDocumentCompressor(max_documents=10)
retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=base_retriever,
)
# Retrieve - automatically keeps most relevant documents
docs = retriever.invoke("What is the capital of France?")
"""
from __future__ import annotations
import logging
import re
from collections.abc import Sequence
from dataclasses import dataclass
from typing import Any
# LangChain imports - these are optional dependencies
try:
from langchain_core.callbacks import Callbacks
from langchain_core.documents import Document
# BaseDocumentCompressor location varies by langchain version
try:
from langchain.retrievers.document_compressors import BaseDocumentCompressor
except ImportError:
try:
from langchain_core.documents.compressors import BaseDocumentCompressor
except ImportError:
# Fallback: create a minimal base class
class BaseDocumentCompressor: # type: ignore[no-redef]
"""Minimal base class for document compression."""
def compress_documents(
self, documents: Sequence[Any], query: str, callbacks: Any = None
) -> Sequence[Any]:
raise NotImplementedError
LANGCHAIN_AVAILABLE = True
except ImportError:
LANGCHAIN_AVAILABLE = False
BaseDocumentCompressor = object # type: ignore[misc,assignment]
Document = object # type: ignore[misc,assignment]
Callbacks = None # type: ignore[misc,assignment]
logger = logging.getLogger(__name__)
def _check_langchain_available() -> None:
"""Raise ImportError if LangChain is not installed."""
if not LANGCHAIN_AVAILABLE:
raise ImportError(
"LangChain is required for this integration. "
"Install with: pip install headroom[langchain] "
"or: pip install langchain-core"
)
@dataclass
class CompressionMetrics:
"""Metrics from document compression."""
documents_before: int
documents_after: int
documents_removed: int
relevance_scores: list[float]
class HeadroomDocumentCompressor(BaseDocumentCompressor):
"""Compresses retrieved documents based on relevance to query.
Uses BM25-style relevance scoring to keep only the most relevant
documents from a larger retrieval set. This allows you to retrieve
many documents initially (for recall) and then compress down to
the most relevant ones (for precision).
Works with LangChain's ContextualCompressionRetriever pattern.
Example:
from langchain.retrievers import ContextualCompressionRetriever
from headroom.integrations import HeadroomDocumentCompressor
compressor = HeadroomDocumentCompressor(
max_documents=10,
min_relevance=0.3,
)
retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=base_retriever, # Any retriever
)
# Retrieves top 10 most relevant docs
docs = retriever.invoke("What is Python?")
Attributes:
max_documents: Maximum documents to return
min_relevance: Minimum relevance score (0-1) to include
prefer_diverse: Whether to prefer diverse results
"""
max_documents: int = 10
min_relevance: float = 0.0
prefer_diverse: bool = False
def __init__(
self,
max_documents: int = 10,
min_relevance: float = 0.0,
prefer_diverse: bool = False,
**kwargs: Any,
):
"""Initialize HeadroomDocumentCompressor.
Args:
max_documents: Maximum number of documents to return. Default 10.
min_relevance: Minimum relevance score (0-1) for a document to
be included. Default 0.0 (no minimum).
prefer_diverse: If True, use MMR-style selection to prefer
diverse results over pure relevance. Default False.
**kwargs: Additional arguments for BaseDocumentCompressor.
"""
_check_langchain_available()
super().__init__(**kwargs)
self.max_documents = max_documents
self.min_relevance = min_relevance
self.prefer_diverse = prefer_diverse
self._last_metrics: CompressionMetrics | None = None
def compress_documents(
self,
documents: Sequence[Document],
query: str,
callbacks: Callbacks = None,
) -> Sequence[Document]:
"""Compress documents based on relevance to query.
Args:
documents: Documents to compress.
query: Query to score relevance against.
callbacks: LangChain callbacks (unused).
Returns:
Compressed list of most relevant documents.
"""
if not documents:
self._last_metrics = CompressionMetrics(
documents_before=0,
documents_after=0,
documents_removed=0,
relevance_scores=[],
)
return []
if len(documents) <= self.max_documents:
# No compression needed
scores = [self._score_document(doc, query) for doc in documents]
self._last_metrics = CompressionMetrics(
documents_before=len(documents),
documents_after=len(documents),
documents_removed=0,
relevance_scores=scores,
)
return list(documents)
# Score all documents
scored = [(doc, self._score_document(doc, query)) for doc in documents]
if self.prefer_diverse:
# Use MMR-style selection for diversity
selected = self._select_diverse(scored, query)
else:
# Sort by relevance score
scored.sort(key=lambda x: x[1], reverse=True)
selected = scored[: self.max_documents]
# Filter by minimum relevance
if self.min_relevance > 0:
selected = [(doc, score) for doc, score in selected if score >= self.min_relevance]
# Track metrics
final_docs = [doc for doc, _ in selected]
final_scores = [score for _, score in selected]
self._last_metrics = CompressionMetrics(
documents_before=len(documents),
documents_after=len(final_docs),
documents_removed=len(documents) - len(final_docs),
relevance_scores=final_scores,
)
logger.info(
f"HeadroomDocumentCompressor: {len(documents)} -> {len(final_docs)} documents "
f"(avg relevance: {sum(final_scores) / len(final_scores) if final_scores else 0:.2f})"
)
return final_docs
def _score_document(self, doc: Document, query: str) -> float:
"""Score a document's relevance to the query using BM25-style scoring.
Args:
doc: Document to score.
query: Query to compare against.
Returns:
Relevance score between 0 and 1.
"""
content = doc.page_content.lower()
query_lower = query.lower()
# Tokenize
query_terms = self._tokenize(query_lower)
doc_terms = self._tokenize(content)
if not query_terms or not doc_terms:
return 0.0
# BM25-style scoring
k1 = 1.5
b = 0.75
avg_dl = 100 # Assume average document length
doc_len = len(doc_terms)
term_freqs: dict[str, int] = {}
for term in doc_terms:
term_freqs[term] = term_freqs.get(term, 0) + 1
score = 0.0
for term in query_terms:
if term in term_freqs:
tf = term_freqs[term]
# Simplified BM25 (without IDF since we don't have corpus stats)
numerator = tf * (k1 + 1)
denominator = tf + k1 * (1 - b + b * (doc_len / avg_dl))
score += numerator / denominator
# Normalize to 0-1 range
max_possible = len(query_terms) * (k1 + 1)
normalized = score / max_possible if max_possible > 0 else 0.0
# Boost for exact phrase matches
if query_lower in content:
normalized = min(1.0, normalized + 0.3)
return min(1.0, normalized)
def _tokenize(self, text: str) -> list[str]:
"""Tokenize text into terms.
Args:
text: Text to tokenize.
Returns:
List of tokens.
"""
# Simple tokenization: split on non-alphanumeric, filter short terms
tokens = re.findall(r"\b\w+\b", text)
return [t for t in tokens if len(t) > 1]
def _select_diverse(
self, scored_docs: list[tuple[Document, float]], query: str
) -> list[tuple[Document, float]]:
"""Select diverse documents using MMR-style approach.
Balances relevance with diversity to avoid redundant results.
Args:
scored_docs: List of (document, relevance_score) tuples.
query: Original query.
Returns:
Selected documents with diversity considered.
"""
if not scored_docs:
return []
# Sort by initial relevance
scored_docs = sorted(scored_docs, key=lambda x: x[1], reverse=True)
# Start with most relevant
selected = [scored_docs[0]]
remaining = scored_docs[1:]
lambda_param = 0.5 # Balance between relevance and diversity
while len(selected) < self.max_documents and remaining:
best_score = -1.0
best_idx = 0
for i, (doc, rel_score) in enumerate(remaining):
# Calculate max similarity to already selected docs
max_sim = max(self._document_similarity(doc, sel_doc) for sel_doc, _ in selected)
# MMR score: lambda * relevance - (1-lambda) * max_similarity
mmr_score = lambda_param * rel_score - (1 - lambda_param) * max_sim
if mmr_score > best_score:
best_score = mmr_score
best_idx = i
selected.append(remaining[best_idx])
remaining.pop(best_idx)
return selected
def _document_similarity(self, doc1: Document, doc2: Document) -> float:
"""Calculate similarity between two documents.
Uses Jaccard similarity on terms for simplicity.
Args:
doc1: First document.
doc2: Second document.
Returns:
Similarity score between 0 and 1.
"""
terms1 = set(self._tokenize(doc1.page_content.lower()))
terms2 = set(self._tokenize(doc2.page_content.lower()))
if not terms1 or not terms2:
return 0.0
intersection = len(terms1 & terms2)
union = len(terms1 | terms2)
return intersection / union if union > 0 else 0.0
@property
def last_metrics(self) -> CompressionMetrics | None:
"""Get metrics from the last compression operation."""
return self._last_metrics
def get_compression_stats(self) -> dict[str, Any]:
"""Get statistics from the last compression.
Returns:
Dictionary with compression metrics, or empty if no compression yet.
"""
if self._last_metrics is None:
return {}
return {
"documents_before": self._last_metrics.documents_before,
"documents_after": self._last_metrics.documents_after,
"documents_removed": self._last_metrics.documents_removed,
"average_relevance": (
sum(self._last_metrics.relevance_scores) / len(self._last_metrics.relevance_scores)
if self._last_metrics.relevance_scores
else 0.0
),
}

View file

@ -0,0 +1,341 @@
"""Streaming metrics tracking for LangChain.
This module provides StreamingMetricsTracker for tracking output tokens
during streaming responses from LangChain models.
Example:
from langchain_openai import ChatOpenAI
from headroom.integrations import HeadroomChatModel, StreamingMetricsTracker
llm = HeadroomChatModel(ChatOpenAI(model="gpt-4o"))
tracker = StreamingMetricsTracker(model="gpt-4o")
for chunk in llm.stream("Tell me a story"):
tracker.add_chunk(chunk)
print(chunk.content, end="", flush=True)
print(f"\\nOutput tokens: {tracker.output_tokens}")
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from datetime import datetime
from typing import Any
# LangChain imports - these are optional dependencies
try:
from langchain_core.messages import AIMessageChunk
from langchain_core.outputs import ChatGenerationChunk
LANGCHAIN_AVAILABLE = True
except ImportError:
LANGCHAIN_AVAILABLE = False
AIMessageChunk = object # type: ignore[misc,assignment]
ChatGenerationChunk = object # type: ignore[misc,assignment]
from headroom.providers import OpenAIProvider
logger = logging.getLogger(__name__)
def _check_langchain_available() -> None:
"""Raise ImportError if LangChain is not installed."""
if not LANGCHAIN_AVAILABLE:
raise ImportError(
"LangChain is required for this integration. "
"Install with: pip install headroom[langchain] "
"or: pip install langchain-core"
)
@dataclass
class StreamingMetrics:
"""Metrics from a streaming response."""
output_tokens: int
chunk_count: int
content_length: int
start_time: datetime
end_time: datetime | None
duration_ms: float | None
def to_dict(self) -> dict[str, Any]:
"""Convert to dictionary."""
return {
"output_tokens": self.output_tokens,
"chunk_count": self.chunk_count,
"content_length": self.content_length,
"start_time": self.start_time.isoformat(),
"end_time": self.end_time.isoformat() if self.end_time else None,
"duration_ms": self.duration_ms,
}
class StreamingMetricsTracker:
"""Tracks output tokens and metrics during streaming.
Accumulates content from streaming chunks and provides accurate
token counting for the streamed output.
Example:
tracker = StreamingMetricsTracker(model="gpt-4o")
async for chunk in llm.astream(messages):
tracker.add_chunk(chunk)
print(chunk.content, end="")
print(f"\\nTokens: {tracker.output_tokens}")
print(f"Duration: {tracker.duration_ms}ms")
Attributes:
model: Model name for token counting
content: Accumulated content from all chunks
output_tokens: Estimated token count for output
chunk_count: Number of chunks received
"""
def __init__(
self,
model: str = "gpt-4o",
provider: Any = None,
):
"""Initialize StreamingMetricsTracker.
Args:
model: Model name for token counting. Default "gpt-4o".
provider: Headroom provider for token counting. Uses
OpenAIProvider if not specified.
"""
_check_langchain_available()
self._model = model
self._provider = provider or OpenAIProvider()
self._content = ""
self._chunk_count = 0
self._start_time: datetime | None = None
self._end_time: datetime | None = None
def add_chunk(self, chunk: Any) -> None:
"""Add a streaming chunk to the tracker.
Extracts content from various chunk types:
- AIMessageChunk
- ChatGenerationChunk
- dict with 'content' key
- string
Args:
chunk: Streaming chunk from LangChain.
"""
if self._start_time is None:
self._start_time = datetime.now()
self._chunk_count += 1
# Extract content from various chunk types
content = self._extract_content(chunk)
if content:
self._content += content
def _extract_content(self, chunk: Any) -> str:
"""Extract string content from a chunk.
Args:
chunk: Streaming chunk of various types.
Returns:
Extracted content string.
"""
# AIMessageChunk
if hasattr(chunk, "content"):
content = chunk.content
if isinstance(content, str):
return content
return str(content) if content else ""
# ChatGenerationChunk
if hasattr(chunk, "message") and hasattr(chunk.message, "content"):
content = chunk.message.content
if isinstance(content, str):
return content
return str(content) if content else ""
# dict
if isinstance(chunk, dict):
return str(chunk.get("content", ""))
# string
if isinstance(chunk, str):
return chunk
return ""
def finish(self) -> StreamingMetrics:
"""Mark streaming as complete and return final metrics.
Returns:
StreamingMetrics with final values.
"""
self._end_time = datetime.now()
duration_ms = None
if self._start_time:
duration_ms = (self._end_time - self._start_time).total_seconds() * 1000
return StreamingMetrics(
output_tokens=self.output_tokens,
chunk_count=self._chunk_count,
content_length=len(self._content),
start_time=self._start_time or self._end_time,
end_time=self._end_time,
duration_ms=duration_ms,
)
@property
def content(self) -> str:
"""Get accumulated content."""
return self._content
@property
def output_tokens(self) -> int:
"""Get estimated output token count."""
if not self._content:
return 0
token_counter = self._provider.get_token_counter(self._model)
return token_counter.count_text(self._content)
@property
def chunk_count(self) -> int:
"""Get number of chunks received."""
return self._chunk_count
@property
def duration_ms(self) -> float | None:
"""Get duration in milliseconds (after finish())."""
if self._start_time is None or self._end_time is None:
return None
return (self._end_time - self._start_time).total_seconds() * 1000
def reset(self) -> None:
"""Reset tracker for reuse."""
self._content = ""
self._chunk_count = 0
self._start_time = None
self._end_time = None
class StreamingMetricsCallback:
"""Context manager for tracking streaming metrics.
Provides a clean interface for tracking a complete streaming
response with automatic timing.
Example:
with StreamingMetricsCallback(model="gpt-4o") as tracker:
for chunk in llm.stream(messages):
tracker.add_chunk(chunk)
print(chunk.content, end="")
print(f"\\nMetrics: {tracker.metrics}")
Attributes:
tracker: The underlying StreamingMetricsTracker
metrics: Final metrics after context exit
"""
def __init__(self, model: str = "gpt-4o", provider: Any = None):
"""Initialize StreamingMetricsCallback.
Args:
model: Model name for token counting.
provider: Headroom provider for token counting.
"""
self._tracker = StreamingMetricsTracker(model=model, provider=provider)
self._metrics: StreamingMetrics | None = None
def __enter__(self) -> StreamingMetricsTracker:
"""Enter context, return tracker."""
return self._tracker
def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
"""Exit context, finalize metrics."""
self._metrics = self._tracker.finish()
@property
def tracker(self) -> StreamingMetricsTracker:
"""Get the tracker."""
return self._tracker
@property
def metrics(self) -> StreamingMetrics | None:
"""Get final metrics (after context exit)."""
return self._metrics
def track_streaming_response(
stream: Any,
model: str = "gpt-4o",
provider: Any = None,
) -> tuple[str, StreamingMetrics]:
"""Track a complete streaming response.
Convenience function that consumes a stream and returns the
accumulated content and metrics.
Args:
stream: Iterable of streaming chunks.
model: Model name for token counting.
provider: Headroom provider for token counting.
Returns:
Tuple of (accumulated_content, metrics).
Example:
content, metrics = track_streaming_response(
llm.stream(messages),
model="gpt-4o"
)
print(f"Content: {content}")
print(f"Tokens: {metrics.output_tokens}")
"""
tracker = StreamingMetricsTracker(model=model, provider=provider)
for chunk in stream:
tracker.add_chunk(chunk)
metrics = tracker.finish()
return tracker.content, metrics
async def track_async_streaming_response(
stream: Any,
model: str = "gpt-4o",
provider: Any = None,
) -> tuple[str, StreamingMetrics]:
"""Track a complete async streaming response.
Async version of track_streaming_response.
Args:
stream: Async iterable of streaming chunks.
model: Model name for token counting.
provider: Headroom provider for token counting.
Returns:
Tuple of (accumulated_content, metrics).
Example:
content, metrics = await track_async_streaming_response(
llm.astream(messages),
model="gpt-4o"
)
"""
tracker = StreamingMetricsTracker(model=model, provider=provider)
async for chunk in stream:
tracker.add_chunk(chunk)
metrics = tracker.finish()
return tracker.content, metrics

View file

@ -0,0 +1,37 @@
"""MCP (Model Context Protocol) integration for Headroom.
This package provides compression utilities for MCP tool results,
helping reduce context usage when tools return large outputs.
Example:
from headroom.integrations.mcp import compress_tool_result
# Compress large tool output
result = compress_tool_result(
tool_name="search",
result=large_json_result,
max_chars=5000,
)
"""
from .server import (
DEFAULT_MCP_PROFILES,
HeadroomMCPClientWrapper,
HeadroomMCPCompressor,
MCPCompressionResult,
MCPToolProfile,
compress_tool_result,
compress_tool_result_with_metrics,
create_headroom_mcp_proxy,
)
__all__ = [
"HeadroomMCPCompressor",
"HeadroomMCPClientWrapper",
"MCPCompressionResult",
"MCPToolProfile",
"compress_tool_result",
"compress_tool_result_with_metrics",
"create_headroom_mcp_proxy",
"DEFAULT_MCP_PROFILES",
]

View file

@ -88,7 +88,8 @@ def _get_llmlingua_compressor(model_name: str, device: str) -> Any:
from llmlingua import PromptCompressor
logger.info(
"Loading LLMLingua-2 model: %s on device: %s (this may take 10-30s on first run)",
"Loading LLMLingua-2 model: %s on device: %s "
"(this may take 10-30s on first run)",
model_name,
device,
)

View file

@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "headroom-ai"
version = "0.2.2"
version = "0.2.3"
description = "The Context Optimization Layer for LLM Applications - Cut costs by 50-90%"
readme = "README.md"
license = "Apache-2.0"

View file

@ -488,7 +488,7 @@ class TestOptimizeMessages:
"""Basic message optimization."""
from headroom.integrations import optimize_messages
with patch("headroom.integrations.langchain.TransformPipeline") as MockPipeline:
with patch("headroom.integrations.langchain.chat_model.TransformPipeline") as MockPipeline:
mock_instance = MagicMock()
mock_result = MagicMock()
mock_result.messages = [
@ -513,7 +513,7 @@ class TestOptimizeMessages:
config = HeadroomConfig(default_mode=HeadroomMode.AUDIT)
with patch("headroom.integrations.langchain.TransformPipeline") as MockPipeline:
with patch("headroom.integrations.langchain.chat_model.TransformPipeline") as MockPipeline:
mock_instance = MagicMock()
mock_result = MagicMock()
mock_result.messages = []
@ -547,7 +547,7 @@ class TestOptimizeMessages:
ToolMessage(content="Sunny", tool_call_id="1"),
]
with patch("headroom.integrations.langchain.TransformPipeline") as MockPipeline:
with patch("headroom.integrations.langchain.chat_model.TransformPipeline") as MockPipeline:
mock_instance = MagicMock()
mock_result = MagicMock()
mock_result.messages = [

View file

@ -0,0 +1,646 @@
"""Tests for extended LangChain integration modules.
Tests cover:
1. langchain_providers - Provider auto-detection
2. langchain_memory - HeadroomChatMessageHistory
3. langchain_retriever - HeadroomDocumentCompressor
4. langchain_agents - HeadroomToolWrapper
5. langchain_langsmith - LangSmith integration
6. langchain_streaming - Streaming metrics
"""
import json
from unittest.mock import MagicMock
import pytest
# Check if LangChain is available
try:
from langchain_core.documents import Document
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.tools import StructuredTool
LANGCHAIN_AVAILABLE = True
except ImportError:
LANGCHAIN_AVAILABLE = False
# Skip all tests if LangChain not installed
pytestmark = pytest.mark.skipif(not LANGCHAIN_AVAILABLE, reason="LangChain not installed")
class TestProviderDetection:
"""Tests for langchain_providers module."""
def test_detect_openai_provider(self):
"""Detect OpenAI from ChatOpenAI class."""
from headroom.integrations.langchain.providers import detect_provider
mock_model = MagicMock()
mock_model.__class__.__name__ = "ChatOpenAI"
mock_model.__class__.__module__ = "langchain_openai.chat_models"
provider = detect_provider(mock_model)
assert provider == "openai"
def test_detect_anthropic_provider(self):
"""Detect Anthropic from ChatAnthropic class."""
from headroom.integrations.langchain.providers import detect_provider
mock_model = MagicMock()
mock_model.__class__.__name__ = "ChatAnthropic"
mock_model.__class__.__module__ = "langchain_anthropic.chat_models"
provider = detect_provider(mock_model)
assert provider == "anthropic"
def test_detect_google_provider(self):
"""Detect Google from ChatGoogleGenerativeAI class."""
from headroom.integrations.langchain.providers import detect_provider
mock_model = MagicMock()
mock_model.__class__.__name__ = "ChatGoogleGenerativeAI"
mock_model.__class__.__module__ = "langchain_google_genai"
provider = detect_provider(mock_model)
assert provider == "google"
def test_detect_fallback_to_openai(self):
"""Fall back to OpenAI for unknown models."""
from headroom.integrations.langchain.providers import detect_provider
mock_model = MagicMock()
mock_model.__class__.__name__ = "CustomChatModel"
mock_model.__class__.__module__ = "my_custom_module"
provider = detect_provider(mock_model)
assert provider == "openai"
def test_detect_from_model_name_claude(self):
"""Detect Anthropic from model name containing 'claude'."""
from headroom.integrations.langchain.providers import detect_provider
mock_model = MagicMock()
mock_model.__class__.__name__ = "CustomModel"
mock_model.__class__.__module__ = "custom"
mock_model.model_name = "claude-3-5-sonnet-20241022"
provider = detect_provider(mock_model)
assert provider == "anthropic"
def test_get_headroom_provider_openai(self):
"""Get OpenAIProvider for OpenAI model."""
from headroom.integrations.langchain.providers import get_headroom_provider
from headroom.providers import OpenAIProvider
mock_model = MagicMock()
mock_model.__class__.__name__ = "ChatOpenAI"
mock_model.__class__.__module__ = "langchain_openai"
provider = get_headroom_provider(mock_model)
assert isinstance(provider, OpenAIProvider)
def test_get_headroom_provider_anthropic(self):
"""Get AnthropicProvider for Anthropic model."""
from headroom.integrations.langchain.providers import get_headroom_provider
from headroom.providers import AnthropicProvider
mock_model = MagicMock()
mock_model.__class__.__name__ = "ChatAnthropic"
mock_model.__class__.__module__ = "langchain_anthropic"
provider = get_headroom_provider(mock_model)
assert isinstance(provider, AnthropicProvider)
def test_get_model_name_from_langchain(self):
"""Extract model name from LangChain model."""
from headroom.integrations.langchain.providers import get_model_name_from_langchain
mock_model = MagicMock()
mock_model.model_name = "gpt-4o"
name = get_model_name_from_langchain(mock_model)
assert name == "gpt-4o"
def test_get_model_name_fallback(self):
"""Fall back when model name not available."""
from headroom.integrations.langchain.providers import get_model_name_from_langchain
mock_model = MagicMock(spec=[])
mock_model.__class__.__name__ = "ChatOpenAI"
name = get_model_name_from_langchain(mock_model)
assert name == "gpt-4o" # Default for OpenAI
class TestHeadroomChatMessageHistory:
"""Tests for HeadroomChatMessageHistory memory wrapper."""
def test_init(self):
"""Initialize with base history."""
from headroom.integrations.langchain.memory import HeadroomChatMessageHistory
mock_history = MagicMock()
mock_history.messages = []
wrapper = HeadroomChatMessageHistory(
mock_history,
compress_threshold_tokens=4000,
keep_recent_turns=5,
)
assert wrapper._base is mock_history
assert wrapper._threshold == 4000
assert wrapper._keep_recent_turns == 5
def test_messages_passthrough_under_threshold(self):
"""Messages pass through when under threshold."""
from headroom.integrations.langchain.memory import HeadroomChatMessageHistory
mock_history = MagicMock()
mock_history.messages = [
HumanMessage(content="Hello"),
AIMessage(content="Hi there!"),
]
wrapper = HeadroomChatMessageHistory(
mock_history,
compress_threshold_tokens=10000, # High threshold
)
messages = wrapper.messages
assert len(messages) == 2
assert messages[0].content == "Hello"
def test_add_message_delegates(self):
"""add_message delegates to base history."""
from headroom.integrations.langchain.memory import HeadroomChatMessageHistory
mock_history = MagicMock()
mock_history.messages = []
wrapper = HeadroomChatMessageHistory(mock_history)
message = HumanMessage(content="Test")
wrapper.add_message(message)
mock_history.add_message.assert_called_once_with(message)
def test_clear_delegates(self):
"""clear delegates to base history."""
from headroom.integrations.langchain.memory import HeadroomChatMessageHistory
mock_history = MagicMock()
mock_history.messages = []
wrapper = HeadroomChatMessageHistory(mock_history)
wrapper.clear()
mock_history.clear.assert_called_once()
def test_get_compression_stats(self):
"""Get compression statistics."""
from headroom.integrations.langchain.memory import HeadroomChatMessageHistory
mock_history = MagicMock()
mock_history.messages = []
wrapper = HeadroomChatMessageHistory(mock_history)
stats = wrapper.get_compression_stats()
assert "compression_count" in stats
assert "total_tokens_saved" in stats
assert stats["compression_count"] == 0
class TestHeadroomDocumentCompressor:
"""Tests for HeadroomDocumentCompressor retriever integration."""
def test_init(self):
"""Initialize with defaults."""
from headroom.integrations.langchain.retriever import HeadroomDocumentCompressor
compressor = HeadroomDocumentCompressor()
assert compressor.max_documents == 10
assert compressor.min_relevance == 0.0
assert compressor.prefer_diverse is False
def test_init_custom(self):
"""Initialize with custom settings."""
from headroom.integrations.langchain.retriever import HeadroomDocumentCompressor
compressor = HeadroomDocumentCompressor(
max_documents=5,
min_relevance=0.5,
prefer_diverse=True,
)
assert compressor.max_documents == 5
assert compressor.min_relevance == 0.5
assert compressor.prefer_diverse is True
def test_compress_passthrough_under_limit(self):
"""Pass through when under max_documents."""
from headroom.integrations.langchain.retriever import HeadroomDocumentCompressor
compressor = HeadroomDocumentCompressor(max_documents=10)
docs = [
Document(page_content="Python is a programming language."),
Document(page_content="JavaScript runs in browsers."),
]
result = compressor.compress_documents(docs, "What is Python?")
assert len(result) == 2
def test_compress_reduces_to_max(self):
"""Compress when over max_documents."""
from headroom.integrations.langchain.retriever import HeadroomDocumentCompressor
compressor = HeadroomDocumentCompressor(max_documents=2)
docs = [
Document(page_content="Python is a programming language."),
Document(page_content="Java is also a language."),
Document(page_content="Weather today is sunny."),
Document(page_content="Cats are cute animals."),
]
result = compressor.compress_documents(docs, "programming language")
assert len(result) == 2
def test_compress_prefers_relevant(self):
"""Keep most relevant documents."""
from headroom.integrations.langchain.retriever import HeadroomDocumentCompressor
compressor = HeadroomDocumentCompressor(max_documents=1)
docs = [
Document(page_content="Weather today is sunny."),
Document(page_content="Python programming tutorial basics."),
Document(page_content="Cats are cute animals."),
]
result = compressor.compress_documents(docs, "Python tutorial")
assert len(result) == 1
assert "Python" in result[0].page_content
def test_metrics_tracked(self):
"""Compression metrics are tracked."""
from headroom.integrations.langchain.retriever import HeadroomDocumentCompressor
compressor = HeadroomDocumentCompressor(max_documents=2)
docs = [
Document(page_content="Doc 1"),
Document(page_content="Doc 2"),
Document(page_content="Doc 3"),
]
compressor.compress_documents(docs, "query")
metrics = compressor.last_metrics
assert metrics is not None
assert metrics.documents_before == 3
assert metrics.documents_after == 2
assert metrics.documents_removed == 1
def test_get_compression_stats(self):
"""Get compression statistics."""
from headroom.integrations.langchain.retriever import HeadroomDocumentCompressor
compressor = HeadroomDocumentCompressor(max_documents=1)
docs = [Document(page_content="A"), Document(page_content="B")]
compressor.compress_documents(docs, "A")
stats = compressor.get_compression_stats()
assert "documents_before" in stats
assert "documents_after" in stats
assert "average_relevance" in stats
class TestHeadroomToolWrapper:
"""Tests for HeadroomToolWrapper agent integration."""
def test_init(self):
"""Initialize wrapper."""
from headroom.integrations.langchain.agents import HeadroomToolWrapper
mock_tool = MagicMock()
mock_tool.name = "test_tool"
mock_tool.description = "A test tool"
wrapper = HeadroomToolWrapper(mock_tool)
assert wrapper.name == "test_tool"
assert wrapper.description == "A test tool"
def test_call_passthrough_small_output(self):
"""Small outputs pass through without compression."""
from headroom.integrations.langchain.agents import HeadroomToolWrapper
mock_tool = MagicMock()
mock_tool.name = "test"
mock_tool.description = "test"
mock_tool.invoke.return_value = "small result"
wrapper = HeadroomToolWrapper(mock_tool, min_chars_to_compress=1000)
result = wrapper("query")
assert result == "small result"
def test_call_compresses_large_json(self):
"""Large JSON outputs get compressed."""
from headroom.integrations.langchain.agents import HeadroomToolWrapper
mock_tool = MagicMock()
mock_tool.name = "search"
mock_tool.description = "search"
# Large JSON output
large_output = json.dumps([{"id": i, "data": "x" * 100} for i in range(50)])
mock_tool.invoke.return_value = large_output
wrapper = HeadroomToolWrapper(mock_tool, min_chars_to_compress=100)
result = wrapper("query")
# Should be smaller after compression
assert len(result) <= len(large_output)
def test_as_langchain_tool(self):
"""Convert to LangChain tool."""
from headroom.integrations.langchain.agents import HeadroomToolWrapper
mock_tool = MagicMock()
mock_tool.name = "test"
mock_tool.description = "test tool"
mock_tool.invoke.return_value = "result"
wrapper = HeadroomToolWrapper(mock_tool)
lc_tool = wrapper.as_langchain_tool()
assert isinstance(lc_tool, StructuredTool)
assert lc_tool.name == "test"
def test_wrap_tools_with_headroom(self):
"""Wrap multiple tools at once."""
from headroom.integrations.langchain.agents import wrap_tools_with_headroom
tools = []
for i in range(3):
mock = MagicMock()
mock.name = f"tool_{i}"
mock.description = f"Tool {i}"
mock.invoke.return_value = "result"
tools.append(mock)
wrapped = wrap_tools_with_headroom(tools)
assert len(wrapped) == 3
assert all(isinstance(t, StructuredTool) for t in wrapped)
def test_metrics_collector(self):
"""Tool metrics are collected."""
from headroom.integrations.langchain.agents import (
HeadroomToolWrapper,
ToolMetricsCollector,
)
collector = ToolMetricsCollector()
mock_tool = MagicMock()
mock_tool.name = "test"
mock_tool.description = "test"
mock_tool.invoke.return_value = "result"
wrapper = HeadroomToolWrapper(mock_tool, metrics_collector=collector)
wrapper("query")
assert len(collector.metrics) == 1
assert collector.metrics[0].tool_name == "test"
class TestHeadroomLangSmithCallbackHandler:
"""Tests for LangSmith integration."""
def test_init(self):
"""Initialize handler."""
from headroom.integrations.langchain.langsmith import (
HeadroomLangSmithCallbackHandler,
)
handler = HeadroomLangSmithCallbackHandler(auto_update_runs=False)
assert handler._auto_update is False
assert handler._pending_metrics == {}
def test_set_headroom_metrics(self):
"""Set metrics for a run."""
from headroom.integrations.langchain.langsmith import (
HeadroomLangSmithCallbackHandler,
)
handler = HeadroomLangSmithCallbackHandler(auto_update_runs=False)
handler.set_headroom_metrics(
run_id="test-run-123",
tokens_before=1000,
tokens_after=800,
transforms_applied=["smart_crusher"],
)
assert "test-run-123" in handler._pending_metrics
metrics = handler._pending_metrics["test-run-123"]
assert metrics.tokens_before == 1000
assert metrics.tokens_after == 800
assert metrics.tokens_saved == 200
assert metrics.savings_percent == 20.0
def test_get_run_metrics(self):
"""Get metrics for a specific run."""
from headroom.integrations.langchain.langsmith import (
HeadroomLangSmithCallbackHandler,
)
handler = HeadroomLangSmithCallbackHandler(auto_update_runs=False)
handler._run_metrics["run-1"] = {"headroom.tokens_saved": 100}
metrics = handler.get_run_metrics("run-1")
assert metrics["headroom.tokens_saved"] == 100
def test_get_summary(self):
"""Get summary statistics."""
from headroom.integrations.langchain.langsmith import (
HeadroomLangSmithCallbackHandler,
)
handler = HeadroomLangSmithCallbackHandler(auto_update_runs=False)
handler._run_metrics = {
"run-1": {"headroom.tokens_saved": 100, "headroom.savings_percent": 20},
"run-2": {"headroom.tokens_saved": 200, "headroom.savings_percent": 30},
}
summary = handler.get_summary()
assert summary["total_runs"] == 2
assert summary["total_tokens_saved"] == 300
assert summary["average_savings_percent"] == 25.0
def test_reset(self):
"""Reset clears all metrics."""
from headroom.integrations.langchain.langsmith import (
HeadroomLangSmithCallbackHandler,
)
handler = HeadroomLangSmithCallbackHandler(auto_update_runs=False)
handler._run_metrics = {"run-1": {}}
handler._pending_metrics = {"run-2": MagicMock()}
handler.reset()
assert handler._run_metrics == {}
assert handler._pending_metrics == {}
class TestStreamingMetricsTracker:
"""Tests for streaming metrics tracking."""
def test_init(self):
"""Initialize tracker."""
from headroom.integrations.langchain.streaming import StreamingMetricsTracker
tracker = StreamingMetricsTracker(model="gpt-4o")
assert tracker._model == "gpt-4o"
assert tracker._content == ""
assert tracker._chunk_count == 0
def test_add_chunk_string(self):
"""Add string chunks."""
from headroom.integrations.langchain.streaming import StreamingMetricsTracker
tracker = StreamingMetricsTracker()
tracker.add_chunk("Hello ")
tracker.add_chunk("world!")
assert tracker.content == "Hello world!"
assert tracker.chunk_count == 2
def test_add_chunk_with_content_attr(self):
"""Add chunks with content attribute."""
from headroom.integrations.langchain.streaming import StreamingMetricsTracker
tracker = StreamingMetricsTracker()
chunk1 = MagicMock()
chunk1.content = "Hello "
chunk2 = MagicMock()
chunk2.content = "world!"
tracker.add_chunk(chunk1)
tracker.add_chunk(chunk2)
assert tracker.content == "Hello world!"
def test_output_tokens(self):
"""Count output tokens."""
from headroom.integrations.langchain.streaming import StreamingMetricsTracker
tracker = StreamingMetricsTracker(model="gpt-4o")
tracker.add_chunk("Hello world, this is a test message.")
tokens = tracker.output_tokens
assert tokens > 0
def test_finish(self):
"""Finish tracking and get metrics."""
from headroom.integrations.langchain.streaming import StreamingMetricsTracker
tracker = StreamingMetricsTracker()
tracker.add_chunk("Test content")
metrics = tracker.finish()
assert metrics.chunk_count == 1
assert metrics.content_length == len("Test content")
assert metrics.duration_ms is not None
assert metrics.end_time is not None
def test_reset(self):
"""Reset tracker for reuse."""
from headroom.integrations.langchain.streaming import StreamingMetricsTracker
tracker = StreamingMetricsTracker()
tracker.add_chunk("Content")
tracker.finish()
tracker.reset()
assert tracker.content == ""
assert tracker.chunk_count == 0
def test_streaming_metrics_callback(self):
"""Test context manager interface."""
from headroom.integrations.langchain.streaming import StreamingMetricsCallback
with StreamingMetricsCallback(model="gpt-4o") as tracker:
tracker.add_chunk("Hello")
tracker.add_chunk(" world")
# After context exit, metrics should be available
# (accessed via the callback object, not the tracker)
def test_track_streaming_response(self):
"""Track a complete streaming response."""
from headroom.integrations.langchain.streaming import track_streaming_response
chunks = ["Hello ", "world", "!"]
content, metrics = track_streaming_response(iter(chunks), model="gpt-4o")
assert content == "Hello world!"
assert metrics.chunk_count == 3
class TestAutoDetectProviderInChatModel:
"""Tests for auto_detect_provider in HeadroomChatModel."""
def test_auto_detect_enabled_by_default(self):
"""auto_detect_provider is True by default."""
from headroom.integrations import HeadroomChatModel
mock_model = MagicMock()
mock_model._llm_type = "test"
mock_model._identifying_params = {}
mock_model.__class__.__name__ = "ChatOpenAI"
mock_model.__class__.__module__ = "langchain_openai"
model = HeadroomChatModel(mock_model)
assert model.auto_detect_provider is True
def test_auto_detect_can_be_disabled(self):
"""auto_detect_provider can be set to False."""
from headroom.integrations import HeadroomChatModel
mock_model = MagicMock()
mock_model._llm_type = "test"
mock_model._identifying_params = {}
model = HeadroomChatModel(mock_model, auto_detect_provider=False)
assert model.auto_detect_provider is False
def test_pipeline_uses_detected_provider(self):
"""Pipeline uses auto-detected provider."""
from headroom.integrations import HeadroomChatModel
from headroom.providers import AnthropicProvider
mock_model = MagicMock()
mock_model._llm_type = "test"
mock_model._identifying_params = {}
mock_model.__class__.__name__ = "ChatAnthropic"
mock_model.__class__.__module__ = "langchain_anthropic"
model = HeadroomChatModel(mock_model)
_ = model.pipeline # Force lazy init
assert isinstance(model._provider, AnthropicProvider)

View file

211
uv.lock generated
View file

@ -6,6 +6,25 @@ resolution-markers = [
"python_full_version < '3.11'",
]
[[package]]
name = "accelerate"
version = "1.12.0"
source = { registry = "https://pypi.netflix.net/simple" }
dependencies = [
{ name = "huggingface-hub" },
{ name = "numpy", version = "2.2.6", source = { registry = "https://pypi.netflix.net/simple" }, marker = "python_full_version < '3.11'" },
{ name = "numpy", version = "2.4.0", source = { registry = "https://pypi.netflix.net/simple" }, marker = "python_full_version >= '3.11'" },
{ name = "packaging" },
{ name = "psutil" },
{ name = "pyyaml" },
{ name = "safetensors" },
{ name = "torch" },
]
sdist = { url = "https://pypi.netflix.net/packages/19372078203/accelerate-1.12.0.tar.gz", hash = "sha256:70988c352feb481887077d2ab845125024b2a137a5090d6d7a32b57d03a45df6", size = 398399 }
wheels = [
{ url = "https://pypi.netflix.net/packages/19372078202/accelerate-1.12.0-py3-none-any.whl", hash = "sha256:3e2091cd341423207e2f084a6654b1efcd250dc326f2a37d6dde446e07cabb11", size = 380935 },
]
[[package]]
name = "annotated-doc"
version = "0.0.4"
@ -362,8 +381,8 @@ wheels = [
]
[[package]]
name = "headroom"
version = "0.2.0"
name = "headroom-ai"
version = "0.2.3"
source = { editable = "." }
dependencies = [
{ name = "pydantic" },
@ -375,11 +394,18 @@ all = [
{ name = "fastapi" },
{ name = "httpx" },
{ name = "jinja2" },
{ name = "llmlingua" },
{ name = "numpy", version = "2.2.6", source = { registry = "https://pypi.netflix.net/simple" }, marker = "python_full_version < '3.11'" },
{ name = "numpy", version = "2.4.0", source = { registry = "https://pypi.netflix.net/simple" }, marker = "python_full_version >= '3.11'" },
{ name = "sentence-transformers" },
{ name = "torch" },
{ name = "transformers" },
{ name = "tree-sitter-language-pack" },
{ name = "uvicorn" },
]
code = [
{ name = "tree-sitter-language-pack" },
]
dev = [
{ name = "anthropic" },
{ name = "mypy" },
@ -389,6 +415,11 @@ dev = [
{ name = "pytest-cov" },
{ name = "ruff" },
]
llmlingua = [
{ name = "llmlingua" },
{ name = "torch" },
{ name = "transformers" },
]
proxy = [
{ name = "fastapi" },
{ name = "httpx" },
@ -407,9 +438,10 @@ reports = [
requires-dist = [
{ name = "anthropic", marker = "extra == 'dev'", specifier = ">=0.18.0" },
{ name = "fastapi", marker = "extra == 'proxy'", specifier = ">=0.100.0" },
{ name = "headroom", extras = ["relevance", "proxy", "reports"], marker = "extra == 'all'" },
{ name = "headroom-ai", extras = ["relevance", "proxy", "reports", "llmlingua", "code"], marker = "extra == 'all'" },
{ name = "httpx", marker = "extra == 'proxy'", specifier = ">=0.24.0" },
{ name = "jinja2", marker = "extra == 'reports'", specifier = ">=3.0.0" },
{ name = "llmlingua", marker = "extra == 'llmlingua'", specifier = ">=0.2.0" },
{ name = "mypy", marker = "extra == 'dev'", specifier = ">=1.0.0" },
{ name = "numpy", marker = "extra == 'relevance'", specifier = ">=1.24.0" },
{ name = "openai", marker = "extra == 'dev'", specifier = ">=1.0.0" },
@ -420,6 +452,9 @@ requires-dist = [
{ name = "ruff", marker = "extra == 'dev'", specifier = ">=0.1.0" },
{ name = "sentence-transformers", marker = "extra == 'relevance'", specifier = ">=2.2.0" },
{ name = "tiktoken", specifier = ">=0.5.0" },
{ name = "torch", marker = "extra == 'llmlingua'", specifier = ">=2.0.0" },
{ name = "transformers", marker = "extra == 'llmlingua'", specifier = ">=4.30.0" },
{ name = "tree-sitter-language-pack", marker = "extra == 'code'", specifier = ">=0.10.0" },
{ name = "uvicorn", marker = "extra == 'proxy'", specifier = ">=0.23.0" },
]
@ -708,6 +743,24 @@ wheels = [
{ url = "https://pypi.netflix.net/packages/19544946795/librt-0.7.7-cp314-cp314t-win_arm64.whl", hash = "sha256:142c2cd91794b79fd0ce113bd658993b7ede0fe93057668c2f98a45ca00b7e91", size = 39724 },
]
[[package]]
name = "llmlingua"
version = "0.2.2"
source = { registry = "https://pypi.netflix.net/simple" }
dependencies = [
{ name = "accelerate" },
{ name = "nltk" },
{ name = "numpy", version = "2.2.6", source = { registry = "https://pypi.netflix.net/simple" }, marker = "python_full_version < '3.11'" },
{ name = "numpy", version = "2.4.0", source = { registry = "https://pypi.netflix.net/simple" }, marker = "python_full_version >= '3.11'" },
{ name = "tiktoken" },
{ name = "torch" },
{ name = "transformers" },
]
sdist = { url = "https://pypi.netflix.net/packages/19606733170/llmlingua-0.2.2.tar.gz", hash = "sha256:1a0caedd8d5a65512a85dadb6bfda6f5b3c4b45e5cb9e7b1c6009573f9058572", size = 59753 }
wheels = [
{ url = "https://pypi.netflix.net/packages/19606733169/llmlingua-0.2.2-py3-none-any.whl", hash = "sha256:da55137efe0db78063b3395396efe8a0dcfe4ae5a09aea0d503c34b7bf1d800c", size = 30536 },
]
[[package]]
name = "markupsafe"
version = "3.0.3"
@ -882,6 +935,21 @@ wheels = [
{ url = "https://pypi.netflix.net/packages/19441125158/networkx-3.6.1-py3-none-any.whl", hash = "sha256:d47fbf302e7d9cbbb9e2555a0d267983d2aa476bac30e90dfbe5669bd57f3762", size = 2068504 },
]
[[package]]
name = "nltk"
version = "3.9.2"
source = { registry = "https://pypi.netflix.net/simple" }
dependencies = [
{ name = "click" },
{ name = "joblib" },
{ name = "regex" },
{ name = "tqdm" },
]
sdist = { url = "https://pypi.netflix.net/packages/19152095449/nltk-3.9.2.tar.gz", hash = "sha256:0f409e9b069ca4177c1903c3e843eef90c7e92992fa4931ae607da6de49e1419", size = 2887629 }
wheels = [
{ url = "https://pypi.netflix.net/packages/19152095448/nltk-3.9.2-py3-none-any.whl", hash = "sha256:1e209d2b3009110635ed9709a67a1a3e33a10f799490fa71cf4bec218c11c88a", size = 1513404 },
]
[[package]]
name = "numpy"
version = "2.2.6"
@ -1225,6 +1293,34 @@ wheels = [
{ url = "https://pypi.netflix.net/packages/18687957486/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538 },
]
[[package]]
name = "psutil"
version = "7.2.1"
source = { registry = "https://pypi.netflix.net/simple" }
sdist = { url = "https://pypi.netflix.net/packages/19533562506/psutil-7.2.1.tar.gz", hash = "sha256:f7583aec590485b43ca601dd9cea0dcd65bd7bb21d30ef4ddbf4ea6b5ed1bdd3", size = 490253 }
wheels = [
{ url = "https://pypi.netflix.net/packages/19533562496/psutil-7.2.1-cp313-cp313t-macosx_10_13_x86_64.whl", hash = "sha256:ba9f33bb525b14c3ea563b2fd521a84d2fa214ec59e3e6a2858f78d0844dd60d", size = 129624 },
{ url = "https://pypi.netflix.net/packages/19533562497/psutil-7.2.1-cp313-cp313t-macosx_11_0_arm64.whl", hash = "sha256:81442dac7abfc2f4f4385ea9e12ddf5a796721c0f6133260687fec5c3780fa49", size = 130132 },
{ url = "https://pypi.netflix.net/packages/19533562498/psutil-7.2.1-cp313-cp313t-manylinux2010_x86_64.manylinux_2_12_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:ea46c0d060491051d39f0d2cff4f98d5c72b288289f57a21556cc7d504db37fc", size = 180612 },
{ url = "https://pypi.netflix.net/packages/19533562499/psutil-7.2.1-cp313-cp313t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:35630d5af80d5d0d49cfc4d64c1c13838baf6717a13effb35869a5919b854cdf", size = 183201 },
{ url = "https://pypi.netflix.net/packages/19533562500/psutil-7.2.1-cp313-cp313t-win_amd64.whl", hash = "sha256:923f8653416604e356073e6e0bccbe7c09990acef442def2f5640dd0faa9689f", size = 139081 },
{ url = "https://pypi.netflix.net/packages/19533562501/psutil-7.2.1-cp313-cp313t-win_arm64.whl", hash = "sha256:cfbe6b40ca48019a51827f20d830887b3107a74a79b01ceb8cc8de4ccb17b672", size = 134767 },
{ url = "https://pypi.netflix.net/packages/19533562502/psutil-7.2.1-cp314-cp314t-macosx_10_15_x86_64.whl", hash = "sha256:494c513ccc53225ae23eec7fe6e1482f1b8a44674241b54561f755a898650679", size = 129716 },
{ url = "https://pypi.netflix.net/packages/19533562503/psutil-7.2.1-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:3fce5f92c22b00cdefd1645aa58ab4877a01679e901555067b1bd77039aa589f", size = 130133 },
{ url = "https://pypi.netflix.net/packages/19533562504/psutil-7.2.1-cp314-cp314t-manylinux2010_x86_64.manylinux_2_12_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:93f3f7b0bb07711b49626e7940d6fe52aa9940ad86e8f7e74842e73189712129", size = 181518 },
{ url = "https://pypi.netflix.net/packages/19533562505/psutil-7.2.1-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:d34d2ca888208eea2b5c68186841336a7f5e0b990edec929be909353a202768a", size = 184348 },
{ url = "https://pypi.netflix.net/packages/19533563921/psutil-7.2.1-cp314-cp314t-win_amd64.whl", hash = "sha256:2ceae842a78d1603753561132d5ad1b2f8a7979cb0c283f5b52fb4e6e14b1a79", size = 140400 },
{ url = "https://pypi.netflix.net/packages/19533563922/psutil-7.2.1-cp314-cp314t-win_arm64.whl", hash = "sha256:08a2f175e48a898c8eb8eace45ce01777f4785bc744c90aa2cc7f2fa5462a266", size = 135430 },
{ url = "https://pypi.netflix.net/packages/19533563923/psutil-7.2.1-cp36-abi3-macosx_10_9_x86_64.whl", hash = "sha256:b2e953fcfaedcfbc952b44744f22d16575d3aa78eb4f51ae74165b4e96e55f42", size = 128137 },
{ url = "https://pypi.netflix.net/packages/19533563924/psutil-7.2.1-cp36-abi3-macosx_11_0_arm64.whl", hash = "sha256:05cc68dbb8c174828624062e73078e7e35406f4ca2d0866c272c2410d8ef06d1", size = 128947 },
{ url = "https://pypi.netflix.net/packages/19533563925/psutil-7.2.1-cp36-abi3-manylinux2010_x86_64.manylinux_2_12_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:5e38404ca2bb30ed7267a46c02f06ff842e92da3bb8c5bfdadbd35a5722314d8", size = 154694 },
{ url = "https://pypi.netflix.net/packages/19533563926/psutil-7.2.1-cp36-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:ab2b98c9fc19f13f59628d94df5cc4cc4844bc572467d113a8b517d634e362c6", size = 156136 },
{ url = "https://pypi.netflix.net/packages/19533563927/psutil-7.2.1-cp36-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:f78baafb38436d5a128f837fab2d92c276dfb48af01a240b861ae02b2413ada8", size = 148108 },
{ url = "https://pypi.netflix.net/packages/19533565348/psutil-7.2.1-cp36-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:99a4cd17a5fdd1f3d014396502daa70b5ec21bf4ffe38393e152f8e449757d67", size = 147402 },
{ url = "https://pypi.netflix.net/packages/19533565349/psutil-7.2.1-cp37-abi3-win_amd64.whl", hash = "sha256:b1b0671619343aa71c20ff9767eced0483e4fc9e1f489d50923738caf6a03c17", size = 136938 },
{ url = "https://pypi.netflix.net/packages/19533565350/psutil-7.2.1-cp37-abi3-win_arm64.whl", hash = "sha256:0d67c1822c355aa6f7314d92018fb4268a76668a536f133599b91edd48759442", size = 133836 },
]
[[package]]
name = "pydantic"
version = "2.12.5"
@ -2193,6 +2289,115 @@ wheels = [
{ url = "https://pypi.netflix.net/packages/19387983499/transformers-4.57.3-py3-none-any.whl", hash = "sha256:c77d353a4851b1880191603d36acb313411d3577f6e2897814f333841f7003f4", size = 11993463 },
]
[[package]]
name = "tree-sitter"
version = "0.25.2"
source = { registry = "https://pypi.netflix.net/simple" }
sdist = { url = "https://pypi.netflix.net/packages/19129803294/tree-sitter-0.25.2.tar.gz", hash = "sha256:fe43c158555da46723b28b52e058ad444195afd1db3ca7720c59a254544e9c20", size = 177961 }
wheels = [
{ url = "https://pypi.netflix.net/packages/19129800490/tree_sitter-0.25.2-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:72a510931c3c25f134aac2daf4eb4feca99ffe37a35896d7150e50ac3eee06c7", size = 146749 },
{ url = "https://pypi.netflix.net/packages/19129800491/tree_sitter-0.25.2-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:44488e0e78146f87baaa009736886516779253d6d6bac3ef636ede72bc6a8234", size = 137766 },
{ url = "https://pypi.netflix.net/packages/19129800492/tree_sitter-0.25.2-cp310-cp310-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:c2f8e7d6b2f8489d4a9885e3adcaef4bc5ff0a275acd990f120e29c4ab3395c5", size = 599809 },
{ url = "https://pypi.netflix.net/packages/19129800493/tree_sitter-0.25.2-cp310-cp310-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:20b570690f87f1da424cd690e51cc56728d21d63f4abd4b326d382a30353acc7", size = 627676 },
{ url = "https://pypi.netflix.net/packages/19129800494/tree_sitter-0.25.2-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:a0ec41b895da717bc218a42a3a7a0bfcfe9a213d7afaa4255353901e0e21f696", size = 624281 },
{ url = "https://pypi.netflix.net/packages/19129800495/tree_sitter-0.25.2-cp310-cp310-win_amd64.whl", hash = "sha256:7712335855b2307a21ae86efe949c76be36c6068d76df34faa27ce9ee40ff444", size = 127295 },
{ url = "https://pypi.netflix.net/packages/19129800496/tree_sitter-0.25.2-cp310-cp310-win_arm64.whl", hash = "sha256:a925364eb7fbb9cdce55a9868f7525a1905af512a559303bd54ef468fd88cb37", size = 113991 },
{ url = "https://pypi.netflix.net/packages/19129800497/tree_sitter-0.25.2-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:b8ca72d841215b6573ed0655b3a5cd1133f9b69a6fa561aecad40dca9029d75b", size = 146752 },
{ url = "https://pypi.netflix.net/packages/19129800498/tree_sitter-0.25.2-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:cc0351cfe5022cec5a77645f647f92a936b38850346ed3f6d6babfbeeeca4d26", size = 137765 },
{ url = "https://pypi.netflix.net/packages/19129800499/tree_sitter-0.25.2-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:1799609636c0193e16c38f366bda5af15b1ce476df79ddaae7dd274df9e44266", size = 604643 },
{ url = "https://pypi.netflix.net/packages/19129800500/tree_sitter-0.25.2-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:3e65ae456ad0d210ee71a89ee112ac7e72e6c2e5aac1b95846ecc7afa68a194c", size = 632229 },
{ url = "https://pypi.netflix.net/packages/19129800501/tree_sitter-0.25.2-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:49ee3c348caa459244ec437ccc7ff3831f35977d143f65311572b8ba0a5f265f", size = 629861 },
{ url = "https://pypi.netflix.net/packages/19129800502/tree_sitter-0.25.2-cp311-cp311-win_amd64.whl", hash = "sha256:56ac6602c7d09c2c507c55e58dc7026b8988e0475bd0002f8a386cce5e8e8adc", size = 127304 },
{ url = "https://pypi.netflix.net/packages/19129800503/tree_sitter-0.25.2-cp311-cp311-win_arm64.whl", hash = "sha256:b3d11a3a3ac89bb8a2543d75597f905a9926f9c806f40fcca8242922d1cc6ad5", size = 113990 },
{ url = "https://pypi.netflix.net/packages/19129801135/tree_sitter-0.25.2-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:ddabfff809ffc983fc9963455ba1cecc90295803e06e140a4c83e94c1fa3d960", size = 146941 },
{ url = "https://pypi.netflix.net/packages/19129801136/tree_sitter-0.25.2-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:c0c0ab5f94938a23fe81928a21cc0fac44143133ccc4eb7eeb1b92f84748331c", size = 137699 },
{ url = "https://pypi.netflix.net/packages/19129801137/tree_sitter-0.25.2-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:dd12d80d91d4114ca097626eb82714618dcdfacd6a5e0955216c6485c350ef99", size = 607125 },
{ url = "https://pypi.netflix.net/packages/19129801138/tree_sitter-0.25.2-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:b43a9e4c89d4d0839de27cd4d6902d33396de700e9ff4c5ab7631f277a85ead9", size = 635418 },
{ url = "https://pypi.netflix.net/packages/19129801139/tree_sitter-0.25.2-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:fbb1706407c0e451c4f8cc016fec27d72d4b211fdd3173320b1ada7a6c74c3ac", size = 631250 },
{ url = "https://pypi.netflix.net/packages/19129801140/tree_sitter-0.25.2-cp312-cp312-win_amd64.whl", hash = "sha256:6d0302550bbe4620a5dc7649517c4409d74ef18558276ce758419cf09e578897", size = 127156 },
{ url = "https://pypi.netflix.net/packages/19129801141/tree_sitter-0.25.2-cp312-cp312-win_arm64.whl", hash = "sha256:0c8b6682cac77e37cfe5cf7ec388844957f48b7bd8d6321d0ca2d852994e10d5", size = 113984 },
{ url = "https://pypi.netflix.net/packages/19129801142/tree_sitter-0.25.2-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:0628671f0de69bb279558ef6b640bcfc97864fe0026d840f872728a86cd6b6cd", size = 146926 },
{ url = "https://pypi.netflix.net/packages/19129801143/tree_sitter-0.25.2-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:f5ddcd3e291a749b62521f71fc953f66f5fd9743973fd6dd962b092773569601", size = 137712 },
{ url = "https://pypi.netflix.net/packages/19129801144/tree_sitter-0.25.2-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:bd88fbb0f6c3a0f28f0a68d72df88e9755cf5215bae146f5a1bdc8362b772053", size = 607873 },
{ url = "https://pypi.netflix.net/packages/19129801145/tree_sitter-0.25.2-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:b878e296e63661c8e124177cc3084b041ba3f5936b43076d57c487822426f614", size = 636313 },
{ url = "https://pypi.netflix.net/packages/19129801146/tree_sitter-0.25.2-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:d77605e0d353ba3fe5627e5490f0fbfe44141bafa4478d88ef7954a61a848dae", size = 631370 },
{ url = "https://pypi.netflix.net/packages/19129803321/tree_sitter-0.25.2-cp313-cp313-win_amd64.whl", hash = "sha256:463c032bd02052d934daa5f45d183e0521ceb783c2548501cf034b0beba92c9b", size = 127157 },
{ url = "https://pypi.netflix.net/packages/19129803322/tree_sitter-0.25.2-cp313-cp313-win_arm64.whl", hash = "sha256:b3f63a1796886249bd22c559a5944d64d05d43f2be72961624278eff0dcc5cb8", size = 113975 },
{ url = "https://pypi.netflix.net/packages/19129803323/tree_sitter-0.25.2-cp314-cp314-macosx_10_13_x86_64.whl", hash = "sha256:65d3c931013ea798b502782acab986bbf47ba2c452610ab0776cf4a8ef150fc0", size = 146776 },
{ url = "https://pypi.netflix.net/packages/19129803324/tree_sitter-0.25.2-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:bda059af9d621918efb813b22fb06b3fe00c3e94079c6143fcb2c565eb44cb87", size = 137732 },
{ url = "https://pypi.netflix.net/packages/19129803325/tree_sitter-0.25.2-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:eac4e8e4c7060c75f395feec46421eb61212cb73998dbe004b7384724f3682ab", size = 609456 },
{ url = "https://pypi.netflix.net/packages/19129803326/tree_sitter-0.25.2-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:260586381b23be33b6191a07cea3d44ecbd6c01aa4c6b027a0439145fcbc3358", size = 636772 },
{ url = "https://pypi.netflix.net/packages/19129803327/tree_sitter-0.25.2-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:7d2ee1acbacebe50ba0f85fff1bc05e65d877958f00880f49f9b2af38dce1af0", size = 631522 },
{ url = "https://pypi.netflix.net/packages/19129803328/tree_sitter-0.25.2-cp314-cp314-win_amd64.whl", hash = "sha256:4973b718fcadfb04e59e746abfbb0288694159c6aeecd2add59320c03368c721", size = 130864 },
{ url = "https://pypi.netflix.net/packages/19129803329/tree_sitter-0.25.2-cp314-cp314-win_arm64.whl", hash = "sha256:b8d4429954a3beb3e844e2872610d2a4800ba4eb42bb1990c6a4b1949b18459f", size = 117470 },
]
[[package]]
name = "tree-sitter-c-sharp"
version = "0.23.1"
source = { registry = "https://pypi.netflix.net/simple" }
sdist = { url = "https://pypi.netflix.net/packages/18519163555/tree_sitter_c_sharp-0.23.1.tar.gz", hash = "sha256:322e2cfd3a547a840375276b2aea3335fa6458aeac082f6c60fec3f745c967eb", size = 1317728 }
wheels = [
{ url = "https://pypi.netflix.net/packages/18519163548/tree_sitter_c_sharp-0.23.1-cp39-abi3-macosx_10_9_x86_64.whl", hash = "sha256:2b612a6e5bd17bb7fa2aab4bb6fc1fba45c94f09cb034ab332e45603b86e32fd", size = 372235 },
{ url = "https://pypi.netflix.net/packages/18519163549/tree_sitter_c_sharp-0.23.1-cp39-abi3-macosx_11_0_arm64.whl", hash = "sha256:1a8b98f62bc53efcd4d971151950c9b9cd5cbe3bacdb0cd69fdccac63350d83e", size = 419046 },
{ url = "https://pypi.netflix.net/packages/18519163550/tree_sitter_c_sharp-0.23.1-cp39-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:986e93d845a438ec3c4416401aa98e6a6f6631d644bbbc2e43fcb915c51d255d", size = 415999 },
{ url = "https://pypi.netflix.net/packages/18519163551/tree_sitter_c_sharp-0.23.1-cp39-abi3-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:a8024e466b2f5611c6dc90321f232d8584893c7fb88b75e4a831992f877616d2", size = 402830 },
{ url = "https://pypi.netflix.net/packages/18519163552/tree_sitter_c_sharp-0.23.1-cp39-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:7f9bf876866835492281d336b9e1f9626ab668737f74e914c31d285261507da7", size = 397880 },
{ url = "https://pypi.netflix.net/packages/18519163553/tree_sitter_c_sharp-0.23.1-cp39-abi3-win_amd64.whl", hash = "sha256:ae9a9e859e8f44e2b07578d44f9a220d3fa25b688966708af6aa55d42abeebb3", size = 377562 },
{ url = "https://pypi.netflix.net/packages/18519163554/tree_sitter_c_sharp-0.23.1-cp39-abi3-win_arm64.whl", hash = "sha256:c81548347a93347be4f48cb63ec7d60ef4b0efa91313330e69641e49aa5a08c5", size = 375157 },
]
[[package]]
name = "tree-sitter-embedded-template"
version = "0.25.0"
source = { registry = "https://pypi.netflix.net/simple" }
sdist = { url = "https://pypi.netflix.net/packages/19023467751/tree_sitter_embedded_template-0.25.0.tar.gz", hash = "sha256:7d72d5e8a1d1d501a7c90e841b51f1449a90cc240be050e4fb85c22dab991d50", size = 14114 }
wheels = [
{ url = "https://pypi.netflix.net/packages/19023467743/tree_sitter_embedded_template-0.25.0-cp310-abi3-macosx_10_9_x86_64.whl", hash = "sha256:fa0d06467199aeb33fb3d6fa0665bf9b7d5a32621ffdaf37fd8249f8a8050649", size = 10266 },
{ url = "https://pypi.netflix.net/packages/19023467744/tree_sitter_embedded_template-0.25.0-cp310-abi3-macosx_11_0_arm64.whl", hash = "sha256:fc7aacbc2985a5d7e7fe7334f44dffe24c38fb0a8295c4188a04cf21a3d64a73", size = 10650 },
{ url = "https://pypi.netflix.net/packages/19023467745/tree_sitter_embedded_template-0.25.0-cp310-abi3-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:a7c88c3dd8b94b3c9efe8ae071ff6b1b936a27ac5f6e651845c3b9631fa4c1c2", size = 18268 },
{ url = "https://pypi.netflix.net/packages/19023467746/tree_sitter_embedded_template-0.25.0-cp310-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:025f7ca84218dcd8455efc901bdbcc2689fb694f3a636c0448e322a23d4bc96b", size = 19068 },
{ url = "https://pypi.netflix.net/packages/19023467747/tree_sitter_embedded_template-0.25.0-cp310-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:b5dc1aef6ffa3fae621fe037d85dd98948b597afba20df29d779c426be813ee5", size = 18518 },
{ url = "https://pypi.netflix.net/packages/19023467748/tree_sitter_embedded_template-0.25.0-cp310-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:d0a35cfe634c44981a516243bc039874580e02a2990669313730187ce83a5bc6", size = 18267 },
{ url = "https://pypi.netflix.net/packages/19023467749/tree_sitter_embedded_template-0.25.0-cp310-abi3-win_amd64.whl", hash = "sha256:3e05a4ac013d54505e75ae48e1a0e9db9aab19949fe15d9f4c7345b11a84a069", size = 13049 },
{ url = "https://pypi.netflix.net/packages/19023467750/tree_sitter_embedded_template-0.25.0-cp310-abi3-win_arm64.whl", hash = "sha256:2751d402179ac0e83f2065b249d8fe6df0718153f1636bcb6a02bde3e5730db9", size = 11978 },
]
[[package]]
name = "tree-sitter-language-pack"
version = "0.13.0"
source = { registry = "https://pypi.netflix.net/simple" }
dependencies = [
{ name = "tree-sitter" },
{ name = "tree-sitter-c-sharp" },
{ name = "tree-sitter-embedded-template" },
{ name = "tree-sitter-yaml" },
]
sdist = { url = "https://pypi.netflix.net/packages/19391792931/tree_sitter_language_pack-0.13.0.tar.gz", hash = "sha256:032034c5e27b1f6e00730b9e7c2dbc8203b4700d0c681fd019d6defcf61183ec", size = 51353370 }
wheels = [
{ url = "https://pypi.netflix.net/packages/19391792760/tree_sitter_language_pack-0.13.0-cp310-abi3-macosx_10_15_universal2.whl", hash = "sha256:0e7eae812b40a2dc8a12eb2f5c55e130eb892706a0bee06215dd76affeb00d07", size = 32991857 },
{ url = "https://pypi.netflix.net/packages/19391792761/tree_sitter_language_pack-0.13.0-cp310-abi3-manylinux2014_aarch64.whl", hash = "sha256:7fdacf383418a845b20772118fcb53ad245f9c5d409bd07dae16acec65151756", size = 20092989 },
{ url = "https://pypi.netflix.net/packages/19391792762/tree_sitter_language_pack-0.13.0-cp310-abi3-manylinux2014_x86_64.whl", hash = "sha256:0d4f261fce387ae040dae7e4d1c1aca63d84c88320afcc0961c123bec0be8377", size = 19952029 },
{ url = "https://pypi.netflix.net/packages/19391792845/tree_sitter_language_pack-0.13.0-cp310-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:78f369dc4d456c5b08d659939e662c2f9b9fba8c0ec5538a1f973e01edfcf04d", size = 19944614 },
{ url = "https://pypi.netflix.net/packages/19391792846/tree_sitter_language_pack-0.13.0-cp310-abi3-win_amd64.whl", hash = "sha256:1cdbc88a03dacd47bec69e56cc20c48eace1fbb6f01371e89c3ee6a2e8f34db1", size = 16896852 },
]
[[package]]
name = "tree-sitter-yaml"
version = "0.7.2"
source = { registry = "https://pypi.netflix.net/simple" }
sdist = { url = "https://pypi.netflix.net/packages/19176087043/tree_sitter_yaml-0.7.2.tar.gz", hash = "sha256:756db4c09c9d9e97c81699e8f941cb8ce4e51104927f6090eefe638ee567d32c", size = 84882 }
wheels = [
{ url = "https://pypi.netflix.net/packages/19176087035/tree_sitter_yaml-0.7.2-cp310-abi3-macosx_10_9_x86_64.whl", hash = "sha256:7e269ddcfcab8edb14fbb1f1d34eed1e1e26888f78f94eedfe7cc98c60f8bc9f", size = 43898 },
{ url = "https://pypi.netflix.net/packages/19176087036/tree_sitter_yaml-0.7.2-cp310-abi3-macosx_11_0_arm64.whl", hash = "sha256:0807b7966e23ddf7dddc4545216e28b5a58cdadedcecca86b8d8c74271a07870", size = 44691 },
{ url = "https://pypi.netflix.net/packages/19176087037/tree_sitter_yaml-0.7.2-cp310-abi3-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:f1a5c60c98b6c4c037aae023569f020d0c489fad8dc26fdfd5510363c9c29a41", size = 91430 },
{ url = "https://pypi.netflix.net/packages/19176087038/tree_sitter_yaml-0.7.2-cp310-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:88636d19d0654fd24f4f242eaaafa90f6f5ebdba8a62e4b32d251ed156c51a2a", size = 92428 },
{ url = "https://pypi.netflix.net/packages/19176087039/tree_sitter_yaml-0.7.2-cp310-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:1d2e8f0bb14aa4537320952d0f9607eef3021d5aada8383c34ebeece17db1e06", size = 90580 },
{ url = "https://pypi.netflix.net/packages/19176087040/tree_sitter_yaml-0.7.2-cp310-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:74ca712c50fc9d7dbc68cb36b4a7811d6e67a5466b5a789f19bf8dd6084ef752", size = 90455 },
{ url = "https://pypi.netflix.net/packages/19176087041/tree_sitter_yaml-0.7.2-cp310-abi3-win_amd64.whl", hash = "sha256:7587b5ca00fc4f9a548eff649697a3b395370b2304b399ceefa2087d8a6c9186", size = 45514 },
{ url = "https://pypi.netflix.net/packages/19176087042/tree_sitter_yaml-0.7.2-cp310-abi3-win_arm64.whl", hash = "sha256:f63c227b18e7ce7587bce124578f0bbf1f890ac63d3e3cd027417574273642c4", size = 44065 },
]
[[package]]
name = "triton"
version = "3.5.1"