层次异步内存中间件

Installation

pip install sawtooth-memory

Optional dependencies:

```bash

Cloud compression providers (install the SDK you use)

pip install langchain-openai langchain-anthropic langchain-google-genai

Automatically compiles the L0, L1.5, L2, and L1 tiers safely across thread boundaries

lc_messages = history.messages

```

Quickstart (V2 Configuration)

The V2 configuration introduces dynamic validation, allowing you to set a single background_model parameter that automatically routes to the respective local or cloud backend. Cloud models (gpt-*, claude-*, gemini-*) read API keys from standard environment variables (OPENAI_API_KEY, ANTHROPIC_API_KEY, GOOGLE_API_KEY).

import asyncio
from sawtooth_memory import ContextManager, ContextManagerConfig

async def main():
    # V2 Simplified Configuration
    config = ContextManagerConfig(
        background_model="gpt-4o-mini",   # Auto-routes to CloudCompressor (or "phi4" for local Ollama)
        soft_limit_tokens=1000,           # Token threshold to trigger background compression
        hard_limit_tokens=2000,           # Emergency truncation limit
        max_unsummarized_turns=10,        # Turn-based batching threshold
        enable_deterministic_ner=True     # Enable local regex extraction for the Entity Ledger
    )

    async with ContextManager(system_prompt="You are a helpful assistant.", config=config) as cm:

        # Optional: Run a health check to verify backend routing and worker status
        await cm.health_check()

        # 1. Instantly ingest messages (Zero-latency on the main thread)
        await cm.add_message("user", "My transaction ID is txn_998877_alpha")
        await cm.add_message("assistant", "I have noted your transaction ID.")

        # 2. Build the optimized prompt to send to your main LLM
        prompt = await cm.build_prompt()
        print(prompt)

if __name__ == "__main__":
    asyncio.run(main())

For explicit cloud configuration without environment variables:

from sawtooth_memory.config import CloudConfig, Provider

config = ContextManagerConfig(
    cloud=CloudConfig(
        provider=Provider.OPENAI,
        model="gpt-4o-mini",
        api_key="sk-...",
    ),
)

---

4. Synchronous API Wrapper (Flask, Django, CLI)

If you are building in a traditional synchronous environment (like a standard Flask or Django view) where you cannot use asyncio or await, Sawtooth provides an enterprise-grade SawtoothSyncWrapper. It uses an AnyIO BlockingPortal to isolate the asynchronous background worker on a safe daemon thread, preventing event loop collisions while maintaining zero-latency writes.

from sawtooth_memory.sync_wrapper import SawtoothSyncWrapper

def my_flask_route():
    # Use standard 'with' - no async required!
    with SawtoothSyncWrapper("You are a helpful assistant.", config=config) as memory:

        # 1. Instantly write to the background thread
        memory.add_message("user", "Hello world!")

        # 2. Safely read the compiled state machine
        prompt = memory.build_prompt()

        return prompt

LangGraph integration

pip install sawtooth-memory[langgraph]

2. LangGraph Integration & ToolMessage Sanitization

Sawtooth provides a native SawtoothLangGraphAdapter to sync state seamlessly.

V2 Safety Feature: Strict cloud APIs (like Anthropic/OpenAI) will crash if a ToolMessage is sent without its parent AIMessage (the tool call request). The LangGraph adapter includes an advanced 3-pass sanitization logic that automatically detects and drops orphaned ToolMessages when their parent AIMessage has been compressed and evicted to L2 Archival Memory.

```python from langgraph.graph import StateGraph from sawtooth_memory.integrations.langgraph import SawtoothLangGraphAdapter

3. Modern LangChain & LCEL Integration

Sawtooth provides a native, pure-Python adapter that fully implements LangChain's modern BaseChatMessageHistory interface. This allows you to drop Sawtooth directly into any LCEL Runnable or Agent executor, bringing background compression and deterministic NER to standard LangChain pipelines without blocking the main thread.

```python from langchain_core.messages import HumanMessage from sawtooth_memory.integrations.langchain_adapter import SawtoothChatMessageHistory

Drop-in replacement for any LangChain memory module

history = SawtoothChatMessageHistory( system_prompt="You are a financial analyst.", config=config )

history.add_message(HumanMessage(content="Analyze these Q3 numbers."))