antaris-context 2.2.0

Context window optimization for AI agents. Zero dependencies.

antaris-context

Zero-dependency context window optimization for AI agents.

Manage context windows, token budgets, turn lifecycle, and message compression without external dependencies. Integrates with antaris-memory for memory-informed priority boosting and antaris-router for adaptive budget allocation. Built for production AI agent systems that need deterministic, configurable context management.

What's New in v2.0.0

• Turn lifecycle API — add_turn(role, content), compact_older_turns(keep_last=20), render(provider='anthropic'|'openai'|'generic'), set_retention_policy(), turn_count
• Provider-ready render — render() produces message lists formatted for OpenAI, Anthropic, or generic clients
• Suite integration — set_memory_client(client) for memory-informed priority boosting; set_router_hints(hints) accepts hints from antaris-router and adjusts section budgets automatically
• Pluggable summarizer — set_summarizer(fn) — plug in any function to compress older turns semantically
• ImportanceWeightedCompressor — priority-aware compression with CompressionResult reporting
• SemanticChunker — sentence-boundary-aware text chunking with configurable overlap
• Cross-session snapshots — export_snapshot(include_importance_above), from_snapshot(dict) for persistence across sessions
• 150 tests (all passing)

See CHANGELOG.md for full version history.

---

Install

pip install antaris-context

---

Quick Start

from antaris_context import ContextManager

# Initialize with a preset template
manager = ContextManager(total_budget=8000, template="code_assistant")
# Templates: chatbot, agent_with_tools, rag_pipeline, code_assistant, balanced

# Add turns (conversation lifecycle)
manager.add_turn("user", "How do I add JWT auth to my Flask API?")
manager.add_turn("assistant", "Use flask-jwt-extended. Here's a minimal example...")

# Check turn count and budget usage
print(f"Turns: {manager.turn_count}")
report = manager.get_usage_report()
print(f"Used: {report['total_used']}/{report['total_budget']} tokens ({report['utilization']:.1%})")

# Compact old turns when context gets full
removed = manager.compact_older_turns(keep_last=20)
print(f"Compacted {removed} turns")

# Render for your LLM provider
messages = manager.render(provider="anthropic")        # → Anthropic message format
messages = manager.render(provider="openai")           # → OpenAI message format
messages = manager.render(provider="generic")          # → generic list of dicts
messages = manager.render(system_prompt="Be concise")  # → inject system prompt

---

OpenClaw Integration

antaris-context is purpose-built for OpenClaw agent sessions. Use it to manage the context window across multi-turn conversations — automatically compressing older turns to make room for memory recall, tool results, and new input.

from antaris_context import ContextManager

ctx = ContextManager(total_budget=8000)
ctx.add_turn("user", user_input)
ctx.add_turn("assistant", agent_response)

# Before the next turn, compact to stay within budget
ctx.compact_older_turns(keep_last=10)
messages = ctx.render()  # Ready for any provider (OpenAI, Anthropic, etc.)

Pairs directly with antaris-memory (inject recalled memories into context budget) and antaris-router (route based on actual token count). Both are wired automatically in antaris-pipeline.

---

Turn Lifecycle

manager = ContextManager(total_budget=16000, template="agent_with_tools")

# Add turns from a conversation
for msg in conversation_history:
    manager.add_turn(msg["role"], msg["content"])

# Compact old turns before hitting the budget limit
removed = manager.compact_older_turns(keep_last=30)

# With a pluggable summarizer (compress rather than drop)
def my_summarizer(turns: list[dict]) -> str:
    """Call your LLM to summarize old turns."""
    # ... call OpenAI/Claude/Ollama ...
    return "Summary of earlier conversation: ..."

manager.set_summarizer(my_summarizer)
manager.compact_older_turns(keep_last=20)
# Older turns are passed to my_summarizer and replaced with the summary

---

Suite Integration

from antaris_context import ContextManager
from antaris_memory import MemorySystem
from antaris_router import Router

# Memory-informed priority boosting
mem = MemorySystem("./workspace")
mem.load()
manager = ContextManager(total_budget=8000)
manager.set_memory_client(mem)
# optimize_context() now boosts sections matching recent memory queries

# Router-driven budget adaptation
router = Router(config_path="./config")
result = router.route(user_input)
manager.set_router_hints(result.routing_hints)
# Section budgets shift based on router's complexity assessment

---

Templates

Built-in section budget presets for common agent patterns:

templates = ContextManager.get_available_templates()
# {
#   'chatbot':          {'system': 800,  'memory': 1500, 'conversation': 5000, 'tools': 700},
#   'agent_with_tools': {'system': 1200, 'memory': 2000, 'conversation': 3500, 'tools': 1300},
#   'rag_pipeline':     {'system': 600,  'memory': 1000, 'conversation': 4500, 'tools': 1900},
#   'code_assistant':   {'system': 1000, 'memory': 1800, 'conversation': 4000, 'tools': 1200},
#   'balanced':         {'system': 1000, 'memory': 2000, 'conversation': 4000, 'tools': 1000},
# }

manager = ContextManager(total_budget=8000, template="agent_with_tools")
manager.apply_template("rag_pipeline")  # Switch template mid-session

---

Content Management

# Add content with priorities
manager.add_content('system', "You are a coding assistant.", priority='critical')
manager.add_content('memory', "User prefers Python examples.", priority='important')
manager.add_content('conversation', messages, priority='normal')
manager.add_content('tools', long_debug_output, priority='optional')

# Priority levels:
# critical  → never truncated (system prompts, safety rules)
# important → removed only when necessary
# normal    → standard selection (conversation history)
# optional  → first to go when space is needed

# Add with query for relevance-based selection
manager.add_content('conversation', messages, query="JWT authentication Flask")

# Set selection strategy
manager.set_strategy('hybrid', recency_weight=0.4, relevance_weight=0.6)
manager.set_strategy('recency', prefer_high_priority=True)
manager.set_strategy('budget', approach='balanced')

# Set compression level
manager.set_compression_level('moderate')  # light, moderate, aggressive

---

Compression

from antaris_context import MessageCompressor, ImportanceWeightedCompressor, SemanticChunker

# Basic message compression
compressor = MessageCompressor('moderate')
compressed = compressor.compress_message_list(messages, max_content_length=500)
output = compressor.compress_tool_output(long_output, max_lines=20, keep_first=10, keep_last=10)
stats = compressor.get_compression_stats()
print(f"Saved {stats['bytes_saved']} bytes ({stats['compression_ratio']:.1%})")

# Priority-aware compression
iwc = ImportanceWeightedCompressor(keep_top_n=5, compress_middle=True, drop_threshold=0.1)

# Sentence-boundary chunking
chunker = SemanticChunker(min_chunk_size=100, max_chunk_size=500)
chunks = chunker.chunk(long_text)  # → list of SemanticChunk

---

Adaptive Budgets

# Track usage patterns over time
manager.track_usage()

# Get reallocation suggestions
suggestions = manager.suggest_adaptive_reallocation()
for section, budget in suggestions['suggested_budgets'].items():
    current = suggestions['current_budgets'][section]
    print(f"{section}: {current} → {budget} tokens")

# Apply automatically
manager.apply_adaptive_reallocation(auto_apply=True, min_improvement_pct=10)

# Enable continuous adaptation
manager.enable_adaptive_budgets(target_utilization=0.85)

---

Cross-Session Snapshots

# Save context state between sessions
manager.save_snapshot("pre-refactor")
snapshot_data = manager.export_snapshot(include_importance_above=0.5)

# Restore later
manager.restore_snapshot("pre-refactor")

# Reconstruct from exported dict
manager2 = ContextManager.from_snapshot(snapshot_data)

# List saved snapshots
for name in manager.list_snapshots():
    print(name)

---

Context Analysis

analysis = manager.analyze_context()
print(f"Efficiency score: {analysis['efficiency_score']:.2f}")

for section, data in analysis['section_analysis'].items():
    print(f"{section}: {data['utilization']:.1%} — {data['status']}")

for suggestion in analysis['optimization_suggestions']:
    print(f"  - {suggestion['description']}")

---

Complete Agent Example

from antaris_context import ContextManager

manager = ContextManager(total_budget=8000, template="code_assistant")

# System prompt (never truncated)
manager.add_content('system',
    "You are a coding assistant. Always provide working examples.",
    priority='critical')

# User memory
for memory in ["User is learning Python", "Prefers concise explanations"]:
    manager.add_content('memory', memory, priority='important')

# Conversation turns
for turn in conversation_history:
    manager.add_turn(turn["role"], turn["content"])

# Compact if needed
if manager.is_over_budget():
    manager.compact_older_turns(keep_last=20)

# Optimize to target utilization
result = manager.optimize_context(query=current_query, target_utilization=0.85)

# Render for your provider
messages = manager.render(provider="openai")
response = openai_client.chat.completions.create(model="gpt-4o", messages=messages)

---

Configuration File

{
  "compression_level": "moderate",
  "strategy": "hybrid",
  "strategy_params": {
    "recency_weight": 0.4,
    "relevance_weight": 0.6
  },
  "section_budgets": {
    "system": 1000,
    "memory": 2000,
    "conversation": 4000,
    "tools": 1000
  },
  "truncation_strategy": "oldest_first",
  "auto_compress": true
}

manager = ContextManager(config_file="config.json")
manager.set_compression_level("aggressive")
manager.save_config("updated_config.json")

---

Token Estimation

Uses character-based approximation (~4 chars/token). Fast and sufficient for budget management. For exact counts, plug in your model's tokenizer:

import tiktoken
enc = tiktoken.encoding_for_model("gpt-4o")
manager._estimate_tokens = lambda text: len(enc.encode(text))

---

What It Doesn't Do

• No actual tokenization — character-based approximation. Plug in your tokenizer for exact counts.
• No LLM calls — purely deterministic. The pluggable set_summarizer() is optional; without it, compaction is structural only.
• No content generation — selects, compresses, and truncates existing content. Won't paraphrase.
• No distributed contexts — manages single context windows. For multi-agent scenarios, use multiple managers.

---

Performance

Operation	Throughput
Token estimation	~100K chars/sec
Message compression	~50K chars/sec
Strategy selection	~10K messages/sec
Context analysis	~1K content items/sec

---

Running Tests

git clone https://github.com/Antaris-Analytics/antaris-context.git
cd antaris-context
python -m pytest tests/ -v

All 150 tests pass with zero external dependencies.

---

Part of the Antaris Analytics Suite

• antaris-memory — Persistent memory for AI agents
• antaris-router — Adaptive model routing with SLA enforcement
• antaris-guard — Security and prompt injection detection
• antaris-context — Context window optimization (this package)
• antaris-pipeline — Agent orchestration pipeline

License

Apache 2.0 — see LICENSE for details.

---

Built with ❤️ by Antaris Analytics
Deterministic infrastructure for AI agents