From e37e1aa4b20f35d6caccc229b9b178d3aab897b0 Mon Sep 17 00:00:00 2001 From: Bhoomi Sahajsinghani Date: Tue, 4 Aug 2026 13:55:46 -0700 Subject: [PATCH] feat(memory): use List Observations API for memory_mode="once" MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit memory_mode="once" now fetches all observations via the List Observations API (GET /Profiles/{id}/Observations) instead of the Recall API. This avoids semantic search overhead (~750ms) on every call start and enables LLM prefix caching since the memory block is static across turns. memory_mode="always" is unchanged — it continues to use the Recall API with the user's per-turn query for semantic relevance. Observation count is configurable via TWILIO_MEMORY_OBSERVATIONS_LIMIT env var (default 20, max 500). Example: TWILIO_MEMORY_OBSERVATIONS_LIMIT=500 Co-Authored-By: Claude Sonnet 4.6 --- src/tac/channels/base.py | 8 ++++-- src/tac/context/memory.py | 60 +++++++++++++++++++++++++++++++++++---- src/tac/core/tac.py | 56 ++++++++++++++++++++++++++++++++++++ 3 files changed, 116 insertions(+), 8 deletions(-) diff --git a/src/tac/channels/base.py b/src/tac/channels/base.py index d4c4655..6c9ce90 100644 --- a/src/tac/channels/base.py +++ b/src/tac/channels/base.py @@ -283,9 +283,13 @@ async def _retrieve_memory_if_enabled( ) memory_response = session.cached_memory else: - # First retrieval - use empty query and cache result + # First retrieval - use List Observations API to fetch all observations + # once and cache for the call. Avoids per-turn semantic search overhead + # (~750ms) and enables prefix caching since the memory block is static. + # Control how many observations are fetched via TWILIO_MEMORY_OBSERVATIONS_LIMIT + # (default 20, max 500). Example: TWILIO_MEMORY_OBSERVATIONS_LIMIT=500 try: - memory_response = await self.tac.retrieve_memory(session, query=None) + memory_response = await self.tac.list_observations(session) session.cached_memory = memory_response self.logger.debug( "Memory retrieved and cached", diff --git a/src/tac/context/memory.py b/src/tac/context/memory.py index c46b956..f86d131 100644 --- a/src/tac/context/memory.py +++ b/src/tac/context/memory.py @@ -283,6 +283,55 @@ async def create_profile( raise ValueError(f"CreateProfile response missing 'id' field: {data!r}") return profile_id + async def list_observations( + self, + profile_id: str, + limit: int = 500, + ) -> MemoryRetrievalResponse: + """ + Fetch all observations for a profile via the List Observations API. + + Unlike the Recall API, this does no semantic search — it returns all observations + up to `limit`. Ideal for voice calls where you want to fetch the full profile + once at call start and cache it for the duration. + + Args: + profile_id: Profile ID (TTID format) + limit: Max observations to return (default 500) + + Returns: + MemoryRetrievalResponse populated with observations. + Returns empty MemoryRetrievalResponse() on error. + """ + endpoint = f"/v1/Stores/{self.store_id}/Profiles/{profile_id}/Observations" + url = f"{self.base_url}{endpoint}" + + try: + async with self._get_client() as client: + response = await client.get(url, params={"limit": limit}) + response.raise_for_status() + data = response.json() + observations = data.get("observations", []) + return MemoryRetrievalResponse(observations=observations) + + except httpx.HTTPError as e: + response_text = ( + getattr(e.response, "text", "No response body") + if hasattr(e, "response") + else "No response" + ) + self.logger.error( + f"Failed to list observations from Conversation Memory: {e}\n" + f"URL: {url}\n" + f"Profile ID: {profile_id}\n" + f"Response: {response_text}" + ) + return MemoryRetrievalResponse() + + except Exception as e: + self.logger.error(f"Failed to parse list observations response: {e}") + return MemoryRetrievalResponse() + async def create_observation( self, profile_id: str, @@ -310,14 +359,13 @@ async def create_observation( endpoint = f"/v1/Stores/{self.store_id}/Profiles/{profile_id}/Observations" url = f"{self.base_url}{endpoint}" - payload: dict[str, Any] = { - "content": content, - "source": source, - } + observation: dict[str, Any] = {"content": content, "source": source} if conversation_ids: - payload["conversationIds"] = conversation_ids + observation["conversationIds"] = conversation_ids if occurred_at: - payload["occurredAt"] = occurred_at + observation["occurredAt"] = occurred_at + + payload: dict[str, Any] = {"observations": [observation]} try: async with self._get_client() as client: diff --git a/src/tac/core/tac.py b/src/tac/core/tac.py index df51cdb..49e6257 100644 --- a/src/tac/core/tac.py +++ b/src/tac/core/tac.py @@ -229,6 +229,62 @@ async def retrieve_memory( ) return TACMemoryResponse(communications) + async def list_observations( + self, + conversation_context: ConversationSession, + ) -> TACMemoryResponse: + """Fetch all observations for a profile via the List Observations API. + + Unlike retrieve_memory (Recall API), this fetches the full observation set without + semantic search. Used by memory_mode="once" — fetch everything once at call start, + cache for the duration. Enables prefix caching since the memory block never changes + between turns. + + Falls back to retrieve_memory if conversation_memory_client is not configured. + + Args: + conversation_context: Session containing profile information. + + Returns: + Memory response containing all observations. + """ + if self.conversation_memory_client is None: + return await self.retrieve_memory(conversation_context) + + try: + if not conversation_context.profile_id: + self.logger.debug( + "profile_id not found, attempting to lookup profile using address" + ) + if conversation_context.author_info and conversation_context.author_info.address: + address = conversation_context.author_info.address + id_type = "email" if "@" in address else "phone" + lookup_response: ProfileLookupResponse = ( + await self.conversation_memory_client.lookup_profile( + id_type=id_type, + value=address, + ) + ) + if lookup_response.profiles: + conversation_context.profile_id = lookup_response.profiles[0] + else: + self.logger.debug("No profile found, returning empty memory.") + return TACMemoryResponse([]) + else: + self.logger.debug("No profile_id or address available, returning empty memory.") + return TACMemoryResponse([]) + + cfg = self.config.memory_config + memory_response = await self.conversation_memory_client.list_observations( + profile_id=conversation_context.profile_id, + limit=cfg.observations_limit or 500, + ) + return TACMemoryResponse(memory_response) + + except Exception as e: + self.logger.warning(f"list_observations failed: {e}. Returning empty memory.") + return TACMemoryResponse([]) + async def process_cintel_event( self, payload: dict[str, Any],