agentaicommunity-agent-context-compression

---

Overview

agentaicommunity-agent-context-compression is a plug-and-play context management layer for AI agent pipelines built on agent-framework. It counts tokens in the conversation history before each LLM call, and when the count approaches your configured limit it automatically summarises the older messages — keeping the session alive without hitting context-window errors.

---

✨ Features

---

📦 Installation

pip install agentaicommunity-agent-context-compression

Or install from source:

cd ContextCompression
pip install -e .

---

🚀 Quick Start

Non-streaming

import asyncio
import logging
from agent_framework.ollama import OllamaChatClient
from context_compression import ContextCompressionMiddleware, TokenThresholdReachedError

logging.basicConfig(level=logging.INFO)

summarizer = ContextCompressionMiddleware(
    llm_client=OllamaChatClient(model="gemma3:4b"),  # LLM used to write the summary
    max_tokens=8000,        # compress when history approaches this size
    trigger_ratio=0.80,     # fire at 80% = 6400 tokens
    keep_last_messages=8,   # always keep the 8 most recent messages verbatim
    logger=logging.getLogger("ContextCompression"),
)

agent = OllamaChatClient(model="gemma3:4b").as_agent(
    name="MyAgent",
    instructions="You are a helpful assistant.",
    middleware=[summarizer],
)
session = agent.create_session()

async def main():
    for message in ["Hi, my name is Vinoth.", "I work in Python.", "What is my name?"]:
        response = await agent.run(message, session=session)
        print(response.text)

asyncio.run(main())

Streaming

async def main():
    messages = ["Hi, my name is Vinoth.", "I work in Python.", "What is my name?"]
    for message in messages:
        stream = agent.run(message, session=session, stream=True)
        async for update in stream:
            chunk = getattr(update, "text", None)
            if chunk:
                print(chunk, end="", flush=True)
        print()
        await stream.get_final_response()  # finalizes stream and populates last_usage

asyncio.run(main())

---

🧑‍💻 Usage

Threshold Callback Payload

Every call to on_threshold_reached receives a dict:

{
    "tokens_used":    87,   # current history token count
    "max_tokens":    100,   # your configured max
    "trigger_tokens": 75,   # the threshold that was crossed
}

Return True → compression proceeds normally.
Return False → request is blocked and TokenThresholdReachedError is raised.

Token Usage After Each Call

last_usage is populated after every call — both streaming and non-streaming:

# Non-streaming
response = await agent.run("Hello", session=session)

# Streaming
stream = agent.run("Hello", session=session, stream=True)
async for update in stream:
    pass
await stream.get_final_response()

# Either way, last_usage is populated:
u = summarizer.last_usage
print(u["input_token_count"])   # tokens sent to LLM
print(u["output_token_count"])  # tokens in the response
print(u["total_token_count"])   # input + output

Handling TokenThresholdReachedError

from context_compression import ContextCompressionMiddleware, TokenThresholdReachedError

try:
    response = await agent.run(message, session=session)
except TokenThresholdReachedError as e:
    print(f"Blocked: {e}")
    # handle gracefully — notify user, end session, etc.

---

⚙️ Configuration

ContextCompressionMiddleware

Blocking runaway sessions

def my_callback(info: dict) -> bool:
    if info["tokens_used"] > 500_000:
        return False   # block — raises TokenThresholdReachedError
    return True        # allow compression

middleware = ContextCompressionMiddleware(
    ...,
    on_threshold_reached=my_callback,
)

---

Provider Compatibility:
Works with any LLM client that implements the agent-framework ChatClient interface.

---

🤝 Contributing

Contributions are welcome! Please open an issue to discuss what you'd like to change before submitting a pull request.

1. Fork the repository
2. Create a feature branch (git checkout -b feature/my-feature)
3. Commit your changes (git commit -m 'Add my feature')
4. Push to the branch (git push origin feature/my-feature)
5. Open a Pull Request

---

👤 Author

Built and maintained by Vinoth Rajendran.

• 🐙 GitHub: github.com/rvinothrajendran — follow for more projects!
• 📺 YouTube: youtube.com/@VinothRajendran — subscribe for tutorials and demos!
• 💼 LinkedIn: linkedin.com/in/rvinothrajendran — let's connect!

---

📄 License

MIT — see LICENSE for details.