Metadata-Version: 2.4
Name: pyagent-compress
Version: 0.2.4
Summary: Inter-agent message compression and token budget management for multi-agent LLM systems
Project-URL: Homepage, https://pyagent.org
Project-URL: Repository, https://github.com/pyagent-core/pyagent
Project-URL: Documentation, https://pyagent.org
Project-URL: Agent Orchestration Patterns, https://pyagent.org/patterns/orchestrator-worker/
Project-URL: Agent Harness Architecture, https://pyagent.org/architecture/multi-agent-harness/
Project-URL: Agent Experience Optimization, https://pyagent.org/concepts/agent-experience-optimization/
Author-email: PyAgent Team <team@pyagent.org>
License: MIT
Keywords: LLM,agent harness,agents,compression,efficiency,multi-agent orchestration,token-efficient agents,tokens
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Typing :: Typed
Requires-Python: >=3.11
Requires-Dist: pyagent-patterns>=0.1.0
Provides-Extra: dev
Requires-Dist: mypy>=1.10; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.5; extra == 'dev'
Description-Content-Type: text/markdown

# pyagent-compress

**Pillar 2 of the PyAgent production stack for multi-agent LLM systems** — inter-agent message compression and token budget management. Reduce costs without losing key information.

[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](../../LICENSE)
[![Python 3.11+](https://img.shields.io/badge/python-3.11+-blue.svg)](https://www.python.org/downloads/)

> **Architecture Pillar: ⚡ Execution**
> Trims the token cost of inter-agent communication in the Execution pillar — message compression, agent pruning, interaction pruning, and budget enforcement so pipelines stay within cost limits.
> Part of the full stack: install `pyagent-all` for all pillars.

## Install

```bash
pip install pyagent-compress
```

Depends on: `pyagent-patterns`.

## Why Compression?

In multi-agent pipelines, verbose intermediate outputs compound quickly. A 5-stage pipeline where each agent produces 2,000 tokens means 10,000 tokens of context consumed before the final stage starts — most of it filler. Compressing intermediate outputs by 40–60% reduces both latency and cost without losing key information.

## MessageCompressor — Compress Individual Messages

Removes filler phrases, ranks sentences by information density, and trims to target ratio.

```python
from pyagent_compress import MessageCompressor

# Default: 50% target compression
compressor = MessageCompressor(target_ratio=0.5)

# Aggressive: 30% target with filler removal
compressor = MessageCompressor(
    target_ratio=0.3,
    min_sentence_length=20,   # drop very short sentences
    remove_filler=True,       # strip "let me think", "basically", "in other words", etc.
)

verbose_output = """
Let me think about this carefully. Okay so basically what we have here is a
situation where the system needs to handle concurrent writes. In other words, we
need to think about race conditions.

I believe that the most important thing is to use transactions. The system should
implement SERIALIZABLE isolation level because it ensures that all transactions
are executed as if they were serial. Studies show that this prevents 100% of dirty reads.
The database must also use row-level locking.

Additionally, we need retry logic for deadlock situations, with exponential backoff
starting at 10ms. Finally, the application should use connection pooling with a
maximum of 20 connections.
"""

result = compressor.compress(verbose_output)
print(result.compressed)           # key sentences only, filler removed
print(f"Saved: {result.savings_pct:.0%}")  # e.g. "Saved: 47%"
print(result.original_tokens)      # 120
print(result.compressed_tokens)    # 64
```

## CompressMiddleware — Wrap Agents with Automatic Compression

```python
from pyagent_compress import CompressMiddleware, MessageCompressor
from pyagent_patterns.base import Agent
from pyagent_patterns.orchestration import Pipeline

compressor = MessageCompressor(target_ratio=0.4)
middleware = CompressMiddleware(compressor=compressor)

pipeline = Pipeline(stages=[
    middleware.wrap(Agent("researcher", llm,
                         system_prompt="Research thoroughly. Include all relevant details.")),
    middleware.wrap(Agent("analyst", llm,
                         system_prompt="Analyse the research findings.")),
    # Final stage: no compression — we want the full output
    Agent("writer", llm, system_prompt="Write the final report."),
])

result = asyncio.run(pipeline.run("Research quantum computing hardware trends"))

# Check compression stats from wrapped agents
for stage in pipeline._stages[:2]:
    if hasattr(stage, "compression_log"):
        for entry in stage.compression_log:
            print(f"{stage.name}: saved {entry['savings_pct']:.0%} "
                  f"({entry['original_tokens']} → {entry['compressed_tokens']} tokens)")
```

## TokenBudget — Enforce Workflow-Wide Token Limits

```python
from pyagent_compress import TokenBudget

# Set a 50k total token budget, 10k per agent
budget = TokenBudget(
    workflow_limit=50_000,
    per_agent_limit=10_000,
    strict=True,  # raises BudgetExceededError if exceeded; False = track only
)

# Custom per-agent limits
budget.register_agent("researcher", limit=15_000)  # researcher gets more
budget.register_agent("writer", limit=5_000)        # writer gets less

# Track consumption
budget.consume("researcher", 3000)
print(budget.remaining())              # 47000 (workflow remaining)
print(budget.remaining("researcher"))  # 12000 (researcher remaining)
print(budget.workflow_utilization)     # 0.06 → 6% of budget used

# Full summary
print(budget.summary())
# {
#     "workflow": {"limit": 50000, "used": 3000, "remaining": 47000, "utilization": 0.06},
#     "researcher": {"limit": 15000, "used": 3000, "remaining": 12000, "utilization": 0.20},
#     "writer": {"limit": 5000, "used": 0, "remaining": 5000, "utilization": 0.0},
# }

# Integrate with CompressMiddleware for automatic budget tracking
middleware = CompressMiddleware(budget=budget)
```

## AgentPruner — Detect Non-Contributing Agents

Identifies agents that repeat what others already said rather than adding new information.

```python
from pyagent_compress import AgentPruner

pruner = AgentPruner(
    min_contribution=0.3,  # agents scoring below 0.3 should be pruned
    window_size=5,         # look at last 5 messages per agent
)

# After running a multi-agent pattern:
scores = pruner.score_agents(result.messages, task="Design a distributed cache")
for score in scores:
    print(f"{score.agent_name}: {score.score:.2f} "
          f"(unique info: {score.unique_info:.2f}, messages: {score.message_count})")
# analyst_1: 0.72 (unique info: 0.65, messages: 3)
# analyst_2: 0.18 (unique info: 0.08, messages: 3)  ← prune this one
# analyst_3: 0.61 (unique info: 0.54, messages: 3)

agents_to_prune = pruner.should_prune(scores)
print(f"Prune: {agents_to_prune}")  # ["analyst_2"]
```

## InteractionPruner — Detect Early Consensus

Skip remaining rounds when agents have converged.

```python
from pyagent_compress import InteractionPruner

interaction_pruner = InteractionPruner(
    consensus_threshold=0.7,  # 70% similarity = consensus reached
    min_rounds=1,             # always run at least 1 round
)

# Check after each round in a multi-round pattern
if interaction_pruner.has_consensus(current_round_outputs, current_round=2):
    print("Consensus reached — skipping remaining rounds")
```

## Integration Example

```python
import asyncio
from pyagent_patterns.base import Agent, MockLLM
from pyagent_patterns.orchestration import Pipeline
from pyagent_compress import MessageCompressor, CompressMiddleware, TokenBudget

llm = MockLLM(responses=[
    "Very detailed research output with lots of filler and verbose explanations...",
    "Thorough analysis building on the research findings...",
    "Final concise report based on the compressed analysis.",
])

budget = TokenBudget(workflow_limit=10_000, per_agent_limit=5_000, strict=True)
middleware = CompressMiddleware(
    compressor=MessageCompressor(target_ratio=0.4),
    budget=budget,
)

pipeline = Pipeline(stages=[
    middleware.wrap(Agent("researcher", llm, system_prompt="Research thoroughly.")),
    middleware.wrap(Agent("analyst", llm, system_prompt="Analyse findings.")),
    Agent("writer", llm, system_prompt="Write final report."),
])

result = asyncio.run(pipeline.run("Analyse quantum computing trends"))
print(f"Budget used: {budget.total_used} / {budget.workflow_limit}")
```

## Architecture

```mermaid
flowchart LR
    subgraph Agent Pipeline
        A1[Agent 1] -->|verbose output| MW[CompressMiddleware]
        MW -->|compressed output| A2[Agent 2]
    end

    subgraph Compression Engine
        MW --> FR[Remove Filler Phrases]
        FR --> SS[Split into Sentences]
        SS --> RK[Rank by Info Density]
        RK --> TP[Keep Top N to Target Ratio]
    end

    subgraph Budget & Tracking
        MW --> TB[TokenBudget]
        MW -.->|emit| TE[TraceEventBus]
    end
```

### How Compression Works (Internal Pipeline)

The `MessageCompressor` processes text through four stages:

1. **Filler removal** — Strips phrases like "let me think", "basically", "in other words", "it's worth noting" that add no information
2. **Sentence splitting** — Breaks text into individual sentences for scoring
3. **Information density ranking** — Scores each sentence by keyword density, specificity, and uniqueness relative to other sentences
4. **Top-N selection** — Keeps the highest-scoring sentences up to the `target_ratio` threshold, preserving their original order

### Compression Result

`MessageCompressor.compress()` returns a `CompressionResult` with:

| Field | Type | Description |
|-------|------|-------------|
| `compressed` | `str` | The compressed text |
| `original_tokens` | `int` | Token count before compression |
| `compressed_tokens` | `int` | Token count after compression |
| `savings_pct` | `float` | Percentage of tokens saved (0.0–1.0) |
| `sentences_kept` | `int` | Number of sentences retained |
| `sentences_removed` | `int` | Number of sentences dropped |

## Integration with pyagent-trace

When `CompressMiddleware` is used alongside a `TraceEventBus`, compression events are emitted for every compressed agent output, enabling visibility in Studio:

```python
from pyagent_trace.events import TraceEventBus
from pyagent_compress import CompressMiddleware, MessageCompressor

bus = TraceEventBus()
middleware = CompressMiddleware(
    compressor=MessageCompressor(target_ratio=0.5),
)

# Each wrapped agent's output triggers a "compression" trace event:
# TraceEvent(event_type="compression", data={
#     "agent": "researcher",
#     "original_tokens": 2000,
#     "compressed_tokens": 1050,
#     "savings_pct": 0.475,
# })
```

This enables:
- **Studio Trace Viewer** — See compression events alongside agent and LLM call events
- **Cost Dashboard** — Track how much compression saves in token costs
- **Budget Monitoring** — Real-time `TokenBudget` utilization in the dashboard

## Integration with pyagent-context

Compression and context work together to manage token budgets across agent communication:

```python
from pyagent_context import ContextLedger, ContextCompressor
from pyagent_compress import MessageCompressor, TokenBudget

# Context-level compression (pyagent-context)
# Manages which context items to keep/drop based on trust and relevance
context_compressor = ContextCompressor(policy="semantic_lossless")
compressed_items = context_compressor.compress(ledger.items(), target_tokens=4000)

# Message-level compression (pyagent-compress)
# Reduces the verbosity of individual agent outputs
msg_compressor = MessageCompressor(target_ratio=0.5)
result = msg_compressor.compress(agent_output)

# Token budget spans both layers
budget = TokenBudget(workflow_limit=50_000, per_agent_limit=10_000)
```

**Key distinction:**
- `pyagent-compress` compresses **inter-agent messages** (output of one agent before it becomes input to the next)
- `pyagent-context` compresses **context items** (the accumulated context ledger that agents read from)

Both contribute to staying within `TokenBudget` limits.

## Integration with pyagent-patterns

`CompressMiddleware.wrap()` returns an agent-compatible wrapper that can be used anywhere a regular `Agent` is expected:

```python
from pyagent_patterns.orchestration import Pipeline, Supervisor
from pyagent_patterns.resolution import Debate
from pyagent_compress import CompressMiddleware, MessageCompressor

middleware = CompressMiddleware(compressor=MessageCompressor(target_ratio=0.4))

# Works with any pattern — Pipeline, Supervisor, Debate, etc.
pipeline = Pipeline(stages=[
    middleware.wrap(agent1),  # output compressed before reaching agent2
    middleware.wrap(agent2),  # output compressed before reaching agent3
    agent3,                   # final stage: no compression (full output)
])

# For multi-round patterns, compression reduces token accumulation
debate = Debate(
    debaters=[middleware.wrap(bull), middleware.wrap(bear)],
    judge=judge,
    rounds=3,
)
```

## Other packages in this pillar

- `pyagent-patterns` — 18 orchestration patterns
- `pyagent-providers` — multi-provider registry
- `pyagent-router` — difficulty-based model routing

## Full stack

Install all pillars at once: `pip install pyagent-all`

→ [pyagent.org](https://pyagent.org) for full documentation.

## Full Documentation

See [pyagent.org](https://pyagent.org) for full API reference and integration guides.

<!-- pyagent-ecosystem-footer:start -->

## The PyAgent ecosystem

PyAgent is a production stack for multi-agent LLM systems. Each package is independent — install
only what you need, or get everything with [`pip install pyagent-all`](https://pyagent.org/getting-started/).

| Package | What it gives you |
|---------|-------------------|
| `pyagent-blueprint` | [Declarative multi-agent blueprints](https://pyagent.org/guides/blueprint/) — compile, validate, and diff agent systems from YAML |
| `pyagent-patterns` | [Reusable multi-agent design patterns](https://pyagent.org/packages/patterns/) — Supervisor, Pipeline, ReAct, and 15 more |
| `pyagent-router` | [Difficulty-aware model routing](https://pyagent.org/guides/router/) — cost-efficient model selection per task |
| `pyagent-compress` | [Token-efficient agent compression](https://pyagent.org/guides/compression/) — inter-agent token budgets |
| `pyagent-providers` | [Multi-provider orchestration](https://pyagent.org/guides/providers/) — fallback chains and capability negotiation |
| `pyagent-context` | [Stateful agent memory](https://pyagent.org/guides/context/) — trust-aware, three-tier context ledger |
| `pyagent-trace` | [Multi-agent observability & tracing](https://pyagent.org/guides/tracing/) — pattern-aware OpenTelemetry spans |
| `pyagent-studio` | [Agent control plane dashboard](https://pyagent.org/guides/studio/) — live traces, cost, and governance |

**Learn the concepts:**
[The Orchestrator-Worker pattern](https://pyagent.org/patterns/orchestrator-worker/) ·
[Engineering a resilient multi-agent harness](https://pyagent.org/architecture/multi-agent-harness/) ·
[Agent Experience Optimization (AXO)](https://pyagent.org/concepts/agent-experience-optimization/)

<!-- pyagent-ecosystem-footer:end -->
