Tracked users: client.events.track and .list, user_id= on complete and sections

README.md

@@ -79,6 +79,21 @@ client.complete([View(shoe)], predict="purchase")  # bought together
 client.complete([View(shoe), Order([shoe])])       # what they want after buying it
 ```
 
+### Tracked users
+
+Instead of sending the history with every call, you can record each event as it happens and pass the user's id. BehaviorGPT then reads that user's history from what you recorded.
+
+```python
+client.events.track("user-42", View(shoe))
+client.events.track("user-42", Order([shoe]), timestamp=bought_at)
+
+client.complete(user_id="user-42")                     # recommendations
+client.complete([Search("socks")], user_id="user-42")  # a search, personalized
+client.sections(user_id="user-42")
+```
+
+`user_id` takes the place of the history: the only event you pass with it is the current search, which is then recorded as the user's too (`register_event=False` leaves it out). `timestamp` defaults to now; pass it when you send events later than they happen. Your events are visible only to your account: `client.events.list(user_id="user-42")` returns that user's latest, newest first.
+
 ## Your own catalog
 
 The model can only return items it knows about. Out of the box the client uses `retail_catalog`, a sample retail catalog. To use the model on your own data, upload your catalog as a parquet file. A catalog doesn't have to be products for sale: artworks, articles, listings or anything else people browse and pick from work the same way.

src/behaviorgpt/_client.py

@@ -13,6 +13,7 @@
 
 from behaviorgpt._exceptions import AuthenticationError, UnboxAIError
 from behaviorgpt.resources.catalogs import AsyncCatalogs, Catalogs
+from behaviorgpt.resources.events import AsyncEvents, Events
 from behaviorgpt.resources.recommendations import AsyncRecommender, Recommender
 from behaviorgpt.resources.requests import UnboxAIRequester
 from behaviorgpt.resources.search import AsyncSearcher, Searcher
@@ -85,13 +86,18 @@ def _complete_request(
         offset: int,
         catalog_id: Optional[str],
         filters: Optional[dict[str, Any]],
-        register_event: bool,
+        register_event: Optional[bool],
         max_trailing_searches: int = DEFAULT_TRAILING_SEARCHES,
         predict: Predict = "view",
+        user_id: Optional[str] = None,
     ) -> tuple[bool, UnboxAIRequest, dict]:
         """The request `complete` sends, whether it goes to search, and its headers."""
         if predict not in ("view", "purchase"):
             raise ValueError(f"predict must be 'view' or 'purchase', not {predict!r}")
+        if user_id is not None:
+            _check_history_for_user(history)
+        if register_event is None:
+            register_event = user_id is not None
         active_catalog, request_headers = self._catalog_headers(catalog_id)
 
         history_events = self.__from_sequence(active_catalog, history)
@@ -112,6 +118,7 @@ def _complete_request(
                 filters=filters,
                 register_event=register_event,
                 max_trailing_searches=max_trailing_searches,
+                user_id=user_id,
             )
             return True, req, request_headers
 
@@ -125,6 +132,7 @@ def _complete_request(
             recommendation_type=(
                 RecommendationType(name="buy_next") if predict == "purchase" else None
             ),
+            user_id=user_id,
         )
         return False, req, request_headers
 
@@ -133,14 +141,15 @@ def _sections_request(
         history: UserHistoryInput,
         catalog_id: Optional[str],
         filters: Optional[dict[str, Any]],
-        register_event: bool,
+        register_event: Optional[bool],
         max_sections: int,
         max_items_per_section: int,
         min_items_per_section: int,
+        user_id: Optional[str] = None,
     ) -> tuple[dict, dict]:
         """The body `sections` sends, and its headers."""
         _, req, headers = self._complete_request(
-            history, 10, 0, catalog_id, filters, register_event
+            history, 10, 0, catalog_id, filters, register_event, user_id=user_id
         )
         body = sections_body(
             req, max_sections, max_items_per_section, min_items_per_section
@@ -255,19 +264,21 @@ def __init__(
         self.recommendations = Recommender(self._http_client)
         self.search = Searcher(self._http_client)
         self.catalogs = Catalogs(self._http_client)
+        self.events = Events(self._http_client, self)
         self._sections = Sectioner(self._http_client)
 
     def complete(
         self,
-        history: UserHistoryInput,
+        history: UserHistoryInput = (),
         limit: int = 10,
         offset: int = 0,
         *,
         catalog_id: Optional[str] = None,
         filters: Optional[dict[str, Any]] = None,
-        register_event: bool = False,
+        register_event: Optional[bool] = None,
         max_trailing_searches: int = DEFAULT_TRAILING_SEARCHES,
         predict: Predict = "view",
+        user_id: Optional[str] = None,
     ) -> UnboxAIResponse:
         """Rank the items most likely to come after `history`.
 
@@ -280,6 +291,12 @@ def complete(
         When the history ends in searches, up to `max_trailing_searches` of
         them in a row, the query included, reach the model, so `nike` then
         `shoes` reads as one intent.
+
+        With `user_id`, the history is that user's events recorded with
+        `client.events.track`: pass no history for recommendations, or
+        `[Search(query)]` to search. The search is then recorded as their
+        event too, so the next call sees it; `register_event=False` leaves it
+        out. Without `user_id` nothing is recorded unless `register_event=True`.
         """
         is_search, req, headers = self._complete_request(
             history,
@@ -290,20 +307,22 @@ def complete(
             register_event,
             max_trailing_searches,
             predict,
+            user_id,
         )
         endpoint = self.search if is_search else self.recommendations
         return endpoint.create(payload=req, headers=headers)
 
     def sections(
         self,
-        history: UserHistoryInput,
+        history: UserHistoryInput = (),
         *,
         catalog_id: Optional[str] = None,
         max_sections: int = 10,
         max_items_per_section: int = 25,
         min_items_per_section: int = 5,
         filters: Optional[dict[str, Any]] = None,
-        register_event: bool = False,
+        register_event: Optional[bool] = None,
+        user_id: Optional[str] = None,
     ) -> SectionsResponse:
         """The items most likely to come after `history`, grouped into titled
         sections such as "black coats", best section first.
@@ -317,6 +336,9 @@ def sections(
         `max_sections` can be at most 15 and `max_items_per_section` at most
         100. Each item's `data` holds its generated search terms in
         `top_terms`, best first.
+
+        `user_id` reads the history from that user's tracked events, as in
+        `complete`.
         """
         body, headers = self._sections_request(
             history,
@@ -326,6 +348,7 @@ def sections(
             max_sections,
             max_items_per_section,
             min_items_per_section,
+            user_id,
         )
         return self._sections.create(body, headers=headers)
 
@@ -559,6 +582,7 @@ def __init__(
         self.recommendations = AsyncRecommender(self._http_client)
         self.search = AsyncSearcher(self._http_client)
         self.catalogs = AsyncCatalogs(self._http_client)
+        self.events = AsyncEvents(self._http_client, self)
         self._sections = AsyncSectioner(self._http_client)
 
     async def __aenter__(self) -> "AsyncUnboxAIClient":
@@ -572,15 +596,16 @@ async def aclose(self) -> None:
 
     async def complete(
         self,
-        history: UserHistoryInput,
+        history: UserHistoryInput = (),
         limit: int = 10,
         offset: int = 0,
         *,
         catalog_id: Optional[str] = None,
         filters: Optional[dict[str, Any]] = None,
-        register_event: bool = False,
+        register_event: Optional[bool] = None,
         max_trailing_searches: int = DEFAULT_TRAILING_SEARCHES,
         predict: Predict = "view",
+        user_id: Optional[str] = None,
     ) -> UnboxAIResponse:
         """Rank the items most likely to come after `history`; see
         `UnboxAIClient.complete`."""
@@ -593,20 +618,22 @@ async def complete(
             register_event,
             max_trailing_searches,
             predict,
+            user_id,
         )
         endpoint = self.search if is_search else self.recommendations
         return await endpoint.create(payload=req, headers=headers)
 
     async def sections(
         self,
-        history: UserHistoryInput,
+        history: UserHistoryInput = (),
         *,
         catalog_id: Optional[str] = None,
         max_sections: int = 10,
         max_items_per_section: int = 25,
         min_items_per_section: int = 5,
         filters: Optional[dict[str, Any]] = None,
-        register_event: bool = False,
+        register_event: Optional[bool] = None,
+        user_id: Optional[str] = None,
     ) -> SectionsResponse:
         """Items grouped into titled sections; see `UnboxAIClient.sections`."""
         body, headers = self._sections_request(
@@ -617,6 +644,7 @@ async def sections(
             max_sections,
             max_items_per_section,
             min_items_per_section,
+            user_id,
         )
         return await self._sections.create(body, headers=headers)
 
@@ -740,6 +768,22 @@ async def umap(self, *, catalog_id: Optional[str] = None, wait: float = 0.0) ->
             await asyncio.sleep(UMAP_POLL)
 
 
+def _check_history_for_user(history: UserHistoryInput) -> None:
+    """With a `user_id` the server reads the history from the user's tracked
+    events, so a history passed along can only hold the current search."""
+    events: list[UserEvent] = (
+        [ev for session in history for ev in session]
+        if history and isinstance(history[0], (list, tuple))
+        else list(cast(Sequence[UserEvent], history))
+    )
+    if len(events) > 1 or (events and not isinstance(events[0], Search)):
+        raise ValueError(
+            "with user_id the history is the user's tracked events: pass no "
+            "history, or only the current Search, and record the rest with "
+            "client.events.track"
+        )
+
+
 def _pick_random(top: UnboxAIResponse, catalog_id: str) -> Item:
     if not top.items:
         raise ValueError(f"catalog {catalog_id!r} returned no items")