Course EN
← back to chapter

Analista de CSV

Haz preguntas sobre un trimestre de órdenes de e-commerce — "which region had

Showcase — Analista de CSV

Haz preguntas sobre un trimestre de órdenes de e-commerce — "which region had the highest average revenue?", "total office units sold in May?" — y el modelo responde encadenando tool calls: list_columns para aprender el schema, y luego filter_rows y aggregate tantas veces como la pregunta lo necesite. Nunca ve la tabla cruda, que es exactamente el patrón de acceso que quieres contra una tabla de warehouse real que no cabe en un prompt.

Córrelo

bash bootstrap-secrets.sh              # reads ../../../../.env, writes secrets/
docker compose up --build              # default: PROVIDER=openai

Abre http://localhost:3000.

Para correrlo contra Gemini:

PROVIDER=gemini docker compose up --build

Qué hay aquí

  • backend/ai_openai.py — la tabla de órdenes embebida, las tres tools, el dict de dispatch y el loop de agente acotado
  • backend/ai_gemini.py — las mismas tools y loop, en la forma de Gemini
  • backend/main.py — loader de FastAPI idéntico; lee PROVIDER y despacha
  • frontend/app/page.tsx — textarea + resultado
  • docker-compose.yml — dos servicios, secrets montados desde ./secrets/

El loop está topado en 8 iteraciones — un loop de agente sin cota es una caída esperando a un modelo confundido.

Detenlo

docker compose down

Ejecú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.

Descargar csv-analyst.zip 

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

Escribe 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


  • .gitignore
  • README.es.md
  • README.md
  • backend/Dockerfile
  • backend/ai_gemini.py
  • backend/ai_openai.py
  • backend/main.py
  • backend/requirements.txt
  • bootstrap-secrets.sh
  • docker-compose.yml
  • frontend/Dockerfile
  • frontend/app/layout.tsx
  • frontend/app/page.tsx
  • frontend/next.config.ts
  • frontend/package.json
  • frontend/tsconfig.json