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

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 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:

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:

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

Run the same agent loop with a local LLM:

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:

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

Then inspect the service and ask a question:

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.

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:

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:

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:

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:

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 and the generated reports under reports/.

Repository layout

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 for the complete evidence boundary.