Examples¶
Every example is a short, runnable script from the
examples/
directory. The offline examples use scripted chat models, so they run without
network access or API keys, and the output shown is the recorded output that
the test suite checks on every change.
To run them, clone the repository and use uv:
The scripts share two small helpers from
examples/_models.py:
scripted_model, a LangChain GenericFakeChatModel that answers with fixed
replies, and live_model, which builds a ChatOpenAI model from environment
variables.
Getting started¶
Quickstart¶
A create_agent agent whose history triggers compaction, a summary and
validation in one turn.
"""Add ContextSage to a LangChain agent and watch it summarize a long history.
Run: python examples/quickstart.py
The agent starts from a history whose large tool result pushes it past the
trigger. Before the next model call, ContextSage compacts the repetitive log
and JSON in that result, LangChain writes the summary, and ContextSage checks
that the incident and transaction IDs survived.
"""
import json
from langchain.agents import create_agent
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from _models import first_line, scripted_model
from contextsage import IntelligentSummarizationMiddleware, SummarizationEvent
events: list[SummarizationEvent] = []
middleware = IntelligentSummarizationMiddleware(
model=scripted_model(
"The payments outage INC-9931 was traced to transaction TX-991 and a "
"saturated connection pool."
),
trigger=("tokens", 2_000),
keep=("messages", 2),
observability_hook=events.append,
)
agent = create_agent(
model=scripted_model("INC-9931 only affects the payments service."),
tools=[],
middleware=[middleware],
)
log = "".join(
f"2026-10-04T03:{index // 60:02d}:{index % 60:02d}Z INFO payments-worker "
"processing queue\n"
for index in range(200)
)
report = (
"Incident INC-9931: elevated latency on the payments service.\n"
+ json.dumps({"orders": [{"status": "queued"}] * 40, "region": "eu-west-1"})
+ "\n"
+ log
+ "2026-10-04T04:00:00Z ERROR connection pool exhausted for TX-991\n"
)
history = [
HumanMessage("Investigate the payments outage."),
AIMessage("", tool_calls=[{"id": "call-1", "name": "incident_report", "args": {}}]),
ToolMessage(report, tool_call_id="call-1"),
AIMessage("INC-9931 is caused by connection pool exhaustion."),
HumanMessage("Is any other service affected?"),
]
result = agent.invoke({"messages": history})
print("History after the agent turn:")
for message in result["messages"]:
print(f" {message.type:<5} {first_line(message.text)}")
event = events[0]
print()
print(f"tokens before: {event.input_tokens}")
print(f"after compaction: {event.prepared_tokens}")
print(f"after summary: {event.output_tokens}")
print(f"compression ratio: {event.compression_ratio:.3f}")
print(f"validation: {event.validation_status}")
print(f"must-keep facts: {event.must_preserve_facts}")
History after the agent turn:
human Here is a summary of the conversation to date:
ai INC-9931 is caused by connection pool exhaustion.
human Is any other service affected?
ai INC-9931 only affects the payments service.
tokens before: 3283
after compaction: 166
after summary: 70
compression ratio: 0.021
validation: passed
must-keep facts: 2
Large tool output¶
A mixed MCP-style tool result: repeated log lines and JSON records collapse, while every distinct line, warning, error and identifier stays.
"""Compact a large, mixed tool result without losing what matters.
Run: python examples/large_tool_output.py
A single MCP-style tool result mixes prose, a JSON payload with repeated
records, a long routine log, warnings and an error. The history is too short
to summarize, so ContextSage only compacts: repeated log lines collapse into
one marker with the omitted line range, identical JSON records collapse into a
counted entry, and every distinct line, warning, error and identifier stays.
"""
import json
from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, ToolMessage
from langgraph.runtime import Runtime
from _models import scripted_model
from contextsage import IntelligentSummarizationMiddleware, SummarizationEvent
events: list[SummarizationEvent] = []
middleware = IntelligentSummarizationMiddleware(
model=scripted_model("unused"),
trigger=("tokens", 1_000),
keep=("messages", 20),
observability_hook=events.append,
)
heartbeats = "".join(
f"2026-10-04T08:{index // 60:02d}:{index % 60:02d}Z "
f"INFO checkout-worker-{index % 4} heartbeat ok\n"
for index in range(150)
)
result = {
"service": "checkout",
"region": "eu-west-1",
"orders": [{"sku": "A-1", "qty": 1, "state": "queued"}] * 25
+ [{"sku": "B-7", "qty": 2, "state": "failed", "order_id": "ORD-5521"}],
}
tool_output = (
"Diagnostics for incident INC-4410 (checkout latency).\n"
+ json.dumps(result)
+ "\n"
+ heartbeats
+ "2026-10-04T08:02:31Z WARN slow query on orders took 2300 ms\n"
+ "2026-10-04T08:02:32Z ERROR payment authorization failed for order ORD-5521\n"
)
history: list[AnyMessage] = [
HumanMessage("Why is checkout slow?"),
AIMessage("", tool_calls=[{"id": "call-1", "name": "diagnostics", "args": {}}]),
ToolMessage(tool_output, tool_call_id="call-1"),
]
update = middleware.before_model({"messages": history}, Runtime())
if update is None:
raise SystemExit("The tool result was not compacted.")
compacted = update["messages"][-1].text
before, after = len(tool_output.splitlines()), len(compacted.splitlines())
print(f"{before} lines before, {after} after")
print()
print(compacted)
event = events[0]
print(f"tokens: {event.input_tokens} -> {event.output_tokens}")
print(f"compacted units: {event.compacted_units}")
print(f"content kinds: {event.content_kinds}")
154 lines before, 7 after
Diagnostics for incident INC-4410 (checkout latency).
{"service":"checkout","region":"eu-west-1","orders":[{"__repeated__":25,"value":{"sku":"A-1","qty":1,"state":"queued"}},{"sku":"B-7","qty":2,"state":"failed","order_id":"ORD-5521"}]}
2026-10-04T08:00:00Z INFO checkout-worker-0 heartbeat ok
... 148 similar lines omitted (lines 4-151) ...
2026-10-04T08:02:29Z INFO checkout-worker-1 heartbeat ok
2026-10-04T08:02:31Z WARN slow query on orders took 2300 ms
2026-10-04T08:02:32Z ERROR payment authorization failed for order ORD-5521
tokens: 2538 -> 173
compacted units: 2
content kinds: {'json': 1, 'log': 2, 'text': 2}
Preservation¶
Restated facts¶
A vague summary loses a standing instruction, identifiers, a correction and a conflict; ContextSage restates them.
"""Keep corrections, instructions and conflicting evidence through summarization.
Run: python examples/preservation.py
The summary model below writes a vague summary that drops everything that
matters. ContextSage notices that the user's correction, a standing
instruction, an identifier tied to a failure and two conflicting sources are
missing, and restates them verbatim inside the summary message.
"""
from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, ToolMessage
from langgraph.runtime import Runtime
from _models import scripted_model
from contextsage import IntelligentSummarizationMiddleware, SummarizationEvent
events: list[SummarizationEvent] = []
middleware = IntelligentSummarizationMiddleware(
model=scripted_model("The user asked about refunds and an order."),
trigger=("messages", 8),
keep=("messages", 2),
observability_hook=events.append,
)
history: list[AnyMessage] = [
HumanMessage("You must never refund more than 500 dollars without approval."),
HumanMessage("Please refund order ORD-1001 for customer 123."),
AIMessage("", tool_calls=[{"id": "c1", "name": "warehouse_status", "args": {}}]),
ToolMessage("warehouse: status=SHIPPED", tool_call_id="c1"),
AIMessage("", tool_calls=[{"id": "c2", "name": "carrier_status", "args": {}}]),
ToolMessage("carrier: status=LOST, refund claim TX-7781 failed", tool_call_id="c2"),
HumanMessage("No, use customer 456 instead of customer 123."),
AIMessage("Understood."),
HumanMessage("What should we do next?"),
]
update = middleware.before_model({"messages": history}, Runtime())
if update is None:
raise SystemExit("The history did not trigger summarization.")
# The update replaces the history: a RemoveMessage, then the new messages.
summary = update["messages"][1]
print(summary.text)
print()
event = events[0]
print(f"validation: {event.validation_status}, recovery: {event.recovery_status}")
Here is a summary of the conversation to date:
The user asked about refunds and an order.
Facts preserved verbatim from the earlier conversation:
- User instruction: "You must never refund more than 500 dollars without approval."
- ORD-1001
- TX-7781
- User correction: "No, use customer 456 instead of customer 123."
- Conflicting values reported for status: SHIPPED (earlier) vs LOST (later)
validation: failed_recovered, recovery: restated_facts
Policies¶
One JSON payload under the three preservation policies.
"""Compare the three preservation policies on the same tool result.
Run: python examples/policies.py
``maximum_preservation`` only minifies JSON. ``balanced`` (the default) also
collapses runs of identical items, which loses nothing. ``maximum_compression``
groups identical items anywhere in a list and samples long lists.
"""
import json
from langchain_core.messages import AnyMessage, HumanMessage, ToolMessage
from langgraph.runtime import Runtime
from _models import scripted_model
from contextsage import IntelligentSummarizationMiddleware
inventory = {
"warehouse": "berlin-2",
"pallets": (
[{"sku": "A-1", "state": "ok"}] * 3
+ [{"sku": "B-2", "state": "ok"}]
+ [{"sku": "A-1", "state": "ok"}] * 3
+ [{"sku": f"C-{index}", "state": "ok"} for index in range(12)]
),
}
history: list[AnyMessage] = [
HumanMessage("Summarize the warehouse inventory."),
ToolMessage(json.dumps(inventory, indent=2), tool_call_id="call-1"),
]
for policy in ("maximum_preservation", "balanced", "maximum_compression"):
middleware = IntelligentSummarizationMiddleware(
model=scripted_model("unused"),
trigger=("tokens", 200),
keep=("messages", 20),
policy=policy,
)
update = middleware.before_model({"messages": history}, Runtime())
if update is None:
raise SystemExit("The tool result was not compacted.")
compacted = update["messages"][-1].text
print(f"{policy} ({len(compacted)} characters):")
print(f" {compacted}")
print()
maximum_preservation (551 characters):
{"warehouse":"berlin-2","pallets":[{"sku":"A-1","state":"ok"},{"sku":"A-1","state":"ok"},{"sku":"A-1","state":"ok"},{"sku":"B-2","state":"ok"},{"sku":"A-1","state":"ok"},{"sku":"A-1","state":"ok"},{"sku":"A-1","state":"ok"},{"sku":"C-0","state":"ok"},{"sku":"C-1","state":"ok"},{"sku":"C-2","state":"ok"},{"sku":"C-3","state":"ok"},{"sku":"C-4","state":"ok"},{"sku":"C-5","state":"ok"},{"sku":"C-6","state":"ok"},{"sku":"C-7","state":"ok"},{"sku":"C-8","state":"ok"},{"sku":"C-9","state":"ok"},{"sku":"C-10","state":"ok"},{"sku":"C-11","state":"ok"}]}
balanced (497 characters):
{"warehouse":"berlin-2","pallets":[{"__repeated__":3,"value":{"sku":"A-1","state":"ok"}},{"sku":"B-2","state":"ok"},{"__repeated__":3,"value":{"sku":"A-1","state":"ok"}},{"sku":"C-0","state":"ok"},{"sku":"C-1","state":"ok"},{"sku":"C-2","state":"ok"},{"sku":"C-3","state":"ok"},{"sku":"C-4","state":"ok"},{"sku":"C-5","state":"ok"},{"sku":"C-6","state":"ok"},{"sku":"C-7","state":"ok"},{"sku":"C-8","state":"ok"},{"sku":"C-9","state":"ok"},{"sku":"C-10","state":"ok"},{"sku":"C-11","state":"ok"}]}
maximum_compression (353 characters):
{"warehouse":"berlin-2","pallets":[{"__repeated__":6,"value":{"sku":"A-1","state":"ok"}},{"sku":"B-2","state":"ok"},{"sku":"C-0","state":"ok"},{"sku":"C-1","state":"ok"},{"sku":"C-2","state":"ok"},{"__omitted__":4},{"sku":"C-7","state":"ok"},{"sku":"C-8","state":"ok"},{"sku":"C-9","state":"ok"},{"sku":"C-10","state":"ok"},{"sku":"C-11","state":"ok"}]}
Recovery¶
The summary model fails, and the agent continues on a trimmed history.
"""Keep the agent running when the summary model fails.
Run: python examples/recovery.py
The summary model is unavailable. Instead of failing the agent turn,
ContextSage trims the history to the ``keep`` window with LangChain's
``trim_messages``, never orphaning a tool result, and restates the facts that
the trimmed messages carried.
"""
from langchain_core.callbacks import CallbackManagerForLLMRun
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import (
AIMessage,
AnyMessage,
BaseMessage,
HumanMessage,
ToolMessage,
)
from langchain_core.outputs import ChatResult
from langgraph.runtime import Runtime
from _models import first_line
from contextsage import IntelligentSummarizationMiddleware, SummarizationEvent
class UnavailableModel(BaseChatModel):
"""A chat model whose provider is down."""
@property
def _llm_type(self) -> str:
return "unavailable"
def _generate(
self,
messages: list[BaseMessage],
stop: list[str] | None = None,
run_manager: CallbackManagerForLLMRun | None = None,
**kwargs: object,
) -> ChatResult:
raise TimeoutError("the summary model did not respond")
events: list[SummarizationEvent] = []
middleware = IntelligentSummarizationMiddleware(
model=UnavailableModel(),
trigger=("messages", 6),
keep=("messages", 2),
observability_hook=events.append,
)
history: list[AnyMessage] = [
HumanMessage("Check order ORD-2207 for customer 812."),
AIMessage("", tool_calls=[{"id": "c1", "name": "order_status", "args": {}}]),
ToolMessage(
"order ORD-2207: payment failed, gateway error GW-5003", tool_call_id="c1"
),
AIMessage("Payment for ORD-2207 failed with GW-5003."),
HumanMessage("Retry the payment."),
AIMessage("", tool_calls=[{"id": "c2", "name": "retry_payment", "args": {}}]),
ToolMessage("retry accepted", tool_call_id="c2"),
]
update = middleware.before_model({"messages": history}, Runtime())
if update is None:
raise SystemExit("The history did not trigger summarization.")
# The update replaces the history: a RemoveMessage, then the new messages.
print("History after the fallback:")
for message in update["messages"][1:]:
calls = ", ".join(call["name"] for call in getattr(message, "tool_calls", []))
print(f" {message.type:<5} {first_line(message.text, 100) or f'calls {calls}'}")
for line in message.text.splitlines()[1:]:
print(f" {line}")
event = events[0]
print()
print(f"validation: {event.validation_status}, recovery: {event.recovery_status}")
History after the fallback:
human Earlier messages were removed because the summary model was unavailable.
The latest user request was:
Retry the payment.
Facts preserved verbatim from the earlier conversation:
- ORD-2207
- customer 812
- GW-5003
ai calls retry_payment
tool retry accepted
validation: skipped, recovery: trimmed_fallback
Parsing, provenance and observability¶
Custom parsing¶
A parsefabric parser, built on sqlglot, teaches ContextSage to recognize SQL.
"""Teach ContextSage a domain format with a parsefabric parser.
Run: python examples/custom_parsing.py
Requires sqlglot (installed with the ``examples`` dependency group). Nothing
here connects to a database or executes SQL.
``routes`` sends matching lines to your own parsefabric parser, and
``fence_routes`` does the same for fenced code blocks. Lines a route claims
become content of that parser's kind: below, the query and the suggested
index stop being generic ``code`` and become ``sql`` content.
"""
import sqlglot
from langchain_core.messages import AnyMessage, HumanMessage, ToolMessage
from langgraph.runtime import Runtime
from parsefabric import ParseContext, ParsedEvent, ParseIssue, ParseResult
from parsefabric.capabilities import ParserCapabilities
from parsefabric.parser import Parser
from parsefabric.patterns import RegexPattern
from sqlglot import exp
from sqlglot.errors import SqlglotError
from _models import scripted_model
from contextsage import IntelligentSummarizationMiddleware, SummarizationEvent
class SQLParser(Parser):
"""Recognize SQL statements with sqlglot and record the tables they touch."""
@property
def name(self) -> str:
return "sql"
@property
def version(self) -> str:
return "1"
@property
def capabilities(self) -> ParserCapabilities:
return ParserCapabilities(
deterministic=True, parallel_safe=True, aggregatable=True
)
async def parse(
self, source: object, context: ParseContext | None = None
) -> ParseResult:
text = source.decode() if isinstance(source, bytes) else str(source)
start = context.source_offset if context and context.source_offset else 0
span = {
"start_offset": start,
"end_offset": start + len(text.encode()),
"line_number": 1,
}
try:
tables = sorted(
{
table.name
for statement in sqlglot.parse(text, read="postgres")
if statement is not None
for table in statement.find_all(exp.Table)
}
)
except SqlglotError as error:
unparsed = ParsedEvent("unparsed", attributes={"text": text})
return ParseResult(
events=(self._with_parser_evidence(unparsed, context, **span),),
errors=(ParseIssue.from_exception(error),),
)
event = ParsedEvent("sql_statement", attributes={"tables": [*tables]})
return ParseResult(events=(self._with_parser_evidence(event, context, **span),))
sql = SQLParser()
tool_output = (
"Slow query report for the orders service.\n"
"2026-10-04T09:00:00Z INFO query planner refreshed statistics\n"
"SELECT o.id, c.email FROM orders o JOIN customers c ON c.id = o.customer_id\n"
"2026-10-04T09:00:02Z WARN sequential scan on orders took 4100 ms\n"
"Suggested index:\n"
"```sql\n"
"CREATE INDEX orders_customer_id ON orders (customer_id);\n"
"```\n"
)
history: list[AnyMessage] = [
HumanMessage("Why is the orders query slow?"),
ToolMessage(tool_output, tool_call_id="call-1"),
]
events: list[SummarizationEvent] = []
summary_model = scripted_model("The user asked why the orders query is slow.")
built_in = IntelligentSummarizationMiddleware(
model=summary_model,
trigger=("messages", 2),
keep=("messages", 1),
observability_hook=events.append,
)
with_sql = IntelligentSummarizationMiddleware(
model=summary_model,
trigger=("messages", 2),
keep=("messages", 1),
routes=[(RegexPattern("select", r"(?i)^\s*(SELECT|WITH)\b"), sql)],
fence_routes={"sql": sql},
observability_hook=events.append,
)
for label, middleware in (
("built-in parsing only", built_in),
("with the SQL parser", with_sql),
):
middleware.before_model({"messages": history}, Runtime())
print(f"{label}: {events[-1].content_kinds}")
Provenance¶
The lineage of a summary of a summary, back to the original messages.
"""Trace a summary of a summary back to the original messages.
Run: python examples/provenance.py
Every summary is linked to the messages it replaced in a langgraph-xai
provenance store (in memory here; use a durable ProvenanceStore in
production). After two rounds of summarization, the lineage of the latest
summary reaches back through the first summary to the original messages.
"""
import asyncio
from langchain_core.messages import AIMessage, AnyMessage, HumanMessage
from langgraph.runtime import Runtime
from _models import scripted_model
from contextsage import IntelligentSummarizationMiddleware, SummarizationEvent
events: list[SummarizationEvent] = []
middleware = IntelligentSummarizationMiddleware(
model=scripted_model("Summary of the shipment discussion."),
trigger=("messages", 5),
keep=("messages", 2),
observability_hook=events.append,
)
def turn(index: int) -> list[AnyMessage]:
"""One question about a shipment and its answer."""
return [
HumanMessage(
f"Question {index} about shipment SHP-{index:03d}.", id=f"user-{index}"
),
AIMessage(f"Answer {index}.", id=f"assistant-{index}"),
]
def summarize(history: list[AnyMessage]) -> list[AnyMessage]:
"""Run the middleware once and return the history the model would see."""
update = middleware.before_model({"messages": history}, Runtime())
if update is None:
raise SystemExit("The history did not trigger summarization.")
# The update replaces the history: a RemoveMessage, then the new messages.
return update["messages"][1:]
first = summarize(turn(1) + turn(2) + turn(3))
summarize([*first, *turn(4), *turn(5)])
names = {event.summary_id: f"summary-{index}" for index, event in enumerate(events, 1)}
latest = events[-1]
if latest.summary_id is None:
raise SystemExit("The second run did not write a summary.")
print(f"{names[latest.summary_id]} is generation {latest.generation}")
links = asyncio.run(middleware.alineage(latest.summary_id))
for link in links:
source = names.get(str(link.source_id), str(link.source_id))
target = names.get(str(link.target_id), str(link.target_id))
print(f" {source} -> {target} ({link.relation})")
summary-2 is generation 2
summary-1 -> summary-2 (derived_from)
user-3 -> summary-2 (derived_from)
assistant-3 -> summary-2 (derived_from)
user-4 -> summary-2 (derived_from)
assistant-4 -> summary-2 (derived_from)
user-1 -> summary-1 (derived_from)
assistant-1 -> summary-1 (derived_from)
user-2 -> summary-1 (derived_from)
assistant-2 -> summary-1 (derived_from)
Logs and a hook¶
The structured event behind every summarization.
"""Observe every summarization through logging and a hook.
Run: python examples/observability.py
Each summarization emits one content-free ``SummarizationEvent``. It is logged
to the ``contextsage`` logger at INFO level, with the event under the
``contextsage`` key of the log record, and passed to ``observability_hook``.
Neither ever contains message content.
"""
import json
import logging
from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, ToolMessage
from langgraph.runtime import Runtime
from _models import scripted_model
from contextsage import IntelligentSummarizationMiddleware, SummarizationEvent
VOLATILE = {"summary_id", "latency_ms"}
class JSONEventHandler(logging.Handler):
"""Print the structured event attached to ContextSage log records."""
def emit(self, record: logging.LogRecord) -> None:
event = {
key: value
for key, value in getattr(record, "contextsage", {}).items()
if key not in VOLATILE
}
print(f"log record '{record.getMessage()}':")
print(json.dumps(event, indent=2))
logger = logging.getLogger("contextsage")
logger.setLevel(logging.INFO)
logger.addHandler(JSONEventHandler())
def alert_on_lost_facts(event: SummarizationEvent) -> None:
"""A hook: flag summaries that needed facts restated."""
if event.recovery_status != "none_needed":
print(f"hook: recovery '{event.recovery_status}' after {event.trigger_reason}")
middleware = IntelligentSummarizationMiddleware(
model=scripted_model("Discussed the invoice."),
trigger=("messages", 6),
keep=("messages", 2),
observability_hook=alert_on_lost_facts,
)
billing_log = "".join(
f"2026-10-04T07:{index // 60:02d}:{index % 60:02d}Z INFO billing-worker charge ok\n"
for index in range(120)
)
history: list[AnyMessage] = [
HumanMessage("Invoice INV-2048 for customer 77 was charged twice."),
AIMessage("", tool_calls=[{"id": "c1", "name": "billing_log", "args": {}}]),
ToolMessage(
billing_log + "2026-10-04T09:00:00Z ERROR duplicate charge for INV-2048\n",
tool_call_id="c1",
),
AIMessage("The billing log shows a duplicate charge for INV-2048."),
HumanMessage("Refund the duplicate."),
AIMessage("Refund issued."),
]
middleware.before_model({"messages": history}, Runtime())
log record 'contextsage summarization':
{
"generation": 1,
"thread_id": null,
"trigger_reason": "6 messages >= 6",
"input_messages": 6,
"input_tokens": 1627,
"available_tokens": 109200,
"overflow_tokens": 0,
"prepared_tokens": 135,
"output_messages": 3,
"output_tokens": 61,
"compression_ratio": 0.037492317148125384,
"compacted_units": 1,
"must_preserve_units": 2,
"must_preserve_facts": 2,
"content_kinds": {
"log": 2,
"text": 4
},
"validation_status": "failed_recovered",
"recovery_status": "restated_facts"
}
hook: recovery 'restated_facts' after 6 messages >= 6
Prometheus metrics¶
Events exported as Prometheus metrics, with a matching scrape configuration.
"""Export ContextSage events as Prometheus metrics.
Run: python examples/observability_prometheus.py
Requires prometheus-client (installed with the ``examples`` dependency group).
The script serves metrics on http://localhost:9109/metrics, runs a few
summarizations, then keeps serving for ``OBSERVABILITY_DEMO_SECONDS`` seconds
(default 60) so a Prometheus server can scrape it. See
``examples/observability_prometheus.yml`` for a scrape configuration.
"""
import os
import time
from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, ToolMessage
from langgraph.runtime import Runtime
from prometheus_client import Counter, Histogram, start_http_server
from _models import scripted_model
from contextsage import IntelligentSummarizationMiddleware, SummarizationEvent
SUMMARIZATIONS = Counter(
"contextsage_summarizations_total",
"Summarizations by validation and recovery outcome.",
labelnames=("validation_status", "recovery_status"),
)
INPUT_TOKENS = Histogram(
"contextsage_input_tokens",
"Tokens in the history before summarization.",
buckets=(250, 500, 1_000, 2_500, 5_000, 10_000, 25_000, 100_000),
)
OUTPUT_TOKENS = Histogram(
"contextsage_output_tokens",
"Tokens in the history after summarization.",
buckets=(250, 500, 1_000, 2_500, 5_000, 10_000, 25_000, 100_000),
)
COMPRESSION = Histogram(
"contextsage_compression_ratio",
"Output tokens divided by input tokens.",
buckets=(0.05, 0.1, 0.2, 0.3, 0.5, 0.75, 1.0, 1.5),
)
LATENCY = Histogram(
"contextsage_latency_ms",
"Time spent in the middleware, in milliseconds.",
buckets=(10, 50, 100, 250, 500, 1_000, 2_500, 10_000, 30_000),
)
def record(event: SummarizationEvent) -> None:
"""Record one event; labels use only low-cardinality status values."""
SUMMARIZATIONS.labels(event.validation_status, event.recovery_status).inc()
INPUT_TOKENS.observe(event.input_tokens)
OUTPUT_TOKENS.observe(event.output_tokens)
COMPRESSION.observe(event.compression_ratio)
LATENCY.observe(event.latency_ms)
print(
f"{event.input_tokens} -> {event.output_tokens} tokens, "
f"validation={event.validation_status}, recovery={event.recovery_status}"
)
def history(size: int) -> list[AnyMessage]:
"""A conversation whose tool result holds ``size`` routine log lines."""
log = "".join(
f"2026-10-04T10:{index // 60:02d}:{index % 60:02d}Z INFO sync batch done\n"
for index in range(size)
)
return [
HumanMessage("Why did the nightly sync for tenant T-310 fail?"),
AIMessage("", tool_calls=[{"id": "c1", "name": "sync_logs", "args": {}}]),
ToolMessage(
log + "2026-10-04T11:00:00Z ERROR sync aborted for T-310\n",
tool_call_id="c1",
),
AIMessage("The sync for T-310 aborted."),
HumanMessage("What should we do?"),
]
start_http_server(9109)
print("Serving metrics on http://localhost:9109/metrics")
for size in (50, 400, 1_500):
middleware = IntelligentSummarizationMiddleware(
model=scripted_model("The nightly sync for tenant T-310 aborted."),
trigger=("tokens", 500),
keep=("messages", 2),
observability_hook=record,
)
middleware.before_model({"messages": history(size)}, Runtime())
time.sleep(int(os.environ.get("OBSERVABILITY_DEMO_SECONDS", "60")))
With a live model¶
These examples call a real model through any OpenAI-compatible chat
completions endpoint. Set CONTEXTSAGE_LLM_API_KEY, and optionally
CONTEXTSAGE_LLM_BASE_URL and CONTEXTSAGE_LLM_MODEL, in the environment or
in a .env file; the
examples README
describes them. Their output depends on the model, so it is not recorded.
Agent with a large tool result¶
"""Run ContextSage inside a live agent loop.
Run: python examples/live_agent.py
Requires CONTEXTSAGE_LLM_API_KEY (and optionally CONTEXTSAGE_LLM_BASE_URL and
CONTEXTSAGE_LLM_MODEL); see .env.example. The agent calls a tool that returns a
large diagnostics report. Before the next model call, ContextSage compacts the
report inside the agent loop, keeping the incident, the error and the
transaction ID, and the model answers from the compacted context.
"""
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from _models import live_model
from contextsage import IntelligentSummarizationMiddleware, SummarizationEvent
@tool
def incident_report(incident_id: str) -> str:
"""Return the diagnostics report for an incident."""
heartbeats = "".join(
f"2026-10-04T03:{index // 60:02d}:{index % 60:02d}Z "
f"INFO payments-worker-{index % 3} processing queue\n"
for index in range(400)
)
return (
f"Incident {incident_id}: elevated latency on the payments service.\n"
f"{heartbeats}"
"2026-10-04T03:59:58Z ERROR connection pool exhausted while settling TX-991\n"
)
model = live_model()
events: list[SummarizationEvent] = []
agent = create_agent(
model=model,
tools=[incident_report],
system_prompt="You are an SRE assistant. Use the tools to investigate incidents.",
middleware=[
IntelligentSummarizationMiddleware(
model=model,
trigger=("tokens", 2_000),
keep=("messages", 6),
observability_hook=events.append,
)
],
)
result = agent.invoke(
{"messages": [HumanMessage("Investigate INC-9931: what is the root cause?")]}
)
print("Answer:", result["messages"][-1].text)
for event in events:
print(
f"summarization: {event.input_tokens} -> {event.output_tokens} tokens, "
f"{event.compacted_units} units compacted, validation {event.validation_status}"
)
Deep agent¶
"""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")
Swarm¶
"""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"
)