Lab 04 — Function-Calling Asset DB

Build a ReAct-style LLM agent that answers natural-language questions about a live pump station database by calling structured SQL tools — no hallucination, every answer grounded in real data.

What you will build:

SQLite asset database (pumps, sensors, alarms across 3 sites)
OpenAI function-calling tool definitions + parameterized SQL executor
Agent loop: LLM → tool calls → results → LLM → final answer
FastAPI server on port 8002 exposing the agent and raw DB endpoints

Prerequisites: Lab 01 server running (Ollama with qwen3-coder:30b).

Step 1 — Populate the database

cd "AI Specialist/lab-04-function-calling"
/Users/s0x/anaconda3/bin/python populate_db.py

Expected output:

Database ready: .../data/assets.db
  pumps           : 8 rows
  sensors         : 24 rows
  alarms (total)  : 10 rows
  alarms (active) : 8 rows

The database contains:

8 pumps across Station-A (3), Station-B (3), Station-C (2)
24 sensors — vibration, temperature, pressure, flow per pump
10 alarms — 8 active (including 1 critical on PUMP-007), 2 resolved

Step 2 — Run the agent CLI

Single query:

/Users/s0x/anaconda3/bin/python agent.py "Which pumps have active critical alarms?"

Expected (2–3 LLM calls, ~15–20s on warm Ollama):

PUMP-007 at Station-C has an active critical alarm: "Vibration critical: 9.4 mm/s
(emergency limit 10.0 mm/s) — imminent bearing failure" triggered at 2026-05-27T08:55:00.

[2 LLM call(s) | 2 tool call(s) | 18.4s]

Show tool calls with --verbose:

/Users/s0x/anaconda3/bin/python agent.py --verbose "What is the system health summary?"

Verbose stderr output (tool calls):

[iter 1] Calling LLM...
  → get_system_summary({})
  ← {"total_pumps": 8, "pump_status": {"online": 6, "offline": 1, "maintenance": 1}, ...
[iter 2] Calling LLM...

Interactive mode:

/Users/s0x/anaconda3/bin/python agent.py --interactive

Asset Agent — interactive mode. Type 'quit' to exit.

Query> How many pumps are online?
Query> What is the vibration status across all pumps?
Query> Tell me everything about PUMP-007
Query> quit

Step 3 — Start the server

/Users/s0x/anaconda3/bin/python asset_server.py

Expected:

Asset server ready — DB: .../data/assets.db
INFO:     Uvicorn running on http://0.0.0.0:8002

REST endpoints (no LLM required)

# System health
curl http://localhost:8002/health | python -m json.tool

# Aggregate stats
curl http://localhost:8002/stats | python -m json.tool

# All online pumps
curl "http://localhost:8002/pumps?status=online" | python -m json.tool

# Active critical alarms
curl "http://localhost:8002/alarms?severity=critical" | python -m json.tool

# Full detail for PUMP-007
curl http://localhost:8002/pumps/PUMP-007 | python -m json.tool

# All vibration sensors
curl "http://localhost:8002/sensor-readings?sensor_type=vibration" | python -m json.tool

Natural-language query endpoint

curl -X POST http://localhost:8002/query \
  -H "Content-Type: application/json" \
  -d '{"query": "Which pump needs immediate attention and why?"}' \
  | python -m json.tool

Expected response shape:

{
  "query": "Which pump needs immediate attention and why?",
  "answer": "PUMP-007 at Station-C needs immediate attention. It has a critical alarm...",
  "tool_calls": [
    {"tool": "get_system_summary", "args": {}, "result": {...}},
    {"tool": "get_pump_details",   "args": {"pump_id": "PUMP-007"}, "result": {...}}
  ],
  "iterations": 2,
  "elapsed_s": 19.3
}

Step 4 — Run the tests

# DB + tool executor tests only (fast, no server needed)
/Users/s0x/anaconda3/bin/python -m pytest test_lab4.py -v -k "not server and not agent"

# All tests except LLM (server must be running)
/Users/s0x/anaconda3/bin/python -m pytest test_lab4.py -v -k "not agent"

# Full suite (server + Ollama running)
/Users/s0x/anaconda3/bin/python -m pytest test_lab4.py -v

Expected (full suite):

test_lab4.py::test_db_pumps_count              PASSED
test_lab4.py::test_db_sensors_count            PASSED
test_lab4.py::test_db_alarms_count             PASSED
test_lab4.py::test_db_critical_alarm_active    PASSED
test_lab4.py::test_tool_list_pumps_all         PASSED
test_lab4.py::test_tool_list_pumps_online      PASSED
test_lab4.py::test_tool_list_pumps_station_a   PASSED
test_lab4.py::test_tool_get_pump_details_found PASSED
test_lab4.py::test_tool_get_pump_details_not_found PASSED
test_lab4.py::test_tool_list_alarms_unresolved PASSED
test_lab4.py::test_tool_get_system_summary     PASSED
test_lab4.py::test_server_health               PASSED
test_lab4.py::test_server_stats                PASSED
test_lab4.py::test_server_pumps_online         PASSED
test_lab4.py::test_server_alarms_critical      PASSED
test_lab4.py::test_server_pump_detail          PASSED
test_lab4.py::test_server_pump_not_found       PASSED
test_lab4.py::test_agent_critical_alarms       PASSED
18 passed in 25.4s

Extension exercises

Add a new tool — schedule_maintenance(pump_id, date) that writes a new row to an maintenance_schedule table. Note how the agent automatically discovers and uses the new tool description.
Parallel tool calls — ask "Compare the vibration readings of PUMP-004 and PUMP-007". Watch whether the model issues two get_pump_details calls in a single turn (parallel) or sequentially. Try a model that supports parallel tool calling.
Integrate with Lab 2 — add a query_documentation(question) tool that calls the Lab 2 RAG server at http://localhost:8001/query. The agent can now answer both "what is the vibration reading on PUMP-007?" (SQL tool) and "what does the pump operations manual say about bearing failure?" (RAG tool) in the same conversation.
Streaming tool calls — change "stream": True in agent.py and parse the SSE chunks to show partial responses. Handle the case where tool call arguments are streamed in fragments.
Write a tool — add export_alarm_report(format="csv") that returns a CSV string of all active alarms. Show how the agent synthesises this with text in the final answer.