pystreamdocuments 0.2.0

Smart context provider for LLMs - Minimal, most-relevant knowledge based on query anticipation and semantic understanding

PyStreamDocuments

Universal Knowledge Ingestion & Retrieval Layer — One API for all sources. Progressive filtering for minimal, relevant context.

Overview

PyStreamDocuments is a unified ingestion and retrieval layer that transforms many knowledge sources into one intelligent context provider for LLMs.

One API, Many Sources:

Local Docs (PDF, DOCX, XLSX) + Databases (PostgreSQL, BigQuery, etc.)
+ MCP Tools (Notion, Slack, GitHub, Salesforce)
+ Web Context (via WebMCP)
+ Code & Configuration
      ↓
  PyStreamDocuments
      ↓
Minimal, Ranked, Deduplicated Context (50 tokens, not 500)

How It Works:

1. Ingest from any source (one API handles all)
2. Extract metadata unified schema (entities, relationships, topics)
3. Filter progressively (4 stages eliminate 90%+ noise)
4. Return minimal context (only what LLM needs)

Why It Matters:

• Traditional RAG: 10 documents (500+ tokens of noise)
• PyStreamDocuments: 1-3 pieces (50 tokens of pure signal)
• 10-100x fewer tokens → 10-100x lower cost → better answers

Key Features

• Multi-Format Support: PDF, DOCX, PPTX, CSV, Google Docs/Sheets/Slides, SharePoint, S3
• Embedded Intelligence: PyStreamPDF for PDFs, PyStreamXL for spreadsheets
• Structure Extraction: Table of contents, headers/footers, appendix, sections as navigable hierarchy
• Entity-Relationship Graphs: Deduplication + cross-document linking with section context
• Query Anticipation: Predict LLM's next question based on document structure
• Minimal Output: Return 50 tokens of pure signal (not 500 tokens of noise)
• Continuous Reranking: Pre-rank at index time; instant O(1) retrieval at query time
• Change Detection: Automatically re-index deltas; maintain ranking coherence
• Semantic Topic Clustering: Live hierarchical topics with automatic grouping
• OKF Integration: Master intelligence index for skip-irrelevant-paths retrieval
• WebMCP Built-In: Browser automation for retrieving web context when local insufficient
• Hybrid Knowledge: Combine local documents + web context seamlessly
• Entity Enrichment: Auto-fetch latest web data for entities (market caps, news, updates)

Quick Start

Installation

pip install pystreamdocuments

Usage

from pystreamdocuments import PyStreamDocuments

# Initialize with your storage source
ingest = Ingest(
    source="s3://my-bucket/documents",
    # or "sharepoint://tenant/sites/site/DocumentLibrary"
    # or "google://My Drive/Documents"
    aws_access_key_id="...",
    aws_secret_access_key="...",
)

# Ingest documents from folder
job = ingest.index_folder("documents/")
await job.wait()  # Metadata-first indexing

# Search with context
results = ingest.search(
    query="What's our Q3 revenue forecast?",
    context={
        "agent_role": "finance_analyst",
        "domain": "finance",
        "time_period": ("2024-Q1", "2024-Q4"),
    }
)

for result in results:
    print(f"{result.entity}: {result.summary}")

REST API

# Start server
pystreamdocuments-server --host 0.0.0.0 --port 8000

# Index a folder
curl -X POST http://localhost:8000/ingest \
  -H "Content-Type: application/json" \
  -d '{"source": "s3://my-bucket/docs", "folder_path": "/"}'

# Search
curl -X POST http://localhost:8000/search \
  -H "Content-Type: application/json" \
  -d '{
    "query": "authorization procedures",
    "context": {"domain": "security", "agent_role": "compliance"},
    "limit": 5
  }'

Architecture

Universal Ingestion Pipeline

One API for All Sources. Source-Specific Handling Internally Hidden.

User Action: await client.ingest("s3://bucket/docs")
             await client.ingest("postgres://db")
             await client.ingest("notion://workspace")
             await client.ingest("github://repo")
             
                        ↓
                        
Ingestion Coordinator:
├─ Detect source type (S3 bucket, PostgreSQL, Notion, GitHub, local)
├─ Route to source-specific handler
└─ Unify output to knowledge objects

                        ↓
                        
Source-Specific Handlers (Parallel):

S3 Bucket / Local Files:
  ├─ Scan: File metadata (ETag, LastModified)
  ├─ Detect format: PDF, DOCX, XLSX, CSV, JSON, YAML, HTML
  ├─ Parse:
  │   ├─ PDF → PyStreamPDF (intelligent structure extraction)
  │   ├─ XLSX → PyStreamXL (query-aware spreadsheet parsing)
  │   ├─ Code → AST parsers (Python, JS, Go, Rust, etc.)
  │   └─ Config → Native parsers (YAML, JSON, TOML, XML)
  └─ Extract: Entities, relationships, structure, metadata

Databases (PostgreSQL, MySQL, MongoDB, BigQuery, Snowflake):
  ├─ Via PyStreamMCP: All schema inspection + queries
  ├─ Schema extraction: Tables, columns, constraints, relationships
  ├─ Data profiling: Row counts, distributions, patterns
  └─ Extract: Database schema as knowledge objects

MCP Tools (Notion, Slack, GitHub, Salesforce, Jira):
  ├─ Via PyStreamMCP: Tool-specific connectors
  ├─ Notion: Extract pages, databases, properties
  ├─ Slack: Extract messages, threads, decisions, participants
  ├─ GitHub: Extract code, issues, PRs, discussions, commits
  └─ Extract: MCP content as unified knowledge objects

Web (via WebMCP):
  ├─ Browser automation: Click, navigate, extract
  ├─ Content extraction: Text, structure, links, metadata
  └─ Extract: Web pages as knowledge objects

                        ↓
                        
Unified Metadata Extraction (All Sources):
├─ Entities: Organizations, people, products, places, dates, metrics
├─ Relationships: Cross-source links, dependencies, references
├─ Structure: Hierarchy, sections, tables, definitions
├─ Classification: Knowledge type (procedure, policy, definition, etc.)
├─ Domain: Finance, engineering, HR, compliance, etc.
├─ Freshness metadata: created_at, modified_at, published_at, effective_date
└─ Deduplication markers: Canonical ID, similar objects

                        ↓
                        
Unified Indexing (All Sources):
├─ BM25 index (keyword search via tantivy)
├─ Metadata index (entity/relationship queries)
├─ Knowledge graph (cross-document relationships)
├─ Topic index (live hierarchical clustering)
└─ Change tracking (what changed and when)

                        ↓
                        
Live Monitoring (All Sources):
├─ Local files: Poll metadata (ETag, LastModified) — incremental
├─ Databases: CDC streams via PyStreamMCP
├─ MCP tools: Periodic syncs (Notion, Slack, GitHub)
└─ Web: Cached with TTL (24-72 hours)

Retrieval Pipeline (Minimal Output via Progressive Filtering)

1. Query Understanding → Intent classification, entity extraction
2. Stage 1 Filter → Source relevance (which sources have answer?)
3. Stage 2 Filter → Entity/Type matching (narrow within sources)
4. Stage 3 Filter → Freshness ranking (use document metadata, not index time)
5. Stage 4 Filter → Semantic reranking (deep understanding, optional embeddings)
6. Deduplication → Same info in multiple sources? Return once
7. Output → Minimal context (50 tokens vs 500 for traditional RAG)

How Progressive Filtering Eliminates Noise

Example Query: "What's the latest authorization policy?"

Starting corpus: 1000 indexed knowledge objects
├─ 500 local documents (PDFs, DOCX)
├─ 100 database tables (schemas via PyStreamMCP)
├─ 200 Notion pages
├─ 100 GitHub issues
├─ 50 Slack messages
└─ 50 web pages

Stage 1 (Source Relevance): Which sources could answer? ($0 cost)
→ Filter to sources with "authorization" topic
→ Remaining: ~100 candidates (5 sources × ~20 each)

Stage 2 (Entity/Type Match): Which objects match entity + type? ($0 cost)
→ Filter to knowledge_type="Policy" AND entities contains "authorization"
→ Remaining: ~40 candidates (high confidence)

Stage 3 (Freshness): Use DOCUMENT metadata, not index time ($0 cost)
→ Rank by: created_at, modified_at, effective_date, deprecated_at
→ Skip stale docs (6+ months old without update)
→ Remaining: ~30 candidates (current + recent)

Stage 4 (Semantic Reranking): Deep understanding of query vs content (~$0.0001 cost)
→ LLM or embeddings: What answer does query need?
→ Rerank top 10 for relevance to specific question
→ Remaining: ~5 top results

Deduplication: Same content from multiple sources?
→ Notion page "Authorization Policy v2.5" + Slack message "see Notion"
→ Return Notion page once (authoritative), skip Slack reference
→ Final: 2-3 results (PDF backup + Notion primary + GitHub implementation)

Output:
"Authorization Policy v2.5 (Notion, updated yesterday):
  - Section 3.2: Policy approval requires...
  - Section 5.1: Escalation process...
  
Implementation (GitHub, merged today):
  - auth/flow.rs: Request validation
  - auth/approval.rs: Approval workflow"

Key Metrics:

• Stage 1-3: Eliminates 97% (metadata filtering, $0 cost)
• Stage 4: Only reranks top 10 (not 1000)
• Deduplication: Returns 1 source per fact (not 5)
• Total cost: <$0.001 per query (vs >$0.01 for traditional RAG)
• Result: 50 tokens (vs 500 for traditional RAG)

Knowledge Categories Supported

Category	Formats	Extracted Knowledge Objects
Documents	PDF, DOCX, PPTX, NOTION, CONFLUENCE, MARKDOWN, HTML	Procedures, policies, definitions, concepts, sections
Structured Data	CSV, XLSX, SQL, Parquet, SQLITE, Databases	Entities, metrics, relationships, datasets, schemas
Code & Configuration	Python, Java, JS, TS, Go, Rust, YAML, JSON, TOML, XML	APIs, classes, functions, configuration rules
Communications	Email, Slack, Teams, Discord, Meeting transcripts	Decisions, discussions, action items, stakeholders
Spatial & Visual	CAD, Diagrams, SVG, Images (PNG, JPEG, WebP), Video	Architecture components, topology, flows, processes

Powered by: PyStreamPDF (Documents), PyStreamXL (Structured Data), AST parsers (Code), Vision models (Spatial)

Storage & Retrieval Backends

Document Sources:

• S3: Direct bucket access
• SharePoint: OAuth2 + Microsoft Graph API
• Google Workspace: OAuth2 + Google Drive API
• Local Filesystem: For testing/development

Database Sources (Via PyStreamMCP):

PyStreamDocuments Role: Schema understanding + metadata indexing PyStreamMCP Role: All database connections, queries, optimization

• PostgreSQL: Schema inspection via PyStreamMCP
• MySQL: Schema inspection via PyStreamMCP
• MongoDB: Collection structure via PyStreamMCP
• BigQuery: Dataset schema via PyStreamMCP
• Snowflake: Warehouse schema via PyStreamMCP
• Redshift: Cluster schema via PyStreamMCP
• Custom databases: Connectors via PyStreamMCP

Workflow:

1. PyStreamMCP inspects schema (introspection queries)
2. PyStreamDocuments indexes schema metadata + structure
3. User queries routed to PyStreamMCP for data retrieval
4. PyStreamMCP applies optimization, cost budgeting, quality gates
5. Results indexed by PyStreamDocuments for future retrieval

Embedding Models (Optional):

• OpenAI (text-embedding-3-small, text-embedding-3-large)
• Anthropic (embeddings API)
• Open-source (Sentence Transformers, all-MiniLM-L6-v2)
• Local models (no API calls, privacy-first)
• Custom models (bring your own)

Vector Databases (Optional):

• Pinecone (serverless, managed)
• Weaviate (open-source, self-hosted)
• Qdrant (open-source, Rust-based)
• Milvus (open-source, scalable)
• Chroma (simple, embedded)
• PostgreSQL + pgvector (traditional)
• Custom backends (pluggable)

Philosophy

Minimal Relevant Context. Embeddings Optional. Zero Vendor Lock-In.

Three Retrieval Modes (You Choose):

Mode 1: No Embeddings (Cheapest)

Metadata index → Topic clusters → BM25 keyword search
Cost: $0 per query
Speed: <50ms
Best for: When structure + keywords sufficient

Mode 2: Optional Embeddings (Balanced)

Metadata index → BM25 → [Optional] Semantic search
Cost: ~$0.02 per 1M tokens (if you enable embeddings)
Speed: <100ms
Best for: When semantic understanding helps

Mode 3: Custom Model + Custom DB (Full Control)

Metadata index → BM25 → Local embeddings → Your vector DB
Cost: $0 (your infrastructure)
Speed: <100ms
Best for: Privacy-critical or cost-optimized deployments

Key Differences from Traditional RAG:

Traditional RAG	PyStreamDocuments
Must embed all docs upfront	Embeddings optional, on-demand
Vendor lock-in (OpenAI/Pinecone)	Your choice of model/DB
All retrieval methods apply to all queries	Smart routing (metadata → BM25 → semantic)
500+ tokens of noise returned	50 tokens of pure signal
Pre-ranked by similarity score	Pre-ranked by topic coherence + entity match

Result: You control cost, privacy, and vendor relationships.

Status

v0.1 (In Development)

• Metadata-first ingestion pipeline
• Multi-format support (PDF, DOCX, PPTX, CSV)
• Format-specific intelligence (PyStreamPDF, PyStreamXL)
• BM25 keyword search
• Change detection & incremental indexing
• REST API

v0.2 (Planned)

• Semantic search with embeddings
• LLM-based reranking
• Query understanding
• Admin UI

v1.0 (Planned)

• Production hardening
• Multi-tenant support
• GraphRAG integration
• Knowledge graph visualization

Development

# Setup
git clone https://github.com/Mullassery/pystreamdocuments.git
cd pystreamdocuments

# Install dependencies
uv pip install -e ".[dev]"

# Build Rust extension
maturin develop

# Run tests
pytest

Contributing

Pull requests welcome. See CLAUDE.md for development guidelines.

License

MIT

---

Built by Georgi Mammen Mullassery

Part of the unified knowledge platform ecosystem:

• PyTerrainMap — Spatial intelligence
• PyStreamMCP — Agent orchestration
• PyStreamPDF — PDF intelligence
• PyStreamXL — Spreadsheet intelligence
• OpenAnchor — Token intelligence
• StatGuardian — Data quality