agentstate-reducer 0.1.0

Framework-agnostic message reducer for AI agent state management. Works with LangGraph, CrewAI, and plain dicts.

agentstate-reducer

Framework-agnostic message reducer for AI agent state management. Works with LangGraph, CrewAI, and plain dicts.

What It Does

Automatically prunes message history when it exceeds a threshold, keeping conversations manageable:

• Windowed pruning: Trigger at max_messages, retain min_messages
• System message preservation: Index 0 (system prompt) is never pruned
• ToolMessage cascade: When an AI message is pruned, linked ToolMessages are pruned too
• Optional summarization: Get a callback with pruned messages to generate summaries

Install

pip install agentstate-reducer

Zero dependencies. Works with Python 3.10+.

Quick Start

from agentstate_reducer import MessageReducer

reducer = MessageReducer(min_messages=10, max_messages=20)

# Works with plain dicts
result = reducer.reduce(
    existing=[
        {"role": "system", "content": "You are helpful"},
        {"role": "human", "content": "Hello"},
        {"role": "ai", "content": "Hi there!"},
    ],
    new=[{"role": "human", "content": "New message"}],
)
print(result.surviving)  # Messages that remain
print(result.pruned)     # Messages that were removed

# Works with LangChain BaseMessage objects too
from langchain_core.messages import HumanMessage, AIMessage
result = reducer.reduce(
    existing=[HumanMessage(content="Hello")],
    new=[AIMessage(content="Hi!")],
)

LangGraph Integration

Use with PrunableStateFactory or directly as a state annotation:

from agentstate_reducer import MessageReducer
from typing_extensions import Annotated, TypedDict

reducer = MessageReducer(min_messages=10, max_messages=20)

class MyState(TypedDict):
    messages: Annotated[list, reducer.as_langgraph_reducer()]

CrewAI Integration

Use with a CosmosDB persistence backend:

from agentstate_reducer import MessageReducer

reducer = MessageReducer(min_messages=10, max_messages=20)
# Pass to your persistence class which calls reducer.reduce()
# on save_state() to prune before storing

Summarization

from agentstate_reducer import MessageReducer, ReducerConfig

def summarize(pruned_messages):
    # Call your LLM to summarize what was removed
    return f"Summary of {len(pruned_messages)} messages"

config = ReducerConfig(
    min_messages=10,
    max_messages=20,
    summarize_fn=summarize,
)
reducer = MessageReducer(config=config)
result = reducer.reduce(existing=messages)
print(result.summary)  # "Summary of 5 messages"

API

MessageReducer(min_messages, max_messages, *, config)

Param	Default	Description
min_messages	0	Messages to retain after pruning
max_messages	None	Threshold to trigger pruning (None = never prune)
config	None	ReducerConfig object (overrides other params)

reducer.reduce(existing, new) -> ReducerResult

Field	Type	Description
surviving	list	Messages that remain
pruned	list	Messages that were removed
summary	str | None	Summary from summarize_fn if configured

reducer.as_langgraph_reducer() -> Callable

Returns a function with signature (existing, new) -> list for use with LangGraph's Annotated[list, fn] pattern.