Let SQL do the counting and AI explain the findings
Calculate civic-data measures in SQL, then give AI a source-linked fact bundle. Reproduce a real example and see how a vendor join can inflate a total.
A civic-research report often needs both a calculation and an explanation. The calculation might count records or total a defined amount. The explanation should tell the reader what that number describes, which sources support it, and what questions it cannot answer.
Give those jobs separate steps. SQL calculates over a documented collection. A model receives the resulting figures, source references, and scope, then prepares a narrative that a reviewer can inspect.
This walkthrough uses a real, deliberately small source extract and locally executed calculations. It also includes a model-written brief based on those results. There was no separately benchmarked external model API call, and the example does not claim that an AI system retrieved the underlying data from AICD API.
Define the collection before asking for a conclusion
The supplied JSON contains three manually transcribed Fort Worth authorization records dated June 24, 2025. They are M&C 25-0563, M&C 25-0573, and M&C 25-0574.
The question is how many selected actions and vendor-name strings the extract contains, and what sum results from their stated annual authorization amounts. The question is not which vendor leads the market or how much the city actually paid.
Save the input collection with a manifest. The included program records a SHA-256 hash, the manual-transcription origin, the research date, and the record count. That makes it possible to tell whether two reports used the same input bytes. A hash identifies the input; it does not certify that the transcription is correct.
Put the amount on the record that owns it
The example stores authorizations in one table and their vendor-name relationships in another. Each authorization's amount appears once in the actions table.
The first calculations are:
SELECT COUNT(*) AS selected_actions,
SUM(amount_cents) AS selected_annual_amount_cents
FROM actions;
SELECT COUNT(DISTINCT vendor) AS named_vendor_strings
FROM action_vendors;
On the supplied extract, the executed SQLite results are three selected actions, four distinct vendor-name strings, and 313,500,000 cents in stated annual amounts.
Keep the unit explicit. These are vendor-name strings in the reviewed extract, not a comprehensive legal-entity registry. The amount is the sum for selected authorization records, not payments, revenue, or a whole-meeting total.
Show how a correct-looking join can change the answer
One authorization names two vendors under a combined annual amount. Joining the action table to every vendor row and summing immediately repeats that action's amount:
-- Deliberately incorrect for an authorization-level total.
SELECT SUM(a.amount_cents)
FROM actions AS a
JOIN action_vendors AS v ON a.id = v.action_id;
The executed result is 325,500,000 cents. It exceeds the action-level sum by 12,000,000 cents, or $120,000, because the two-vendor action appears twice after the join.
This is not an arithmetic mistake a better prose prompt can fix. The query is calculating at the wrong unit. Correct the data operation before asking a model to explain its result.
Do not replace the query with SUM(DISTINCT amount_cents). Two unrelated actions can legitimately have the same amount. Deduplicate by the actual record identity, or calculate directly from the table that contains one row per action.
Pass a fact bundle to the model
The program writes analysis_results.json with the computed values, a scope description, a measure label, and source references. A shortened view of the actual calculated fields is:
{
"selected_actions": 3,
"named_vendor_strings": 4,
"selected_annual_authorization_amount_cents": 313500000,
"incorrect_vendor_join_total_cents": 325500000,
"claim_type": "sum of stated annual amounts for selected actions, NOT payments or revenue"
}
Keep the source references and full scope alongside these fields when supplying the model. The narrative should not have to infer what the amounts mean from column names alone.
For a larger workflow, include the query version and any reviewed exclusions. If some records were omitted because their amount type could not be established, say so in the fact bundle rather than making the resulting total look comprehensive.
Ask for an explanation, not a replacement calculation
A reusable prompt is:
Write a short research brief using only the supplied calculation results and source references. Describe the selected collection and the unit being counted. Preserve the distinction between authorization amounts and payments. Do not allocate a combined amount between vendors, infer exercised renewal options, or extrapolate to the whole city. Explain the demonstrated join error and link factual findings to the corresponding source records. Identify any question the inputs do not answer.
The model-written example prepared from the executed output reads:
In the three selected Fort Worth authorizations dated June 24, 2025, the stated annual amounts sum to $3,135,000. Four distinct vendor-name strings appear across those records. One action names two vendors under a single combined annual ceiling, so repeating the amount once per vendor overstates the selected total by $120,000. These figures describe the selected authorizations, not payments or vendor revenue.
That paragraph is an explanation of the local calculation. It is not a separate measurement of model accuracy or evidence that an autonomous agent completed the research.
Review the meaning as well as the digits
A reviewer should compare every numeric statement with the fact bundle and every source-dependent statement with its cited record. Matching digits alone is insufficient: changing “annual authorization amount” to “money spent” would preserve the number while changing the claim.
A production application can reject unfamiliar source IDs or missing scope fields before generating prose. It can also require the model to return structured claim references alongside its narrative. Those are useful design checks, but they do not eliminate the need to review the explanation.
Keep the generation step unable to modify the input collection or approved calculations. When a reviewer finds a data or query error, fix that upstream component, regenerate the fact bundle, and produce a new report version with the change recorded.
Reproduce the calculation locally
Save the source files from the example code section below in one directory and run:
python aicd_examples.py test
python aicd_examples.py analyze
The program uses Python's standard library and SQLite. It writes the calculation JSON, a source-linked detail CSV, a relationship-query result, and a data manifest. The example code section below holds the source extract and program as code blocks to save as files; the prepared narrative is the example quoted above.
The package's thirteen Python tests include the correct sum and the deliberately inflated join result. Both assertions passed under Python 3.13.5 and SQLite 3.46.1. They establish behavior on this controlled extract, not correctness for every possible civic dataset.
Connect the same pattern to a larger AICD API workflow
Use the AICD API reference to retrieve an appropriately scoped supported collection. Preserve original responses and transform them into a documented analytical model before calculating.
Keep the stages distinct: retrieval establishes what was obtained, review establishes the record meaning and identity, SQL calculates the chosen measures, and the model prepares an explanation. A scheduled application can repeat that sequence while retaining the inputs and report versions.
The useful result is a narrative whose important numbers can be reproduced and whose claims can be checked. Let SQL finish the counting task, then give the model a clearly bounded explanation task rather than asking one prompt to do everything invisibly.
Example code
Save each code block below using its heading as the filename, in one directory. The dataset is a manually transcribed extract of three official Fort Worth authorization records, not an AICD API response. Thirteen Python unit tests passed using Python 3.13.5 and SQLite 3.46.1. The program also includes a local serve command that binds only to 127.0.0.1:8765 by default and is not for production.
authorizations.json
[
{
"record_id": "fw-mc-25-0563",
"agency": "City of Fort Worth",
"action_date": "2025-06-24",
"category": "Telecommunications",
"annual_ceiling_cents": 155000000,
"vendors": ["AT&T Corporation"],
"source_url": "https://fortworthgov.legistar.com/LegislationDetail.aspx?G=5D78EE16-987A-40AB-8FB5-726D3DBA5713&GUID=98F18D76-29EC-47B8-98A4-C0E497FFE570&ID=7440706&Options=&Search=",
"source_file": "M&C 25-0563",
"measure": "stated annual authorization amount",
"source_status": "Passed"
},
{
"record_id": "fw-mc-25-0573",
"agency": "City of Fort Worth",
"action_date": "2025-06-24",
"category": "HVAC equipment and installation",
"annual_ceiling_cents": 12000000,
"vendors": ["Baker Distributing Company LLC", "EPMA Corporation DBA Johnstone Supply"],
"source_url": "https://fortworthgov.legistar.com/LegislationDetail.aspx?G=5D78EE16-987A-40AB-8FB5-726D3DBA5713&GUID=41CC6E8D-6E48-4246-A420-D2C696EE9F55&ID=7440754&Options=&Search=",
"source_file": "M&C 25-0573",
"measure": "combined annual authorization ceiling",
"source_status": "Passed"
},
{
"record_id": "fw-mc-25-0574",
"agency": "City of Fort Worth",
"action_date": "2025-06-24",
"category": "HVAC services",
"annual_ceiling_cents": 146500000,
"vendors": ["Texas AirSystems, LLC"],
"source_url": "https://fortworthgov.legistar.com/LegislationDetail.aspx?G=5D78EE16-987A-40AB-8FB5-726D3DBA5713&GUID=14801F53-DE07-48C6-8022-A30072C5EEE4&ID=7440760&Options=&Search=",
"source_file": "M&C 25-0574",
"measure": "combined annual authorization ceiling",
"source_status": "Passed"
}
]
aicd_examples.py
"""Local article examples. Standard library only; not a production web server."""
from __future__ import annotations
import argparse
import csv
import hashlib
import html
import json
import sqlite3
import sys
import unittest
from datetime import date, timedelta
from decimal import Decimal
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, urlparse
from urllib.request import Request, urlopen
ROOT = Path(__file__).resolve().parent
PUBLIC_SAMPLE = 'https://aicdapi.com/api/v1/public/agenda-hits?limit=3'
MAX_RESPONSE = 1_048_576
def https_url(value: object) -> str:
if not isinstance(value, str):
raise ValueError('Source URL must be a string')
parsed = urlparse(value)
if parsed.scheme != 'https' or not parsed.netloc or parsed.username or parsed.password:
raise ValueError('Only credential-free HTTPS source URLs are accepted')
return value
def read_records() -> list[dict]:
values = json.loads((ROOT/'authorizations.json').read_text())
seen = set()
for row in values:
if row['record_id'] in seen:
raise ValueError('Duplicate source-extract record ID')
seen.add(row['record_id'])
https_url(row['source_url'])
date.fromisoformat(row['action_date'])
if type(row['annual_ceiling_cents']) is not int or row['annual_ceiling_cents'] < 0:
raise ValueError('Amount must be nonnegative integer cents')
return values
def sample_cards(payload: object) -> list[dict]:
"""Reference-based adapter. Live API schema has NOT been reverified."""
if not isinstance(payload, dict) or not isinstance(payload.get('data'), list):
raise ValueError('Expected a data array; inspect the current API reference')
if len(payload['data']) > 3:
raise ValueError('Unexpected public-sample size; inspect the response')
cards = []
for row in payload['data']:
if not isinstance(row, dict) or not isinstance(row.get('title'), str):
raise ValueError('Each hit needs a title')
source = https_url(row.get('sourceUrl'))
for key in ('summary','evidenceQuote'):
if key in row and row[key] is not None and not isinstance(row[key], str):
raise ValueError(f'{key} must be text when present')
cards.append({'title':row['title'], 'summary':row.get('summary') or '',
'evidence':row.get('evidenceQuote') or '', 'source_url':source})
return cards
def fetch_public_cards() -> list[dict]:
request = Request(PUBLIC_SAMPLE, headers={'Accept':'application/json'})
with urlopen(request, timeout=10) as response:
if urlparse(response.geturl()).hostname != 'aicdapi.com':
raise ValueError('Unexpected redirect host')
raw = response.read(MAX_RESPONSE+1)
if len(raw) > MAX_RESPONSE:
raise ValueError('Sample response exceeded the local size limit')
return sample_cards(json.loads(raw))
def curated_cards() -> list[dict]:
return [{'title':r['source_file']+' | '+r['category'],
'summary':r['agency']+'; source action date '+r['action_date']+
'; selected authorization record, not payment data.',
'evidence':'Named in the source: '+ '; '.join(r['vendors']),
'source_url':r['source_url']} for r in read_records()]
def render(cards: list[dict], query: str = '', label: str = 'Curated historical source extract') -> bytes:
escape = html.escape
q = query.casefold()
selected = [c for c in cards if q in (c['title']+' '+c['summary']+' '+c['evidence']).casefold()]
blocks = ''.join('<article><h2>'+escape(c['title'])+'</h2><p>'+escape(c['summary'])+
'</p><p>'+escape(c['evidence'])+'</p><a rel="noopener noreferrer" href="'+
escape(https_url(c['source_url']), quote=True)+'">Open official source</a></article>'
for c in selected)
if not selected:
blocks = '<p>No matches in this small loaded sample. This is not a coverage finding.</p>'
doc = '''<!doctype html><html lang="en"><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>Civic evidence cards</title><style>body{font:17px/1.55 system-ui;max-width:850px;margin:3rem auto;padding:0 1rem}article{border:1px solid #bbb;border-radius:8px;margin:1rem 0;padding:1rem}input,button{font:inherit;padding:.4rem}label{display:block}</style><h1>Civic evidence cards</h1>'''
doc += '<p>'+escape(label)+'. Source-linked learning example, not a complete city dataset.</p>'
doc += '<form><label for="q">Filter the loaded cards</label><input id="q" name="q" value="'+escape(query,quote=True)+'"><button>Filter</button></form>'
doc += '<p>'+str(len(selected))+' card(s) shown.</p>'+blocks+'</html>'
return doc.encode()
def database() -> sqlite3.Connection:
db = sqlite3.connect(':memory:')
db.execute('PRAGMA foreign_keys=ON')
db.executescript('''CREATE TABLE actions(id TEXT PRIMARY KEY, agency TEXT NOT NULL,
action_date TEXT NOT NULL, category TEXT NOT NULL, amount_cents INTEGER NOT NULL,
source_url TEXT NOT NULL); CREATE TABLE action_vendors(action_id TEXT NOT NULL,
vendor TEXT NOT NULL, PRIMARY KEY(action_id,vendor),
FOREIGN KEY(action_id) REFERENCES actions(id));''')
for r in read_records():
db.execute('INSERT INTO actions VALUES (?,?,?,?,?,?)',
(r['record_id'],r['agency'],r['action_date'],r['category'],r['annual_ceiling_cents'],r['source_url']))
db.executemany('INSERT INTO action_vendors VALUES (?,?)',[(r['record_id'],v) for v in r['vendors']])
db.commit()
return db
def analyze() -> dict:
with database() as db:
count, total = db.execute('SELECT COUNT(*),SUM(amount_cents) FROM actions').fetchone()
vendors = db.execute('SELECT COUNT(DISTINCT vendor) FROM action_vendors').fetchone()[0]
naive = db.execute('SELECT SUM(a.amount_cents) FROM actions a JOIN action_vendors v ON a.id=v.action_id').fetchone()[0]
return {'scope':'three deliberately selected Fort Worth authorizations dated 2025-06-24; not a census',
'selected_actions':count,'named_vendor_strings':vendors,
'selected_annual_authorization_amount_cents':total,
'incorrect_vendor_join_total_cents':naive,
'claim_type':'sum of stated annual amounts for selected actions, NOT payments or revenue',
'source_records':[{'id':r['record_id'],'source_url':r['source_url']} for r in read_records()]}
def graph_rows(vendor: str) -> list[dict]:
"""A typed two-hop graph expressed as relational joins, preserving edge evidence."""
with database() as db:
rows = db.execute('''SELECT a.agency,a.id,v.vendor,a.source_url
FROM actions a JOIN action_vendors v ON a.id=v.action_id
WHERE v.vendor=? ORDER BY a.id''',(vendor,)).fetchall()
return [{'agency':a,'action_id':i,'relation':'names_vendor','vendor':v,'evidence_url':u}
for a,i,v,u in rows]
def write_outputs() -> None:
report = analyze()
(ROOT/'analysis_results.json').write_text(json.dumps(report, indent=2)+'\n')
(ROOT/'graph_results.json').write_text(json.dumps(graph_rows('EPMA Corporation DBA Johnstone Supply'),indent=2)+'\n')
with (ROOT/'authorization_report.csv').open('w',newline='') as f:
writer = csv.DictWriter(f, fieldnames=['record_id','action_date','category','annual_ceiling_cents','source_url'])
writer.writeheader()
for r in read_records(): writer.writerow({k:r[k] for k in writer.fieldnames})
digest = hashlib.sha256((ROOT/'authorizations.json').read_bytes()).hexdigest()
(ROOT/'data_manifest.json').write_text(json.dumps({'fixture_sha256':digest,
'origin':'Manual transcription of three official Legistar authorization records',
'api_response':False,'research_date':'2026-10-03','record_count':len(read_records())},indent=2)+'\n')
def serve(port: int, live: bool) -> None:
cards = fetch_public_cards() if live else curated_cards()
label = 'AICD public sample fetched at startup' if live else 'Curated historical source extract'
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
parsed = urlparse(self.path)
if parsed.path != '/':
self.send_error(404); return
body = render(cards, parse_qs(parsed.query).get('q',[''])[0], label)
self.send_response(200)
self.send_header('Content-Type','text/html; charset=utf-8')
self.send_header('Content-Length',str(len(body)))
self.send_header('X-Content-Type-Options','nosniff')
self.send_header('Content-Security-Policy',"default-src 'none'; style-src 'unsafe-inline'; form-action 'self'; base-uri 'none'; frame-ancestors 'none'")
self.end_headers(); self.wfile.write(body)
print(f'Local-only example: http://127.0.0.1:{port}',flush=True)
ThreadingHTTPServer(('127.0.0.1',port),Handler).serve_forever()
class Tests(unittest.TestCase):
def test_fixture(self): self.assertEqual(len(read_records()),3)
def test_integer_totals(self): self.assertEqual(analyze()['selected_annual_authorization_amount_cents'],313500000)
def test_join_inflation(self): self.assertEqual(analyze()['incorrect_vendor_join_total_cents'],325500000)
def test_graph(self): self.assertEqual(graph_rows('EPMA Corporation DBA Johnstone Supply')[0]['action_id'],'fw-mc-25-0573')
def test_unknown_vendor(self): self.assertEqual(graph_rows("' OR 1=1 --"),[])
def test_escaping(self):
page = render([{'title':'<script>alert(1)</script>','summary':'','evidence':'','source_url':'https://example.com'}]).decode()
self.assertNotIn('<script>',page); self.assertIn('<script>',page)
def test_unsafe_url(self):
with self.assertRaises(ValueError): https_url('javascript:alert(1)')
def test_wrong_envelope(self):
with self.assertRaises(ValueError): sample_cards({'results':[]})
def test_empty_sample(self): self.assertEqual(sample_cards({'data':[]}),[])
def test_filter(self): self.assertIn(b'No matches',render(curated_cards(),'zzzzzz'))
def test_adapter(self): self.assertEqual(sample_cards({'data':[{'title':'Synthetic test','sourceUrl':'https://example.com','summary':'x'}]})[0]['summary'],'x')
def test_review_date(self): self.assertEqual(date(2026,4,12)-timedelta(days=90),date(2026,1,12))
def test_payment_fixture(self): self.assertEqual(sum(map(Decimal,['8000.00','1200.00','-200.00']),Decimal('0')),Decimal('9000.00'))
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument('command',choices=['test','analyze','graph','serve'])
parser.add_argument('--port',type=int,default=8765)
parser.add_argument('--live',action='store_true')
args = parser.parse_args()
if args.command == 'test': unittest.main(argv=[sys.argv[0]],verbosity=2)
elif args.command == 'analyze': write_outputs(); print(json.dumps(analyze(),indent=2))
elif args.command == 'graph': print(json.dumps(graph_rows('EPMA Corporation DBA Johnstone Supply'),indent=2))
else: serve(args.port,args.live)