From a7252bad6c8a632bd1aec34ba5ce3a312006b5df Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rickard=20Br=C3=BCel=20Gabrielsson?= Date: Fri, 23 Oct 2026 08:03:53 -0400 Subject: [PATCH] Tracked users: client.events.track and .list, user_id= on complete and sections Co-authored-by: Cursor --- README.md | 15 ++ src/behaviorgpt/_client.py | 66 ++++++-- src/behaviorgpt/resources/events.py | 159 +++++++++++++++++ src/behaviorgpt/resources/requests.py | 2 + src/behaviorgpt/types/__init__.py | 2 + src/behaviorgpt/types/http.py | 25 ++- tests/test_events.py | 235 ++++++++++++++++++++++++++ 7 files changed, 492 insertions(+), 12 deletions(-) create mode 100644 src/behaviorgpt/resources/events.py create mode 100644 tests/test_events.py diff --git a/README.md b/README.md index a77d81e..16c8ace 100644 --- a/README.md +++ b/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. diff --git a/src/behaviorgpt/_client.py b/src/behaviorgpt/_client.py index 04f5c85..7c5a062 100644 --- a/src/behaviorgpt/_client.py +++ b/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") diff --git a/src/behaviorgpt/resources/events.py b/src/behaviorgpt/resources/events.py new file mode 100644 index 0000000..76c6421 --- /dev/null +++ b/src/behaviorgpt/resources/events.py @@ -0,0 +1,159 @@ +from datetime import UTC, datetime +from typing import Optional, Protocol, Sequence +from uuid import uuid4 + +import httpx + +from behaviorgpt.resources._shared import handle_response +from behaviorgpt.types import Domains, Order, UserEvent +from behaviorgpt.types.http import TrackedEvent + +PATH = "/events" +OWN_PATH = "/events/own" +# the server's limit +MAX_LISTED = 200 + + +class EventContext(Protocol): + domains: Domains + timezone: str + default_catalog_id: str + + +def event_body( + context: EventContext, + user_id: str, + event: UserEvent, + *, + catalog_id: Optional[str], + timestamp: Optional[datetime], +) -> dict: + """The body `track` posts for `event`.""" + if timestamp is None: + timestamp = datetime.now(tz=UTC) + elif timestamp.tzinfo is None: + raise ValueError("timestamp must be timezone-aware") + + payload = event.model_dump(mode="json", exclude_none=True) + if isinstance(event, Order): + if not event.items: + raise ValueError("track an Order with the items bought: Order([item, ...])") + # The API stores orders with a cart id and a unit price per item; the + # model reads only which items were bought. + payload["cart_id"] = event.cart_id or str(uuid4()) + payload["items"] = [ + { + "product": entry.item, + "quantity": entry.quantity, + "unit_selling_price": 0.0, + } + for entry in event.items + ] + + return { + "user_id": user_id, + "store_id": catalog_id or context.default_catalog_id, + "event": payload, + "domains": context.domains.model_dump(mode="json", exclude_none=True), + "timezone": context.timezone, + "timestamp": timestamp.isoformat(), + } + + +def list_params( + user_id: Optional[str], limit: int, types: Optional[Sequence[str]] +) -> list[tuple[str, str]]: + if not 1 <= limit <= MAX_LISTED: + raise ValueError(f"limit must be between 1 and {MAX_LISTED}, got {limit}") + params: list[tuple[str, str]] = [("limit", str(limit))] + if user_id is not None: + params.append(("user_id", user_id)) + params.extend(("type", t) for t in types or ()) + return params + + +def tracked_events(data: dict) -> list[TrackedEvent]: + return [TrackedEvent(**entry) for entry in data["events"]] + + +class Events: + """Your users' events: record them with `track`, and BehaviorGPT reads them + as the history when you pass `user_id=` to `complete` or `sections`.""" + + def __init__(self, http_client: httpx.Client, context: EventContext): + self._client = http_client + self._context = context + + def track( + self, + user_id: str, + event: UserEvent, + *, + catalog_id: Optional[str] = None, + timestamp: Optional[datetime] = None, + ) -> str: + """Record that `user_id` did `event`, and return the event's id. + + `timestamp` is when it happened, now by default. Pass it when you send + events later than they happen, so that they stay in order. + """ + body = event_body( + self._context, + user_id, + event, + catalog_id=catalog_id, + timestamp=timestamp, + ) + return handle_response(self._client.post(PATH, json=body))["id"] + + def list( + self, + *, + user_id: Optional[str] = None, + limit: int = 50, + types: Optional[Sequence[str]] = None, + ) -> list[TrackedEvent]: + """Your account's latest events, newest first: all your users', or + `user_id`'s. `types` keeps only those event types, such as + `["viewItem", "search"]`. `limit` can be at most 200.""" + params = list_params(user_id, limit, types) + response = self._client.get(OWN_PATH, params=params) + return tracked_events(handle_response(response)) + + +class AsyncEvents: + """`Events` for `AsyncUnboxAIClient`.""" + + def __init__(self, http_client: httpx.AsyncClient, context: EventContext): + self._client = http_client + self._context = context + + async def track( + self, + user_id: str, + event: UserEvent, + *, + catalog_id: Optional[str] = None, + timestamp: Optional[datetime] = None, + ) -> str: + """See `Events.track`.""" + body = event_body( + self._context, + user_id, + event, + catalog_id=catalog_id, + timestamp=timestamp, + ) + return handle_response(await self._client.post(PATH, json=body))["id"] + + async def list( + self, + *, + user_id: Optional[str] = None, + limit: int = 50, + types: Optional[Sequence[str]] = None, + ) -> list[TrackedEvent]: + """See `Events.list`.""" + params = list_params(user_id, limit, types) + response = await self._client.get(OWN_PATH, params=params) + return tracked_events(handle_response(response)) diff --git a/src/behaviorgpt/resources/requests.py b/src/behaviorgpt/resources/requests.py index 4e437a1..f59d9c6 100644 --- a/src/behaviorgpt/resources/requests.py +++ b/src/behaviorgpt/resources/requests.py @@ -28,10 +28,12 @@ def make( num_candidates: Optional[int] = None, max_trailing_searches: Optional[int] = None, recommendation_type: Optional[RecommendationType] = None, + user_id: Optional[str] = None, ) -> UnboxAIRequest: return UnboxAIRequest( query=query, history=history or [], + user_id=user_id, store_id=catalog_id, domains=domains or self.domains, timezone=self.timezone, diff --git a/src/behaviorgpt/types/__init__.py b/src/behaviorgpt/types/__init__.py index ea4034f..a40d99d 100644 --- a/src/behaviorgpt/types/__init__.py +++ b/src/behaviorgpt/types/__init__.py @@ -27,6 +27,7 @@ Section, SectionsResponse, SimilarItemsRequest, + TrackedEvent, UnboxAIRequest, UnboxAIResponse, ) @@ -56,6 +57,7 @@ "CartItem", "Section", "SectionsResponse", + "TrackedEvent", ] diff --git a/src/behaviorgpt/types/http.py b/src/behaviorgpt/types/http.py index 07d83be..566a8b8 100644 --- a/src/behaviorgpt/types/http.py +++ b/src/behaviorgpt/types/http.py @@ -48,7 +48,10 @@ class RecommendationType(BaseModel): class UnboxAIRequest(BaseModel): query: Optional[str] = Field(default=None) - history: List[SessionEvent] = Field(default_factory=list) + # With `user_id`, the server reads that user's tracked events as the + # history, and `history` is left out: it accepts one or the other. + history: Optional[List[SessionEvent]] = Field(default_factory=list) + user_id: Optional[str] = None store_id: str domains: Domains timezone: str = "UTC" @@ -81,6 +84,26 @@ def validate_pagination(self) -> "UnboxAIRequest": raise ValueError("Offset + limit cannot exceed 10,000") return self + @model_validator(mode="after") + def user_id_or_history(self) -> "UnboxAIRequest": + if self.user_id is not None: + if self.history: + raise ValueError( + "pass a user_id or a history, not both: with a user_id the " + "history is that user's tracked events" + ) + self.history = None + return self + + +class TrackedEvent(BaseModel): + """An event as the API stored it.""" + + user_id: str + timestamp: datetime + event: dict + store_id: Optional[str] = None + class UnboxAIResponse(BaseModel): items: List[Item] diff --git a/tests/test_events.py b/tests/test_events.py new file mode 100644 index 0000000..5222c79 --- /dev/null +++ b/tests/test_events.py @@ -0,0 +1,235 @@ +"""Tracked events: `events.track` and `events.list`, and `user_id=` on +`complete` and `sections`, which read the history from them.""" + +import asyncio +import json +from datetime import UTC, datetime + +import httpx +import pytest +from pydantic import ValidationError + +from behaviorgpt import AsyncUnboxAIClient, Order, Search, UnboxAIClient, View +from behaviorgpt.types import UnboxAIRequest +from behaviorgpt.types.domains import Domains +from behaviorgpt.types.events import SessionEvent + +BASE = "http://sdk.test/v1" +MOMENT = datetime(2026, 10, 23, 12, 0, tzinfo=UTC) +PAGE = {"products": {"items": [], "offset": 0, "limit": 10}} +LISTED = { + "events": [ + { + "user_id": "u1", + "timestamp": "2026-10-23T12:03:00Z", + "store_id": "cat", + "event": {"type": "viewItem", "product": "p3"}, + "domains": {"market": "US"}, + "timezone": "UTC", + } + ] +} + + +def handler_for(requests: list[httpx.Request]): + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + if request.url.path == "/v1/events": + return httpx.Response(201, json={"id": "ev-1"}) + if request.url.path == "/v1/events/own": + return httpx.Response(200, json=LISTED) + return httpx.Response(200, json=PAGE) + + return handler + + +def mock_client() -> tuple[UnboxAIClient, list[httpx.Request]]: + requests: list[httpx.Request] = [] + client = UnboxAIClient("US", api_key="test", base_url=BASE) + client._http_client._transport = httpx.MockTransport(handler_for(requests)) + return client, requests + + +def body(request: httpx.Request) -> dict: + return json.loads(request.content) + + +def test_track_posts_the_event_and_returns_its_id(): + client, requests = mock_client() + event_id = client.events.track("u1", View("sku-1"), timestamp=MOMENT) + + assert event_id == "ev-1" + assert requests[0].method == "POST" and requests[0].url.path == "/v1/events" + assert body(requests[0]) == { + "user_id": "u1", + "store_id": "retail_catalog", + "event": {"type": "viewItem", "product": "sku-1"}, + "domains": { + "market": "US", + "postal_code": "unk", + "gender": "unk", + "age_group": "unk", + }, + "timezone": "UTC", + "timestamp": "2026-10-23T12:00:00+00:00", + } + + +def test_track_sends_searches_and_the_catalog(): + client, requests = mock_client() + client.events.track("u1", Search("sofa"), catalog_id="cat") + sent = body(requests[0]) + assert sent["event"] == {"type": "search", "query": "sofa"} + assert sent["store_id"] == "cat" + assert datetime.fromisoformat(sent["timestamp"]).tzinfo is not None + + +def test_track_sends_orders_as_the_api_stores_them(): + client, requests = mock_client() + client.events.track("u1", Order(["a", ("b", 2)], cart_id="cart-7")) + assert body(requests[0])["event"] == { + "type": "order", + "cart_id": "cart-7", + "items": [ + {"product": "a", "quantity": 1, "unit_selling_price": 0.0}, + {"product": "b", "quantity": 2, "unit_selling_price": 0.0}, + ], + } + + +def test_track_gives_an_order_a_cart_id(): + client, requests = mock_client() + client.events.track("u1", Order(["a"])) + assert body(requests[0])["event"]["cart_id"] + + +def test_track_refuses_an_order_without_items(): + client, requests = mock_client() + with pytest.raises(ValueError, match="items bought"): + client.events.track("u1", Order([])) + assert requests == [] + + +def test_track_refuses_a_naive_timestamp(): + client, _ = mock_client() + with pytest.raises(ValueError, match="timezone-aware"): + client.events.track("u1", View("a"), timestamp=datetime(2026, 10, 23)) + + +def test_list_reads_the_accounts_events(): + client, requests = mock_client() + events = client.events.list(user_id="u1", limit=10, types=["viewItem", "search"]) + + assert requests[0].method == "GET" and requests[0].url.path == "/v1/events/own" + assert requests[0].url.params.multi_items() == [ + ("limit", "10"), + ("user_id", "u1"), + ("type", "viewItem"), + ("type", "search"), + ] + assert [(ev.user_id, ev.event) for ev in events] == [ + ("u1", {"type": "viewItem", "product": "p3"}) + ] + assert events[0].timestamp == datetime(2026, 10, 23, 12, 3, tzinfo=UTC) + + +@pytest.mark.parametrize("limit", [0, 201]) +def test_list_refuses_a_limit_out_of_range(limit): + client, _ = mock_client() + with pytest.raises(ValueError, match="limit"): + client.events.list(limit=limit) + + +def test_complete_for_a_user_sends_the_user_and_no_history(): + client, requests = mock_client() + client.complete(user_id="u1", catalog_id="cat") + sent = body(requests[0]) + assert requests[0].url.path == "/v1/vector/recommendations" + assert sent["user_id"] == "u1" + assert "history" not in sent + + +def test_search_for_a_user_is_recorded_by_default(): + client, requests = mock_client() + client.complete([Search("sofa")], user_id="u1") + sent = body(requests[0]) + assert requests[0].url.path == "/v1/vector/search" + assert sent["query"] == "sofa" and sent["user_id"] == "u1" + assert "history" not in sent + assert sent["register_event"] is True + + +def test_search_for_a_user_can_be_left_unrecorded(): + client, requests = mock_client() + client.complete([Search("sofa")], user_id="u1", register_event=False) + assert body(requests[0])["register_event"] is False + + +def test_without_a_user_nothing_is_recorded_by_default(): + client, requests = mock_client() + client.complete([View("a"), Search("sofa")]) + sent = body(requests[0]) + assert sent["register_event"] is False + assert "user_id" not in sent + + +@pytest.mark.parametrize( + "history", + [[View("a")], [Search("a"), Search("b")], [[View("a")], [Search("b")]]], +) +def test_a_user_with_more_history_than_the_current_search_is_refused(history): + client, requests = mock_client() + with pytest.raises(ValueError, match="events.track"): + client.complete(history, user_id="u1") + assert requests == [] + + +def test_sections_for_a_user_send_the_user(): + client, requests = mock_client() + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(200, json={"sections": []}) + + client._http_client._transport = httpx.MockTransport(handler) + client.sections(user_id="u1") + sent = body(requests[0]) + assert requests[0].url.path == "/v1/vector/sections" + assert sent["user_id"] == "u1" and "history" not in sent + + +def test_request_refuses_a_user_and_a_history(): + with pytest.raises(ValidationError, match="not both"): + UnboxAIRequest( + user_id="u1", + store_id="cat", + domains={"market": "US"}, + history=[ + SessionEvent( + store_id="cat", + domains=Domains(market="US"), + timezone="UTC", + event=Search("a"), + session_id="s", + ) + ], + ) + + +def test_async_client_tracks_lists_and_completes_for_a_user(): + requests: list[httpx.Request] = [] + + async def run() -> None: + async with AsyncUnboxAIClient("US", api_key="test", base_url=BASE) as client: + client._http_client._transport = httpx.MockTransport(handler_for(requests)) + assert await client.events.track("u1", View("a")) == "ev-1" + assert len(await client.events.list()) == 1 + await client.complete([Search("sofa")], user_id="u1") + + asyncio.run(run()) + assert [r.url.path for r in requests] == [ + "/v1/events", + "/v1/events/own", + "/v1/vector/search", + ] + assert body(requests[2])["user_id"] == "u1"