MexicoCSV analyst
Ask questions about a quarter of e-commerce orders — "which region had the
Showcase — CSV analyst
Ask questions about a quarter of e-commerce orders — "which region had the
highest average revenue?", "total office units sold in May?" — and the model
answers by chaining tool calls: list_columns to learn the schema, then
filter_rows and aggregate as many times as the question needs. It never
sees the raw table, which is exactly the access pattern you want against a
real warehouse table that doesn't fit in a prompt.
Run
bash bootstrap-secrets.sh # reads ../../../../.env, writes secrets/
docker compose up --build # default: PROVIDER=openaiOpen http://localhost:3000.
To run against Gemini instead:
PROVIDER=gemini docker compose up --buildWhat's where
backend/ai_openai.py— the embedded orders table, the three tools, the dispatch dict, and the bounded agent loopbackend/ai_gemini.py— same tools and loop, Gemini spellingbackend/main.py— identical FastAPI loader; reads PROVIDER and dispatchesfrontend/app/page.tsx— textarea + resultdocker-compose.yml— two services, secrets mounted from./secrets/
The loop is capped at 8 iterations — an agent loop without a bound is an outage waiting for a confused model.
Stop
docker compose downRun locally
Download the project as a ZIP and run it with Docker. Brings up a FastAPI backend + Next.js frontend on localhost:3000.
unzip csv-analyst.zip
cd csv-analyst
bash bootstrap-secrets.sh # one-time: pulls API keys into ./secrets
docker compose up --build # default provider: openai
# or: PROVIDER=gemini docker compose up --buildType some input, pick a provider, and run the same code shown in Source against the live API. Sign-in required.
The same modules the Run button hits. The whole project (frontend, Dockerfile, compose) is in the ZIP under README.
backend/ai_openai.py
"""Week 6 - Showcase 1 (OpenAI): question-answering over a table with chained tools.
The model never sees the whole dataset. It discovers the schema, filters,
and aggregates through three tools, chaining as many calls as the question
needs — the same access pattern you'd want against a real warehouse table
that doesn't fit in a prompt.
"""
import json
from openai import OpenAI
_client = OpenAI()
_MODEL = "gpt-5.4-nano"
# A quarter of e-commerce orders. Small enough to read, rich enough that
# real questions take several tool calls to answer.
_ROWS = [
{"order_id": 1001, "region": "north", "category": "electronics", "units": 3, "unit_price": 199.0, "month": "april"},
{"order_id": 1002, "region": "south", "category": "furniture", "units": 1, "unit_price": 649.0, "month": "april"},
{"order_id": 1003, "region": "north", "category": "office", "units": 12, "unit_price": 8.5, "month": "april"},
{"order_id": 1004, "region": "west", "category": "electronics", "units": 2, "unit_price": 349.0, "month": "april"},
{"order_id": 1005, "region": "south", "category": "office", "units": 30, "unit_price": 4.2, "month": "may"},
{"order_id": 1006, "region": "west", "category": "furniture", "units": 2, "unit_price": 289.0, "month": "may"},
{"order_id": 1007, "region": "north", "category": "electronics", "units": 1, "unit_price": 999.0, "month": "may"},
{"order_id": 1008, "region": "east", "category": "office", "units": 6, "unit_price": 12.0, "month": "may"},
{"order_id": 1009, "region": "east", "category": "electronics", "units": 4, "unit_price": 149.0, "month": "june"},
{"order_id": 1010, "region": "south", "category": "electronics", "units": 2, "unit_price": 449.0, "month": "june"},
{"order_id": 1011, "region": "west", "category": "office", "units": 20, "unit_price": 6.8, "month": "june"},
{"order_id": 1012, "region": "north", "category": "furniture", "units": 1, "unit_price": 1149.0, "month": "june"},
{"order_id": 1013, "region": "east", "category": "furniture", "units": 3, "unit_price": 219.0, "month": "june"},
{"order_id": 1014, "region": "south", "category": "office", "units": 15, "unit_price": 9.9, "month": "june"},
{"order_id": 1015, "region": "west", "category": "electronics", "units": 5, "unit_price": 89.0, "month": "june"},
]
_NUMERIC = {"units", "unit_price", "revenue"}
def _with_revenue(row: dict) -> dict:
return {**row, "revenue": round(row["units"] * row["unit_price"], 2)}
def _list_columns() -> dict:
return {
"columns": ["order_id", "region", "category", "units", "unit_price", "month", "revenue"],
"note": "revenue is units * unit_price, precomputed per row",
"row_count": len(_ROWS),
}
def _filter_rows(column: str, op: str, value) -> dict:
ops = {"eq": lambda a, b: a == b, "ne": lambda a, b: a != b,
"gt": lambda a, b: a > b, "lt": lambda a, b: a < b}
if op not in ops:
return {"error": f"op must be one of {sorted(ops)}"}
rows = [_with_revenue(r) for r in _ROWS]
if column not in rows[0]:
return {"error": f"no column {column!r}"}
if column in _NUMERIC:
try:
value = float(value)
except (TypeError, ValueError):
return {"error": f"column {column!r} is numeric; got {value!r}"}
matched = [r for r in rows if ops[op](r[column], value)]
return {"count": len(matched), "rows": matched}
def _aggregate(column: str, op: str, group_by: str = "") -> dict:
if column not in _NUMERIC:
return {"error": f"aggregate column must be numeric: {sorted(_NUMERIC)}"}
funcs = {"sum": sum, "min": min, "max": max, "count": len,
"avg": lambda v: round(sum(v) / len(v), 2) if v else 0}
if op not in funcs:
return {"error": f"op must be one of {sorted(funcs)}"}
rows = [_with_revenue(r) for r in _ROWS]
if not group_by:
return {op: funcs[op]([r[column] for r in rows])}
if group_by not in rows[0]:
return {"error": f"no column {group_by!r}"}
groups: dict = {}
for r in rows:
groups.setdefault(r[group_by], []).append(r[column])
return {str(k): funcs[op](v) for k, v in sorted(groups.items())}
_DISPATCH = {"list_columns": _list_columns, "filter_rows": _filter_rows, "aggregate": _aggregate}
_TOOLS = [
{
"type": "function",
"name": "list_columns",
"description": "The table's columns, types note, and row count. Call this first.",
"parameters": {"type": "object", "properties": {}},
},
{
"type": "function",
"name": "filter_rows",
"description": "Rows where <column> <op> <value>. Ops: eq, ne, gt, lt.",
"parameters": {
"type": "object",
"properties": {
"column": {"type": "string"},
"op": {"type": "string", "enum": ["eq", "ne", "gt", "lt"]},
"value": {"type": ["string", "number"]},
},
"required": ["column", "op", "value"],
},
},
{
"type": "function",
"name": "aggregate",
"description": "Aggregate a numeric column: sum, avg, min, max, count — "
"optionally grouped by another column.",
"parameters": {
"type": "object",
"properties": {
"column": {"type": "string"},
"op": {"type": "string", "enum": ["sum", "avg", "min", "max", "count"]},
"group_by": {"type": "string"},
},
"required": ["column", "op"],
},
},
]
def run(question: str) -> str:
input_list = [{"role": "user", "content": question.strip()}]
response = None
for _ in range(8): # bounded — never ship an open loop
response = _client.responses.create(
model=_MODEL,
instructions="Answer questions about the orders table using the "
"tools. Start from list_columns if unsure of the schema. "
"Base every number on tool results.",
input=input_list,
tools=_TOOLS,
)
calls = [item for item in response.output if item.type == "function_call"]
if not calls:
break
input_list += response.output
for call in calls:
args = json.loads(call.arguments)
fn = _DISPATCH.get(call.name)
result = fn(**args) if fn else {"error": f"unknown tool {call.name}"}
input_list.append({
"type": "function_call_output",
"call_id": call.call_id,
"output": json.dumps(result),
})
return response.output_text if response else ""
backend/ai_gemini.py
"""Week 6 - Showcase 1 (Gemini): question-answering over a table with chained tools."""
import os
from google import genai
from google.genai import types
_client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
_MODEL = "gemini-3.1-flash-lite"
_ROWS = [
{"order_id": 1001, "region": "north", "category": "electronics", "units": 3, "unit_price": 199.0, "month": "april"},
{"order_id": 1002, "region": "south", "category": "furniture", "units": 1, "unit_price": 649.0, "month": "april"},
{"order_id": 1003, "region": "north", "category": "office", "units": 12, "unit_price": 8.5, "month": "april"},
{"order_id": 1004, "region": "west", "category": "electronics", "units": 2, "unit_price": 349.0, "month": "april"},
{"order_id": 1005, "region": "south", "category": "office", "units": 30, "unit_price": 4.2, "month": "may"},
{"order_id": 1006, "region": "west", "category": "furniture", "units": 2, "unit_price": 289.0, "month": "may"},
{"order_id": 1007, "region": "north", "category": "electronics", "units": 1, "unit_price": 999.0, "month": "may"},
{"order_id": 1008, "region": "east", "category": "office", "units": 6, "unit_price": 12.0, "month": "may"},
{"order_id": 1009, "region": "east", "category": "electronics", "units": 4, "unit_price": 149.0, "month": "june"},
{"order_id": 1010, "region": "south", "category": "electronics", "units": 2, "unit_price": 449.0, "month": "june"},
{"order_id": 1011, "region": "west", "category": "office", "units": 20, "unit_price": 6.8, "month": "june"},
{"order_id": 1012, "region": "north", "category": "furniture", "units": 1, "unit_price": 1149.0, "month": "june"},
{"order_id": 1013, "region": "east", "category": "furniture", "units": 3, "unit_price": 219.0, "month": "june"},
{"order_id": 1014, "region": "south", "category": "office", "units": 15, "unit_price": 9.9, "month": "june"},
{"order_id": 1015, "region": "west", "category": "electronics", "units": 5, "unit_price": 89.0, "month": "june"},
]
_NUMERIC = {"units", "unit_price", "revenue"}
def _with_revenue(row: dict) -> dict:
return {**row, "revenue": round(row["units"] * row["unit_price"], 2)}
def _list_columns() -> dict:
return {
"columns": ["order_id", "region", "category", "units", "unit_price", "month", "revenue"],
"note": "revenue is units * unit_price, precomputed per row",
"row_count": len(_ROWS),
}
def _filter_rows(column: str, op: str, value) -> dict:
ops = {"eq": lambda a, b: a == b, "ne": lambda a, b: a != b,
"gt": lambda a, b: a > b, "lt": lambda a, b: a < b}
if op not in ops:
return {"error": f"op must be one of {sorted(ops)}"}
rows = [_with_revenue(r) for r in _ROWS]
if column not in rows[0]:
return {"error": f"no column {column!r}"}
if column in _NUMERIC:
try:
value = float(value)
except (TypeError, ValueError):
return {"error": f"column {column!r} is numeric; got {value!r}"}
matched = [r for r in rows if ops[op](r[column], value)]
return {"count": len(matched), "rows": matched}
def _aggregate(column: str, op: str, group_by: str = "") -> dict:
if column not in _NUMERIC:
return {"error": f"aggregate column must be numeric: {sorted(_NUMERIC)}"}
funcs = {"sum": sum, "min": min, "max": max, "count": len,
"avg": lambda v: round(sum(v) / len(v), 2) if v else 0}
if op not in funcs:
return {"error": f"op must be one of {sorted(funcs)}"}
rows = [_with_revenue(r) for r in _ROWS]
if not group_by:
return {op: funcs[op]([r[column] for r in rows])}
if group_by not in rows[0]:
return {"error": f"no column {group_by!r}"}
groups: dict = {}
for r in rows:
groups.setdefault(r[group_by], []).append(r[column])
return {str(k): funcs[op](v) for k, v in sorted(groups.items())}
_DISPATCH = {"list_columns": _list_columns, "filter_rows": _filter_rows, "aggregate": _aggregate}
_CONFIG = types.GenerateContentConfig(
system_instruction="Answer questions about the orders table using the tools. "
"Start from list_columns if unsure of the schema. "
"Base every number on tool results.",
tools=[types.Tool(function_declarations=[
types.FunctionDeclaration(
name="list_columns",
description="The table's columns, types note, and row count. Call this first.",
parameters=types.Schema(type=types.Type.OBJECT, properties={}),
),
types.FunctionDeclaration(
name="filter_rows",
description="Rows where <column> <op> <value>. Ops: eq, ne, gt, lt.",
parameters=types.Schema(
type=types.Type.OBJECT,
properties={
"column": types.Schema(type=types.Type.STRING),
"op": types.Schema(type=types.Type.STRING, enum=["eq", "ne", "gt", "lt"]),
"value": types.Schema(type=types.Type.STRING,
description="Comparison value; numbers as plain digits."),
},
required=["column", "op", "value"],
),
),
types.FunctionDeclaration(
name="aggregate",
description="Aggregate a numeric column: sum, avg, min, max, count — "
"optionally grouped by another column.",
parameters=types.Schema(
type=types.Type.OBJECT,
properties={
"column": types.Schema(type=types.Type.STRING),
"op": types.Schema(type=types.Type.STRING,
enum=["sum", "avg", "min", "max", "count"]),
"group_by": types.Schema(type=types.Type.STRING),
},
required=["column", "op"],
),
),
])],
)
def run(question: str) -> str:
contents = [types.Content(role="user", parts=[types.Part(text=question.strip())])]
response = None
for _ in range(8): # bounded — never ship an open loop
response = _client.models.generate_content(
model=_MODEL, contents=contents, config=_CONFIG,
)
if not response.function_calls:
break
contents.append(response.candidates[0].content)
parts = []
for fc in response.function_calls:
fn = _DISPATCH.get(fc.name)
result = fn(**fc.args) if fn else {"error": f"unknown tool {fc.name}"}
parts.append(types.Part.from_function_response(name=fc.name, response=result))
contents.append(types.Content(role="user", parts=parts))
return (response.text or "") if response else ""
Project files
.gitignoreREADME.mdbackend/Dockerfilebackend/ai_gemini.pybackend/ai_openai.pybackend/main.pybackend/requirements.txtbootstrap-secrets.shdocker-compose.ymlfrontend/Dockerfilefrontend/app/layout.tsxfrontend/app/page.tsxfrontend/next.config.tsfrontend/package.jsonfrontend/tsconfig.json