Production-grade context compression for LLM agents
60-70% fewer tokens. Same answers. Faster, cheaper, smarter.
Quick Start β’ Architecture β’ Features β’ Benchmarks β’ API β’ Development
|
Tests Passing |
Modules |
Lines of Code |
Compression Algorithms |
Token Reduction |
COMPACT is a production-grade context compression system that sits between your application and LLM providers. It intelligently compresses conversation history, tool outputs, and context before they reach the LLM β reducing token usage by 60-70% while preserving answer quality.
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
β THE PROBLEM β
β β
β Your App βββ [Raw Context: 50K tokens] βββ LLM β
β πΈ Expensive β
β π Slow β
β β οΈ Context window limits β
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
β¬οΈ COMPACT
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
β THE SOLUTION β
β β
β Your App βββ [COMPACT] βββ [Compressed: 15K tokens] βββ LLM β
β π§ Smart compression β
β π° 60-70% cost reduction β
β β‘ Faster responses β
β β
Same answer quality β
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
# Clone the repository
git clone https://github.com/your-org/compact.git
cd compact
# Install with all extras
pip install -e ".[all]"import asyncio
from context_system import ContextManager
async def main():
mgr = ContextManager()
# Send messages β COMPACT handles compression automatically
r1 = await mgr.process_request("demo", "What is Python?")
r2 = await mgr.process_request("demo", "How do I install it?")
r3 = await mgr.process_request("demo", "What are the best libraries?")
# See compression stats
stats = mgr.get_stats("demo")
print(f"Tokens saved: {stats['total_tokens']}")
print(f"Strategy used: {r3.compression_strategy}")
asyncio.run(main())# Start the FastAPI gateway
context-system serve --port 8000
# Or with Docker
docker-compose up# Interactive chat
context-system chat "Explain async Python"
# Compress text directly
context-system compress "Your long text here..." --strategy truncation
# Run benchmarks
context-system benchmark --test allflowchart TD
A[π₯οΈ Client Request] -->|HTTP/WebSocket/CLI| B[β‘ FastAPI Gateway]
B --> C[π§ ContextManager Orchestrator]
C --> D{π Usage Check}
D -->|> 90%| E[π Auto-Compaction]
D -->|< 90%| F[π¦ Strategy Selection]
E --> F
F -->|< 70%| G[βοΈ Truncation]
F -->|70-95%| H[π Summarization]
F -->|> 95%| I[π― Token Compression]
F -->|Any| J[π RAG Retrieval]
G --> K[π§ Tool Output Reduction]
H --> K
I --> K
J --> K
K --> L[π Observation Masking]
K --> M[π Targeted Summary]
L --> N[ποΈ Build OptimizedContext]
M --> N
N --> O[β
Ready for LLM]
style A fill:#e0f2fe,stroke:#0284c7
style B fill:#dbeafe,stroke:#2563eb
style C fill:#ede9fe,stroke:#7c3aed
style E fill:#fef3c7,stroke:#d97706
style G fill:#dcfce7,stroke:#16a34a
style H fill:#fce7f3,stroke:#db2777
style I fill:#ffedd5,stroke:#ea580c
style J fill:#f0fdf4,stroke:#15803d
style O fill:#d1fae5,stroke:#059669
sequenceDiagram
participant C as π₯οΈ Client
participant G as β‘ Gateway
participant M as π§ ContextManager
participant AC as π AutoCompactor
participant HR as π¦ HistoryReducer
participant TR as π§ ToolReducer
participant LLM as π€ LLM Provider
C->>G: POST /v1/chat
G->>M: process_request()
alt Usage > 90%
M->>AC: compact(state)
AC-->>M: CompactedState
end
M->>HR: reduce(state, budget)
HR-->>M: CompressedHistory
M->>TR: reduce(state, budget)
TR-->>M: OptimizedContext
M-->>G: OptimizedContext
G-->>C: ChatResponse
|
|
|||||||||||||||||||||||||||||
|
|
context_system/
βββ core/ # π§ Engine, config, types, errors
β βββ engine.py # ContextManager orchestrator
β βββ config.py # Pydantic config (YAML-loadable)
β βββ types.py # ContextState, Message, ToolObservation
β βββ errors.py # Error hierarchy
β
βββ context/ # π¦ Compression algorithms
β βββ history/ # 4 history reducers
β β βββ truncation.py # Sliding window
β β βββ summarization.py # LLM-based
β β βββ token_compression.py # Signal-based
β β βββ rag_retrieval.py # Vector DB retrieval
β βββ tool/ # 2 tool reducers
β β βββ masking.py # Observation masking
β β βββ targeted_summary.py # LLM compress
β βββ compaction/
β βββ auto_compaction.py # ACON trigger
β
βββ mcp/ # π Model Context Protocol
β βββ server.py # MCP server
β βββ client.py # MCP client
β βββ bridge.py # MCP-to-internal bridge
β
βββ memory/ # πΎ Persistence
β βββ store.py # ChromaDB + Redis (with fallback)
β
βββ gateway/ # β‘ FastAPI server
β βββ server.py # 5 REST endpoints
β
βββ clients/ # π± Client interfaces
β βββ interfaces.py # Terminal, Web, Mobile
β
βββ benchmarks/ # π Headroom comparison
β βββ runner.py # 4 benchmark tests
β
βββ utils/ # π§ Shared utilities
βββ tokens.py # tiktoken counting
βββ async_helpers.py # retry, timeout, gather
βββ format.py # Result formatting
|
100 search results |
Incident logs |
Per request |
GSM8K + TruthfulQA |
| Metric | COMPACT | Headroom | Winner |
|---|---|---|---|
| Code Search (100 results) | 65% reduction | 92% reduction | Headroom |
| SRE Debugging (logs) | 65% reduction | 92% reduction | Headroom |
| Latency (per request) | 15ms | 3ms | Headroom |
| Accuracy (GSM8K) | Β±0.02 | Β±0.000 | Headroom |
| Setup Complexity | Zero config | Requires setup | COMPACT |
| Customizability | Full control | Limited | COMPACT |
| MCP Native | β Built-in | β Yes | Tie |
| Local-First | β Yes | β Yes | Tie |
| Reversible | β Yes | β Yes | Tie |
Trade-off: Headroom achieves higher compression via ML models but requires more setup. COMPACT prioritizes zero-config deployment, full customizability, and production readiness.
Send a message and get optimized context.
curl -X POST http://localhost:8000/v1/chat \
-H "Content-Type: application/json" \
-d '{
"conversation_id": "my-conv",
"message": "What is Python?",
"system_prompt": "You are a helpful assistant"
}'Response:
{
"conversation_id": "my-conv",
"system_prompt": "You are a helpful assistant...",
"messages": [{"role": "user", "content": "What is Python?"}],
"tool_observations": [],
"summary": "",
"key_decisions": [],
"key_findings": [],
"token_metrics": {},
"compaction_count": 0,
"compression_strategy": "truncation",
"total_tokens": 42
}Compress text directly.
curl -X POST http://localhost:8000/v1/compress \
-H "Content-Type: application/json" \
-d '{"text": "Your long text...", "strategy": "truncation"}'Get compression statistics.
Clear a conversation.
Health check endpoint.
# docker-compose.yml
version: "3.8"
services:
compact:
build: .
ports:
- "8000:8000"
environment:
- REDIS_URL=redis://redis:6379
- OPENAI_API_KEY=${OPENAI_API_KEY}
depends_on:
- redis
redis:
image: redis:7-alpine
ports:
- "6379:6379"docker-compose up -d# Install production dependencies
pip install -e ".[all]"
# Start with multiple workers
uvicorn context_system.gateway.server:app \
--host 0.0.0.0 \
--port 8000 \
--workers 4Edit configs/default.yaml:
max_tokens: 128000
compaction_threshold: 0.90
strategy:
truncation_threshold: 0.70
summarization_threshold: 0.95
history:
truncation_window_tokens: 4000
summarization_model: "gpt-4o-mini"
rag_top_k: 10
tool_output:
masking_enabled: true
recent_observations_verbatim: 5
compaction:
enabled: true
keep_last_n_messages: 20
llm:
provider: "openai"
model: "gpt-4o"
mcp:
enabled: true
server_name: "context-engine"
memory:
vector_store:
backend: "chromadb"
cache:
backend: "redis"
url: "redis://localhost:6379"# Install dev dependencies
pip install -e ".[dev]"
# Run all tests (310 tests)
pytest
# Run with coverage
pytest --cov=context_system --cov-report=html
# Run feature tests only
pytest tests/feature_tests/ -v
# Run specific test
pytest tests/feature_tests/test_feature_history.py -v
# Lint
ruff check .
# Type check
mypy context_system/tests/
βββ feature_tests/ # 255 feature tests
β βββ test_feature_types.py # Core types
β βββ test_feature_tokens.py # Token utilities
β βββ test_feature_config.py # Configuration
β βββ test_feature_errors.py # Error hierarchy
β βββ test_feature_history.py # 4 history reducers
β βββ test_feature_tool_output.py # Tool reducers
β βββ test_feature_compaction.py # ACON
β βββ test_feature_mcp.py # MCP server/client/bridge
β βββ test_feature_memory.py # VectorStore + CacheLayer
β βββ test_feature_gateway.py # API endpoints
β βββ test_feature_utils.py # Async + format utils
β βββ test_e2e_integration.py # Full pipeline tests
βββ test_*.module # 55 original unit tests
| Phase | Status | Description |
|---|---|---|
| Phase 1 | β Done | Core engine, truncation, observation masking |
| Phase 2 | β Done | Summarization, RAG, ACON, MCP |
| Phase 3 | β Done | Gateway, clients, benchmarks, Docker |
| Phase 4 | π Next | Streaming support, parallel tool execution |
| Phase 5 | π Planned | Dashboard UI, real-time metrics |
Apache License 2.0 β see LICENSE for details.
Built with β€οΈ for the LLM agent community