PageIndex/examples/local_demo.py
Ray 6f200927cb fix: enforce collection membership on cloud query doc_ids
Cloud query()/query_stream() now run the same _require_document membership
guard the four doc-scoped ops received in 929b3df, so a doc_id from another
collection raises DocumentNotFoundError (matching the local backend's
_scoped_docs) instead of silently answering from another collection's
document. The guard runs after doc_ids normalization and before the
completion request; query_stream runs it via asyncio.to_thread. Folders
unavailable on the plan -> guard skips, as with the sibling ops.

Also rename the streaming QueryEvent types answer_delta/answer_done ->
text_delta/text_done: text_done fires once per assistant text message, which
is honest for the local agent loop's multi-message stream (the last one
before the stream ends carries the final answer). Corrects the cloud
backend's now-inaccurate "single terminal contract" comment. reasoning
stays reserved for future model-reasoning tokens. QueryEvent is unreleased
API (0.3.0.dev1), so the rename is not a breaking change.
2026-07-16 14:40:19 +08:00

69 lines
2.2 KiB
Python

"""
Agentic Vectorless RAG with PageIndex SDK - Local Demo
A simple example of using LocalClient for self-hosted document indexing
and agent-based QA. The agent uses OpenAI Agents SDK to reason over
the document's tree structure index.
Steps:
1 — Download and index a PDF
2 — Stream a question with tool call visibility
Requirements:
pip install pageindex
export OPENAI_API_KEY=your-api-key # or any LiteLLM-supported provider
"""
import asyncio
from pathlib import Path
import requests
from pageindex import LocalClient
_EXAMPLES_DIR = Path(__file__).parent
PDF_URL = "https://arxiv.org/pdf/1706.03762.pdf"
PDF_PATH = _EXAMPLES_DIR / "documents" / "attention.pdf"
WORKSPACE = _EXAMPLES_DIR / "workspace"
MODEL = "gpt-4o-2024-11-20" # any LiteLLM-supported model
# Download PDF if needed
if not PDF_PATH.exists():
print(f"Downloading {PDF_URL} ...")
PDF_PATH.parent.mkdir(parents=True, exist_ok=True)
with requests.get(PDF_URL, stream=True, timeout=30) as r:
r.raise_for_status()
with open(PDF_PATH, "wb") as f:
for chunk in r.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
print("Download complete.\n")
client = LocalClient(model=MODEL, storage_path=str(WORKSPACE))
col = client.collection()
doc_id = col.add(str(PDF_PATH))
print(f"Indexed: {doc_id}\n")
# Streaming query
stream = col.query(
"What is the main architecture proposed in this paper and how does self-attention work?",
stream=True,
)
async def main():
streamed_text = False
async for event in stream:
if event.type == "text_delta":
print(event.data, end="", flush=True)
streamed_text = True
elif event.type == "tool_call":
if streamed_text:
print()
streamed_text = False
print(f"[tool call] {event.data['name']}")
elif event.type == "tool_result":
preview = str(event.data)[:200] + "..." if len(str(event.data)) > 200 else event.data
print(f"[tool output] {preview}")
elif event.type == "text_done":
print()
streamed_text = False
asyncio.run(main())