Context Window Management
Automatic token limit handling, context optimization, and intelligent truncation for efficient LLM usage.
from praisonaiagents import Agent
agent = Agent(name="long-context", instructions="Handle large inputs safely.")
agent.start("Summarise this 50-page report")The user sends a large prompt; context management trims or summarises input to fit the model window.
How It Works
Local models — window comes from the server
Since PR #4924, when PraisonAI has already probed a locally-served model it uses the window the server reported, not the litellm default.
from praisonaiagents import Agent
# Ollama's qwen3:0.6b reports a 40960-token window via /api/show
agent = Agent(llm="ollama/qwen3:0.6b", instructions="Summarize")
agent.start("...long document...")
# Compaction now targets 40960, not the generic 128000 default.Before the fix, a 4096-context local model was budgeted at 128000 tokens — the server silently dropped the head of the conversation and returned an answer built from the tail alone.
| Model | Before | After (server reports) |
|---|---|---|
ollama/qwen3:0.6b | 128000 | 40960 |
gpt-4o (hosted) | 128000 | 128000 (unchanged) |
Any local model whose /api/show reports a shorter window | litellm default | server value |
Hosted providers are untouched: gpt-4o still resolves through litellm.
sequenceDiagram
participant Agent
participant Budgeter as get_model_limit
participant Local as local layer (probed)
participant LiteLLM as litellm registry
Agent->>Budgeter: get_model_limit("ollama/qwen3:0.6b")
Budgeter->>Local: context_length_for(...)
Local-->>Budgeter: 40960 (from /api/show)
Budgeter-->>Agent: 40960
Note over Agent,LiteLLM: litellm is only consulted<br/>when the local cache misses
classDef agent fill:#8B0000,stroke:#7C90A0,color:#fff
classDef process fill:#189AB4,stroke:#7C90A0,color:#fff
classDef output fill:#10B981,stroke:#7C90A0,color:#fff
class Agent agent
class Budgeter,Local process
class LiteLLM outputThe probed window is a process-local, no-I/O cache — populated when the local layer first records what /api/show reported, then read on every budgeting call without touching the network.
Quick Start
- Install Package
Install PraisonAI Agents:
bash pip install praisonaiagents - Import Context Manager
Import context management utilities:
python from praisonaiagents import ( ContextManager, TokenCounter, ContextOptimizer ) - Create Example
Create
context_management.py:python from praisonaiagents import Agent, Task, AgentTeam from praisonaiagents import ContextManager # Initialize context manager context_manager = ContextManager( model="gpt-4", max_tokens=8192, # Model's context limit reserve_tokens=1000, # Reserve for response optimization_strategy="smart" ) # Create agent with context management smart_agent = Agent( name="Context-Aware Agent", role="Document Processor", goal="Process large documents efficiently", context_manager=context_manager, instructions="Analyze documents while managing context limits" ) # Create task with large context large_document = "..." * 10000 # Very long document analysis_task = Task( description="Analyze this extensive document", expected_output="Comprehensive analysis", agent=smart_agent, context={ "document": large_document, "additional_context": "Focus on key findings" } ) # The context manager automatically handles the large input workflow = AgentTeam( agents=[smart_agent], tasks=[analysis_task], context_management=True ) results = workflow.start() # Check context usage usage = context_manager.get_usage_stats() print(f"Tokens used: {usage['tokens_used']}/{usage['max_tokens']}") print(f"Optimization applied: {usage['optimization_method']}") - Run Example
Execute the context management example:
bash python context_management.py
Core Features
Token Counting
from praisonaiagents import TokenCounter
# Initialize token counter for specific model
counter = TokenCounter(model="gpt-4")
# Count tokens in text
text = "This is a sample text for token counting."
token_count = counter.count_tokens(text)
print(f"Token count: {token_count}")
# Count tokens with special formatting
messages = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Hello, how are you?"}
]
message_tokens = counter.count_message_tokens(messages)
print(f"Message tokens: {message_tokens}")
# Estimate tokens before encoding
estimated_tokens = counter.estimate_tokens(text)
print(f"Estimated tokens: {estimated_tokens}")Context Optimization
from praisonaiagents import ContextOptimizer
# Initialize optimizer
optimizer = ContextOptimizer(
max_tokens=4096,
preserve_ratio=0.8, # Keep 80% of important content
strategy="intelligent"
)
# Optimize large context
large_context = {
"background": "Long background information...",
"data": "Extensive data points...",
"instructions": "Detailed instructions...",
"examples": "Multiple examples..."
}
# Optimize with priorities
optimized = optimizer.optimize_context(
context=large_context,
priorities={
"instructions": 1.0, # Highest priority
"data": 0.8,
"examples": 0.6,
"background": 0.4 # Lowest priority
}
)
print(f"Original tokens: {optimizer.count_tokens(large_context)}")
print(f"Optimized tokens: {optimizer.count_tokens(optimized)}")Smart Truncation
from praisonaiagents import SmartTruncator
# Initialize smart truncator
truncator = SmartTruncator(
max_tokens=2048,
preserve_structure=True,
keep_boundaries=True
)
# Truncate while preserving meaning
long_text = """
Introduction: This document discusses...
[5000 words of content]
Conclusion: In summary...
"""
truncated = truncator.truncate_smart(
text=long_text,
keep_sections=["Introduction", "Conclusion"],
summarize_middle=True
)
# Truncate with different strategies
strategies = {
"head": truncator.truncate_head, # Keep beginning
"tail": truncator.truncate_tail, # Keep end
"middle": truncator.truncate_middle, # Keep middle
"smart": truncator.truncate_smart # Intelligent truncation
}
for name, strategy in strategies.items():
result = strategy(long_text, max_tokens=1000)
print(f"{name}: {len(result)} chars")Advanced Context Management
Dynamic Context Allocation
from praisonaiagents import DynamicContextManager
# Create dynamic context manager
dynamic_manager = DynamicContextManager(
base_model="gpt-4",
fallback_models=["gpt-3.5-turbo-16k", "claude-2"],
auto_switch=True
)
# Process with dynamic allocation
class ContextAwareAgent(Agent):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.context_manager = dynamic_manager
def process_task(self, task):
# Automatically selects appropriate model based on context size
context_size = self.context_manager.calculate_context_size(task)
if context_size > 8192:
# Automatically switches to model with larger context
self.context_manager.switch_to_larger_context()
# Process with optimized context
optimized_task = self.context_manager.optimize_task(task)
return super().process_task(optimized_task)Context Windowing
from praisonaiagents import ContextWindow
# Sliding window for continuous conversations
window = ContextWindow(
max_size=4096,
overlap=512,
compression_ratio=0.7
)
# Process long conversation
conversation_history = []
for user_input in user_inputs:
# Add new input to window
window.add_message({
"role": "user",
"content": user_input
})
# Get optimized context for current turn
current_context = window.get_current_context()
# Process with agent
response = agent.process(current_context)
# Add response to window
window.add_message({
"role": "assistant",
"content": response
})
# Window automatically manages size and removes old messages
stats = window.get_stats()
print(f"Window usage: {stats['current_tokens']}/{stats['max_tokens']}")Hierarchical Context Management
from praisonaiagents import HierarchicalContextManager
# Manage context at multiple levels
hierarchy_manager = HierarchicalContextManager(
levels={
"global": 1024, # Global context
"session": 2048, # Session context
"task": 4096, # Task-specific context
"immediate": 8192 # Immediate context
}
)
# Add context at different levels
hierarchy_manager.add_context(
level="global",
content="System instructions and global rules",
priority=1.0
)
hierarchy_manager.add_context(
level="session",
content="User preferences and session data",
priority=0.8
)
hierarchy_manager.add_context(
level="task",
content="Current task details and requirements",
priority=0.9
)
# Get optimized context for processing
processing_context = hierarchy_manager.get_optimized_context(
available_tokens=6000,
required_levels=["global", "task", "immediate"]
)Context Strategies
Summarization Strategy
from praisonaiagents import SummarizationStrategy
# Use summarization for large contexts
summarizer = SummarizationStrategy(
summary_model="gpt-3.5-turbo",
summary_ratio=0.2, # Reduce to 20% of original
preserve_key_points=True
)
# Summarize large document sections
document_sections = {
"introduction": "Long introduction text...",
"methodology": "Detailed methodology...",
"results": "Extensive results...",
"conclusion": "Detailed conclusion..."
}
summarized = summarizer.summarize_sections(
sections=document_sections,
importance_weights={
"results": 1.0,
"conclusion": 0.9,
"methodology": 0.7,
"introduction": 0.5
}
)
# Adaptive summarization based on available space
available_tokens = 2000
adaptive_summary = summarizer.summarize_adaptive(
content=large_document,
target_tokens=available_tokens,
preserve_structure=True
)Chunking Strategy
from praisonaiagents import ChunkingStrategy
# Process large documents in chunks
chunker = ChunkingStrategy(
chunk_size=2048,
overlap=256,
maintain_coherence=True
)
# Process document in chunks
chunks = chunker.create_chunks(large_document)
results = []
for i, chunk in enumerate(chunks):
# Process each chunk with context from previous chunks
chunk_context = chunker.get_chunk_context(
current_chunk=i,
previous_results=results[-3:] if results else []
)
result = agent.process(chunk_context)
results.append(result)
# Combine results
final_result = chunker.combine_results(results)Priority-Based Selection
from praisonaiagents import PriorityContextSelector
# Select context based on priorities
selector = PriorityContextSelector(
max_tokens=4096,
selection_strategy="weighted"
)
# Define context items with priorities
context_items = [
{
"content": "Critical instructions",
"priority": 1.0,
"category": "instructions"
},
{
"content": "Recent conversation history",
"priority": 0.9,
"category": "history"
},
{
"content": "Background information",
"priority": 0.5,
"category": "background"
},
{
"content": "Examples and samples",
"priority": 0.3,
"category": "examples"
}
]
# Select context within token limit
selected_context = selector.select_context(
items=context_items,
required_categories=["instructions"],
preferred_categories=["history"]
)Model-Specific Management
from praisonaiagents import ModelContextManager
# Configure for different models
model_configs = {
"gpt-4": {
"max_tokens": 8192,
"encoding": "cl100k_base",
"reserve_tokens": 1000
},
"gpt-3.5-turbo": {
"max_tokens": 4096,
"encoding": "cl100k_base",
"reserve_tokens": 500
},
"claude-2": {
"max_tokens": 100000,
"encoding": "claude",
"reserve_tokens": 2000
},
"llama-2": {
"max_tokens": 4096,
"encoding": "llama",
"reserve_tokens": 500
}
}
# Automatic model-specific handling
manager = ModelContextManager(model_configs)
# Process with automatic model detection
def process_with_model(content, model_name):
context = manager.prepare_context(
content=content,
model=model_name
)
if context.requires_optimization:
context = manager.optimize_for_model(context, model_name)
return contextMonitoring and Analytics
from praisonaiagents import ContextAnalytics
# Monitor context usage
analytics = ContextAnalytics()
# Track usage across multiple interactions
@analytics.track_context_usage
def process_request(request):
# Your processing logic
return response
# Get analytics
stats = analytics.get_statistics()
print(f"Average context usage: {stats['avg_usage']}%")
print(f"Peak usage: {stats['peak_usage']} tokens")
print(f"Optimization rate: {stats['optimization_rate']}%")
# Visualize usage patterns
analytics.plot_usage_over_time()
analytics.generate_optimization_report()Best Practices
Reserve Tokens
Always reserve tokens for model responses:
context_config = {
"max_input_tokens": 3000, # 75% of 4096
"reserved_output": 1000, # 25% for response
"safety_margin": 96 # Extra buffer
}Context Prioritization
Prioritize context elements:
priority_order = [
"current_task", # Highest
"recent_history",
"user_preferences",
"system_instructions",
"examples", # Lowest
]Optimization Strategies
Choose appropriate optimization:
optimization_matrix = {
"small_overflow": "truncation",
"medium_overflow": "summarization",
"large_overflow": "chunking",
"critical_content": "priority_selection"
}Troubleshooting
If hitting token limits: – Enable aggressive optimization – Reduce context items – Use summarization – Switch to larger context model
Context LossIf losing important context: – Adjust priority weights – Increase overlap in chunking – Use hierarchical management – Enable context compression
Next Steps
Learn about model-specific context limits
Memory SystemsExplore long-term context preservation
Tool Output StoreRecover full tool outputs after truncation