MongoSense

Ask your MongoDB database questions in plain English. MongoSense translates natural-language questions into safe, validated MongoDB aggregation pipelines and returns human-readable answers.

async with MongoSense(config) as ms:
    result = await ms.query("How many orders were placed last month?")
    print(result.formatted)
    # → "96,432 orders were placed in May 2026."

---

Requirements

• Python 3.9+
• MongoDB (Atlas or self-hosted, replica set)
• An LLM provider API key — Gemini, OpenAI, or any OpenAI-compatible endpoint (Kimi, DeepSeek, Groq, Together AI, OpenRouter, local Ollama/vLLM, etc.)

---

Installation

Not yet published to PyPI. Install from a clone in editable mode:

git clone https://github.com/VivekCh-003/mongoSense.git
cd mongoSense
pip install -e .

Once published, pip install mongosense will work directly.

---

Quickstart

import asyncio
from mongosense import MongoSense, MongoSenseConfig, GeminiBackend

config = MongoSenseConfig(
    mongo_uri="mongodb+srv://...",
    db_name="my_database",
    llm_backend=GeminiBackend(
        api_key="YOUR_GEMINI_API_KEY",
        models=["gemini-2.5-flash-lite"],
    ),
)

async def main():
    async with MongoSense(config) as ms:
        result = await ms.query("Show me the top 5 customers by order count")
        print(result.formatted)

asyncio.run(main())

MongoSense auto-discovers your collections at startup. No manual schema configuration required.

---

QueryResult

Every ms.query() call returns a QueryResult:

Field	Type	Description
docs	list[dict]	Raw documents returned from MongoDB
formatted	str | None	Plain-English answer (when response_format is "formatted" or "both")
mql	list[dict]	The aggregation pipeline that was executed
collection	str	Collection that was queried
confidence	float	0–1 score of how clearly the question mapped to the schema
interpretation	str	One-sentence description of what the library understood
requires_approval	bool	True when confidence is below hitl_threshold
alternatives	list[HITLAlternative]	Offered when requires_approval=True
pending_id	str | None	ID used to resume a pending HITL result

---

Handling Low-Confidence Queries (HITL)

When a question is ambiguous, query() returns early with requires_approval=True and three plain-English alternatives instead of executing a query. Call resume() with the chosen index to proceed.

result = await ms.query("Show me problem orders")

if result.requires_approval:
    print(f"Ambiguous ({result.confidence:.2f}) — did you mean:")
    for i, alt in enumerate(result.alternatives, 1):
        print(f"  {i}. {alt.plain_english}")

    choice = int(input("Pick 1/2/3: ")) - 1
    result = await ms.resume(result.pending_id, choice)

print(result.formatted)

The confidence threshold defaults to 0.8. Queries scoring at or above this execute automatically.

---

Configuration

from mongosense import MongoSenseConfig, SafetyConfig, CollectionConfig, GeminiBackend

config = MongoSenseConfig(
    mongo_uri="mongodb+srv://...",
    db_name="my_database",

    # Any LLMBackend implementation — GeminiBackend and OpenAICompatibleBackend ship built-in
    llm_backend=GeminiBackend(
        api_key="YOUR_GEMINI_API_KEY",
        models=["gemini-2.5-flash-lite", "gemini-2.0-flash"],  # tried in order on rate-limit errors
    ),

    # What to return: "raw", "formatted", or "both"
    response_format="both",

    # How the formatter generates answers
    # "flexible" — sends documents to LLM for free-form answer (default)
    # "strict"   — Python-computed stats, no LLM hallucination risk
    formatter_mode="flexible",

    # Confidence below this triggers HITL instead of execution
    hitl_threshold=0.8,

    # Restrict to specific collections (None = auto-discover all)
    collections=None,

    # Store successful (question, pipeline) pairs for few-shot retrieval.
    # In-memory by default; set True to persist to a collection in your own database.
    persist_query_memory=False,
)

SafetyConfig

config = MongoSenseConfig(
    ...
    safety=SafetyConfig(
        # Operators that are always blocked ($lookup is banned by default for Atlas M0 compatibility)
        banned_operators=["$lookup", "$out", "$merge", "$where", "$function", "$accumulator"],

        # Default result cap injected into every pipeline
        default_limit=500,

        # Ratio of docs examined to docs returned that triggers Gate 2 rejection
        max_docs_examined_ratio=1000,

        # Absolute docs-examined and collection-size ceilings, checked independently of the ratio
        max_docs_examined_absolute=10_000_000,
        max_scan_collection_size=10_000_000,

        # What to do when Gate 2 flags an inefficient query: "reject" (default) or "warn"
        on_inefficient_query="reject",
    ),
)

CollectionConfig (optional semantic hints)

By default MongoSense uses MongoDB's $jsonSchema validators (or sample documents) to build schema context. You can supplement this with plain-English descriptions:

from mongosense import CollectionConfig, RelationshipConfig

config = MongoSenseConfig(
    ...
    collections={
        "orders": CollectionConfig(
            description="Customer purchase transactions",
            fields={
                "status": "order lifecycle state: created, approved, invoiced, processing, shipped, delivered, unavailable, canceled"
            },
            # Optional — declare a known foreign key instead of relying on automatic inference
            relationships=[
                RelationshipConfig(field="customer_id", to="customers", to_field="customer_id"),
            ],
            # Optional — excluded from the sample values shown to the LLM and from formatter output
            redacted_fields=["customer_email"],
        ),
        "customers": CollectionConfig(
            description="Registered customer accounts",
        ),
    },
)

Fields listed here supplement (not replace) the schema inferred from MongoDB. Relationships you don't declare are inferred automatically at startup from field-naming patterns and cross-collection sampling.

---

Safety

Every query passes through three gates before execution. Failures raise MongoSenseError.

Gate	What it checks
Gate 1	Static analysis: banned operators, missing $ prefix on stage names, $graphLookup without maxDepth, $skip above threshold
Gate 2	explain() pre-flight: rejects pipelines that fail the docs-examined/returned ratio, an absolute docs-examined ceiling, or a collection-size ceiling (all independently configurable via on_inefficient_query)
Gate 3	Constraint injection: appends $limit if none is present, clamps any existing $limit above default_limit

Gates run sequentially. A query that passes all three gates is safe to execute.

from mongosense import MongoSenseError

try:
    result = await ms.query("Drop the orders collection")
except MongoSenseError as e:
    print(e.gate)       # 1
    print(e.violation)  # "banned operator '$drop'"

---

LLM Backends & Fallback

MongoSense doesn't hard-code a provider. Any LLMBackend implementation works — GeminiBackend and OpenAICompatibleBackend ship built-in.

from mongosense import GeminiBackend, OpenAICompatibleBackend

# Gemini
llm_backend = GeminiBackend(
    api_key="YOUR_GEMINI_API_KEY",
    models=["gemini-2.5-flash-lite", "gemini-2.0-flash"],
)

# OpenAI, or any OpenAI-compatible Chat Completions endpoint
# (Kimi, DeepSeek, Groq, Together AI, OpenRouter, local Ollama/vLLM, LM Studio, ...)
llm_backend = OpenAICompatibleBackend(
    base_url="https://api.openai.com/v1",
    api_key="YOUR_API_KEY",
    models=["gpt-4o-mini"],
)

Each backend takes a models list. If a call fails with a rate-limit error, MongoSense retries automatically with the next model in the list — this fallback logic is provider-agnostic and lives outside the backend implementations.

To use a different provider entirely, implement the LLMBackend protocol — a single async method, generate(model, system_prompt, user_message) -> LLMResponse — and pass an instance as llm_backend.

---

Logging

MongoSense uses structlog but does not configure it itself — a library shouldn't override your application's logging setup. Configure structlog yourself before using MongoSense:

import logging
import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.dev.ConsoleRenderer(),  # or structlog.processors.JSONRenderer() in production
    ],
    wrapper_class=structlog.make_filtering_bound_logger(logging.DEBUG),
)

If you skip this, structlog's own default configuration applies. MongoSense logs at debug for gate-by-gate detail (schema source, graph inference, safety checks) and info for query lifecycle events (query start/done, HITL, token usage per call).

---

Known Limitations

• Cross-collection joins — $lookup is blocked by default (required for Atlas M0). Questions that need data from multiple collections will return a best-effort single-collection answer with a lower confidence score.
• Date strings — If your dataset stores timestamps as strings rather than BSON Date objects, date-range queries may return unexpected results.
• Query memory doesn't dedupe — asking the same question twice (even worded identically) stores two separate entries. This bloats the few-shot prompt over time and can bias retrieval toward whichever duplicate happens to score higher.
• Redaction is top-level only — redacted_fields in CollectionConfig doesn't recurse into nested documents. PII in sub-documents (e.g. address.email) is not stripped from formatter input.
• Fixed read preference — secondaryPreferred is hardcoded in the Mongo adapter. Standalone (non-replica-set) MongoDB deployments will error on connect.
• HITL state is in-process only — resume(pending_id, ...) requires the same process that returned the original requires_approval=True result. Pending approvals are not shared across a load-balanced or multi-worker deployment.

---

Planned Features

Forward-looking work, not yet implemented. Listed here so expectations are set correctly — none of this ships today.

• Follow-up / context memory — carry conversational context across queries so a user can ask "and what about last month?" without restating the full question.
• Query memory deduplication — collapse repeated identical questions into a single stored entry instead of accumulating duplicates.
• Multi-step agentic query execution — a planner + builder pattern to correctly answer cross-collection questions without relying on $lookup.
• Deterministic date-arithmetic safety check — statically detect $dateDiff/$dateAdd/etc. applied to string-typed date fields and reject or auto-cast, instead of relying on a prompt rule alone.
• HITL rejection self-heal — when Gate 2 rejects a pipeline, feed the rejection back to the LLM for one corrected retry instead of failing outright.
• Cursor-based pagination — replace $skip/$limit paging with cursor-based paging for large result sets.
• Dry-run cost preview — surface an explain()-based cost estimate (docs examined, plan shape) when execute=False, instead of just returning the pipeline.
• Nested-field redaction — extend redacted_fields to support dotted paths so PII in sub-documents is also stripped.
• Configurable read preference — expose read_preference in MongoSenseConfig instead of hardcoding secondaryPreferred.
• Cross-process HITL state — persist pending HITL approvals to a collection so resume() works across load-balanced deployments.
• Audit log — append-only record of every query() call (question, pipeline, outcome) for traceability.
• Authorized-approver HITL — require a different, authorized person (not the requester) to approve heavy or intensive queries before execution.
• MCP server integration — expose MongoSense as an MCP tool with a two-tier standard/admin agent access model.