diff --git a/pyproject.toml b/pyproject.toml index aa778f3e..a1342a66 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -34,6 +34,7 @@ dependencies = [ extensibility = ["a2a-sdk>=0.2.0"] starlette = ["starlette>=0.40.0"] langchain = ["langchain-core>=1.2.7"] +langgraph = ["langgraph>=1.0.0"] [build-system] requires = ["hatchling"] diff --git a/src/sap_cloud_sdk/agent_memory/factory/__init__.py b/src/sap_cloud_sdk/agent_memory/factory/__init__.py new file mode 100644 index 00000000..525b1b2f --- /dev/null +++ b/src/sap_cloud_sdk/agent_memory/factory/__init__.py @@ -0,0 +1,14 @@ +"""Factory subpackage for SAP Agent Memory framework adapters. + +Each module in this package provides a factory function that creates the +appropriate framework-specific implementation for the current environment. + +Available factories: + +- :func:`sap_cloud_sdk.agent_memory.factory.langgraph_checkpoint.create_checkpointer` + — returns a LangGraph ``BaseCheckpointSaver`` (short-term memory) + +Usage: + + from sap_cloud_sdk.agent_memory.factory.langgraph_checkpoint import create_checkpointer +""" diff --git a/src/sap_cloud_sdk/agent_memory/factory/_timed_memory.py b/src/sap_cloud_sdk/agent_memory/factory/_timed_memory.py new file mode 100644 index 00000000..791d1c45 --- /dev/null +++ b/src/sap_cloud_sdk/agent_memory/factory/_timed_memory.py @@ -0,0 +1,111 @@ +"""TimedInMemorySaver — InMemorySaver with background TTL-based thread eviction. + +Not intended for direct use. Use ``create_checkpointer(ttl_seconds=...)`` instead. +""" + +import logging +import threading +import time +from typing import Any + +from langchain_core.runnables import RunnableConfig +from langgraph.checkpoint.base import ChannelVersions, Checkpoint, CheckpointMetadata +from langgraph.checkpoint.memory import InMemorySaver + +logger = logging.getLogger(__name__) + +_SWEEP_INTERVAL_SECONDS = 60 + + +class TimedInMemorySaver(InMemorySaver): + """InMemorySaver with background TTL-based thread eviction. + + Tracks last-active time per thread and evicts inactive threads via a + daemon background sweep. The sweep is fully decoupled from the read/write + path — ``put()`` only updates the activity timestamp. + + This approximates server-side TTL semantics for in-process storage: + - ``put()`` records last-active timestamp (hot path stays clean) + - A daemon sweep thread evicts inactive threads every 60 seconds + - Eviction is best-effort — a thread may live up to + ``ttl_seconds + 60`` seconds before deletion + - State does not survive process restarts (inherent to InMemorySaver) + + Args: + ttl_seconds: Inactivity threshold in seconds. Threads inactive for + longer than this are evicted. Default: 3600 (1 hour). + + Example:: + + from sap_cloud_sdk.agent_memory.factory.langgraph_checkpoint import ( + create_checkpointer, + ) + + checkpointer = create_checkpointer(ttl_seconds=3600) + """ + + def __init__(self, *, ttl_seconds: int = 3600, **kwargs: Any) -> None: + super().__init__(**kwargs) + self.ttl_seconds = ttl_seconds + self._last_active: dict[str, float] = {} + self._lock = threading.Lock() + self._sweeper = threading.Thread( + target=self._sweep_loop, + daemon=True, + name="TimedInMemorySaver-sweeper", + ) + self._sweeper.start() + logger.debug( + "TimedInMemorySaver started with ttl_seconds=%d, sweep_interval=%ds", + ttl_seconds, + _SWEEP_INTERVAL_SECONDS, + ) + + def put( + self, + config: RunnableConfig, + checkpoint: Checkpoint, + metadata: CheckpointMetadata, + new_versions: ChannelVersions, + ) -> RunnableConfig: + """Save checkpoint and refresh thread activity timestamp.""" + thread_id: str = config["configurable"]["thread_id"] + with self._lock: + self._last_active[thread_id] = time.monotonic() + return super().put(config, checkpoint, metadata, new_versions) + + async def aput( + self, + config: RunnableConfig, + checkpoint: Checkpoint, + metadata: CheckpointMetadata, + new_versions: ChannelVersions, + ) -> RunnableConfig: + """Async version of put — refreshes thread activity timestamp.""" + thread_id: str = config["configurable"]["thread_id"] + with self._lock: + self._last_active[thread_id] = time.monotonic() + return await super().aput(config, checkpoint, metadata, new_versions) + + def _sweep_loop(self) -> None: + while True: + time.sleep(_SWEEP_INTERVAL_SECONDS) + self._evict_expired() + + def _evict_expired(self) -> None: + now = time.monotonic() + with self._lock: + expired = [ + tid + for tid, ts in self._last_active.items() + if now - ts > self.ttl_seconds + ] + for tid in expired: + self.delete_thread(tid) + with self._lock: + self._last_active.pop(tid, None) + logger.info( + "TimedInMemorySaver: evicted inactive thread '%s' (inactive for >%ds)", + tid, + self.ttl_seconds, + ) diff --git a/src/sap_cloud_sdk/agent_memory/factory/langgraph_checkpoint.py b/src/sap_cloud_sdk/agent_memory/factory/langgraph_checkpoint.py new file mode 100644 index 00000000..379e6e97 --- /dev/null +++ b/src/sap_cloud_sdk/agent_memory/factory/langgraph_checkpoint.py @@ -0,0 +1,75 @@ +"""LangGraph checkpointer factory for SAP Agent Memory. + +Usage:: + + from sap_cloud_sdk.agent_memory.factory.langgraph_checkpoint import create_checkpointer + + # No TTL — plain InMemorySaver + checkpointer = create_checkpointer() + + # With TTL — TimedInMemorySaver evicts inactive threads automatically + checkpointer = create_checkpointer(ttl_seconds=3600) + + app = workflow.compile(checkpointer=checkpointer) + + # Or with LangChain create_agent: + from langchain.agents import create_agent + agent = create_agent(model="...", tools=[...], checkpointer=checkpointer) +""" + +import logging +from typing import Optional + +logger = logging.getLogger(__name__) + + +def create_checkpointer(*, ttl_seconds: Optional[int] = None): + """Create a LangGraph checkpointer for the current environment. + + Returns LangGraph's ``InMemorySaver`` (no TTL) or ``TimedInMemorySaver`` + (with TTL-based thread eviction). State is held in-process and does not + survive restarts. + + Args: + ttl_seconds: Evict threads inactive for this many seconds. + ``None`` (default) disables eviction. + + Returns: + BaseCheckpointSaver instance. + + Raises: + ImportError: If langgraph is not installed. + + Example — no TTL:: + + checkpointer = create_checkpointer() + app = workflow.compile(checkpointer=checkpointer) + + Example — evict threads inactive for 1 hour:: + + checkpointer = create_checkpointer(ttl_seconds=3600) + app = workflow.compile(checkpointer=checkpointer) + """ + try: + from langgraph.checkpoint.memory import InMemorySaver + except ImportError: + raise ImportError( + "langgraph is required for create_checkpointer(). " + "Install it with: pip install sap-cloud-sdk[langgraph] or pip install langgraph" + ) + + if ttl_seconds is not None: + from sap_cloud_sdk.agent_memory.factory._timed_memory import TimedInMemorySaver + + logger.warning( + "create_checkpointer(): using TimedInMemorySaver(ttl_seconds=%d) — " + "session state is in-process only and will be lost on process exit.", + ttl_seconds, + ) + return TimedInMemorySaver(ttl_seconds=ttl_seconds) + + logger.warning( + "create_checkpointer(): using InMemorySaver — " + "session state is in-process only and will be lost on process exit." + ) + return InMemorySaver() diff --git a/src/sap_cloud_sdk/agent_memory/user-guide.md b/src/sap_cloud_sdk/agent_memory/user-guide.md index caf98fc2..7eb4d56a 100644 --- a/src/sap_cloud_sdk/agent_memory/user-guide.md +++ b/src/sap_cloud_sdk/agent_memory/user-guide.md @@ -57,6 +57,16 @@ plain text, and the service makes it searchable by meaning. - [`AgentMemoryNotFoundError` when fetching a resource](#agentmemorynotfounderror-when-fetching-a-resource) - [`AgentMemoryHttpError` with status 401](#agentmemoryhttperror-with-status-401) - [Configuration](#configuration) + - [Service Binding](#service-binding) + - [Mounted Secrets (Kubernetes)](#mounted-secrets-kubernetes) + - [Environment Variables](#environment-variables) + - [UAA JSON Schema](#uaa-json-schema) + - [LangGraph Checkpointer](#langgraph-checkpointer) + - [Prerequisites](#prerequisites) + - [Import](#import-1) + - [Usage with LangGraph StateGraph](#usage-with-langgraph-stategraph) + - [Usage with LangChain create\_agent](#usage-with-langchain-create_agent) + - [Thread TTL](#thread-ttl) ## Installation @@ -806,3 +816,116 @@ The `uaa` key must contain a JSON string with the XSUAA credentials: "url": "https://subdomain.authentication.region.hana.ondemand.com" } ``` + +## LangGraph Checkpointer + +> This section covers the LangGraph-specific +> factory for short-term memory (checkpointing). It is only relevant if your agent +> is built with LangGraph or LangChain's `create_agent()`. The core +> `AgentMemoryClient` and `create_client()` above are framework-agnostic and work +> independently of this section. + +For LangGraph agents, the `agent_memory` module provides a `create_checkpointer()` +factory in the `factory` subpackage. It returns a `BaseCheckpointSaver` that +manages short-term session memory — conversation state, thread continuity, and +HITL support — natively within LangGraph. + +> [!NOTE] +> The current implementation uses LangGraph's `InMemorySaver` or +> `TimedInMemorySaver` (when `ttl_seconds` is set). State is held in-process +> and does not survive restarts. Persistent checkpointing backed by the +> Agent Memory Service is not yet supported. + +### Prerequisites + +`langgraph` is an optional dependency. Install it via the SDK extra or directly: + +```bash +# Via SDK optional extra (recommended) +pip install "sap-cloud-sdk[langgraph]" + +# Or install langgraph directly +pip install langgraph +``` + +### Import + +```python +from sap_cloud_sdk.agent_memory.factory.langgraph_checkpoint import create_checkpointer +``` + +### Usage with LangGraph StateGraph + +```python +from sap_cloud_sdk.agent_memory.factory.langgraph_checkpoint import create_checkpointer + +checkpointer = create_checkpointer() +app = workflow.compile(checkpointer=checkpointer) + +result = app.invoke( + {"messages": [{"role": "user", "content": "hello"}]}, + {"configurable": {"thread_id": "session-1"}}, +) +``` + +### Usage with LangChain create_agent + +```python +from langchain.agents import create_agent +from sap_cloud_sdk.agent_memory.factory.langgraph_checkpoint import create_checkpointer + +agent = create_agent( + model="...", + tools=[...], + checkpointer=create_checkpointer(), +) +``` + +### Thread TTL + +Pass `ttl_seconds` to evict threads that have been inactive for the given +period. This prevents unbounded memory growth in long-running processes. + +```python +# Evict threads inactive for more than 1 hour +checkpointer = create_checkpointer(ttl_seconds=3600) +``` + +When `ttl_seconds` is set, the factory returns a `TimedInMemorySaver` that +tracks last-active time per thread and evicts inactive threads via a +background daemon sweep. Eviction is best-effort — a thread may live up to +`ttl_seconds + 60` seconds before deletion. + +#### Exposing TTL as a configurable parameter with `@agent_config` + +Use `@agent_config` from `sap_cloud_sdk.agent_decorators` to expose +`ttl_seconds` as an operator-adjustable configuration field. The key +`config.checkpointer.ttl_seconds` groups it with other checkpointer settings +so external tooling can surface it in the low-code UI alongside model +selection and temperature. + +```python +from sap_cloud_sdk.agent_decorators import agent_config +from sap_cloud_sdk.agent_memory.factory.langgraph_checkpoint import create_checkpointer + + +@agent_config( + key="config.checkpointer.ttl_seconds", + label="Thread TTL (seconds)", + description="Evict inactive conversation threads after this period of " + "inactivity. Set to 0 to disable eviction.", +) +def thread_ttl_seconds() -> int: + return 3600 # 1 hour + + +class MyAgent: + def __init__(self): + ttl = thread_ttl_seconds() + self._checkpointer = create_checkpointer(ttl_seconds=ttl or None) +``` + +> [!NOTE] +> `TimedInMemorySaver` state does not survive process restarts — the TTL +> applies to in-process memory only. Persistent TTL enforcement will be +> available when the Agent Memory Service checkpointer ships. diff --git a/tests/agent_memory/unit/test_langgraph_checkpointer.py b/tests/agent_memory/unit/test_langgraph_checkpointer.py new file mode 100644 index 00000000..5885961c --- /dev/null +++ b/tests/agent_memory/unit/test_langgraph_checkpointer.py @@ -0,0 +1,117 @@ +"""Unit tests for the create_checkpointer() LangGraph factory.""" + +import builtins +from typing import Any, TypedDict +from unittest.mock import patch + +import pytest +from langchain_core.runnables import RunnableConfig +from langgraph.checkpoint.memory import InMemorySaver +from langgraph.graph import END, START, StateGraph + +from sap_cloud_sdk.agent_memory.factory.langgraph_checkpoint import create_checkpointer + + +class TestCreateCheckpointer: + """Tests for create_checkpointer() factory.""" + + # ── Return type ─────────────────────────────────────────────────────────── + + def test_returns_in_memory_saver(self): + """Factory returns LangGraph's InMemorySaver.""" + result = create_checkpointer() + assert isinstance(result, InMemorySaver) + + def test_ttl_seconds_still_returns_in_memory_saver(self): + """ttl_seconds returns TimedInMemorySaver.""" + from sap_cloud_sdk.agent_memory.factory._timed_memory import TimedInMemorySaver + result = create_checkpointer(ttl_seconds=3600) + assert isinstance(result, TimedInMemorySaver) + + def test_ttl_seconds_logs_warning(self, caplog): + """ttl_seconds logs a warning about in-process state.""" + import logging + with caplog.at_level( + logging.WARNING, + logger="sap_cloud_sdk.agent_memory.factory.langgraph_checkpoint", + ): + create_checkpointer(ttl_seconds=3600) + assert "TimedInMemorySaver" in caplog.text + + # ── Missing langgraph ───────────────────────────────────────────────────── + + def test_missing_langgraph_raises_import_error(self): + """Clear ImportError when langgraph is not installed.""" + original_import = builtins.__import__ + + def mock_import(name, *args, **kwargs): + if name == "langgraph.checkpoint.memory": + raise ImportError("No module named 'langgraph'") + return original_import(name, *args, **kwargs) + + with patch("builtins.__import__", side_effect=mock_import): + with pytest.raises(ImportError, match="langgraph is required"): + create_checkpointer() + + # ── LangGraph integration ───────────────────────────────────────────────── + + def test_checkpointer_compiles_with_langgraph_graph(self): + """Returned checkpointer can compile a LangGraph StateGraph.""" + class SimpleState(TypedDict): + value: str + + def noop(state: SimpleState) -> SimpleState: + return state + + builder = StateGraph(SimpleState) # type: ignore + builder.add_node("noop", noop) + builder.add_edge(START, "noop") + builder.add_edge("noop", END) + + app: Any = builder.compile(checkpointer=create_checkpointer()) + assert app is not None + + def test_checkpointer_persists_state_across_invocations(self): + """State is preserved across invocations on the same thread_id.""" + class TickState(TypedDict): + values: list + + def append_node(state: TickState) -> TickState: + return {"values": state["values"] + ["tick"]} + + builder = StateGraph(TickState) # type: ignore + builder.add_node("append", append_node) + builder.add_edge(START, "append") + builder.add_edge("append", END) + + app: Any = builder.compile(checkpointer=create_checkpointer()) + config: RunnableConfig = {"configurable": {"thread_id": "test-thread-persist"}} + + result1 = app.invoke({"values": []}, config) + assert result1["values"] == ["tick"] + + result2 = app.invoke({"values": result1["values"]}, config) + assert result2["values"] == ["tick", "tick"] + + def test_different_thread_ids_are_isolated(self): + """Two thread IDs maintain independent state.""" + class NameState(TypedDict): + name: str + + def noop(state: NameState) -> NameState: + return state + + builder = StateGraph(NameState) # type: ignore + builder.add_node("noop", noop) + builder.add_edge(START, "noop") + builder.add_edge("noop", END) + + app: Any = builder.compile(checkpointer=create_checkpointer()) + config_a: RunnableConfig = {"configurable": {"thread_id": "thread-isolation-a"}} + config_b: RunnableConfig = {"configurable": {"thread_id": "thread-isolation-b"}} + + app.invoke({"name": "alice"}, config_a) + app.invoke({"name": "bob"}, config_b) + + assert app.get_state(config_a).values["name"] == "alice" + assert app.get_state(config_b).values["name"] == "bob" diff --git a/tests/agent_memory/unit/test_timed_memory.py b/tests/agent_memory/unit/test_timed_memory.py new file mode 100644 index 00000000..9aa6b2b3 --- /dev/null +++ b/tests/agent_memory/unit/test_timed_memory.py @@ -0,0 +1,167 @@ +"""Unit tests for TimedInMemorySaver and create_checkpointer(ttl_seconds=...).""" + +import time +from typing import Any, TypedDict +from langchain_core.runnables import RunnableConfig +from langgraph.checkpoint.memory import InMemorySaver +from langgraph.graph import END, START, StateGraph + +from sap_cloud_sdk.agent_memory.factory._timed_memory import TimedInMemorySaver +from sap_cloud_sdk.agent_memory.factory.langgraph_checkpoint import create_checkpointer + + +class TestCreateCheckpointerWithTTL: + """Tests for create_checkpointer(ttl_seconds=...) factory.""" + + def test_no_ttl_returns_in_memory_saver(self): + """Default returns plain InMemorySaver when ttl_seconds is None.""" + result = create_checkpointer() + assert isinstance(result, InMemorySaver) + assert not isinstance(result, TimedInMemorySaver) + + def test_ttl_returns_timed_in_memory_saver(self): + """ttl_seconds returns TimedInMemorySaver.""" + result = create_checkpointer(ttl_seconds=3600) + assert isinstance(result, TimedInMemorySaver) + + def test_ttl_value_is_set(self): + """ttl_seconds value is stored on the returned saver.""" + result = create_checkpointer(ttl_seconds=7200) + assert isinstance(result, TimedInMemorySaver) + assert result.ttl_seconds == 7200 + + def test_timed_saver_is_also_in_memory_saver(self): + """TimedInMemorySaver is a valid BaseCheckpointSaver subtype.""" + result = create_checkpointer(ttl_seconds=3600) + assert isinstance(result, InMemorySaver) + + def test_compiles_with_langgraph_graph(self): + """Timed checkpointer compiles a StateGraph correctly.""" + class State(TypedDict): + value: str + + def noop(state: State) -> State: + return state + + builder = StateGraph(State) # type: ignore + builder.add_node("noop", noop) + builder.add_edge(START, "noop") + builder.add_edge("noop", END) + + app: Any = builder.compile(checkpointer=create_checkpointer(ttl_seconds=3600)) + assert app is not None + + +class TestTimedInMemorySaver: + """Tests for TimedInMemorySaver behaviour.""" + + def test_default_ttl(self): + """Default TTL is 3600 seconds.""" + saver = TimedInMemorySaver() + assert saver.ttl_seconds == 3600 + + def test_custom_ttl(self): + """Custom TTL is stored correctly.""" + saver = TimedInMemorySaver(ttl_seconds=1800) + assert saver.ttl_seconds == 1800 + + def test_sweeper_thread_is_daemon(self): + """Background sweep thread is a daemon — dies with the process.""" + saver = TimedInMemorySaver() + assert saver._sweeper.daemon is True + assert saver._sweeper.is_alive() + + def test_put_updates_last_active(self): + """put() updates last-active timestamp for the thread.""" + class State(TypedDict): + x: int + + def noop(state: State) -> State: + return state + + builder = StateGraph(State) # type: ignore + builder.add_node("noop", noop) + builder.add_edge(START, "noop") + builder.add_edge("noop", END) + + saver = TimedInMemorySaver(ttl_seconds=3600) + app: Any = builder.compile(checkpointer=saver) + config: RunnableConfig = {"configurable": {"thread_id": "ttl-test-1"}} + + app.invoke({"x": 1}, config) + + assert "ttl-test-1" in saver._last_active + + def test_expired_thread_evicted(self): + """Threads inactive beyond ttl_seconds are evicted by _evict_expired.""" + class State(TypedDict): + x: int + + def noop(state: State) -> State: + return state + + builder = StateGraph(State) # type: ignore + builder.add_node("noop", noop) + builder.add_edge(START, "noop") + builder.add_edge("noop", END) + + saver = TimedInMemorySaver(ttl_seconds=1) + app: Any = builder.compile(checkpointer=saver) + config: RunnableConfig = {"configurable": {"thread_id": "evict-me"}} + + app.invoke({"x": 1}, config) + assert "evict-me" in saver._last_active + + # Backdate last-active to simulate 2 seconds of inactivity + with saver._lock: + saver._last_active["evict-me"] = time.monotonic() - 2 + + saver._evict_expired() + + assert "evict-me" not in saver._last_active + assert saver.storage.get("evict-me") is None + + def test_active_thread_not_evicted(self): + """Threads active within ttl_seconds are not evicted.""" + class State(TypedDict): + x: int + + def noop(state: State) -> State: + return state + + builder = StateGraph(State) # type: ignore + builder.add_node("noop", noop) + builder.add_edge(START, "noop") + builder.add_edge("noop", END) + + saver = TimedInMemorySaver(ttl_seconds=3600) + app: Any = builder.compile(checkpointer=saver) + config: RunnableConfig = {"configurable": {"thread_id": "keep-me"}} + + app.invoke({"x": 1}, config) + saver._evict_expired() + + assert "keep-me" in saver._last_active + assert saver.storage.get("keep-me") is not None + + def test_state_preserved_across_invocations(self): + """TimedInMemorySaver preserves state across invocations.""" + class State(TypedDict): + values: list + + def append_node(state: State) -> State: + return {"values": state["values"] + ["tick"]} + + builder = StateGraph(State) # type: ignore + builder.add_node("append", append_node) + builder.add_edge(START, "append") + builder.add_edge("append", END) + + app: Any = builder.compile(checkpointer=TimedInMemorySaver(ttl_seconds=3600)) + config: RunnableConfig = {"configurable": {"thread_id": "persist-test"}} + + result1 = app.invoke({"values": []}, config) + assert result1["values"] == ["tick"] + + result2 = app.invoke({"values": result1["values"]}, config) + assert result2["values"] == ["tick", "tick"]