google / google/adk-python

BaseMemoryService has no efficient way to access cross-session events — forces N+1 queries on every custom implementation

Ouverte
#5,300 4 commentaires 0 réactions 2 personnes assignées Réclamée par @wyf7107 Voir sur GitHub
needs review services spam
Langage dominant
Python
Étoiles
21.5k
Forks
4k
Merge moyen
1 j 14 h
PR mergées (30 j)
37

Description

# BaseMemoryService has no efficient way to access cross-session events — forces N+1 queries on every custom implementation

## The problem affects all custom BaseMemoryService implementations

Any `BaseMemoryService` that needs cross-session context (conversation history across multiple sessions for the same user) is forced into an N+1 query pattern. There is no bulk method on `BaseSessionService` to retrieve events across sessions — you must call `list_sessions()` (which explicitly returns sessions **without events**) followed by `get_session()` for each session individually.

This is not a niche use case. Long-term memory, user preference extraction, cross-conversation summarization, and observational memory all require seeing events beyond the single session passed to `add_session_to_memory(session)`. Every implementation of these patterns on ADK today must work around this gap.

LangGraph avoids this entirely by decoupling its Store (cross-thread memory) from the Checkpointer (per-thread state) — the Store never reads from session records. Mastra avoids it by owning its own `mastra_messages` table and querying across threads via `listMessagesByResourceId()`. ADK has neither escape hatch.

## Reproduction

Minimal `BaseMemoryService` that needs cross-session history:

```python
from google.adk.memory import BaseMemoryService
from google.adk.sessions import BaseSessionService, Session

class CrossSessionMemoryService(BaseMemoryService):
"""Memory service that compresses conversation history across sessions."""

def __init__(self, session_service: BaseSessionService):
self._session_service = session_service

async def add_session_to_memory(self, session: Session) -> None:
# We need events from ALL of this user's sessions, not just the current one.
# But `session.events` only contains the current session's events.

# Step 1: list_sessions returns sessions WITHOUT events or state.
# (ListSessionsResponse docstring: "The events and states are not set
# within each Session object.")
list_resp = await self._session_service.list_sessions(
app_name=session.app_name,
user_id=session.user_id,
)

# Step 2: N+1 — must call get_session() individually for each session
# to get their events. No bulk alternative exists.
all_events = []
for s in list_resp.sessions:
if s.id == session.id:
continue # current session's events already in hand
full = await self._session_service.get_session(
app_name=session.app_name,
user_id=session.user_id,
session_id=s.id,
)
if full and full.events:
all_events.extend(full.events) # one DB round-trip per session

all_events.extend(session.events)
# ... compress, summarize, store ...

async def search_memory(self, *, app_name: str, user_id: str, query: str):
return {"memory": "..."}
```

The N+1 cost scales with the number of sessions per user. For `DatabaseSessionService`, each `get_session()` call executes a separate `SELECT` against the `app_events` table filtered by `session_id`. The table already has `app_name` and `user_id` columns that could support a single cross-session query — but no method exposes this.

## Proposed fix

Add an optional `get_user_events()` method to `BaseSessionService` with a default N+1 fallback, so concrete backends can override with a single query:

```python
# google/adk/sessions/base_session_service.py

class BaseSessionService(ABC):
# ... existing methods ...

async def get_user_events(
self,
*,
app_name: str,
user_id: str,
since: float | None = None,
) -> list[Event]:
"""Returns events across all sessions for a user.

Args:
app_name: The name of the app.
user_id: The ID of the user.
since: If provided, only return events with timestamp > since.
Enables efficient delta queries for memory services that
track their last observation time.

Returns:
Events sorted chronologically across all sessions.

Override in concrete implementations for single-query efficiency.
Default implementation falls back to list_sessions + get_session (N+1).
"""
list_resp = await self.list_sessions(
app_name=app_name, user_id=user_id
)
all_events: list[Event] = []
for s in list_resp.sessions:
if since and s.last_update_time <= since:
continue
full = await self.get_session(
app_name=app_name, user_id=user_id, session_id=s.id
)
if full:
all_events.extend(full.events)
all_events.sort(key=lambda e: getattr(e, "timestamp", 0))
if since:
all_events = [
e for e in all_events
if getattr(e, "timestamp", 0) > since
]
return all_events
```

```python
# google/adk/sessions/database_session_service.py

class DatabaseSessionService(BaseSessionService):
# ... existing methods ...

async def get_user_events(
self,
*,
app_name: str,
user_id: str,
since: float | None = None,
) -> list[Event]:
"""Single-query override — reads directly from app_events table."""
async with self._engine.begin() as conn:
stmt = (
select(self._event_table)
.where(self._event_table.c.app_name == app_name)
.where(self._event_table.c.user_id == user_id)
)
if since:
stmt = stmt.where(self._event_table.c.timestamp > since)
stmt = stmt.order_by(self._event_table.c.timestamp)
rows = (await conn.execute(stmt)).fetchall()
return [self._row_to_event(row) for row in rows]
```

This lets any `BaseMemoryService` replace:
```python
# Before: N+1 queries
list_resp = await session_service.list_sessions(...)
for s in list_resp.sessions:
full = await session_service.get_session(...)
```

With:
```python
# After: 1 query (on DatabaseSessionService), N+1 fallback (on others)
events = await session_service.get_user_events(
app_name=session.app_name,
user_id=session.user_id,
since=last_observed_at,
)
```

## Context

- **Affected**: Every custom `BaseMemoryService` that needs cross-session context
- **Not affected**: `InMemoryMemoryService` (test-only, no cross-session), `VertexAiMemoryBankService` (uses Vertex AI RAG, not session events)
- **Related**: #4735 (proposes `DatabaseMemoryService` with its own event table — sidesteps N+1 by owning event data, but doesn't fix the interface gap), #2524 (requests DatabaseMemoryService/RedisMemoryService)
- **Comparison**: LangGraph's `Store` is decoupled from `Checkpointer` (no cross-session reads needed). Mastra owns `mastra_messages` table with `listMessagesByResourceId()` (single cross-thread query). ADK's `BaseMemoryService` is the only major framework where the memory service must reconstruct history from sessions via the session service.

## Environment

- ADK version: 2.0.0a3
- Python: 3.14
- Session backend: `DatabaseSessionService` (PostgreSQL via asyncpg)

Guide de contribution

Ouvrir le guide de contribution

Évaluation

Cette issue n'a pas encore été évaluée.

Recevez les nouvelles issues par e-mail

Un résumé court des issues GitHub adaptées aux débutants.