# Mobility Operations Agent Platform

Mobility Operations Agent Platform is a Python service for grounded analysis over synthetic electric-taxi operations data. It combines an LLM tool-calling loop, policy retrieval, structured warehouse tools, workflow guardrails, evaluation gates, MCP exposure, and a local Spark SQL data pipeline.

The repository focuses on engineering boundaries that matter when an AI feature can influence operations: models do not receive arbitrary database access, retrieved policy text is cited, personal-data requests are blocked, tool calls are traced, and workflow writes require explicit human approval.

No DRIVR affiliation, real taxi data, production deployment, or autonomous operational action is claimed.

## What the project demonstrates

- a multi-step agent loop with function calling and an Ollama adapter tested against `qwen2.5:7b-instruct`
- retrieval-augmented generation using a versioned mobility-policy corpus and BM25 ranking
- allowlisted tools for aggregated trip metrics, charging risk, policy retrieval, and draft case creation
- input, output, SQL, personal-data, prompt-injection, and side-effect guardrails
- FastAPI endpoints and an official MCP Python SDK server over the same service functions
- a DuckDB analytical warehouse seeded with deterministic synthetic trips and charging sessions
- a local PySpark pipeline that validates trip data and writes silver and gold Parquet datasets
- deterministic regression evaluations for tool choice, retrieval, approval boundaries, and refusals
- unit, integration, API, MCP, Spark, lint, Docker, and GitHub Actions checks

## Architecture

```mermaid
flowchart LR
  U[Operations question] --> G[Input guardrails]
  G --> R[BM25 policy retrieval]
  R --> A[Agent loop]
  A --> L[Ollama or deterministic model]
  L --> T{Allowlisted tool call}
  T --> W[DuckDB warehouse]
  T --> P[Policy index]
  T --> C[Human-approved draft case]
  W --> A
  P --> A
  C --> A
  A --> O[Grounded answer and tool trace]
  M[MCP clients] --> T
  F[FastAPI clients] --> A
  S[Spark SQL pipeline] --> D[Silver and gold Parquet]
```

The agent never executes model-generated SQL. Tool arguments are validated with Pydantic and compiled into fixed, parameterised analytical queries. See [docs/architecture.md](docs/architecture.md) for the request and data flows.

## Quick start

Prerequisites:

- Python 3.11 or 3.12
- `uv`
- Java 17 for the optional local Spark pipeline
- Ollama with `qwen2.5:7b-instruct` for the live LLM path

Install and seed the deterministic local profile:

```bash
uv sync --extra dev --extra spark
uv run mobility-ai seed --rows 10000
uv run mobility-ai profile
```

Ask through the reproducible deterministic orchestration model:

```bash
uv run mobility-ai ask "What is the average pickup wait by city?"
```

Run the same agent loop with a local LLM:

```bash
ollama pull qwen2.5:7b-instruct
uv run mobility-ai ask --provider ollama "Which depot-day has the highest charging failure rate?"
```

The deterministic provider exists for repeatable tests and CI. It is not presented as an LLM. The live smoke report is produced separately with the Ollama provider.

## FastAPI

Start the API:

```bash
uv run uvicorn mobility_ai.api:app --host 0.0.0.0 --port 8000
```

Then inspect the service and ask a question:

```bash
curl http://127.0.0.1:8000/health
curl -X POST http://127.0.0.1:8000/v1/chat \
  -H 'content-type: application/json' \
  -d '{"question":"What threshold applies to accessible-trip pickup waits?"}'
```

API details are documented in [docs/api.md](docs/api.md).

## MCP server

The MCP server uses `mcp.server.fastmcp.FastMCP` from the official Python SDK and exposes aggregated metric, charging-risk, and policy-retrieval tools:

```bash
uv run mobility-ai-mcp
```

It defaults to stdio so an MCP client controls the transport lifecycle. The MCP tools call the same warehouse and retrieval functions used by the FastAPI agent.

## Spark SQL pipeline

Export the seeded warehouse and build local silver/gold Parquet tables:

```bash
uv run mobility-ai export-spark-source --destination data/runtime/trips.csv
uv run mobility-ai spark-pipeline \
  --source data/runtime/trips.csv \
  --output data/runtime/lakehouse
```

The pipeline runs Spark locally, not on Databricks or a production cluster. It validates required fields and non-negative measures, partitions the silver table by city, and writes daily city metrics to the gold layer.

## Evaluation and tests

Run the reproducible quality gates:

```bash
uv run ruff check .
uv run pytest -m "not spark and not live_llm" \
  --cov=mobility_ai --cov-report=term-missing --cov-fail-under=85
uv run pytest -m spark
uv run mobility-ai eval --cases evals/cases.json --report-dir reports
```

Run the live local-model smoke scenarios:

```bash
uv run python scripts/live_ollama_smoke.py
```

The deterministic evaluation set covers metric tools, policy retrieval, tool chaining, human approval, prompt injection, personal-data requests, direct side effects, and arbitrary-SQL requests. See [docs/evaluation.md](docs/evaluation.md) and the generated reports under `reports/`.

## Repository layout

```text
src/mobility_ai/
  agent.py           bounded tool-calling loop and RAG context
  api.py             FastAPI surface
  evaluation.py      regression evaluation runner and reports
  guardrails.py      prompt, personal-data, and side-effect policies
  llm.py             deterministic and Ollama providers
  mcp_server.py      official MCP SDK tools
  retrieval.py       BM25 policy index
  spark_pipeline.py  local silver/gold Spark SQL pipeline
  tools.py           schemas, validation, and tool allowlist
  warehouse.py       DuckDB schema, seed data, and analytical queries
data/policies/       versioned synthetic operating policies
evals/               regression cases
tests/               unit and integration tests
```

## Limitations

- All trip, charging, policy, and workflow data are synthetic.
- The repository has no connection to DRIVR systems and does not model their internal architecture.
- Ollama verification is a local model smoke test, not a hosted production LLM deployment.
- Spark runs in local mode and writes Parquet; no Databricks workspace or distributed cluster is claimed.
- DuckDB is appropriate for this reproducible portfolio scope, not a substitute for a production warehouse governance model.
- Draft operations cases are the only write workflow and require explicit human approval.

See [docs/limitations.md](docs/limitations.md) for the complete evidence boundary.