Release 0.6.4: umap(catalog_ids=...)

README.md

@@ -117,6 +117,14 @@ for item in res.items:
     print(item.catalog_id, item.data["name"])
 ```
 
+`catalog_ids` also draws several catalogs on one map, one colour each, so you can see where they overlap. The first request takes a few minutes; later ones for the same catalogs return the stored map:
+
+```python
+page = client.umap(catalog_ids=[job.catalog_id, "retail_catalog"], wait=300)
+with open("map.html", "w") as f:
+    f.write(page)
+```
+
 [`docs/catalogs.md`](https://github.com/Unbox-AI/behaviorgpt/blob/main/docs/catalogs.md) has the details: catalog slots, what replacing an item keeps, how long keywords take, what happens when a job fails, and how a history names items of several catalogs.
 
 If mapping your data to the format is hard, open a coding agent in this directory and ask it to embed your catalog. The repo ships a skill at [`.claude/skills/embed-catalog/SKILL.md`](https://github.com/Unbox-AI/behaviorgpt/blob/main/.claude/skills/embed-catalog/SKILL.md) that walks the agent through mapping your columns, building and validating the parquet, checking image URLs, and uploading. Claude Code picks it up automatically; for other agents, point them at that file.

docs/catalog-format.md

@@ -64,6 +64,8 @@ Listed in order of impact.
 
 `client.umap()` returns a 2D map of the catalog's items. The map is built only for catalogs of up to 100,000 items, within about a minute after the catalog reads `ready`. Laying out more items than that does not fit in the embedding worker's memory, so larger catalogs, such as the shared `retail_catalog`, never get one. Embedding and every query work the same at any size; only the map is skipped. For a catalog without a map, `umap()` raises `UnboxAIError` with status 404. Right after an embed, pass `wait=` (seconds) to keep asking until the map is built, for example `client.umap(catalog_id=job.catalog_id, wait=180)`. Items added with `client.add_items()` are not drawn on the map until the next full upload of the catalog.
 
+`client.umap(catalog_ids=[...])` draws one map of several catalogs instead, one colour each, at most 5 of them. It is drawn on request and works at any size: above 100,000 items in total, every catalog is sampled evenly, and a small catalog next to a large one is shown whole. A single large catalog can be mapped this way too, for example `client.umap(catalog_ids=["retail_catalog"], wait=300)`. The first request takes a few minutes, so pass `wait=`; a map still being drawn when the wait is up raises `UnboxAIError` with status 202, and asking again later finds it. Later requests for the same catalogs return the stored map at once, until one of them is uploaded again or gets items. To draw the map yourself, `client.catalogs.create_map(catalog_ids=[...])` returns its `map_id`, and once `client.catalogs.get_map(map_id).status` is `"ready"`, `client.catalogs.get_map_points(map_id)` returns every item's position, name, category and image, with the catalog it is in.
+
 ## Hosting your catalog
 
 The parquet itself is uploaded, but images are fetched from their `image_url` by UnboxAI's servers. Every URL must therefore be public, `https://`, and reachable from a server, not only from a browser. Some sites put images behind bot protection (Cloudflare, for example) that serves browsers but blocks servers; the job then fails when all image fetches come back 403. Test a few URLs with `curl` from a cloud machine if in doubt.

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "behaviorgpt"
-version = "0.6.3"
+version = "0.6.4"
 description = "Python client for BehaviorGPT, UnboxAI's Large Behavioral Model for search, recommendations and personalization"
 readme = "README.md"
 license = "MIT"

src/behaviorgpt/_client.py

@@ -23,6 +23,7 @@
     AddKeywordsJobDetails,
     AddToCart,
     CartItem,
+    CatalogMap,
     CatalogSummary,
     Domains,
     EmbedJobDetails,
@@ -623,7 +624,13 @@ def random_product(self, *args: Any, **kwargs: Any) -> Item:
         )
         return self.random_item(*args, **kwargs)
 
-    def umap(self, *, catalog_id: Optional[str] = None, wait: float = 0.0) -> str:
+    def umap(
+        self,
+        *,
+        catalog_id: Optional[str] = None,
+        catalog_ids: Optional[Sequence[str]] = None,
+        wait: float = 0.0,
+    ) -> str:
         """The catalog's UMAP, a 2D map of its items, as an HTML page.
 
         Maps exist only for catalogs of up to 100,000 items, and appear within
@@ -632,12 +639,27 @@ def umap(self, *, catalog_id: Optional[str] = None, wait: float = 0.0) -> str:
         (such as the shared `retail_catalog`) never get one. A missing map
         raises `UnboxAIError` with status 404 and says which case applies.
 
+        `catalog_ids` lays several catalogs out on one map instead, one colour
+        each, so items of different catalogs that sit close together are
+        alike. It is drawn on request, within a few minutes, and kept until one
+        of the catalogs is uploaded again. Above 100,000 items every catalog is
+        sampled evenly, so any catalog can be mapped this way, `retail_catalog`
+        too, alone or with others. A map still being drawn when `wait` is up
+        raises `UnboxAIError` with status 202; asking again later finds it.
+
         `wait` is how many seconds to keep asking while the map is missing,
         for a catalog that just became ready. A catalog that never gets a map
         raises only once that time is up.
         """
-        active_catalog: str = catalog_id or self.default_catalog_id
         deadline = time.monotonic() + wait
+        if catalog_ids is not None:
+            ids = self._target(catalog_id, catalog_ids).catalog_ids
+            joint = self.catalogs.create_map(catalog_ids=ids)
+            while not _map_ready(joint, deadline):
+                time.sleep(UMAP_POLL)
+                joint = self.catalogs.get_map(joint.map_id)
+            return self.catalogs.get_map_page(joint.map_id)
+        active_catalog: str = catalog_id or self.default_catalog_id
         while True:
             try:
                 return self.catalogs.get_umap(catalog_id=active_catalog)
@@ -864,10 +886,23 @@ async def random_item(
         top = await self.complete(history=[], catalog_id=catalog_id, limit=window)
         return _pick_random(top, catalog_id or self.default_catalog_id)
 
-    async def umap(self, *, catalog_id: Optional[str] = None, wait: float = 0.0) -> str:
+    async def umap(
+        self,
+        *,
+        catalog_id: Optional[str] = None,
+        catalog_ids: Optional[Sequence[str]] = None,
+        wait: float = 0.0,
+    ) -> str:
         """The catalog's UMAP as an HTML page; see `UnboxAIClient.umap`."""
-        active_catalog: str = catalog_id or self.default_catalog_id
         deadline = time.monotonic() + wait
+        if catalog_ids is not None:
+            ids = self._target(catalog_id, catalog_ids).catalog_ids
+            joint = await self.catalogs.create_map(catalog_ids=ids)
+            while not _map_ready(joint, deadline):
+                await asyncio.sleep(UMAP_POLL)
+                joint = await self.catalogs.get_map(joint.map_id)
+            return await self.catalogs.get_map_page(joint.map_id)
+        active_catalog: str = catalog_id or self.default_catalog_id
         while True:
             try:
                 return await self.catalogs.get_umap(catalog_id=active_catalog)
@@ -877,6 +912,27 @@ async def umap(self, *, catalog_id: Optional[str] = None, wait: float = 0.0) ->
             await asyncio.sleep(UMAP_POLL)
 
 
+def _map_ready(joint: CatalogMap, deadline: float) -> bool:
+    """Whether the map can be read; raises when it failed, or when it is still
+    being drawn and polling once more would pass `deadline`."""
+    if joint.status == "ready":
+        return True
+    catalogs = ", ".join(joint.catalog_ids)
+    if joint.status == "failed":
+        raise UnboxAIError(
+            f"the map of {catalogs} failed: {joint.error}",
+            response=joint.model_dump(mode="json"),
+        )
+    if time.monotonic() + UMAP_POLL > deadline:
+        raise UnboxAIError(
+            f"the map of {catalogs} is still being drawn ({joint.map_id}); pass "
+            "wait= (seconds) to wait for it, e.g. wait=300, or ask again later",
+            status_code=202,
+            response=joint.model_dump(mode="json"),
+        )
+    return False
+
+
 def _by_catalog(event: UserEvent, target: _Target) -> list[tuple[str, UserEvent]]:
     """An event as the API reads it, each with the catalog its items are in.
     An event names one catalog, so an order of items from several becomes one

src/behaviorgpt/resources/catalogs.py

@@ -5,7 +5,7 @@
 import warnings
 from datetime import UTC, datetime
 from pathlib import Path
-from typing import Any, Callable, Optional, TextIO
+from typing import Any, Callable, Optional, Sequence, TextIO
 from urllib.parse import quote
 
 import httpx
@@ -19,6 +19,7 @@
 from behaviorgpt.types import (
     AddItemsJobDetails,
     AddKeywordsJobDetails,
+    CatalogMap,
     CatalogSummary,
     EmbedJobDetails,
     JobStatus,
@@ -355,6 +356,16 @@ def umap_page(response: httpx.Response) -> str:
     return response.text
 
 
+def map_body(catalog_ids: Sequence[str]) -> dict:
+    if isinstance(catalog_ids, str):
+        raise TypeError("catalog_ids takes a list of catalog ids, not a string")
+    return {"catalog_ids": list(dict.fromkeys(catalog_ids))}
+
+
+def map_path(map_id: str, part: str = "") -> str:
+    return f"/maps/{quote(map_id, safe='')}" + (f"/{part}" if part else "")
+
+
 class JobWatcher:
     """The rules `wait_for_job` polls by, for the sync and the async client alike.
 
@@ -650,6 +661,24 @@ def get_similar_products(
     def get_umap(self, *, catalog_id: str):
         return umap_page(self._client.get(f"/{catalog_id}/umap"))
 
+    def create_map(self, *, catalog_ids: Sequence[str]) -> CatalogMap:
+        """Ask for one map of these catalogs; it is drawn in the background
+        unless it already exists. `UnboxAIClient.umap(catalog_ids=...)` waits
+        for it and returns its page."""
+        response = self._client.post("/maps", json=map_body(catalog_ids))
+        return CatalogMap(**handle_response(response))
+
+    def get_map(self, map_id: str) -> CatalogMap:
+        return CatalogMap(**handle_response(self._client.get(map_path(map_id))))
+
+    def get_map_page(self, map_id: str) -> str:
+        return umap_page(self._client.get(map_path(map_id, "page")))
+
+    def get_map_points(self, map_id: str) -> dict:
+        """The map as data, to draw it yourself: one list per field and one
+        entry per item, `catalog` indexing `catalogs`."""
+        return handle_response(self._client.get(map_path(map_id, "points")))
+
     def list_catalogs(self) -> list[CatalogSummary]:
         return catalog_summaries(self._client.get("/catalogs"))
 
@@ -800,6 +829,19 @@ async def get_similar_items(
     async def get_umap(self, *, catalog_id: str) -> str:
         return umap_page(await self._client.get(f"/{catalog_id}/umap"))
 
+    async def create_map(self, *, catalog_ids: Sequence[str]) -> CatalogMap:
+        response = await self._client.post("/maps", json=map_body(catalog_ids))
+        return CatalogMap(**handle_response(response))
+
+    async def get_map(self, map_id: str) -> CatalogMap:
+        return CatalogMap(**handle_response(await self._client.get(map_path(map_id))))
+
+    async def get_map_page(self, map_id: str) -> str:
+        return umap_page(await self._client.get(map_path(map_id, "page")))
+
+    async def get_map_points(self, map_id: str) -> dict:
+        return handle_response(await self._client.get(map_path(map_id, "points")))
+
     async def list_catalogs(self) -> list[CatalogSummary]:
         return catalog_summaries(await self._client.get("/catalogs"))
 

src/behaviorgpt/types/__init__.py

@@ -17,6 +17,7 @@
 from behaviorgpt.types.http import (
     AddItemsJobDetails,
     AddKeywordsJobDetails,
+    CatalogMap,
     CatalogSummary,
     EmbedJobDetails,
     Item,
@@ -47,6 +48,7 @@
     "EmbedJobDetails",
     "AddItemsJobDetails",
     "AddKeywordsJobDetails",
+    "CatalogMap",
     "CatalogSummary",
     "JobStatus",
     "ItemsRejected",

src/behaviorgpt/types/http.py

@@ -186,6 +186,19 @@ class CatalogSummary(BaseModel):
     owned: bool
 
 
+class CatalogMap(BaseModel):
+    """A map of several catalogs laid out together (`POST /maps`)."""
+
+    # the same for the same catalogs until one of them is uploaded again
+    map_id: str
+    # "building", "ready" or "failed"
+    status: str
+    catalog_ids: list[str]
+    requested_at: Optional[datetime] = None
+    # why the map failed; asking for it again retries it
+    error: Optional[str] = None
+
+
 class SimilarItemsRequest(BaseModel):
     model_config = ConfigDict(serialize_by_alias=True)
 

tests/test_maps.py

@@ -0,0 +1,145 @@
+"""Joint maps of several catalogs: `umap(catalog_ids=...)` asks for the map,
+polls until it is drawn and returns its page; the catalogs resource reads its
+status and points."""
+
+import asyncio
+import gzip
+import json
+
+import httpx
+import pytest
+
+from behaviorgpt import AsyncUnboxAIClient, UnboxAIClient, UnboxAIError
+
+BASE = "http://sdk.test/v1"
+MAP_ID = "map_0123456789abcdef01234567"
+IDS = ["cat_a", "retail_catalog"]
+
+
+def joint(status: str, error: str | None = None, code: int = 200) -> httpx.Response:
+    return httpx.Response(
+        code,
+        json={
+            "map_id": MAP_ID,
+            "status": status,
+            "catalog_ids": IDS,
+            "requested_at": "2026-10-24T21:16:28Z",
+            "error": error,
+        },
+    )
+
+
+def serve(statuses: list[httpx.Response], client_class=UnboxAIClient):
+    """POST /maps answers the first status, each GET /maps/{id} the next (the
+    last repeats), and the page and points answer as stored."""
+    calls: list[httpx.Request] = []
+    pending = list(statuses)
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        calls.append(request)
+        path = request.url.path
+        if path.endswith("/page"):
+            return httpx.Response(200, text="<html>joint</html>")
+        if path.endswith("/points"):
+            body = gzip.compress(json.dumps({"catalog": [0, 1]}).encode())
+            gzipped = {"content-encoding": "gzip"}
+            return httpx.Response(200, content=body, headers=gzipped)
+        return pending.pop(0) if len(pending) > 1 else pending[0]
+
+    client = client_class("US", api_key="test", base_url=BASE)
+    client._http_client._transport = httpx.MockTransport(handler)
+    return client, calls
+
+
+@pytest.fixture
+def clock(monkeypatch):
+    now = [0.0]
+    monkeypatch.setattr("behaviorgpt._client.time.monotonic", lambda: now[0])
+    monkeypatch.setattr(
+        "behaviorgpt._client.time.sleep", lambda s: now.__setitem__(0, now[0] + s)
+    )
+    return now
+
+
+def test_a_ready_map_is_read_at_once():
+    client, calls = serve([joint("ready")])
+    assert client.umap(catalog_ids=IDS) == "<html>joint</html>"
+    post, page = calls
+    assert (post.method, post.url.path) == ("POST", "/v1/maps")
+    assert json.loads(post.content) == {"catalog_ids": IDS}
+    assert page.url.path == f"/v1/maps/{MAP_ID}/page"
+
+
+def test_waits_while_the_map_is_drawn(clock):
+    client, calls = serve(
+        [joint("building", code=202), joint("building"), joint("ready")]
+    )
+    assert client.umap(catalog_ids=IDS, wait=60) == "<html>joint</html>"
+    assert [c.url.path for c in calls] == [
+        "/v1/maps",
+        f"/v1/maps/{MAP_ID}",
+        f"/v1/maps/{MAP_ID}",
+        f"/v1/maps/{MAP_ID}/page",
+    ]
+
+
+def test_a_map_still_drawn_at_the_deadline_raises_202(clock):
+    client, calls = serve([joint("building", code=202)])
+    with pytest.raises(UnboxAIError, match="still being drawn") as exc:
+        client.umap(catalog_ids=IDS, wait=12)
+    assert exc.value.status_code == 202
+    assert exc.value.response["map_id"] == MAP_ID
+    assert len(calls) == 3
+
+
+def test_without_wait_a_map_being_drawn_raises_at_once():
+    client, calls = serve([joint("building", code=202)])
+    with pytest.raises(UnboxAIError, match="wait="):
+        client.umap(catalog_ids=IDS)
+    assert len(calls) == 1
+
+
+def test_a_failed_map_raises_its_reason(clock):
+    client, _ = serve([joint("building", code=202), joint("failed", "too few")])
+    with pytest.raises(UnboxAIError, match="failed: too few"):
+        client.umap(catalog_ids=IDS, wait=60)
+
+
+def test_catalog_errors_raise_as_they_stand():
+    client, _ = serve([httpx.Response(404, json={"detail": "unknown catalog: x"})])
+    with pytest.raises(UnboxAIError, match="unknown catalog") as exc:
+        client.umap(catalog_ids=["x"])
+    assert exc.value.status_code == 404
+
+
+def test_catalog_id_and_catalog_ids_are_exclusive():
+    client, calls = serve([joint("ready")])
+    with pytest.raises(ValueError, match="not both"):
+        client.umap(catalog_id="cat_a", catalog_ids=IDS)
+    with pytest.raises(TypeError):
+        client.umap(catalog_ids="cat_a")
+    assert calls == []
+
+
+def test_the_resource_reads_status_and_points():
+    client, _ = serve([joint("ready")])
+    status = client.catalogs.get_map(MAP_ID)
+    assert (status.status, status.catalog_ids) == ("ready", IDS)
+    assert status.requested_at is not None
+    assert client.catalogs.get_map_points(MAP_ID) == {"catalog": [0, 1]}
+
+
+def test_async_waits_while_the_map_is_drawn(monkeypatch):
+    async def no_sleep(_):
+        pass
+
+    monkeypatch.setattr("behaviorgpt._client.asyncio.sleep", no_sleep)
+    client, calls = serve(
+        [joint("building", code=202), joint("ready")], AsyncUnboxAIClient
+    )
+    page = asyncio.run(client.umap(catalog_ids=IDS, wait=60))
+    assert page == "<html>joint</html>"
+    assert len(calls) == 3
+
+    points = asyncio.run(client.catalogs.get_map_points(MAP_ID))
+    assert points == {"catalog": [0, 1]}