feat: chat() front door and Claude cache marking on LiteLLM routes

pageindex/client.py

@@ -4,7 +4,7 @@
 import os
 import time
 import warnings
-from typing import Any, Callable, Iterator, Optional, Union
+from typing import Any, Callable, Iterator, Optional, Union, cast
 
 from .errors import PageIndexAPIError
 
@@ -345,7 +345,44 @@ def get_retrieval(self, retrieval_id: str) -> dict[str, Any]:
  "favor of chat completions; use chat_completions instead."
  ).get_retrieval(retrieval_id=retrieval_id)
 
- # ---------- CHAT COMPLETIONS ----------
+ # ---------- CHAT ----------
+
+ 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,
+ ) -> Union[str, Iterator[str]]:
+ """
+ Ask a question about your documents, get the answer.
+
+ Thin sugar over ``chat_completions()`` in both modes — 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()``.
+
+ Args:
+ messages: A question string, or role/content conversation
+ history.
+ doc_id: Document ID or list of IDs to scope the conversation.
+ Keep it identical across a conversation's calls.
+ stream: Yield the answer as text chunks as it is produced.
+ model: Local only — backend model name (defaults to
+ ``retrieve_model``).
+
+ Returns:
+ - stream=False: the answer string
+ - stream=True: iterator of text chunks
+ """
+ result = self.chat_completions(messages, stream=stream,
+ doc_id=doc_id, model=model)
+ if stream:
+ return cast(Iterator[str], result)
+ envelope = cast(dict[str, Any], result)
+ return envelope["choices"][0]["message"]["content"] or ""
 
  def chat_completions(
  self,
@@ -368,7 +405,9 @@ def chat_completions(
  backend, so any OpenAI-compatible server works; a ``/`` in the
  model name means LiteLLM provider routing, so prefix ``openai/``
  when the backend itself serves slashed ids, e.g.
- ``openai/Qwen/...`` on vLLM). The non-stream
+ ``openai/Qwen/...`` on vLLM; LiteLLM-routed Claude models —
+ Anthropic direct, Bedrock, Vertex — get the managed prompt
+ prefix cache-marked automatically). The non-stream
  response carries the final answer only; streaming yields the
  agent's visible text as it is produced, including narration before
  tool calls. ``finish_reason`` reports loop completion ("stop") —
@@ -380,7 +419,10 @@ def chat_completions(
  messages: Conversation messages with 'role' and 'content' keys,
  or a bare query string (it becomes a single user message).
  Local also accepts system/developer messages — their content
- is appended to the managed system prompt.
+ is appended to the managed system prompt. Local takes text
+ history only: tool-role turns are rejected (the cloud
+ endpoint forwards them verbatim), and message fields beyond
+ role/content are dropped.
  stream: Enable streaming responses.
  doc_id: Document ID or list of IDs to scope the conversation.
  Keep it identical across a conversation's calls — the

pageindex/local_chat.py

@@ -206,7 +206,8 @@ def _require_openai_agents(method: str) -> None:
  except ImportError as exc:
  raise PageIndexAPIError(
  f"{method} in local mode requires the OpenAI Agents SDK — "
- "pip install openai-agents (or pip install 'pageindex[openai]')."
+ "pip install openai-agents (or pip install 'pageindex[openai]'). "
+ "messages() runs on the anthropic extra instead."
  ) from exc
 
 
@@ -271,6 +272,29 @@ def _reported_model(model_name: str) -> str:
  return model_name.removeprefix("litellm/").removeprefix("openai/")
 
 
+def _cache_extra_args(model_name: str) -> Optional[dict]:
+ """Claude's prompt caching is opt-in per request: on Claude models
+ routed through LiteLLM (Anthropic direct, Bedrock, Vertex — each
+ channel live-verified), mark the managed system prefix via LiteLLM's
+ injection param so the loop's later turns and a conversation's next
+ calls read it instead of repaying full price. Provider resolution is
+ LiteLLM's own, so this predicate can never disagree with where the
+ request actually routes."""
+ if "/" not in model_name or model_name.startswith("openai/"):
+ return None
+ try:
+ from litellm import get_llm_provider
+ model, provider, _, _ = get_llm_provider(
+ model=model_name.removeprefix("litellm/"))
+ except Exception:
+ return None
+ if provider == "anthropic" or (provider in ("bedrock", "vertex_ai")
+ and "claude" in model.lower()):
+ return {"cache_control_injection_points": [
+ {"location": "message", "role": "system"}]}
+ return None
+
+
 def _openai_agent(client, protocol: str, model_name: str, instructions: str,
  temperature, top_p, doc_ids=None):
  from agents import Agent, ModelSettings
@@ -280,7 +304,8 @@ def _openai_agent(client, protocol: str, model_name: str, instructions: str,
  instructions=instructions,
  tools=build_openai_tools(client, doc_ids=doc_ids),
  model=_openai_model(protocol, model_name),
- model_settings=ModelSettings(temperature=temperature, top_p=top_p),
+ model_settings=ModelSettings(temperature=temperature, top_p=top_p,
+ extra_args=_cache_extra_args(model_name)),
  )
 
 

tests/test_local_chat.py

@@ -280,6 +280,127 @@ def test_cloud_guards():
  max_tokens=10)
 
 
+@needs_agents
+def test_anthropic_routed_models_mark_managed_prefix_for_cache(
+ client, store_path, fake_model):
+ fake_model([[_msg_item("ok")]])
+ from pageindex.local_chat import _openai_agent
+ marked = {"cache_control_injection_points": [
+ {"location": "message", "role": "system"}]}
+ for name in ("anthropic/claude-x", "litellm/anthropic/claude-x",
+ "bedrock/us.anthropic.claude-sonnet-5",
+ "vertex_ai/claude-sonnet-4-5"):
+ agent = _openai_agent(client, "chat", name, "sys", None, None)
+ assert agent.model_settings.extra_args == marked
+ for name in ("gpt-5", "openai/Qwen/x", "litellm/groq/x",
+ "bedrock/meta.llama3-70b-instruct-v1:0",
+ "vertex_ai/gemini-2.5-pro"):
+ agent = _openai_agent(client, "chat", name, "sys", None, None)
+ assert agent.model_settings.extra_args is None
+
+
+@needs_agents
+def test_status_recorder_attaches_to_the_real_responses_model(monkeypatch):
+ # Guards the private-attribute chain the recorder rides
+ # (agent.model._client.responses.create): a vendor rename turns the
+ # recorder into a silent no-op and truncation reports as completion.
+ monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
+ import openai
+ from agents.models.openai_responses import OpenAIResponsesModel
+ backend = openai.AsyncOpenAI()
+ model = OpenAIResponsesModel("gpt-test", openai_client=backend)
+ original = backend.responses.create
+ local_chat._record_response_status(types.SimpleNamespace(model=model), {})
+ assert backend.responses.create is not original
+ asyncio.run(backend.close())
+
+
+@needs_agents
+def test_cache_marker_reaches_the_anthropic_wire(client, store_path,
+ monkeypatch):
+ # End-to-end guard for the injection flag: through the real
+ # LitellmModel and litellm's request build, the marker must appear in
+ # the HTTP body — a regression in either vendor hop silently reverts
+ # anthropic-routed calls to full price.
+ pytest.importorskip("litellm")
+ from litellm.llms.custom_httpx.http_handler import (AsyncHTTPHandler,
+ HTTPHandler)
+ monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test")
+ captured = {}
+ reply = {"id": "msg_01", "type": "message", "role": "assistant",
+ "model": "claude-3-5-sonnet-20240620",
+ "content": [{"type": "text", "text": "ok"}],
+ "stop_reason": "end_turn", "stop_sequence": None,
+ "usage": {"input_tokens": 10, "output_tokens": 2}}
+
+ def _capture(url, kwargs):
+ body = kwargs.get("json")
+ if body is None and kwargs.get("data") is not None:
+ body = json.loads(kwargs["data"])
+ captured["url"] = str(url)
+ captured["body"] = body
+ return httpx.Response(200, json=reply,
+ request=httpx.Request("POST", str(url)))
+
+ async def fake_apost(self, url=None, *args, **kwargs):
+ return _capture(url, kwargs)
+
+ def fake_post(self, url=None, *args, **kwargs):
+ return _capture(url, kwargs)
+
+ monkeypatch.setattr(AsyncHTTPHandler, "post", fake_apost)
+ monkeypatch.setattr(HTTPHandler, "post", fake_post)
+ result = client.chat_completions(
+ "hi", model="anthropic/claude-3-5-sonnet-20240620")
+ assert "/v1/messages" in captured["url"]
+ assert '"cache_control"' in json.dumps(captured["body"])
+ assert result["choices"][0]["message"]["content"] == "ok"
+
+
+# ── chat (front door) ──
+
+@needs_agents
+def test_chat_returns_answer_string(client, store_path, fake_model):
+ doc_id = seed_doc(store_path, "pi-a", "report.pdf")
+ fake = fake_model([
+ [_call_item("get_document", {"doc_name": "report.pdf"})],
+ [_msg_item("The answer")],
+ ])
+ assert client.chat("What status?", doc_id=doc_id) == "The answer"
+ first_item = fake.inputs[0][0]
+ assert "The user has specified document: report.pdf" in first_item["content"]
+
+
+@needs_agents
+def test_chat_stream_yields_text_chunks(client, store_path, fake_model):
+ fake_model([[_msg_item("The answer")]])
+ assert list(client.chat("q", stream=True)) == ["The ", "answer"]
+
+
+@needs_agents
+def test_chat_multi_turn_history(client, store_path, fake_model):
+ fake = fake_model([[_msg_item("Chapter 4 covers pears")]])
+ history = [
+ {"role": "user", "content": "What about chapter 3?"},
+ {"role": "assistant", "content": "Chapter 3 covers apples"},
+ {"role": "user", "content": "And chapter 4?"},
+ ]
+ assert client.chat(history) == "Chapter 4 covers pears"
+ assert fake.inputs[0][-3:] == history
+
+
+def test_chat_cloud_unwraps_envelope(monkeypatch):
+ cloud = PageIndexCloudClient(api_key="pi-test-key")
+
+ def fake_cc(**kwargs):
+ assert kwargs["messages"] == [{"role": "user", "content": "q"}]
+ return {"choices": [{"message": {"role": "assistant",
+ "content": "cloud answer"}}]}
+
+ monkeypatch.setattr(cloud._api, "chat_completions", fake_cc)
+ assert cloud.chat("q") == "cloud answer"
+
+
 # ── responses ──
 
 @needs_agents

pageindex/client.py

@@ -4,7 +4,7 @@
 import os
 import time
 import warnings
-from typing import Any, Callable, Iterator, Optional, Union
+from typing import Any, Callable, Iterator, Optional, Union, cast
 
 from .errors import PageIndexAPIError
 
@@ -345,7 +345,44 @@ def get_retrieval(self, retrieval_id: str) -> dict[str, Any]:
  "favor of chat completions; use chat_completions instead."
  ).get_retrieval(retrieval_id=retrieval_id)
 
- # ---------- CHAT COMPLETIONS ----------
+ # ---------- CHAT ----------
+
+ 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,
+ ) -> Union[str, Iterator[str]]:
+ """
+ Ask a question about your documents, get the answer.
+
+ Thin sugar over ``chat_completions()`` in both modes — 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()``.
+
+ Args:
+ messages: A question string, or role/content conversation
+ history.
+ doc_id: Document ID or list of IDs to scope the conversation.
+ Keep it identical across a conversation's calls.
+ stream: Yield the answer as text chunks as it is produced.
+ model: Local only — backend model name (defaults to
+ ``retrieve_model``).
+
+ Returns:
+ - stream=False: the answer string
+ - stream=True: iterator of text chunks
+ """
+ result = self.chat_completions(messages, stream=stream,
+ doc_id=doc_id, model=model)
+ if stream:
+ return cast(Iterator[str], result)
+ envelope = cast(dict[str, Any], result)
+ return envelope["choices"][0]["message"]["content"] or ""
 
  def chat_completions(
  self,
@@ -368,7 +405,9 @@ def chat_completions(
  backend, so any OpenAI-compatible server works; a ``/`` in the
  model name means LiteLLM provider routing, so prefix ``openai/``
  when the backend itself serves slashed ids, e.g.
- ``openai/Qwen/...`` on vLLM). The non-stream
+ ``openai/Qwen/...`` on vLLM; LiteLLM-routed Claude models —
+ Anthropic direct, Bedrock, Vertex — get the managed prompt
+ prefix cache-marked automatically). The non-stream
  response carries the final answer only; streaming yields the
  agent's visible text as it is produced, including narration before
  tool calls. ``finish_reason`` reports loop completion ("stop") —
@@ -380,7 +419,10 @@ def chat_completions(
  messages: Conversation messages with 'role' and 'content' keys,
  or a bare query string (it becomes a single user message).
  Local also accepts system/developer messages — their content
- is appended to the managed system prompt.
+ is appended to the managed system prompt. Local takes text
+ history only: tool-role turns are rejected (the cloud
+ endpoint forwards them verbatim), and message fields beyond
+ role/content are dropped.
  stream: Enable streaming responses.
  doc_id: Document ID or list of IDs to scope the conversation.
  Keep it identical across a conversation's calls — the

pageindex/local_chat.py

@@ -206,7 +206,8 @@ def _require_openai_agents(method: str) -> None:
  except ImportError as exc:
  raise PageIndexAPIError(
  f"{method} in local mode requires the OpenAI Agents SDK — "
- "pip install openai-agents (or pip install 'pageindex[openai]')."
+ "pip install openai-agents (or pip install 'pageindex[openai]'). "
+ "messages() runs on the anthropic extra instead."
  ) from exc
 
 
@@ -271,6 +272,29 @@ def _reported_model(model_name: str) -> str:
  return model_name.removeprefix("litellm/").removeprefix("openai/")
 
 
+def _cache_extra_args(model_name: str) -> Optional[dict]:
+ """Claude's prompt caching is opt-in per request: on Claude models
+ routed through LiteLLM (Anthropic direct, Bedrock, Vertex — each
+ channel live-verified), mark the managed system prefix via LiteLLM's
+ injection param so the loop's later turns and a conversation's next
+ calls read it instead of repaying full price. Provider resolution is
+ LiteLLM's own, so this predicate can never disagree with where the
+ request actually routes."""
+ if "/" not in model_name or model_name.startswith("openai/"):
+ return None
+ try:
+ from litellm import get_llm_provider
+ model, provider, _, _ = get_llm_provider(
+ model=model_name.removeprefix("litellm/"))
+ except Exception:
+ return None
+ if provider == "anthropic" or (provider in ("bedrock", "vertex_ai")
+ and "claude" in model.lower()):
+ return {"cache_control_injection_points": [
+ {"location": "message", "role": "system"}]}
+ return None
+
+
 def _openai_agent(client, protocol: str, model_name: str, instructions: str,
  temperature, top_p, doc_ids=None):
  from agents import Agent, ModelSettings
@@ -280,7 +304,8 @@ def _openai_agent(client, protocol: str, model_name: str, instructions: str,
  instructions=instructions,
  tools=build_openai_tools(client, doc_ids=doc_ids),
  model=_openai_model(protocol, model_name),
- model_settings=ModelSettings(temperature=temperature, top_p=top_p),
+ model_settings=ModelSettings(temperature=temperature, top_p=top_p,
+ extra_args=_cache_extra_args(model_name)),
  )
 
 

tests/test_local_chat.py

@@ -280,6 +280,127 @@ def test_cloud_guards():
  max_tokens=10)
 
 
+@needs_agents
+def test_anthropic_routed_models_mark_managed_prefix_for_cache(
+ client, store_path, fake_model):
+ fake_model([[_msg_item("ok")]])
+ from pageindex.local_chat import _openai_agent
+ marked = {"cache_control_injection_points": [
+ {"location": "message", "role": "system"}]}
+ for name in ("anthropic/claude-x", "litellm/anthropic/claude-x",
+ "bedrock/us.anthropic.claude-sonnet-5",
+ "vertex_ai/claude-sonnet-4-5"):
+ agent = _openai_agent(client, "chat", name, "sys", None, None)
+ assert agent.model_settings.extra_args == marked
+ for name in ("gpt-5", "openai/Qwen/x", "litellm/groq/x",
+ "bedrock/meta.llama3-70b-instruct-v1:0",
+ "vertex_ai/gemini-2.5-pro"):
+ agent = _openai_agent(client, "chat", name, "sys", None, None)
+ assert agent.model_settings.extra_args is None
+
+
+@needs_agents
+def test_status_recorder_attaches_to_the_real_responses_model(monkeypatch):
+ # Guards the private-attribute chain the recorder rides
+ # (agent.model._client.responses.create): a vendor rename turns the
+ # recorder into a silent no-op and truncation reports as completion.
+ monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
+ import openai
+ from agents.models.openai_responses import OpenAIResponsesModel
+ backend = openai.AsyncOpenAI()
+ model = OpenAIResponsesModel("gpt-test", openai_client=backend)
+ original = backend.responses.create
+ local_chat._record_response_status(types.SimpleNamespace(model=model), {})
+ assert backend.responses.create is not original
+ asyncio.run(backend.close())
+
+
+@needs_agents
+def test_cache_marker_reaches_the_anthropic_wire(client, store_path,
+ monkeypatch):
+ # End-to-end guard for the injection flag: through the real
+ # LitellmModel and litellm's request build, the marker must appear in
+ # the HTTP body — a regression in either vendor hop silently reverts
+ # anthropic-routed calls to full price.
+ pytest.importorskip("litellm")
+ from litellm.llms.custom_httpx.http_handler import (AsyncHTTPHandler,
+ HTTPHandler)
+ monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test")
+ captured = {}
+ reply = {"id": "msg_01", "type": "message", "role": "assistant",
+ "model": "claude-3-5-sonnet-20240620",
+ "content": [{"type": "text", "text": "ok"}],
+ "stop_reason": "end_turn", "stop_sequence": None,
+ "usage": {"input_tokens": 10, "output_tokens": 2}}
+
+ def _capture(url, kwargs):
+ body = kwargs.get("json")
+ if body is None and kwargs.get("data") is not None:
+ body = json.loads(kwargs["data"])
+ captured["url"] = str(url)
+ captured["body"] = body
+ return httpx.Response(200, json=reply,
+ request=httpx.Request("POST", str(url)))
+
+ async def fake_apost(self, url=None, *args, **kwargs):
+ return _capture(url, kwargs)
+
+ def fake_post(self, url=None, *args, **kwargs):
+ return _capture(url, kwargs)
+
+ monkeypatch.setattr(AsyncHTTPHandler, "post", fake_apost)
+ monkeypatch.setattr(HTTPHandler, "post", fake_post)
+ result = client.chat_completions(
+ "hi", model="anthropic/claude-3-5-sonnet-20240620")
+ assert "/v1/messages" in captured["url"]
+ assert '"cache_control"' in json.dumps(captured["body"])
+ assert result["choices"][0]["message"]["content"] == "ok"
+
+
+# ── chat (front door) ──
+
+@needs_agents
+def test_chat_returns_answer_string(client, store_path, fake_model):
+ doc_id = seed_doc(store_path, "pi-a", "report.pdf")
+ fake = fake_model([
+ [_call_item("get_document", {"doc_name": "report.pdf"})],
+ [_msg_item("The answer")],
+ ])
+ assert client.chat("What status?", doc_id=doc_id) == "The answer"
+ first_item = fake.inputs[0][0]
+ assert "The user has specified document: report.pdf" in first_item["content"]
+
+
+@needs_agents
+def test_chat_stream_yields_text_chunks(client, store_path, fake_model):
+ fake_model([[_msg_item("The answer")]])
+ assert list(client.chat("q", stream=True)) == ["The ", "answer"]
+
+
+@needs_agents
+def test_chat_multi_turn_history(client, store_path, fake_model):
+ fake = fake_model([[_msg_item("Chapter 4 covers pears")]])
+ history = [
+ {"role": "user", "content": "What about chapter 3?"},
+ {"role": "assistant", "content": "Chapter 3 covers apples"},
+ {"role": "user", "content": "And chapter 4?"},
+ ]
+ assert client.chat(history) == "Chapter 4 covers pears"
+ assert fake.inputs[0][-3:] == history
+
+
+def test_chat_cloud_unwraps_envelope(monkeypatch):
+ cloud = PageIndexCloudClient(api_key="pi-test-key")
+
+ def fake_cc(**kwargs):
+ assert kwargs["messages"] == [{"role": "user", "content": "q"}]
+ return {"choices": [{"message": {"role": "assistant",
+ "content": "cloud answer"}}]}
+
+ monkeypatch.setattr(cloud._api, "chat_completions", fake_cc)
+ assert cloud.chat("q") == "cloud answer"
+
+
 # ── responses ──
 
 @needs_agents

pageindex/client.py

@@ -4,7 +4,7 @@
 import os
 import time
 import warnings
-from typing import Any, Callable, Iterator, Optional, Union
+from typing import Any, Callable, Iterator, Optional, Union, cast
 
 from .errors import PageIndexAPIError
 
@@ -345,7 +345,44 @@ def get_retrieval(self, retrieval_id: str) -> dict[str, Any]:
  "favor of chat completions; use chat_completions instead."
  ).get_retrieval(retrieval_id=retrieval_id)
 
- # ---------- CHAT COMPLETIONS ----------
+ # ---------- CHAT ----------
+
+ 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,
+ ) -> Union[str, Iterator[str]]:
+ """
+ Ask a question about your documents, get the answer.
+
+ Thin sugar over ``chat_completions()`` in both modes — 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()``.
+
+ Args:
+ messages: A question string, or role/content conversation
+ history.
+ doc_id: Document ID or list of IDs to scope the conversation.
+ Keep it identical across a conversation's calls.
+ stream: Yield the answer as text chunks as it is produced.
+ model: Local only — backend model name (defaults to
+ ``retrieve_model``).
+
+ Returns:
+ - stream=False: the answer string
+ - stream=True: iterator of text chunks
+ """
+ result = self.chat_completions(messages, stream=stream,
+ doc_id=doc_id, model=model)
+ if stream:
+ return cast(Iterator[str], result)
+ envelope = cast(dict[str, Any], result)
+ return envelope["choices"][0]["message"]["content"] or ""
 
  def chat_completions(
  self,
@@ -368,7 +405,9 @@ def chat_completions(
  backend, so any OpenAI-compatible server works; a ``/`` in the
  model name means LiteLLM provider routing, so prefix ``openai/``
  when the backend itself serves slashed ids, e.g.
- ``openai/Qwen/...`` on vLLM). The non-stream
+ ``openai/Qwen/...`` on vLLM; LiteLLM-routed Claude models —
+ Anthropic direct, Bedrock, Vertex — get the managed prompt
+ prefix cache-marked automatically). The non-stream
  response carries the final answer only; streaming yields the
  agent's visible text as it is produced, including narration before
  tool calls. ``finish_reason`` reports loop completion ("stop") —
@@ -380,7 +419,10 @@ def chat_completions(
  messages: Conversation messages with 'role' and 'content' keys,
  or a bare query string (it becomes a single user message).
  Local also accepts system/developer messages — their content
- is appended to the managed system prompt.
+ is appended to the managed system prompt. Local takes text
+ history only: tool-role turns are rejected (the cloud
+ endpoint forwards them verbatim), and message fields beyond
+ role/content are dropped.
  stream: Enable streaming responses.
  doc_id: Document ID or list of IDs to scope the conversation.
  Keep it identical across a conversation's calls — the

pageindex/local_chat.py

@@ -206,7 +206,8 @@ def _require_openai_agents(method: str) -> None:
  except ImportError as exc:
  raise PageIndexAPIError(
  f"{method} in local mode requires the OpenAI Agents SDK — "
- "pip install openai-agents (or pip install 'pageindex[openai]')."
+ "pip install openai-agents (or pip install 'pageindex[openai]'). "
+ "messages() runs on the anthropic extra instead."
  ) from exc
 
 
@@ -271,6 +272,29 @@ def _reported_model(model_name: str) -> str:
  return model_name.removeprefix("litellm/").removeprefix("openai/")
 
 
+def _cache_extra_args(model_name: str) -> Optional[dict]:
+ """Claude's prompt caching is opt-in per request: on Claude models
+ routed through LiteLLM (Anthropic direct, Bedrock, Vertex — each
+ channel live-verified), mark the managed system prefix via LiteLLM's
+ injection param so the loop's later turns and a conversation's next
+ calls read it instead of repaying full price. Provider resolution is
+ LiteLLM's own, so this predicate can never disagree with where the
+ request actually routes."""
+ if "/" not in model_name or model_name.startswith("openai/"):
+ return None
+ try:
+ from litellm import get_llm_provider
+ model, provider, _, _ = get_llm_provider(
+ model=model_name.removeprefix("litellm/"))
+ except Exception:
+ return None
+ if provider == "anthropic" or (provider in ("bedrock", "vertex_ai")
+ and "claude" in model.lower()):
+ return {"cache_control_injection_points": [
+ {"location": "message", "role": "system"}]}
+ return None
+
+
 def _openai_agent(client, protocol: str, model_name: str, instructions: str,
  temperature, top_p, doc_ids=None):
  from agents import Agent, ModelSettings
@@ -280,7 +304,8 @@ def _openai_agent(client, protocol: str, model_name: str, instructions: str,
  instructions=instructions,
  tools=build_openai_tools(client, doc_ids=doc_ids),
  model=_openai_model(protocol, model_name),
- model_settings=ModelSettings(temperature=temperature, top_p=top_p),
+ model_settings=ModelSettings(temperature=temperature, top_p=top_p,
+ extra_args=_cache_extra_args(model_name)),
  )
 
 

tests/test_local_chat.py

@@ -280,6 +280,127 @@ def test_cloud_guards():
  max_tokens=10)
 
 
+@needs_agents
+def test_anthropic_routed_models_mark_managed_prefix_for_cache(
+ client, store_path, fake_model):
+ fake_model([[_msg_item("ok")]])
+ from pageindex.local_chat import _openai_agent
+ marked = {"cache_control_injection_points": [
+ {"location": "message", "role": "system"}]}
+ for name in ("anthropic/claude-x", "litellm/anthropic/claude-x",
+ "bedrock/us.anthropic.claude-sonnet-5",
+ "vertex_ai/claude-sonnet-4-5"):
+ agent = _openai_agent(client, "chat", name, "sys", None, None)
+ assert agent.model_settings.extra_args == marked
+ for name in ("gpt-5", "openai/Qwen/x", "litellm/groq/x",
+ "bedrock/meta.llama3-70b-instruct-v1:0",
+ "vertex_ai/gemini-2.5-pro"):
+ agent = _openai_agent(client, "chat", name, "sys", None, None)
+ assert agent.model_settings.extra_args is None
+
+
+@needs_agents
+def test_status_recorder_attaches_to_the_real_responses_model(monkeypatch):
+ # Guards the private-attribute chain the recorder rides
+ # (agent.model._client.responses.create): a vendor rename turns the
+ # recorder into a silent no-op and truncation reports as completion.
+ monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
+ import openai
+ from agents.models.openai_responses import OpenAIResponsesModel
+ backend = openai.AsyncOpenAI()
+ model = OpenAIResponsesModel("gpt-test", openai_client=backend)
+ original = backend.responses.create
+ local_chat._record_response_status(types.SimpleNamespace(model=model), {})
+ assert backend.responses.create is not original
+ asyncio.run(backend.close())
+
+
+@needs_agents
+def test_cache_marker_reaches_the_anthropic_wire(client, store_path,
+ monkeypatch):
+ # End-to-end guard for the injection flag: through the real
+ # LitellmModel and litellm's request build, the marker must appear in
+ # the HTTP body — a regression in either vendor hop silently reverts
+ # anthropic-routed calls to full price.
+ pytest.importorskip("litellm")
+ from litellm.llms.custom_httpx.http_handler import (AsyncHTTPHandler,
+ HTTPHandler)
+ monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test")
+ captured = {}
+ reply = {"id": "msg_01", "type": "message", "role": "assistant",
+ "model": "claude-3-5-sonnet-20240620",
+ "content": [{"type": "text", "text": "ok"}],
+ "stop_reason": "end_turn", "stop_sequence": None,
+ "usage": {"input_tokens": 10, "output_tokens": 2}}
+
+ def _capture(url, kwargs):
+ body = kwargs.get("json")
+ if body is None and kwargs.get("data") is not None:
+ body = json.loads(kwargs["data"])
+ captured["url"] = str(url)
+ captured["body"] = body
+ return httpx.Response(200, json=reply,
+ request=httpx.Request("POST", str(url)))
+
+ async def fake_apost(self, url=None, *args, **kwargs):
+ return _capture(url, kwargs)
+
+ def fake_post(self, url=None, *args, **kwargs):
+ return _capture(url, kwargs)
+
+ monkeypatch.setattr(AsyncHTTPHandler, "post", fake_apost)
+ monkeypatch.setattr(HTTPHandler, "post", fake_post)
+ result = client.chat_completions(
+ "hi", model="anthropic/claude-3-5-sonnet-20240620")
+ assert "/v1/messages" in captured["url"]
+ assert '"cache_control"' in json.dumps(captured["body"])
+ assert result["choices"][0]["message"]["content"] == "ok"
+
+
+# ── chat (front door) ──
+
+@needs_agents
+def test_chat_returns_answer_string(client, store_path, fake_model):
+ doc_id = seed_doc(store_path, "pi-a", "report.pdf")
+ fake = fake_model([
+ [_call_item("get_document", {"doc_name": "report.pdf"})],
+ [_msg_item("The answer")],
+ ])
+ assert client.chat("What status?", doc_id=doc_id) == "The answer"
+ first_item = fake.inputs[0][0]
+ assert "The user has specified document: report.pdf" in first_item["content"]
+
+
+@needs_agents
+def test_chat_stream_yields_text_chunks(client, store_path, fake_model):
+ fake_model([[_msg_item("The answer")]])
+ assert list(client.chat("q", stream=True)) == ["The ", "answer"]
+
+
+@needs_agents
+def test_chat_multi_turn_history(client, store_path, fake_model):
+ fake = fake_model([[_msg_item("Chapter 4 covers pears")]])
+ history = [
+ {"role": "user", "content": "What about chapter 3?"},
+ {"role": "assistant", "content": "Chapter 3 covers apples"},
+ {"role": "user", "content": "And chapter 4?"},
+ ]
+ assert client.chat(history) == "Chapter 4 covers pears"
+ assert fake.inputs[0][-3:] == history
+
+
+def test_chat_cloud_unwraps_envelope(monkeypatch):
+ cloud = PageIndexCloudClient(api_key="pi-test-key")
+
+ def fake_cc(**kwargs):
+ assert kwargs["messages"] == [{"role": "user", "content": "q"}]
+ return {"choices": [{"message": {"role": "assistant",
+ "content": "cloud answer"}}]}
+
+ monkeypatch.setattr(cloud._api, "chat_completions", fake_cc)
+ assert cloud.chat("q") == "cloud answer"
+
+
 # ── responses ──
 
 @needs_agents

pageindex/client.py

@@ -4,7 +4,7 @@
 import os
 import time
 import warnings
-from typing import Any, Callable, Iterator, Optional, Union
+from typing import Any, Callable, Iterator, Optional, Union, cast
 
 from .errors import PageIndexAPIError
 
@@ -345,7 +345,44 @@ def get_retrieval(self, retrieval_id: str) -> dict[str, Any]:
  "favor of chat completions; use chat_completions instead."
  ).get_retrieval(retrieval_id=retrieval_id)
 
- # ---------- CHAT COMPLETIONS ----------
+ # ---------- CHAT ----------
+
+ 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,
+ ) -> Union[str, Iterator[str]]:
+ """
+ Ask a question about your documents, get the answer.
+
+ Thin sugar over ``chat_completions()`` in both modes — 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()``.
+
+ Args:
+ messages: A question string, or role/content conversation
+ history.
+ doc_id: Document ID or list of IDs to scope the conversation.
+ Keep it identical across a conversation's calls.
+ stream: Yield the answer as text chunks as it is produced.
+ model: Local only — backend model name (defaults to
+ ``retrieve_model``).
+
+ Returns:
+ - stream=False: the answer string
+ - stream=True: iterator of text chunks
+ """
+ result = self.chat_completions(messages, stream=stream,
+ doc_id=doc_id, model=model)
+ if stream:
+ return cast(Iterator[str], result)
+ envelope = cast(dict[str, Any], result)
+ return envelope["choices"][0]["message"]["content"] or ""
 
  def chat_completions(
  self,
@@ -368,7 +405,9 @@ def chat_completions(
  backend, so any OpenAI-compatible server works; a ``/`` in the
  model name means LiteLLM provider routing, so prefix ``openai/``
  when the backend itself serves slashed ids, e.g.
- ``openai/Qwen/...`` on vLLM). The non-stream
+ ``openai/Qwen/...`` on vLLM; LiteLLM-routed Claude models —
+ Anthropic direct, Bedrock, Vertex — get the managed prompt
+ prefix cache-marked automatically). The non-stream
  response carries the final answer only; streaming yields the
  agent's visible text as it is produced, including narration before
  tool calls. ``finish_reason`` reports loop completion ("stop") —
@@ -380,7 +419,10 @@ def chat_completions(
  messages: Conversation messages with 'role' and 'content' keys,
  or a bare query string (it becomes a single user message).
  Local also accepts system/developer messages — their content
- is appended to the managed system prompt.
+ is appended to the managed system prompt. Local takes text
+ history only: tool-role turns are rejected (the cloud
+ endpoint forwards them verbatim), and message fields beyond
+ role/content are dropped.
  stream: Enable streaming responses.
  doc_id: Document ID or list of IDs to scope the conversation.
  Keep it identical across a conversation's calls — the

pageindex/local_chat.py

@@ -206,7 +206,8 @@ def _require_openai_agents(method: str) -> None:
  except ImportError as exc:
  raise PageIndexAPIError(
  f"{method} in local mode requires the OpenAI Agents SDK — "
- "pip install openai-agents (or pip install 'pageindex[openai]')."
+ "pip install openai-agents (or pip install 'pageindex[openai]'). "
+ "messages() runs on the anthropic extra instead."
  ) from exc
 
 
@@ -271,6 +272,29 @@ def _reported_model(model_name: str) -> str:
  return model_name.removeprefix("litellm/").removeprefix("openai/")
 
 
+def _cache_extra_args(model_name: str) -> Optional[dict]:
+ """Claude's prompt caching is opt-in per request: on Claude models
+ routed through LiteLLM (Anthropic direct, Bedrock, Vertex — each
+ channel live-verified), mark the managed system prefix via LiteLLM's
+ injection param so the loop's later turns and a conversation's next
+ calls read it instead of repaying full price. Provider resolution is
+ LiteLLM's own, so this predicate can never disagree with where the
+ request actually routes."""
+ if "/" not in model_name or model_name.startswith("openai/"):
+ return None
+ try:
+ from litellm import get_llm_provider
+ model, provider, _, _ = get_llm_provider(
+ model=model_name.removeprefix("litellm/"))
+ except Exception:
+ return None
+ if provider == "anthropic" or (provider in ("bedrock", "vertex_ai")
+ and "claude" in model.lower()):
+ return {"cache_control_injection_points": [
+ {"location": "message", "role": "system"}]}
+ return None
+
+
 def _openai_agent(client, protocol: str, model_name: str, instructions: str,
  temperature, top_p, doc_ids=None):
  from agents import Agent, ModelSettings
@@ -280,7 +304,8 @@ def _openai_agent(client, protocol: str, model_name: str, instructions: str,
  instructions=instructions,
  tools=build_openai_tools(client, doc_ids=doc_ids),
  model=_openai_model(protocol, model_name),
- model_settings=ModelSettings(temperature=temperature, top_p=top_p),
+ model_settings=ModelSettings(temperature=temperature, top_p=top_p,
+ extra_args=_cache_extra_args(model_name)),
  )
 
 

tests/test_local_chat.py

@@ -280,6 +280,127 @@ def test_cloud_guards():
  max_tokens=10)
 
 
+@needs_agents
+def test_anthropic_routed_models_mark_managed_prefix_for_cache(
+ client, store_path, fake_model):
+ fake_model([[_msg_item("ok")]])
+ from pageindex.local_chat import _openai_agent
+ marked = {"cache_control_injection_points": [
+ {"location": "message", "role": "system"}]}
+ for name in ("anthropic/claude-x", "litellm/anthropic/claude-x",
+ "bedrock/us.anthropic.claude-sonnet-5",
+ "vertex_ai/claude-sonnet-4-5"):
+ agent = _openai_agent(client, "chat", name, "sys", None, None)
+ assert agent.model_settings.extra_args == marked
+ for name in ("gpt-5", "openai/Qwen/x", "litellm/groq/x",
+ "bedrock/meta.llama3-70b-instruct-v1:0",
+ "vertex_ai/gemini-2.5-pro"):
+ agent = _openai_agent(client, "chat", name, "sys", None, None)
+ assert agent.model_settings.extra_args is None
+
+
+@needs_agents
+def test_status_recorder_attaches_to_the_real_responses_model(monkeypatch):
+ # Guards the private-attribute chain the recorder rides
+ # (agent.model._client.responses.create): a vendor rename turns the
+ # recorder into a silent no-op and truncation reports as completion.
+ monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
+ import openai
+ from agents.models.openai_responses import OpenAIResponsesModel
+ backend = openai.AsyncOpenAI()
+ model = OpenAIResponsesModel("gpt-test", openai_client=backend)
+ original = backend.responses.create
+ local_chat._record_response_status(types.SimpleNamespace(model=model), {})
+ assert backend.responses.create is not original
+ asyncio.run(backend.close())
+
+
+@needs_agents
+def test_cache_marker_reaches_the_anthropic_wire(client, store_path,
+ monkeypatch):
+ # End-to-end guard for the injection flag: through the real
+ # LitellmModel and litellm's request build, the marker must appear in
+ # the HTTP body — a regression in either vendor hop silently reverts
+ # anthropic-routed calls to full price.
+ pytest.importorskip("litellm")
+ from litellm.llms.custom_httpx.http_handler import (AsyncHTTPHandler,
+ HTTPHandler)
+ monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test")
+ captured = {}
+ reply = {"id": "msg_01", "type": "message", "role": "assistant",
+ "model": "claude-3-5-sonnet-20240620",
+ "content": [{"type": "text", "text": "ok"}],
+ "stop_reason": "end_turn", "stop_sequence": None,
+ "usage": {"input_tokens": 10, "output_tokens": 2}}
+
+ def _capture(url, kwargs):
+ body = kwargs.get("json")
+ if body is None and kwargs.get("data") is not None:
+ body = json.loads(kwargs["data"])
+ captured["url"] = str(url)
+ captured["body"] = body
+ return httpx.Response(200, json=reply,
+ request=httpx.Request("POST", str(url)))
+
+ async def fake_apost(self, url=None, *args, **kwargs):
+ return _capture(url, kwargs)
+
+ def fake_post(self, url=None, *args, **kwargs):
+ return _capture(url, kwargs)
+
+ monkeypatch.setattr(AsyncHTTPHandler, "post", fake_apost)
+ monkeypatch.setattr(HTTPHandler, "post", fake_post)
+ result = client.chat_completions(
+ "hi", model="anthropic/claude-3-5-sonnet-20240620")
+ assert "/v1/messages" in captured["url"]
+ assert '"cache_control"' in json.dumps(captured["body"])
+ assert result["choices"][0]["message"]["content"] == "ok"
+
+
+# ── chat (front door) ──
+
+@needs_agents
+def test_chat_returns_answer_string(client, store_path, fake_model):
+ doc_id = seed_doc(store_path, "pi-a", "report.pdf")
+ fake = fake_model([
+ [_call_item("get_document", {"doc_name": "report.pdf"})],
+ [_msg_item("The answer")],
+ ])
+ assert client.chat("What status?", doc_id=doc_id) == "The answer"
+ first_item = fake.inputs[0][0]
+ assert "The user has specified document: report.pdf" in first_item["content"]
+
+
+@needs_agents
+def test_chat_stream_yields_text_chunks(client, store_path, fake_model):
+ fake_model([[_msg_item("The answer")]])
+ assert list(client.chat("q", stream=True)) == ["The ", "answer"]
+
+
+@needs_agents
+def test_chat_multi_turn_history(client, store_path, fake_model):
+ fake = fake_model([[_msg_item("Chapter 4 covers pears")]])
+ history = [
+ {"role": "user", "content": "What about chapter 3?"},
+ {"role": "assistant", "content": "Chapter 3 covers apples"},
+ {"role": "user", "content": "And chapter 4?"},
+ ]
+ assert client.chat(history) == "Chapter 4 covers pears"
+ assert fake.inputs[0][-3:] == history
+
+
+def test_chat_cloud_unwraps_envelope(monkeypatch):
+ cloud = PageIndexCloudClient(api_key="pi-test-key")
+
+ def fake_cc(**kwargs):
+ assert kwargs["messages"] == [{"role": "user", "content": "q"}]
+ return {"choices": [{"message": {"role": "assistant",
+ "content": "cloud answer"}}]}
+
+ monkeypatch.setattr(cloud._api, "chat_completions", fake_cc)
+ assert cloud.chat("q") == "cloud answer"
+
+
 # ── responses ──
 
 @needs_agents

pageindex/client.py

@@ -4,7 +4,7 @@
 import os
 import time
 import warnings
-from typing import Any, Callable, Iterator, Optional, Union
+from typing import Any, Callable, Iterator, Optional, Union, cast
 
 from .errors import PageIndexAPIError
 
@@ -345,7 +345,44 @@ def get_retrieval(self, retrieval_id: str) -> dict[str, Any]:
  "favor of chat completions; use chat_completions instead."
  ).get_retrieval(retrieval_id=retrieval_id)
 
- # ---------- CHAT COMPLETIONS ----------
+ # ---------- CHAT ----------
+
+ 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,
+ ) -> Union[str, Iterator[str]]:
+ """
+ Ask a question about your documents, get the answer.
+
+ Thin sugar over ``chat_completions()`` in both modes — 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()``.
+
+ Args:
+ messages: A question string, or role/content conversation
+ history.
+ doc_id: Document ID or list of IDs to scope the conversation.
+ Keep it identical across a conversation's calls.
+ stream: Yield the answer as text chunks as it is produced.
+ model: Local only — backend model name (defaults to
+ ``retrieve_model``).
+
+ Returns:
+ - stream=False: the answer string
+ - stream=True: iterator of text chunks
+ """
+ result = self.chat_completions(messages, stream=stream,
+ doc_id=doc_id, model=model)
+ if stream:
+ return cast(Iterator[str], result)
+ envelope = cast(dict[str, Any], result)
+ return envelope["choices"][0]["message"]["content"] or ""
 
  def chat_completions(
  self,
@@ -368,7 +405,9 @@ def chat_completions(
  backend, so any OpenAI-compatible server works; a ``/`` in the
  model name means LiteLLM provider routing, so prefix ``openai/``
  when the backend itself serves slashed ids, e.g.
- ``openai/Qwen/...`` on vLLM). The non-stream
+ ``openai/Qwen/...`` on vLLM; LiteLLM-routed Claude models —
+ Anthropic direct, Bedrock, Vertex — get the managed prompt
+ prefix cache-marked automatically). The non-stream
  response carries the final answer only; streaming yields the
  agent's visible text as it is produced, including narration before
  tool calls. ``finish_reason`` reports loop completion ("stop") —
@@ -380,7 +419,10 @@ def chat_completions(
  messages: Conversation messages with 'role' and 'content' keys,
  or a bare query string (it becomes a single user message).
  Local also accepts system/developer messages — their content
- is appended to the managed system prompt.
+ is appended to the managed system prompt. Local takes text
+ history only: tool-role turns are rejected (the cloud
+ endpoint forwards them verbatim), and message fields beyond
+ role/content are dropped.
  stream: Enable streaming responses.
  doc_id: Document ID or list of IDs to scope the conversation.
  Keep it identical across a conversation's calls — the

pageindex/local_chat.py

@@ -206,7 +206,8 @@ def _require_openai_agents(method: str) -> None:
  except ImportError as exc:
  raise PageIndexAPIError(
  f"{method} in local mode requires the OpenAI Agents SDK — "
- "pip install openai-agents (or pip install 'pageindex[openai]')."
+ "pip install openai-agents (or pip install 'pageindex[openai]'). "
+ "messages() runs on the anthropic extra instead."
  ) from exc
 
 
@@ -271,6 +272,29 @@ def _reported_model(model_name: str) -> str:
  return model_name.removeprefix("litellm/").removeprefix("openai/")
 
 
+def _cache_extra_args(model_name: str) -> Optional[dict]:
+ """Claude's prompt caching is opt-in per request: on Claude models
+ routed through LiteLLM (Anthropic direct, Bedrock, Vertex — each
+ channel live-verified), mark the managed system prefix via LiteLLM's
+ injection param so the loop's later turns and a conversation's next
+ calls read it instead of repaying full price. Provider resolution is
+ LiteLLM's own, so this predicate can never disagree with where the
+ request actually routes."""
+ if "/" not in model_name or model_name.startswith("openai/"):
+ return None
+ try:
+ from litellm import get_llm_provider
+ model, provider, _, _ = get_llm_provider(
+ model=model_name.removeprefix("litellm/"))
+ except Exception:
+ return None
+ if provider == "anthropic" or (provider in ("bedrock", "vertex_ai")
+ and "claude" in model.lower()):
+ return {"cache_control_injection_points": [
+ {"location": "message", "role": "system"}]}
+ return None
+
+
 def _openai_agent(client, protocol: str, model_name: str, instructions: str,
  temperature, top_p, doc_ids=None):
  from agents import Agent, ModelSettings
@@ -280,7 +304,8 @@ def _openai_agent(client, protocol: str, model_name: str, instructions: str,
  instructions=instructions,
  tools=build_openai_tools(client, doc_ids=doc_ids),
  model=_openai_model(protocol, model_name),
- model_settings=ModelSettings(temperature=temperature, top_p=top_p),
+ model_settings=ModelSettings(temperature=temperature, top_p=top_p,
+ extra_args=_cache_extra_args(model_name)),
  )
 
 

tests/test_local_chat.py

@@ -280,6 +280,127 @@ def test_cloud_guards():
  max_tokens=10)
 
 
+@needs_agents
+def test_anthropic_routed_models_mark_managed_prefix_for_cache(
+ client, store_path, fake_model):
+ fake_model([[_msg_item("ok")]])
+ from pageindex.local_chat import _openai_agent
+ marked = {"cache_control_injection_points": [
+ {"location": "message", "role": "system"}]}
+ for name in ("anthropic/claude-x", "litellm/anthropic/claude-x",
+ "bedrock/us.anthropic.claude-sonnet-5",
+ "vertex_ai/claude-sonnet-4-5"):
+ agent = _openai_agent(client, "chat", name, "sys", None, None)
+ assert agent.model_settings.extra_args == marked
+ for name in ("gpt-5", "openai/Qwen/x", "litellm/groq/x",
+ "bedrock/meta.llama3-70b-instruct-v1:0",
+ "vertex_ai/gemini-2.5-pro"):
+ agent = _openai_agent(client, "chat", name, "sys", None, None)
+ assert agent.model_settings.extra_args is None
+
+
+@needs_agents
+def test_status_recorder_attaches_to_the_real_responses_model(monkeypatch):
+ # Guards the private-attribute chain the recorder rides
+ # (agent.model._client.responses.create): a vendor rename turns the
+ # recorder into a silent no-op and truncation reports as completion.
+ monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
+ import openai
+ from agents.models.openai_responses import OpenAIResponsesModel
+ backend = openai.AsyncOpenAI()
+ model = OpenAIResponsesModel("gpt-test", openai_client=backend)
+ original = backend.responses.create
+ local_chat._record_response_status(types.SimpleNamespace(model=model), {})
+ assert backend.responses.create is not original
+ asyncio.run(backend.close())
+
+
+@needs_agents
+def test_cache_marker_reaches_the_anthropic_wire(client, store_path,
+ monkeypatch):
+ # End-to-end guard for the injection flag: through the real
+ # LitellmModel and litellm's request build, the marker must appear in
+ # the HTTP body — a regression in either vendor hop silently reverts
+ # anthropic-routed calls to full price.
+ pytest.importorskip("litellm")
+ from litellm.llms.custom_httpx.http_handler import (AsyncHTTPHandler,
+ HTTPHandler)
+ monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test")
+ captured = {}
+ reply = {"id": "msg_01", "type": "message", "role": "assistant",
+ "model": "claude-3-5-sonnet-20240620",
+ "content": [{"type": "text", "text": "ok"}],
+ "stop_reason": "end_turn", "stop_sequence": None,
+ "usage": {"input_tokens": 10, "output_tokens": 2}}
+
+ def _capture(url, kwargs):
+ body = kwargs.get("json")
+ if body is None and kwargs.get("data") is not None:
+ body = json.loads(kwargs["data"])
+ captured["url"] = str(url)
+ captured["body"] = body
+ return httpx.Response(200, json=reply,
+ request=httpx.Request("POST", str(url)))
+
+ async def fake_apost(self, url=None, *args, **kwargs):
+ return _capture(url, kwargs)
+
+ def fake_post(self, url=None, *args, **kwargs):
+ return _capture(url, kwargs)
+
+ monkeypatch.setattr(AsyncHTTPHandler, "post", fake_apost)
+ monkeypatch.setattr(HTTPHandler, "post", fake_post)
+ result = client.chat_completions(
+ "hi", model="anthropic/claude-3-5-sonnet-20240620")
+ assert "/v1/messages" in captured["url"]
+ assert '"cache_control"' in json.dumps(captured["body"])
+ assert result["choices"][0]["message"]["content"] == "ok"
+
+
+# ── chat (front door) ──
+
+@needs_agents
+def test_chat_returns_answer_string(client, store_path, fake_model):
+ doc_id = seed_doc(store_path, "pi-a", "report.pdf")
+ fake = fake_model([
+ [_call_item("get_document", {"doc_name": "report.pdf"})],
+ [_msg_item("The answer")],
+ ])
+ assert client.chat("What status?", doc_id=doc_id) == "The answer"
+ first_item = fake.inputs[0][0]
+ assert "The user has specified document: report.pdf" in first_item["content"]
+
+
+@needs_agents
+def test_chat_stream_yields_text_chunks(client, store_path, fake_model):
+ fake_model([[_msg_item("The answer")]])
+ assert list(client.chat("q", stream=True)) == ["The ", "answer"]
+
+
+@needs_agents
+def test_chat_multi_turn_history(client, store_path, fake_model):
+ fake = fake_model([[_msg_item("Chapter 4 covers pears")]])
+ history = [
+ {"role": "user", "content": "What about chapter 3?"},
+ {"role": "assistant", "content": "Chapter 3 covers apples"},
+ {"role": "user", "content": "And chapter 4?"},
+ ]
+ assert client.chat(history) == "Chapter 4 covers pears"
+ assert fake.inputs[0][-3:] == history
+
+
+def test_chat_cloud_unwraps_envelope(monkeypatch):
+ cloud = PageIndexCloudClient(api_key="pi-test-key")
+
+ def fake_cc(**kwargs):
+ assert kwargs["messages"] == [{"role": "user", "content": "q"}]
+ return {"choices": [{"message": {"role": "assistant",
+ "content": "cloud answer"}}]}
+
+ monkeypatch.setattr(cloud._api, "chat_completions", fake_cc)
+ assert cloud.chat("q") == "cloud answer"
+
+
 # ── responses ──
 
 @needs_agents

pageindex/client.py

@@ -4,7 +4,7 @@
 import os
 import time
 import warnings
-from typing import Any, Callable, Iterator, Optional, Union
+from typing import Any, Callable, Iterator, Optional, Union, cast
 
 from .errors import PageIndexAPIError
 
@@ -345,7 +345,44 @@ def get_retrieval(self, retrieval_id: str) -> dict[str, Any]:
  "favor of chat completions; use chat_completions instead."
  ).get_retrieval(retrieval_id=retrieval_id)
 
- # ---------- CHAT COMPLETIONS ----------
+ # ---------- CHAT ----------
+
+ 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,
+ ) -> Union[str, Iterator[str]]:
+ """
+ Ask a question about your documents, get the answer.
+
+ Thin sugar over ``chat_completions()`` in both modes — 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()``.
+
+ Args:
+ messages: A question string, or role/content conversation
+ history.
+ doc_id: Document ID or list of IDs to scope the conversation.
+ Keep it identical across a conversation's calls.
+ stream: Yield the answer as text chunks as it is produced.
+ model: Local only — backend model name (defaults to
+ ``retrieve_model``).
+
+ Returns:
+ - stream=False: the answer string
+ - stream=True: iterator of text chunks
+ """
+ result = self.chat_completions(messages, stream=stream,
+ doc_id=doc_id, model=model)
+ if stream:
+ return cast(Iterator[str], result)
+ envelope = cast(dict[str, Any], result)
+ return envelope["choices"][0]["message"]["content"] or ""
 
  def chat_completions(
  self,
@@ -368,7 +405,9 @@ def chat_completions(
  backend, so any OpenAI-compatible server works; a ``/`` in the
  model name means LiteLLM provider routing, so prefix ``openai/``
  when the backend itself serves slashed ids, e.g.
- ``openai/Qwen/...`` on vLLM). The non-stream
+ ``openai/Qwen/...`` on vLLM; LiteLLM-routed Claude models —
+ Anthropic direct, Bedrock, Vertex — get the managed prompt
+ prefix cache-marked automatically). The non-stream
  response carries the final answer only; streaming yields the
  agent's visible text as it is produced, including narration before
  tool calls. ``finish_reason`` reports loop completion ("stop") —
@@ -380,7 +419,10 @@ def chat_completions(
  messages: Conversation messages with 'role' and 'content' keys,
  or a bare query string (it becomes a single user message).
  Local also accepts system/developer messages — their content
- is appended to the managed system prompt.
+ is appended to the managed system prompt. Local takes text
+ history only: tool-role turns are rejected (the cloud
+ endpoint forwards them verbatim), and message fields beyond
+ role/content are dropped.
  stream: Enable streaming responses.
  doc_id: Document ID or list of IDs to scope the conversation.
  Keep it identical across a conversation's calls — the

pageindex/local_chat.py

@@ -206,7 +206,8 @@ def _require_openai_agents(method: str) -> None:
  except ImportError as exc:
  raise PageIndexAPIError(
  f"{method} in local mode requires the OpenAI Agents SDK — "
- "pip install openai-agents (or pip install 'pageindex[openai]')."
+ "pip install openai-agents (or pip install 'pageindex[openai]'). "
+ "messages() runs on the anthropic extra instead."
  ) from exc
 
 
@@ -271,6 +272,29 @@ def _reported_model(model_name: str) -> str:
  return model_name.removeprefix("litellm/").removeprefix("openai/")
 
 
+def _cache_extra_args(model_name: str) -> Optional[dict]:
+ """Claude's prompt caching is opt-in per request: on Claude models
+ routed through LiteLLM (Anthropic direct, Bedrock, Vertex — each
+ channel live-verified), mark the managed system prefix via LiteLLM's
+ injection param so the loop's later turns and a conversation's next
+ calls read it instead of repaying full price. Provider resolution is
+ LiteLLM's own, so this predicate can never disagree with where the
+ request actually routes."""
+ if "/" not in model_name or model_name.startswith("openai/"):
+ return None
+ try:
+ from litellm import get_llm_provider
+ model, provider, _, _ = get_llm_provider(
+ model=model_name.removeprefix("litellm/"))
+ except Exception:
+ return None
+ if provider == "anthropic" or (provider in ("bedrock", "vertex_ai")
+ and "claude" in model.lower()):
+ return {"cache_control_injection_points": [
+ {"location": "message", "role": "system"}]}
+ return None
+
+
 def _openai_agent(client, protocol: str, model_name: str, instructions: str,
  temperature, top_p, doc_ids=None):
  from agents import Agent, ModelSettings
@@ -280,7 +304,8 @@ def _openai_agent(client, protocol: str, model_name: str, instructions: str,
  instructions=instructions,
  tools=build_openai_tools(client, doc_ids=doc_ids),
  model=_openai_model(protocol, model_name),
- model_settings=ModelSettings(temperature=temperature, top_p=top_p),
+ model_settings=ModelSettings(temperature=temperature, top_p=top_p,
+ extra_args=_cache_extra_args(model_name)),
  )
 
 

tests/test_local_chat.py

@@ -280,6 +280,127 @@ def test_cloud_guards():
  max_tokens=10)
 
 
+@needs_agents
+def test_anthropic_routed_models_mark_managed_prefix_for_cache(
+ client, store_path, fake_model):
+ fake_model([[_msg_item("ok")]])
+ from pageindex.local_chat import _openai_agent
+ marked = {"cache_control_injection_points": [
+ {"location": "message", "role": "system"}]}
+ for name in ("anthropic/claude-x", "litellm/anthropic/claude-x",
+ "bedrock/us.anthropic.claude-sonnet-5",
+ "vertex_ai/claude-sonnet-4-5"):
+ agent = _openai_agent(client, "chat", name, "sys", None, None)
+ assert agent.model_settings.extra_args == marked
+ for name in ("gpt-5", "openai/Qwen/x", "litellm/groq/x",
+ "bedrock/meta.llama3-70b-instruct-v1:0",
+ "vertex_ai/gemini-2.5-pro"):
+ agent = _openai_agent(client, "chat", name, "sys", None, None)
+ assert agent.model_settings.extra_args is None
+
+
+@needs_agents
+def test_status_recorder_attaches_to_the_real_responses_model(monkeypatch):
+ # Guards the private-attribute chain the recorder rides
+ # (agent.model._client.responses.create): a vendor rename turns the
+ # recorder into a silent no-op and truncation reports as completion.
+ monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
+ import openai
+ from agents.models.openai_responses import OpenAIResponsesModel
+ backend = openai.AsyncOpenAI()
+ model = OpenAIResponsesModel("gpt-test", openai_client=backend)
+ original = backend.responses.create
+ local_chat._record_response_status(types.SimpleNamespace(model=model), {})
+ assert backend.responses.create is not original
+ asyncio.run(backend.close())
+
+
+@needs_agents
+def test_cache_marker_reaches_the_anthropic_wire(client, store_path,
+ monkeypatch):
+ # End-to-end guard for the injection flag: through the real
+ # LitellmModel and litellm's request build, the marker must appear in
+ # the HTTP body — a regression in either vendor hop silently reverts
+ # anthropic-routed calls to full price.
+ pytest.importorskip("litellm")
+ from litellm.llms.custom_httpx.http_handler import (AsyncHTTPHandler,
+ HTTPHandler)
+ monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test")
+ captured = {}
+ reply = {"id": "msg_01", "type": "message", "role": "assistant",
+ "model": "claude-3-5-sonnet-20240620",
+ "content": [{"type": "text", "text": "ok"}],
+ "stop_reason": "end_turn", "stop_sequence": None,
+ "usage": {"input_tokens": 10, "output_tokens": 2}}
+
+ def _capture(url, kwargs):
+ body = kwargs.get("json")
+ if body is None and kwargs.get("data") is not None:
+ body = json.loads(kwargs["data"])
+ captured["url"] = str(url)
+ captured["body"] = body
+ return httpx.Response(200, json=reply,
+ request=httpx.Request("POST", str(url)))
+
+ async def fake_apost(self, url=None, *args, **kwargs):
+ return _capture(url, kwargs)
+
+ def fake_post(self, url=None, *args, **kwargs):
+ return _capture(url, kwargs)
+
+ monkeypatch.setattr(AsyncHTTPHandler, "post", fake_apost)
+ monkeypatch.setattr(HTTPHandler, "post", fake_post)
+ result = client.chat_completions(
+ "hi", model="anthropic/claude-3-5-sonnet-20240620")
+ assert "/v1/messages" in captured["url"]
+ assert '"cache_control"' in json.dumps(captured["body"])
+ assert result["choices"][0]["message"]["content"] == "ok"
+
+
+# ── chat (front door) ──
+
+@needs_agents
+def test_chat_returns_answer_string(client, store_path, fake_model):
+ doc_id = seed_doc(store_path, "pi-a", "report.pdf")
+ fake = fake_model([
+ [_call_item("get_document", {"doc_name": "report.pdf"})],
+ [_msg_item("The answer")],
+ ])
+ assert client.chat("What status?", doc_id=doc_id) == "The answer"
+ first_item = fake.inputs[0][0]
+ assert "The user has specified document: report.pdf" in first_item["content"]
+
+
+@needs_agents
+def test_chat_stream_yields_text_chunks(client, store_path, fake_model):
+ fake_model([[_msg_item("The answer")]])
+ assert list(client.chat("q", stream=True)) == ["The ", "answer"]
+
+
+@needs_agents
+def test_chat_multi_turn_history(client, store_path, fake_model):
+ fake = fake_model([[_msg_item("Chapter 4 covers pears")]])
+ history = [
+ {"role": "user", "content": "What about chapter 3?"},
+ {"role": "assistant", "content": "Chapter 3 covers apples"},
+ {"role": "user", "content": "And chapter 4?"},
+ ]
+ assert client.chat(history) == "Chapter 4 covers pears"
+ assert fake.inputs[0][-3:] == history
+
+
+def test_chat_cloud_unwraps_envelope(monkeypatch):
+ cloud = PageIndexCloudClient(api_key="pi-test-key")
+
+ def fake_cc(**kwargs):
+ assert kwargs["messages"] == [{"role": "user", "content": "q"}]
+ return {"choices": [{"message": {"role": "assistant",
+ "content": "cloud answer"}}]}
+
+ monkeypatch.setattr(cloud._api, "chat_completions", fake_cc)
+ assert cloud.chat("q") == "cloud answer"
+
+
 # ── responses ──
 
 @needs_agents

pageindex/client.py

@@ -4,7 +4,7 @@
 import os
 import time
 import warnings
-from typing import Any, Callable, Iterator, Optional, Union
+from typing import Any, Callable, Iterator, Optional, Union, cast
 
 from .errors import PageIndexAPIError
 
@@ -345,7 +345,44 @@ def get_retrieval(self, retrieval_id: str) -> dict[str, Any]:
  "favor of chat completions; use chat_completions instead."
  ).get_retrieval(retrieval_id=retrieval_id)
 
- # ---------- CHAT COMPLETIONS ----------
+ # ---------- CHAT ----------
+
+ 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,
+ ) -> Union[str, Iterator[str]]:
+ """
+ Ask a question about your documents, get the answer.
+
+ Thin sugar over ``chat_completions()`` in both modes — 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()``.
+
+ Args:
+ messages: A question string, or role/content conversation
+ history.
+ doc_id: Document ID or list of IDs to scope the conversation.
+ Keep it identical across a conversation's calls.
+ stream: Yield the answer as text chunks as it is produced.
+ model: Local only — backend model name (defaults to
+ ``retrieve_model``).
+
+ Returns:
+ - stream=False: the answer string
+ - stream=True: iterator of text chunks
+ """
+ result = self.chat_completions(messages, stream=stream,
+ doc_id=doc_id, model=model)
+ if stream:
+ return cast(Iterator[str], result)
+ envelope = cast(dict[str, Any], result)
+ return envelope["choices"][0]["message"]["content"] or ""
 
  def chat_completions(
  self,
@@ -368,7 +405,9 @@ def chat_completions(
  backend, so any OpenAI-compatible server works; a ``/`` in the
  model name means LiteLLM provider routing, so prefix ``openai/``
  when the backend itself serves slashed ids, e.g.
- ``openai/Qwen/...`` on vLLM). The non-stream
+ ``openai/Qwen/...`` on vLLM; LiteLLM-routed Claude models —
+ Anthropic direct, Bedrock, Vertex — get the managed prompt
+ prefix cache-marked automatically). The non-stream
  response carries the final answer only; streaming yields the
  agent's visible text as it is produced, including narration before
  tool calls. ``finish_reason`` reports loop completion ("stop") —
@@ -380,7 +419,10 @@ def chat_completions(
  messages: Conversation messages with 'role' and 'content' keys,
  or a bare query string (it becomes a single user message).
  Local also accepts system/developer messages — their content
- is appended to the managed system prompt.
+ is appended to the managed system prompt. Local takes text
+ history only: tool-role turns are rejected (the cloud
+ endpoint forwards them verbatim), and message fields beyond
+ role/content are dropped.
  stream: Enable streaming responses.
  doc_id: Document ID or list of IDs to scope the conversation.
  Keep it identical across a conversation's calls — the

pageindex/local_chat.py

@@ -206,7 +206,8 @@ def _require_openai_agents(method: str) -> None:
  except ImportError as exc:
  raise PageIndexAPIError(
  f"{method} in local mode requires the OpenAI Agents SDK — "
- "pip install openai-agents (or pip install 'pageindex[openai]')."
+ "pip install openai-agents (or pip install 'pageindex[openai]'). "
+ "messages() runs on the anthropic extra instead."
  ) from exc
 
 
@@ -271,6 +272,29 @@ def _reported_model(model_name: str) -> str:
  return model_name.removeprefix("litellm/").removeprefix("openai/")
 
 
+def _cache_extra_args(model_name: str) -> Optional[dict]:
+ """Claude's prompt caching is opt-in per request: on Claude models
+ routed through LiteLLM (Anthropic direct, Bedrock, Vertex — each
+ channel live-verified), mark the managed system prefix via LiteLLM's
+ injection param so the loop's later turns and a conversation's next
+ calls read it instead of repaying full price. Provider resolution is
+ LiteLLM's own, so this predicate can never disagree with where the
+ request actually routes."""
+ if "/" not in model_name or model_name.startswith("openai/"):
+ return None
+ try:
+ from litellm import get_llm_provider
+ model, provider, _, _ = get_llm_provider(
+ model=model_name.removeprefix("litellm/"))
+ except Exception:
+ return None
+ if provider == "anthropic" or (provider in ("bedrock", "vertex_ai")
+ and "claude" in model.lower()):
+ return {"cache_control_injection_points": [
+ {"location": "message", "role": "system"}]}
+ return None
+
+
 def _openai_agent(client, protocol: str, model_name: str, instructions: str,
  temperature, top_p, doc_ids=None):
  from agents import Agent, ModelSettings
@@ -280,7 +304,8 @@ def _openai_agent(client, protocol: str, model_name: str, instructions: str,
  instructions=instructions,
  tools=build_openai_tools(client, doc_ids=doc_ids),
  model=_openai_model(protocol, model_name),
- model_settings=ModelSettings(temperature=temperature, top_p=top_p),
+ model_settings=ModelSettings(temperature=temperature, top_p=top_p,
+ extra_args=_cache_extra_args(model_name)),
  )
 
 

tests/test_local_chat.py

@@ -280,6 +280,127 @@ def test_cloud_guards():
  max_tokens=10)
 
 
+@needs_agents
+def test_anthropic_routed_models_mark_managed_prefix_for_cache(
+ client, store_path, fake_model):
+ fake_model([[_msg_item("ok")]])
+ from pageindex.local_chat import _openai_agent
+ marked = {"cache_control_injection_points": [
+ {"location": "message", "role": "system"}]}
+ for name in ("anthropic/claude-x", "litellm/anthropic/claude-x",
+ "bedrock/us.anthropic.claude-sonnet-5",
+ "vertex_ai/claude-sonnet-4-5"):
+ agent = _openai_agent(client, "chat", name, "sys", None, None)
+ assert agent.model_settings.extra_args == marked
+ for name in ("gpt-5", "openai/Qwen/x", "litellm/groq/x",
+ "bedrock/meta.llama3-70b-instruct-v1:0",
+ "vertex_ai/gemini-2.5-pro"):
+ agent = _openai_agent(client, "chat", name, "sys", None, None)
+ assert agent.model_settings.extra_args is None
+
+
+@needs_agents
+def test_status_recorder_attaches_to_the_real_responses_model(monkeypatch):
+ # Guards the private-attribute chain the recorder rides
+ # (agent.model._client.responses.create): a vendor rename turns the
+ # recorder into a silent no-op and truncation reports as completion.
+ monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
+ import openai
+ from agents.models.openai_responses import OpenAIResponsesModel
+ backend = openai.AsyncOpenAI()
+ model = OpenAIResponsesModel("gpt-test", openai_client=backend)
+ original = backend.responses.create
+ local_chat._record_response_status(types.SimpleNamespace(model=model), {})
+ assert backend.responses.create is not original
+ asyncio.run(backend.close())
+
+
+@needs_agents
+def test_cache_marker_reaches_the_anthropic_wire(client, store_path,
+ monkeypatch):
+ # End-to-end guard for the injection flag: through the real
+ # LitellmModel and litellm's request build, the marker must appear in
+ # the HTTP body — a regression in either vendor hop silently reverts
+ # anthropic-routed calls to full price.
+ pytest.importorskip("litellm")
+ from litellm.llms.custom_httpx.http_handler import (AsyncHTTPHandler,
+ HTTPHandler)
+ monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test")
+ captured = {}
+ reply = {"id": "msg_01", "type": "message", "role": "assistant",
+ "model": "claude-3-5-sonnet-20240620",
+ "content": [{"type": "text", "text": "ok"}],
+ "stop_reason": "end_turn", "stop_sequence": None,
+ "usage": {"input_tokens": 10, "output_tokens": 2}}
+
+ def _capture(url, kwargs):
+ body = kwargs.get("json")
+ if body is None and kwargs.get("data") is not None:
+ body = json.loads(kwargs["data"])
+ captured["url"] = str(url)
+ captured["body"] = body
+ return httpx.Response(200, json=reply,
+ request=httpx.Request("POST", str(url)))
+
+ async def fake_apost(self, url=None, *args, **kwargs):
+ return _capture(url, kwargs)
+
+ def fake_post(self, url=None, *args, **kwargs):
+ return _capture(url, kwargs)
+
+ monkeypatch.setattr(AsyncHTTPHandler, "post", fake_apost)
+ monkeypatch.setattr(HTTPHandler, "post", fake_post)
+ result = client.chat_completions(
+ "hi", model="anthropic/claude-3-5-sonnet-20240620")
+ assert "/v1/messages" in captured["url"]
+ assert '"cache_control"' in json.dumps(captured["body"])
+ assert result["choices"][0]["message"]["content"] == "ok"
+
+
+# ── chat (front door) ──
+
+@needs_agents
+def test_chat_returns_answer_string(client, store_path, fake_model):
+ doc_id = seed_doc(store_path, "pi-a", "report.pdf")
+ fake = fake_model([
+ [_call_item("get_document", {"doc_name": "report.pdf"})],
+ [_msg_item("The answer")],
+ ])
+ assert client.chat("What status?", doc_id=doc_id) == "The answer"
+ first_item = fake.inputs[0][0]
+ assert "The user has specified document: report.pdf" in first_item["content"]
+
+
+@needs_agents
+def test_chat_stream_yields_text_chunks(client, store_path, fake_model):
+ fake_model([[_msg_item("The answer")]])
+ assert list(client.chat("q", stream=True)) == ["The ", "answer"]
+
+
+@needs_agents
+def test_chat_multi_turn_history(client, store_path, fake_model):
+ fake = fake_model([[_msg_item("Chapter 4 covers pears")]])
+ history = [
+ {"role": "user", "content": "What about chapter 3?"},
+ {"role": "assistant", "content": "Chapter 3 covers apples"},
+ {"role": "user", "content": "And chapter 4?"},
+ ]
+ assert client.chat(history) == "Chapter 4 covers pears"
+ assert fake.inputs[0][-3:] == history
+
+
+def test_chat_cloud_unwraps_envelope(monkeypatch):
+ cloud = PageIndexCloudClient(api_key="pi-test-key")
+
+ def fake_cc(**kwargs):
+ assert kwargs["messages"] == [{"role": "user", "content": "q"}]
+ return {"choices": [{"message": {"role": "assistant",
+ "content": "cloud answer"}}]}
+
+ monkeypatch.setattr(cloud._api, "chat_completions", fake_cc)
+ assert cloud.chat("q") == "cloud answer"
+
+
 # ── responses ──
 
 @needs_agents

pageindex/client.py

@@ -4,7 +4,7 @@
 import os
 import time
 import warnings
-from typing import Any, Callable, Iterator, Optional, Union
+from typing import Any, Callable, Iterator, Optional, Union, cast
 
 from .errors import PageIndexAPIError
 
@@ -345,7 +345,44 @@ def get_retrieval(self, retrieval_id: str) -> dict[str, Any]:
  "favor of chat completions; use chat_completions instead."
  ).get_retrieval(retrieval_id=retrieval_id)
 
- # ---------- CHAT COMPLETIONS ----------
+ # ---------- CHAT ----------
+
+ 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,
+ ) -> Union[str, Iterator[str]]:
+ """
+ Ask a question about your documents, get the answer.
+
+ Thin sugar over ``chat_completions()`` in both modes — 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()``.
+
+ Args:
+ messages: A question string, or role/content conversation
+ history.
+ doc_id: Document ID or list of IDs to scope the conversation.
+ Keep it identical across a conversation's calls.
+ stream: Yield the answer as text chunks as it is produced.
+ model: Local only — backend model name (defaults to
+ ``retrieve_model``).
+
+ Returns:
+ - stream=False: the answer string
+ - stream=True: iterator of text chunks
+ """
+ result = self.chat_completions(messages, stream=stream,
+ doc_id=doc_id, model=model)
+ if stream:
+ return cast(Iterator[str], result)
+ envelope = cast(dict[str, Any], result)
+ return envelope["choices"][0]["message"]["content"] or ""
 
  def chat_completions(
  self,
@@ -368,7 +405,9 @@ def chat_completions(
  backend, so any OpenAI-compatible server works; a ``/`` in the
  model name means LiteLLM provider routing, so prefix ``openai/``
  when the backend itself serves slashed ids, e.g.
- ``openai/Qwen/...`` on vLLM). The non-stream
+ ``openai/Qwen/...`` on vLLM; LiteLLM-routed Claude models —
+ Anthropic direct, Bedrock, Vertex — get the managed prompt
+ prefix cache-marked automatically). The non-stream
  response carries the final answer only; streaming yields the
  agent's visible text as it is produced, including narration before
  tool calls. ``finish_reason`` reports loop completion ("stop") —
@@ -380,7 +419,10 @@ def chat_completions(
  messages: Conversation messages with 'role' and 'content' keys,
  or a bare query string (it becomes a single user message).
  Local also accepts system/developer messages — their content
- is appended to the managed system prompt.
+ is appended to the managed system prompt. Local takes text
+ history only: tool-role turns are rejected (the cloud
+ endpoint forwards them verbatim), and message fields beyond
+ role/content are dropped.
  stream: Enable streaming responses.
  doc_id: Document ID or list of IDs to scope the conversation.
  Keep it identical across a conversation's calls — the

pageindex/local_chat.py

@@ -206,7 +206,8 @@ def _require_openai_agents(method: str) -> None:
  except ImportError as exc:
  raise PageIndexAPIError(
  f"{method} in local mode requires the OpenAI Agents SDK — "
- "pip install openai-agents (or pip install 'pageindex[openai]')."
+ "pip install openai-agents (or pip install 'pageindex[openai]'). "
+ "messages() runs on the anthropic extra instead."
  ) from exc
 
 
@@ -271,6 +272,29 @@ def _reported_model(model_name: str) -> str:
  return model_name.removeprefix("litellm/").removeprefix("openai/")
 
 
+def _cache_extra_args(model_name: str) -> Optional[dict]:
+ """Claude's prompt caching is opt-in per request: on Claude models
+ routed through LiteLLM (Anthropic direct, Bedrock, Vertex — each
+ channel live-verified), mark the managed system prefix via LiteLLM's
+ injection param so the loop's later turns and a conversation's next
+ calls read it instead of repaying full price. Provider resolution is
+ LiteLLM's own, so this predicate can never disagree with where the
+ request actually routes."""
+ if "/" not in model_name or model_name.startswith("openai/"):
+ return None
+ try:
+ from litellm import get_llm_provider
+ model, provider, _, _ = get_llm_provider(
+ model=model_name.removeprefix("litellm/"))
+ except Exception:
+ return None
+ if provider == "anthropic" or (provider in ("bedrock", "vertex_ai")
+ and "claude" in model.lower()):
+ return {"cache_control_injection_points": [
+ {"location": "message", "role": "system"}]}
+ return None
+
+
 def _openai_agent(client, protocol: str, model_name: str, instructions: str,
  temperature, top_p, doc_ids=None):
  from agents import Agent, ModelSettings
@@ -280,7 +304,8 @@ def _openai_agent(client, protocol: str, model_name: str, instructions: str,
  instructions=instructions,
  tools=build_openai_tools(client, doc_ids=doc_ids),
  model=_openai_model(protocol, model_name),
- model_settings=ModelSettings(temperature=temperature, top_p=top_p),
+ model_settings=ModelSettings(temperature=temperature, top_p=top_p,
+ extra_args=_cache_extra_args(model_name)),
  )
 
 

tests/test_local_chat.py

@@ -280,6 +280,127 @@ def test_cloud_guards():
  max_tokens=10)
 
 
+@needs_agents
+def test_anthropic_routed_models_mark_managed_prefix_for_cache(
+ client, store_path, fake_model):
+ fake_model([[_msg_item("ok")]])
+ from pageindex.local_chat import _openai_agent
+ marked = {"cache_control_injection_points": [
+ {"location": "message", "role": "system"}]}
+ for name in ("anthropic/claude-x", "litellm/anthropic/claude-x",
+ "bedrock/us.anthropic.claude-sonnet-5",
+ "vertex_ai/claude-sonnet-4-5"):
+ agent = _openai_agent(client, "chat", name, "sys", None, None)
+ assert agent.model_settings.extra_args == marked
+ for name in ("gpt-5", "openai/Qwen/x", "litellm/groq/x",
+ "bedrock/meta.llama3-70b-instruct-v1:0",
+ "vertex_ai/gemini-2.5-pro"):
+ agent = _openai_agent(client, "chat", name, "sys", None, None)
+ assert agent.model_settings.extra_args is None
+
+
+@needs_agents
+def test_status_recorder_attaches_to_the_real_responses_model(monkeypatch):
+ # Guards the private-attribute chain the recorder rides
+ # (agent.model._client.responses.create): a vendor rename turns the
+ # recorder into a silent no-op and truncation reports as completion.
+ monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
+ import openai
+ from agents.models.openai_responses import OpenAIResponsesModel
+ backend = openai.AsyncOpenAI()
+ model = OpenAIResponsesModel("gpt-test", openai_client=backend)
+ original = backend.responses.create
+ local_chat._record_response_status(types.SimpleNamespace(model=model), {})
+ assert backend.responses.create is not original
+ asyncio.run(backend.close())
+
+
+@needs_agents
+def test_cache_marker_reaches_the_anthropic_wire(client, store_path,
+ monkeypatch):
+ # End-to-end guard for the injection flag: through the real
+ # LitellmModel and litellm's request build, the marker must appear in
+ # the HTTP body — a regression in either vendor hop silently reverts
+ # anthropic-routed calls to full price.
+ pytest.importorskip("litellm")
+ from litellm.llms.custom_httpx.http_handler import (AsyncHTTPHandler,
+ HTTPHandler)
+ monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test")
+ captured = {}
+ reply = {"id": "msg_01", "type": "message", "role": "assistant",
+ "model": "claude-3-5-sonnet-20240620",
+ "content": [{"type": "text", "text": "ok"}],
+ "stop_reason": "end_turn", "stop_sequence": None,
+ "usage": {"input_tokens": 10, "output_tokens": 2}}
+
+ def _capture(url, kwargs):
+ body = kwargs.get("json")
+ if body is None and kwargs.get("data") is not None:
+ body = json.loads(kwargs["data"])
+ captured["url"] = str(url)
+ captured["body"] = body
+ return httpx.Response(200, json=reply,
+ request=httpx.Request("POST", str(url)))
+
+ async def fake_apost(self, url=None, *args, **kwargs):
+ return _capture(url, kwargs)
+
+ def fake_post(self, url=None, *args, **kwargs):
+ return _capture(url, kwargs)
+
+ monkeypatch.setattr(AsyncHTTPHandler, "post", fake_apost)
+ monkeypatch.setattr(HTTPHandler, "post", fake_post)
+ result = client.chat_completions(
+ "hi", model="anthropic/claude-3-5-sonnet-20240620")
+ assert "/v1/messages" in captured["url"]
+ assert '"cache_control"' in json.dumps(captured["body"])
+ assert result["choices"][0]["message"]["content"] == "ok"
+
+
+# ── chat (front door) ──
+
+@needs_agents
+def test_chat_returns_answer_string(client, store_path, fake_model):
+ doc_id = seed_doc(store_path, "pi-a", "report.pdf")
+ fake = fake_model([
+ [_call_item("get_document", {"doc_name": "report.pdf"})],
+ [_msg_item("The answer")],
+ ])
+ assert client.chat("What status?", doc_id=doc_id) == "The answer"
+ first_item = fake.inputs[0][0]
+ assert "The user has specified document: report.pdf" in first_item["content"]
+
+
+@needs_agents
+def test_chat_stream_yields_text_chunks(client, store_path, fake_model):
+ fake_model([[_msg_item("The answer")]])
+ assert list(client.chat("q", stream=True)) == ["The ", "answer"]
+
+
+@needs_agents
+def test_chat_multi_turn_history(client, store_path, fake_model):
+ fake = fake_model([[_msg_item("Chapter 4 covers pears")]])
+ history = [
+ {"role": "user", "content": "What about chapter 3?"},
+ {"role": "assistant", "content": "Chapter 3 covers apples"},
+ {"role": "user", "content": "And chapter 4?"},
+ ]
+ assert client.chat(history) == "Chapter 4 covers pears"
+ assert fake.inputs[0][-3:] == history
+
+
+def test_chat_cloud_unwraps_envelope(monkeypatch):
+ cloud = PageIndexCloudClient(api_key="pi-test-key")
+
+ def fake_cc(**kwargs):
+ assert kwargs["messages"] == [{"role": "user", "content": "q"}]
+ return {"choices": [{"message": {"role": "assistant",
+ "content": "cloud answer"}}]}
+
+ monkeypatch.setattr(cloud._api, "chat_completions", fake_cc)
+ assert cloud.chat("q") == "cloud answer"
+
+
 # ── responses ──
 
 @needs_agents