Compression by framework¶

Every framework keeps its own message list. compress_for converts that list to the adapter's Message, compresses it and converts it back to the same type. Messages that are kept come back as the very same objects; changed ones keep every field but the text. Tool-call ids stay paired, and pinned messages stay verbatim. Approval items stay too when they are marked: the adapter's approval results, MCP approval requests and responses, and any item whose type names an approval or confirmation (ADK adk_request_confirmation). Pass pinned= for a framework's approval state that carries no such marker.

flowchart LR
  F["framework messages<br/>(dicts, BaseMessage, Content)"] --> C["compress_for(messages, policy)"]
  C --> T["to Message"] --> S["compress()"] --> B["back to the same type"]
  B --> M["the framework's model call"]

Format	`format=`	Covers
`Message`	`"agent_adapter"`	`run_tool_loop`, `CallableModel`, your own code
LangChain `BaseMessage`	`"langchain"`	LangChain, LangGraph
OpenAI chat dicts and Responses items	`"openai"`	OpenAI, OpenAI Agents SDK, any OpenAI-compatible API
Anthropic Messages dicts	`"anthropic"`	Anthropic
Gemini `contents` (dicts or `types.Content`)	`"gemini"`	Gemini, Google ADK
Pydantic AI, Agno, Strands, smolagents, AutoGen and AG2 messages	`"pydantic_ai"`, `"agno"`, `"strands"`, `"smolagents"`, `"autogen"`, `"ag2"`	registered when you import that framework's integration module
LlamaIndex `ChatMessage`, Semantic Kernel `ChatMessageContent`, Agent Framework `Message`, CAMEL-AI `MemoryRecord`	`"llamaindex"`, `"semantic_kernel"`, `"agent_framework"`, `"camel"`	registered when you import that framework's integration module
CrewAI `LLMMessage`, the messages a DSPy adapter builds	`"openai"`	they are OpenAI chat dicts

format="auto" (the default) detects it; plain {"role", "content"} dicts count as OpenAI. Pin native messages with pinned=lambda message: ...: no marker is added to them. Your own integration can add a format with register_format(name, converter, detect=...) from agent_adapter.context.formats.

Where to hook¶

Integration	Where to hook	Format	Snippet
No framework (FastAPI, a script)	`LLMRequest(compression=)`	`agent_adapter`	how-to
The adapter's own loop	`run_tool_loop(compression=)`	`agent_adapter`	how-to
OpenAI	before `chat.completions.create`	`openai`	OpenAI
Anthropic	before `messages.create`	`anthropic`	Anthropic
Gemini	before `models.generate_content`	`gemini`	Gemini
LangChain	before `model.invoke(messages)`	`langchain`	LangChain
LangGraph	in your agent node: `langgraph_compressor(policy)`	`langchain`	LangGraph
OpenAI Agents SDK	`RunConfig(call_model_input_filter=...)`	`openai`	OpenAI Agents
Google ADK	`LlmAgent(before_model_callback=...)`	`gemini`	ADK
Haystack	`ChatMessage.to_openai_dict_format()` before, `ChatMessage.from_openai_dict_format()` after	`openai`	OpenAI
Pydantic AI	`Agent(capabilities=[ProcessHistory(pydantic_ai_history_processor(policy))])`	`pydantic_ai`	Agent frameworks
Agno	`Agent(model=agno_compressed_model(model, policy))`	`agno`	Agent frameworks
Strands Agents	`agent.add_middleware(InvokeModelStage.Input, strands_compress_middleware(policy))`	`strands`	Agent frameworks
smolagents	`CodeAgent` or `ToolCallingAgent(model=smolagents_compressed_model(model, policy))`	`smolagents`	Agent frameworks
AutoGen	`AssistantAgent(model_context=autogen_model_context(policy))`	`autogen`	Agent frameworks
AG2	`Agent(assembly=[ag2_compress_policy(policy)])`	`ag2`	Agent frameworks
CrewAI	`register_before_llm_call_hook(...)`	`openai`	More frameworks
DSPy	`dspy.configure(adapter=...)`, a `ChatAdapter` subclass	`openai`	More frameworks
LlamaIndex	before `llm.achat(messages)`, or on an agent's `chat_history=`	`llamaindex`	More frameworks
Semantic Kernel	before `get_chat_message_contents(chat_history=...)`	`semantic_kernel`	More frameworks
CAMEL-AI	`ChatAgent(memory=ChatHistoryMemory(...))` with a context creator	`camel`	More frameworks
Microsoft Agent Framework	`Agent(middleware=[...])`, a chat middleware	`agent_framework`	More frameworks
Claude Agent SDK	its harness compacts its own context	—	—
MCP, MCP tasks, A2A, AG-UI, OpenAPI	tools and protocols: no model history here	—	—

OpenAI¶

def ask_openai(client: Any, messages: list[dict]) -> Any:
    smaller, _ = compress_for_sync(messages, policy)  # chat dicts in, chat dicts out
    return client.chat.completions.create(model="gpt-4.1", messages=smaller)

Anthropic¶

The system prompt is a separate argument, so it is never in the list:

def ask_anthropic(client: Any, system: str, messages: list[dict]) -> Any:
    smaller, _ = compress_for_sync(messages, policy, format="anthropic")
    return client.messages.create(
        model="claude-sonnet-4-5", system=system, messages=smaller, max_tokens=1024
    )

Gemini¶

def ask_gemini(client: Any, contents: list[Any]) -> Any:
    smaller, _ = compress_for_sync(contents, policy, format="gemini")  # dicts or types.Content
    return client.models.generate_content(model="gemini-2.5-flash", contents=smaller)

LangGraph and LangChain¶

In a LangGraph node, compress what the model gets; the graph state keeps the full history:

compressor = langgraph_compressor(policy)


async def agent(state: State) -> dict[str, Any]:
    messages = await compressor(state["messages"])  # or compressor.sync(...) in a sync node
    sent.append(len(messages))
    return {"messages": [await llm.ainvoke(messages)]}  # the state keeps the full history

With LangChain alone, compress before the call:

def ask(messages: list[Any]) -> Any:
    smaller, _ = compress_for_sync(messages, policy)  # BaseMessage in, BaseMessage out
    return llm.invoke(smaller)

OpenAI Agents SDK¶

The filter gets the Responses input items of each model call, reasoning items included:

async def compress_model_input(data: Any) -> Any:
    """RunConfig(call_model_input_filter=compress_model_input): Responses items each turn."""
    smaller, _ = await compress_for(data.model_data.input, policy, format="openai")
    return dataclasses.replace(data.model_data, input=smaller)

Google ADK¶

def before_model(callback_context: Any, llm_request: Any) -> None:
    """LlmAgent(before_model_callback=before_model): shorten what this model call sends."""
    llm_request.contents, _ = compress_for_sync(llm_request.contents, policy, format="gemini")

Agent frameworks¶

Import the hook from the framework's integration module; that import also registers the format for compress_for(format="auto"). Pydantic AI keeps the processed history for the rest of the run; the others shorten only what each model call sends.

def pydantic_ai_agent(model: Any) -> Any:
    from pydantic_ai import Agent
    from pydantic_ai.capabilities import ProcessHistory

    return Agent(model, capabilities=[ProcessHistory(pydantic_ai_history_processor(policy))])


def agno_agent(model: Any) -> Any:  # model: any Agno model, e.g. OpenAIChat(id="gpt-4.1")
    from agno.agent import Agent

    return Agent(model=agno_compressed_model(model, policy))


def strands_agent(agent: Any) -> Any:  # agent: a strands.Agent
    from strands.middleware import InvokeModelStage

    agent.add_middleware(InvokeModelStage.Input, strands_compress_middleware(policy))
    return agent


def smolagents_agent(model: Any, tools: list[Any]) -> Any:
    from smolagents import ToolCallingAgent

    return ToolCallingAgent(tools=tools, model=smolagents_compressed_model(model, policy))


def autogen_agent(model_client: Any) -> Any:
    from autogen_agentchat.agents import AssistantAgent

    context = autogen_model_context(policy)
    return AssistantAgent("assistant", model_client=model_client, model_context=context)


def ag2_agent(config: Any) -> Any:  # config: an AG2 ModelConfig
    from ag2 import Agent

    return Agent("assistant", config=config, assembly=[ag2_compress_policy(policy)])

More agent frameworks¶

CrewAI and DSPy send OpenAI chat dicts, so they use openai. The CrewAI hook shortens the agent's own list in place. The other four send the model a shorter copy, and the history keeps every message. Tool rounds that Semantic Kernel or a LlamaIndex agent run inside one call are not seen by these hooks; Agent Framework middleware sees every call. A framework's own compaction (CrewAI respect_context_window, CAMEL summarize_threshold, Agent Framework compaction_strategy) can run instead; use one, not both.

def compress_crewai_messages(context: Any) -> None:
    """`register_before_llm_call_hook(compress_crewai_messages)`, or decorate `@before_llm_call`."""
    smaller, _ = compress_for_sync(context.messages, policy, format="openai")
    context.messages[:] = smaller  # in place: CrewAI keeps this very list as the history


class CompressingChatAdapter(dspy.ChatAdapter):
    """`dspy.configure(adapter=CompressingChatAdapter())`: history, demos and the new input."""

    def format(self, signature: Any, demos: list[Any], inputs: dict[str, Any]) -> list[Any]:
        messages = super().format(signature, demos, inputs)
        smaller, _ = compress_for_sync(messages, policy, format="openai")  # OpenAI chat dicts
        return smaller


import agent_adapter.integrations.llamaindex  # noqa: E402, F401  registers its format


async def ask_llamaindex(llm: Any, messages: list[Any]) -> Any:
    """`messages` is a `list[ChatMessage]`; compress an agent's `chat_history=` the same way."""
    smaller, _ = await compress_for(messages, policy, format="llamaindex")
    return await llm.achat(smaller)


import agent_adapter.integrations.semantic_kernel  # noqa: E402, F401  registers its format


async def ask_semantic_kernel(service: Any, history: Any, settings: Any) -> Any:
    """The service gets a shorter copy; `history` itself keeps every message."""
    smaller, _ = await compress_for(history.messages, policy, format="semantic_kernel")
    shorter = history.model_copy(update={"messages": smaller})
    return await service.get_chat_message_contents(chat_history=shorter, settings=settings)


import agent_adapter.integrations.camel  # noqa: E402, F401  registers its format


class CompressingContextCreator(ScoreBasedContextCreator):
    """`ChatAgent(memory=ChatHistoryMemory(CompressingContextCreator(counter, limit)))`."""

    def create_context(self, records: list[ContextRecord]) -> Any:
        memory = [record.memory_record for record in records]
        smaller, _ = compress_for_sync(memory, policy, format="camel")
        kept = [ContextRecord(memory_record=r, score=1.0, timestamp=r.timestamp) for r in smaller]
        return super().create_context(kept)  # the memory itself keeps every record


import agent_adapter.integrations.agent_framework  # noqa: E402, F401  registers its format


@chat_middleware
async def compress_messages(context: ChatContext, call_next: Any) -> None:
    """`Agent(client=..., middleware=[compress_messages])`: the session keeps the full history."""
    context.messages, _ = await compress_for(context.messages, policy, format="agent_framework")
    await call_next()

Rules for every framework¶

compress_for never calls your main model; AI steps call the models you gave them.
In async code use await compress_for(...); compress_for_sync is for sync code.
A framework that compacts on its own (OpenAI's OpenAIResponsesCompactionSession, LangChain's summarisation middleware) can run instead; use one, not both.