Agent integrations¶
IntelligentSummarizationMiddleware is LangChain agent middleware. It works
with agents built by create_agent and with frameworks built on it, such as
Deep Agents and LangGraph Swarm.
LangChain agents¶
Pass the middleware to create_agent, as in the
quickstart. Use one summarization
middleware per agent: ContextSage replaces LangChain's
SummarizationMiddleware rather than running alongside it.
When an agent has several middleware, their before_model hooks run in list
order. Put ContextSage where you would put SummarizationMiddleware,
typically after middleware that adds messages to the history and before
middleware that only inspects the model call.
Synchronous and asynchronous agents¶
invoke and stream run the synchronous hook, and ainvoke and astream
the asynchronous one; both behave identically. Analysis is CPU-bound, so the
asynchronous hook runs it in a worker thread and never blocks the event loop.
parsefabric parses asynchronously, so the synchronous hook runs it on a
private event loop, which leaves any event loop your thread has set
untouched.
Checkpointers and threads¶
With a checkpointer, the rewritten history is saved in the thread's state like any other update, and each later turn starts from it. A summary that is summarized again becomes the next generation of the same lineage, and provenance links are scoped to the thread; see Provenance and lineage.
Streaming¶
When you stream an agent with stream_events(version="v3"), the summary
model's tokens are kept out of the agent's message stream: ContextSage marks
the summary call as internal, as LangChain does for its own summarization.
Deep Agents¶
Deep Agents add LangChain's SummarizationMiddleware to every deep agent.
Exclude it with a harness profile for your model provider, so that
ContextSage replaces it instead of running alongside it:
"""Replace a deep agent's built-in summarization with ContextSage.
Run: python examples/deep_agent.py
Requires CONTEXTSAGE_LLM_API_KEY (and optionally CONTEXTSAGE_LLM_BASE_URL and
CONTEXTSAGE_LLM_MODEL); see .env.example. ``deepagents`` adds LangChain's
``SummarizationMiddleware`` to every deep agent. A harness profile removes it
so that ContextSage replaces it rather than running alongside it.
"""
from deepagents import HarnessProfile, create_deep_agent, register_harness_profile
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from _models import live_model
from contextsage import IntelligentSummarizationMiddleware, SummarizationEvent
@tool
def deployment_log(service: str) -> str:
"""Return the deployment log of a service."""
steps = "".join(
f"2026-10-04T12:{index // 60:02d}:{index % 60:02d}Z INFO {service} rollout "
"step ok\n"
for index in range(300)
)
return (
f"Deployment DEP-7310 of {service}.\n{steps}"
"2026-10-04T12:05:01Z ERROR health check failed: readiness probe timeout\n"
)
register_harness_profile(
"openai",
HarnessProfile(excluded_middleware=frozenset({"SummarizationMiddleware"})),
)
model = live_model()
events: list[SummarizationEvent] = []
agent = create_deep_agent(
model=model,
tools=[deployment_log],
system_prompt="You investigate failed deployments. Be concise.",
middleware=[
IntelligentSummarizationMiddleware(
model=model,
trigger=("tokens", 2_000),
keep=("messages", 6),
observability_hook=events.append,
)
],
)
result = agent.invoke(
{"messages": [HumanMessage("Why did the checkout deployment fail?")]}
)
print("Answer:", result["messages"][-1].text)
for event in events:
print(f"summarization: {event.input_tokens} -> {event.output_tokens} tokens")
LangGraph Swarm¶
In a swarm, add ContextSage to the agents whose histories grow large, such as those that call tools with large results. Each agent's middleware sees the shared conversation before that agent's model calls:
"""Use ContextSage on one agent of a multi-agent swarm.
Run: python examples/swarm.py
Requires CONTEXTSAGE_LLM_API_KEY (and optionally CONTEXTSAGE_LLM_BASE_URL and
CONTEXTSAGE_LLM_MODEL); see .env.example. A triage agent hands the
conversation to an investigator whose tool returns a large report; only the
investigator runs ContextSage, and the swarm keeps its state in a
checkpointer across turns.
"""
from collections.abc import Callable
from langchain.agents import create_agent
from langchain.agents.middleware import ModelRequest, ModelResponse, wrap_model_call
from langchain_core.messages import HumanMessage, ToolMessage
from langchain_core.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph_swarm import create_handoff_tool, create_swarm
from _models import live_model
from contextsage import IntelligentSummarizationMiddleware, SummarizationEvent
@wrap_model_call
def without_speaker_names(
request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
"""Drop the agent name that swarm agents stamp on their messages.
Some OpenAI-compatible endpoints accept ``name`` only on tool messages.
"""
messages = [
message
if isinstance(message, ToolMessage) or message.name is None
else message.model_copy(update={"name": None})
for message in request.messages
]
return handler(request.override(messages=messages))
@tool
def billing_records(account: str) -> str:
"""Return recent billing records for an account."""
rows = "".join(
f"2026-10-0{1 + index % 3}T08:00:{index % 60:02d}Z "
f"INFO invoice batch {account} ok\n"
for index in range(300)
)
return (
f"Billing records for {account}.\n{rows}"
"2026-10-04T08:59:59Z ERROR invoice INV-8812 charged twice for account "
f"{account}\n"
)
model = live_model()
events: list[SummarizationEvent] = []
triage = create_agent(
model,
tools=[
create_handoff_tool(
agent_name="investigator",
description="Hand billing questions to the investigator.",
)
],
system_prompt="You triage support requests. Hand billing questions over.",
name="triage",
middleware=[without_speaker_names],
)
investigator = create_agent(
model,
tools=[billing_records],
system_prompt="You investigate billing problems with the billing_records tool.",
name="investigator",
middleware=[
IntelligentSummarizationMiddleware(
model=model,
trigger=("tokens", 2_000),
keep=("messages", 6),
observability_hook=events.append,
),
without_speaker_names,
],
)
swarm = create_swarm([triage, investigator], default_active_agent="triage").compile(
checkpointer=InMemorySaver()
)
config = {"configurable": {"thread_id": "billing-1"}}
result = swarm.invoke(
{"messages": [HumanMessage("Account ACC-42 was charged twice. Why?")]}, config
)
print("Answer:", result["messages"][-1].text)
for event in events:
print(
f"summarization in thread {event.thread_id}: "
f"{event.input_tokens} -> {event.output_tokens} tokens"
)
without_speaker_names is only needed for OpenAI-compatible endpoints that
reject the name that swarm agents put on their messages; the OpenAI API
accepts it.
Run the live examples¶
These examples call a real model. Configure an OpenAI-compatible endpoint
through the environment, or a local .env file, as described in the
examples README,
then run, for example: