3tears-langgraph
Three-tier LangGraph checkpoint saver: L1 (SQLite) -> L2 (NATS KV) -> L3 (PostgreSQL).
L1 and L2 are optional cache layers that degrade gracefully on failure. L3 (PostgreSQL) is the source of truth, reached through the AsyncQueryExecutor protocol so the same saver serves trusted services (direct asyncpg pool) and sandboxed agents (NATS L3 proxy).
L3 failures reach the caller, with one channel-scoped carve-out: aput_writes is called from LangGraph's executor teardown, where raising kills a turn that has already answered, so a failed crash-recovery write is logged and degraded. Writes on LangGraph's control channels (__interrupt__, __resume__, __error__, __scheduled__) still raise: losing one changes what the run does, and a lost __interrupt__ would silently skip a human-approval gate. A control-channel write also overwrites an earlier write on the same channel, matching the reference saver, so a second Command(resume=...) replaces the first rather than being dropped.
Installation
pip install 3tears-langgraph
Usage
from threetears.langgraph import (
AsyncpgPoolAdapter,
CheckpointScope,
ThreeTierCheckpointSaver,
)
# Trusted service with direct asyncpg.Pool: wrap once
saver = ThreeTierCheckpointSaver(
executor=AsyncpgPoolAdapter(pool),
scope=CheckpointScope.for_customer(customer_id),
)
# Sandboxed agent: NatsProxyL3Backend already implements
# AsyncQueryExecutor, pass it straight through
saver = ThreeTierCheckpointSaver(
executor=nats_l3_backend,
scope=CheckpointScope.for_customer(customer_id),
)
graph = builder.compile(checkpointer=saver)
Scope is required
scope has no default. A saver either names the customer whose checkpoints it addresses, or says in writing that it deliberately names none:
saver = ThreeTierCheckpointSaver(
executor=AsyncpgPoolAdapter(pool),
scope=CheckpointScope.unscoped(reason="single-tenant deployment"),
)
CheckpointScope.for_customer(...) folds the customer into the stored thread_id, and therefore into the L3 bound parameter, the L2 bucket key, and the L1 thread key — a saver scoped to one customer cannot name another customer's row at any tier. It also unlocks adelete_customer_threads(), the whole-tenant purge, which refuses on an unscoped saver.
CheckpointScope.unscoped(...) is a legitimate answer, not a placeholder: it produces byte-identical keys and statements to a pre-tenancy saver, so an existing deployment adopts the required parameter by adding this one argument and migrating no data. The reason is mandatory, logged at WARNING on construction, and greppable in source, so "which deployments still run unscoped, and why" has an answer.
Adopting a real customer later is a data change rather than a code change: existing rows live under a bare thread id and a scoped saver will not find them, so they must be re-keyed (UPDATE checkpoints SET thread_id = $customer || '/' || thread_id, likewise checkpoint_writes, plus L2 invalidation). No re-key script ships here and none can — which customer owns which thread lives in the host's own tables, which this library has never seen.
Person erasure: anonymize, never delete
result = await saver.aanonymize_threads(conversation_ids)
Rewrites every stored checkpoint and pending write of those threads in place: a human
message's name (the sender's display name) and every value under
IDENTIFYING_METADATA_KEYS (external_user_name, external_user_id) become
ANONYMIZED_MARKER. Message text, every id, and the blob format are unchanged, so the
graph still loads and resumes. Idempotent, paged by batch_size, and it evicts the
threads' L1 and shared L2 bundles, raising if an eviction fails. An unknown metadata key
is kept: the metadata channel is working state the graph reads on resume. Run it on
threads that are not in a live turn.
Middleware
The package ships platform-level AgentMiddleware for langchain.agents.create_agent — the framework-aligned successor to the old hand-rolled AgentNodeHook / ToolNodeHook protocols. Consumer-specific policy lives in each consumer as its own middleware; only the reusable platform seams live here:
PromptCachingMiddleware(wrap_model_call) — annotates a leading bare-string system message with Anthropiccache_control={"type": "ephemeral"}when the model supports it, then normalizes cache-hit/creation counters ontousage_metadata["cache_usage"]. Non-Anthropic adapters degrade silently to bare-string system messages.ToolResultOffloadMiddleware(wrap_tool_call) — when aToolResultOffloaderis injected onconfig["configurable"]and a tool result exceedsoffload_threshold_chars, stores the full content out-of-band and shows the model"<summary>\n\n[ctx:<handle>]"(the structuredartifactis preserved). Opt-in: no offloader ⇒ byte-for-byte no-op.ObjectCatalogMiddleware(wrap_tool_call) — when a tool returns anObjectHandlein its result artifact and anObjectCatalogeris injected, persists a catalog record under the verified call identity. Soft-fail side-effect: a catalog error never breaks the tool result.- Two summarizers. Choose by whether the conversation's history must survive:
SummarizationMiddleware(before_model) — once the message COUNT passestrigger_messages, rewrites the checkpointedmessageschannel to[summary, *last keep_messages]. The older messages are gone from the checkpoint.RollingSummaryMiddleware(awrap_model_call, async only) — NON-destructive. Give itstore(built per turn) orstore_for(request)(an agent compiled once). The checkpointer keeps every message, and only the model's input is trimmed, once the messages since the last fold pass a TOKEN budget. The rolling summary and a cursor (the last folded message's id) live behind aSummaryStore, whose save is a compare-and-swap. It folds at most once per turn, never stores a failed or empty summary over the prior one, never leaves an orphaned tool result at the head of the kept tail, and marks its summary callNOSTREAM_TAGplusmetadata["threetears.usage.purpose"] = "summarization".threetears.conversations.ConversationSummaryStoreis the store over a conversations row, anddispatch_conversation_summarizedis itson_summarized.
ToolLoopBudgetMiddleware(wrap_model_call, plus abefore_agentnode) — bounds a turn's tool loop and always ends the turn with text. A turn is one invocation; its record lives in a private agent-state field the budget resets inbefore_agent, so a middleware that rewrites or appends tomessages(the summarizer's summaryHumanMessage, a reminder) cannot reset the count. At mostmax_tool_calls(16) per turn; wrap-up nudges at thewarn/urgenttiers (a share of the cap, or of the model's context window); atforcethe model is told to answer from what it has, its tools still declared (Anthropic refuses tool history with none) withforced_tool_choice("none"where the provider takes it), and the reply is rebuilt as text only with its usage and metadata kept; an empty or reasoning-only answer is retried once, thenfallback_textstreams through a stand-in model; a runaway guard for rounds with no text whose calls all failed. Put it first in the list and run with{"recursion_limit": recursion_limit_for(middleware=<the same list>)}: the cap comes from the budget in the list, everybefore_model/after_modelhook adds its node per round, so the budget, not LangGraph's recursion error, ends a loop.
from langchain.agents import create_agent
from threetears.langgraph import (
ObjectCatalogMiddleware,
PromptCachingMiddleware,
ToolLoopBudgetMiddleware,
ToolResultOffloadMiddleware,
recursion_limit_for,
)
middleware = [
ToolLoopBudgetMiddleware(),
PromptCachingMiddleware(),
ToolResultOffloadMiddleware(),
ObjectCatalogMiddleware(),
]
agent = create_agent(model=chat_anthropic, tools=tools, middleware=middleware)
config = {"recursion_limit": recursion_limit_for(middleware=middleware)}
# after a run, PromptCachingMiddleware has stamped:
# message.usage_metadata["cache_usage"]
# == {"cache_read_input_tokens": ..., "cache_creation_input_tokens": ..., "cached_tokens": ...}
The offload / catalog contracts (ToolResultOffloader, ObjectCataloger) are pure structural Protocols exported from threetears.langgraph.offload / threetears.langgraph.catalog; a consumer injects a concrete implementation on config["configurable"] (e.g. tool_result_offloader, object_cataloger) without the package taking any dependency on the consumer's context store.
See 3tears/docs/prompt-caching.md for the full caching contract, summarization interaction, downstream wiring checklist, and a worked example.
Streaming
The package ships StreamingResponse, a transport-agnostic primitive that owns the lifecycle of one streaming response: start -> any number of emit_token / emit_tool_call_* -> mutually-exclusive end (success) or error (failure) terminal. run_graph(compiled_graph, state, config) consumes a LangGraph astream_events(version="v2") loop with the start/end ordering managed; on graph exception it fires error(code="AGENT_FAILED", ...) and re-raises so the caller still sees the failure on the synchronous path.
The wire vocabulary is fixed: StreamStartEvent / StreamTokenEvent / StreamEndEvent / StreamErrorEvent / ToolCallStartEvent / ToolCallEndEvent / ToolCallProgressEvent, dispatched via the StreamEvent discriminated union and the parse_stream_event(payload) adapter. The transport seam is the StreamTransport Protocol -- one method, async def publish(self, payload: bytes) -> None. Any wire (NATS subject, websocket, chunked HTTP body) satisfies it.
from threetears.langgraph import StreamingResponse, StreamTransport
class WebSocketStreamTransport:
"""example transport for a websocket consumer."""
def __init__(self, ws): self._ws = ws
async def publish(self, payload: bytes) -> None:
await self._ws.send_bytes(payload)
stream = StreamingResponse(
transport=WebSocketStreamTransport(ws),
correlation_id=correlation_id,
conversation_id=conversation_id,
start_time_monotonic=request_start,
)
final_state = await stream.run_graph(compiled_graph, state, config)
A reference adapter can bind the primitive to a per-correlation-id stream subject via nc.publish_raw. Tool-call observation envelopes flow through ToolCallProgressHook reading the active StreamingResponse from config["configurable"]["streaming_response"].
1 maintainer
Metadata
1 maintainer
Release files for 3tears-langgraph 0.73.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | Provenance | |
|---|---|---|---|---|
| 3tears_langgraph-0.73.0.tar.gz | 234.6 kB | With provenance | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| 3tears_langgraph-0.73.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 385.6 kB
Release files / 3tears_langgraph-0.73.0.tar.gz
| Download URL | 3tears_langgraph-0.73.0.tar.gz |
|---|---|
| Size | 234.6 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
7afe3037c4463c0547aa79b4a82111dc2485d0f211fc8d40ec881a53c5578176
|
|
BLAKE2b-256 checksum How to use checksums |
26a9ea0b154609e515991a5876a574fd9c06273efced95a626201144c270f7e3
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.16
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 20, 2026.
Transparency logRelease files / 3tears_langgraph-0.73.0-py3-none-any.whl
| Download URL | 3tears_langgraph-0.73.0-py3-none-any.whl |
|---|---|
| Size | 151.0 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
f35750a5344b8df9f69e369c4382a7d569aac6b64764d33606ff216ae9f83ad8
|
|
BLAKE2b-256 checksum How to use checksums |
029ab4428d42d3d61005565cdd31f8b72a0716fdcd49c353fcc06965b833ee38
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.16
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 20, 2026.
Transparency logSecurity for 3tears-langgraph 0.73.0
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
Full provenance
All 2 files in this release have been uploaded with provenance metadata.
Attestation for 3tears_langgraph-0.73.0.tar.gz
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 20, 2026.
Transparency logA note on package security
PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.
