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 downEjecútalo en tu máquina
Descarga el proyecto como ZIP y córrelo con Docker. Levanta un backend FastAPI y un frontend Next.js en 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 --buildEscribe algo, elige un proveedor y ejecuta el mismo código de Código contra la API real. Requiere iniciar sesión.
Los mismos módulos que ejecuta el botón Run. El proyecto completo (frontend, Dockerfile, compose) está en el ZIP, pestaña 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 ""
Archivos del proyecto
.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