Review view: chat process display

pageindex/__init__.py

@@ -1,10 +1,11 @@
 """PageIndex SDK."""
 from typing import TYPE_CHECKING as _TYPE_CHECKING
 
+from .chat_stream import ChatStream
 from .client import PageIndexClient, PageIndexCloudClient, PageIndexLocalClient
 from .errors import PageIndexAPIError
-from .types import (ChatConfig, CloudIndexConfig, IndexConfig,
-                    LocalIndexConfig)
+from .types import (ChatConfig, ChatProcessOptions, CloudIndexConfig,
+                    IndexConfig, LocalIndexConfig)
 
 if _TYPE_CHECKING:
     from .flash import page_index_flash
@@ -16,6 +17,7 @@
     "PageIndexClient", "PageIndexCloudClient", "PageIndexLocalClient",
     "PageIndexAPIError",
     "IndexConfig", "CloudIndexConfig", "LocalIndexConfig", "ChatConfig",
+    "ChatProcessOptions", "ChatStream",
     "page_index", "page_index_main", "page_index_flash",
     "optimize_tree", "md_to_tree",
 ]
@@ -25,10 +27,10 @@
     "optimize_tree": ".tree_optimize",
     "md_to_tree": ".page_index_md",
 }
-_SUBMODULES = {"agent_tools", "client", "cloud_api", "errors", "flash",
-               "integrations", "local_api", "local_chat", "local_store",
-               "mcp_bridge", "page_index_classic", "page_index_md",
-               "tree_optimize", "types", "utils"}
+_SUBMODULES = {"agent_tools", "chat_stream", "client", "cloud_api", "errors",
+               "flash", "integrations", "local_api", "local_chat",
+               "local_store", "mcp_bridge", "page_index_classic",
+               "page_index_md", "tree_optimize", "types", "utils"}
 
 
 def __getattr__(name):

pageindex/chat_stream.py

@@ -0,0 +1,68 @@
+"""chat(stream=True)'s return type: one run, one view — text or events."""
+from __future__ import annotations
+
+from typing import Any, Iterator, Optional
+
+from .errors import PageIndexAPIError
+
+
+class ChatStream:
+    """chat(stream=True)'s stream: iterate it for the answer text pieces
+    (with show_process, the woven display); read ``.events`` instead for
+    the typed process event dicts. One underlying run — consume exactly
+    one view; call chat() again for the other."""
+
+    def __init__(self, text, events):
+        self._text = text      # () -> Iterator[str]
+        self._events = events  # () -> Iterator[dict], or the refusal text
+        self._view: Optional[str] = None
+        self._it: Any = None
+        self._closed = False
+
+    def _claim(self, view: str) -> None:
+        if self._view is not None and self._view != view:
+            raise PageIndexAPIError(
+                f"This chat stream is being consumed as {self._view}; one "
+                "run serves one view — call chat() again for the other.")
+        self._view = view
+
+    def __iter__(self) -> "ChatStream":
+        return self
+
+    def __next__(self) -> str:
+        self._claim("text")
+        if self._it is None:
+            if self._closed:
+                raise StopIteration
+            self._it = self._text()
+        return next(self._it)
+
+    @property
+    def events(self) -> Iterator[dict]:
+        """The run as typed event dicts: {"type": "thinking"|"answer",
+        "delta": ...}, {"type": "tool_call", "call_id", "name",
+        "arguments"}, {"type": "tool_result", "call_id", "name",
+        "output"} — full data, never clipped. Consuming — not merely
+        reading the attribute — claims the view, so debugger panes and
+        getattr probing stay side-effect free."""
+        def consume():
+            if isinstance(self._events, str):
+                raise PageIndexAPIError(self._events)
+            self._claim("events")
+            if self._it is None:
+                if self._closed:
+                    return
+                self._it = self._events()
+            # no `yield from`: a dropped handle must not close the run
+            for ev in self._it:
+                yield ev
+        return consume()
+
+    def close(self) -> None:
+        """Stop the run: closes the open view, and the stream is dead
+        afterwards, like a closed generator (own-model chat: a run never
+        consumed never starts)."""
+        self._closed = True
+        close = getattr(self._it, "close", None)
+        if close is not None:
+            close()

pageindex/client.py

@@ -6,8 +6,10 @@
 import threading
 import time
 import warnings
-from typing import Any, Callable, Iterator, Mapping, Optional, Union, cast
+from typing import (Any, Callable, Iterator, Literal, Mapping, Optional, Union,
+                    cast, overload)
 
+from .chat_stream import ChatStream
 from .errors import PageIndexAPIError
 
 
@@ -749,23 +751,62 @@ def get_retrieval(self, retrieval_id: str) -> dict[str, Any]:
 
     # ---------- CHAT ----------
 
+    # stream picks the return type: the docstring's `.events` usage must
+    # type-check for py.typed consumers
+    @overload
+    def chat(
+        self,
+        messages: Union[str, list[dict[str, str]]],
+        doc_id: Optional[Union[str, list[str]]] = None,
+        stream: Literal[False] = False,
+        model: Optional[str] = None,
+        reasoning_effort: Optional[str] = None,
+        show_process: Union[bool, Mapping[str, Any], None] = None,
+    ) -> str: ...
+
+    @overload
+    def chat(
+        self,
+        messages: Union[str, list[dict[str, str]]],
+        doc_id: Optional[Union[str, list[str]]] = None,
+        *,
+        stream: Literal[True],
+        model: Optional[str] = None,
+        reasoning_effort: Optional[str] = None,
+        show_process: Union[bool, Mapping[str, Any], None] = None,
+    ) -> ChatStream: ...
+
+    @overload
     def chat(
         self,
         messages: Union[str, list[dict[str, str]]],
         doc_id: Optional[Union[str, list[str]]] = None,
         stream: bool = False,
         model: Optional[str] = None,
         reasoning_effort: Optional[str] = None,
-    ) -> Union[str, Iterator[str]]:
+        show_process: Union[bool, Mapping[str, Any], None] = None,
+    ) -> Union[str, ChatStream]: ...
+
+    def chat(
+        self,
+        messages: Union[str, list[dict[str, str]]],
+        doc_id: Optional[Union[str, list[str]]] = None,
+        stream: bool = False,
+        model: Optional[str] = None,
+        reasoning_effort: Optional[str] = None,
+        show_process: Union[bool, Mapping[str, Any], None] = None,
+    ) -> Union[str, ChatStream]:
         """
         Ask a question about your documents, get the answer.
 
-        Thin sugar over ``chat_completions()`` in every mode — same
-        engine, same wire, minus the envelope. Multi-turn: keep your own
-        role/content list of the visible conversation (append each answer
-        as an assistant message) and pass it back. For usage accounting,
-        streaming metadata, or the tool-use process, use the protocol
-        surfaces: ``chat_completions()``, ``responses()``, ``messages()``.
+        Thin sugar over the same engine as ``chat_completions()`` in
+        every mode — same wire, minus the envelope. Multi-turn: keep your
+        own role/content list of the visible conversation (append each
+        answer as an assistant message; join a stream into one only with
+        ``show_process=False``) and pass it back. For usage
+        accounting, streaming metadata, or the full protocol transcripts,
+        use the protocol surfaces: ``chat_completions()``,
+        ``responses()``, ``messages()``.
 
         Args:
             messages: A question string, or role/content conversation
@@ -775,23 +816,81 @@ def chat(
                 documents: also enforced at the tool layer, not just
                 prompted. Cloud documents: the managed chat scopes
                 server-side; own-model chat targets at the prompt level.
-            stream: Yield the answer as text chunks as it is produced.
+            stream: Return a ``ChatStream``: iterate it for the answer
+                as text chunks as they are produced (``show_process``
+                is on by default, so the run's process arrives woven
+                in; ``show_process=False`` gives the bare answer), or
+                read its ``.events`` property instead for the run as
+                typed event dicts — thinking/answer deltas, each tool
+                call and its full result (own-model chat only; never
+                clipped). One run serves one view.
             model: Own-model chat only — backend model name (defaults
                 to ``chat_model``).
             reasoning_effort: Own-model chat only — how hard the model
                 thinks (``"low"`` / ``"medium"`` / ``"high"``; what a
                 backend accepts is its own). Unset sends nothing — the
                 model's default behavior applies.
+            show_process: Streamed chat — weave the run into
+                the text stream for display: thinking flows as
+                "[thinking] " sections, each tool call as a "[tool_call]
+                name arguments" line with its "[tool_result]" line, and
+                the answer unlabeled. **On by default**, weaving what the mode
+                serves: the in-process agent's full run; on a managed
+                client, the tool calls the endpoint streams (its wire
+                carries no thinking and no tool results). Pass ``False``
+                for the bare answer stream — do that before appending a
+                streamed answer to the conversation history.
+                ``True`` shows everything; a dict (typed as
+                ``pageindex.ChatProcessOptions``) selects the parts —
+                ``thinking`` / ``tool_call`` / ``tool_result``, bools
+                defaulting on — and sets ``max_chars``, the per-line
+                summary cap in characters (default 200). Omitted keys
+                keep their defaults, so ``{"thinking": False}`` hides
+                only thinking and ``{}`` equals ``True``.
+                Thinking appears when the backend streams it
+                (e.g. Claude models with ``reasoning_effort``; OpenAI
+                models expose none on the chat protocol). The labels are
+                not a parse format, and a process stream must not be
+                appended back as conversation history — for the
+                machine-readable process use ``.events``, ``responses()``
+                or ``messages()``.
 
         Returns:
             - stream=False: the answer string
-            - stream=True: iterator of text chunks
+            - stream=True: a ``ChatStream`` — iterating it yields text
+              chunks (with show_process, on by default, the run's process
+              woven in as labeled sections); ``.events`` yields typed
+              event dicts:
+              ``{"type": "thinking"|"answer", "delta": ...}``,
+              ``{"type": "tool_call", "call_id", "name", "arguments"}``,
+              ``{"type": "tool_result", "call_id", "name", "output"}``
         """
-        result = self.chat_completions(messages, stream=stream,
-                                       doc_id=doc_id, model=model,
-                                       reasoning_effort=reasoning_effort)
+        if show_process is not False and show_process is not None:
+            from .local_chat import _process_options
+            _process_options(show_process)  # a bad value chokes first
+            if not stream:
+                raise PageIndexAPIError(
+                    "show_process shows the run as it happens and requires "
+                    "stream=True; only show_process=False (or None) means "
+                    f"off — got {show_process!r}.")
         if stream:
-            return cast(Iterator[str], result)
+            # the default means "on where available"
+            resolved = True if show_process is None else show_process
+            if self._local_chat:
+                from .local_chat import run_chat_stream
+                return run_chat_stream(self, messages, doc_id=doc_id,
+                                       model=model,
+                                       reasoning_effort=reasoning_effort,
+                                       show_process=resolved)
+            from .local_chat import run_cloud_chat_stream
+            chunks = self.chat_completions(messages, stream=True,
+                                           stream_metadata=True,
+                                           doc_id=doc_id, model=model,
+                                           reasoning_effort=reasoning_effort)
+            return run_cloud_chat_stream(
+                cast(Iterator[dict[str, Any]], chunks), resolved)
+        result = self.chat_completions(messages, doc_id=doc_id, model=model,
+                                       reasoning_effort=reasoning_effort)
         envelope = cast(dict[str, Any], result)
         try:
             return envelope["choices"][0]["message"]["content"] or ""

pageindex/cloud_api.py

@@ -272,6 +272,10 @@ def _stream_chat_response(self, response: requests.Response) -> Iterator[str]:
 
                         try:
                             chunk = json.loads(data)
+                            if chunk.get("error"):
+                                raise PageIndexAPIError(
+                                    "Chat completion failed mid-stream: "
+                                    f"{chunk['error']}")
                             choices = chunk.get("choices") or [{}]
                             content = choices[0].get("delta", {}).get("content", "")
                             if content:
@@ -294,6 +298,10 @@ def _stream_chat_response_raw(self, response: requests.Response) -> Iterator[Dic
 
                         try:
                             chunk = json.loads(data)
+                            if chunk.get("error"):
+                                raise PageIndexAPIError(
+                                    "Chat completion failed mid-stream: "
+                                    f"{chunk['error']}")
                             yield chunk
                         except json.JSONDecodeError:
                             continue