Support assistant

The whole course in one app: moderation guard (wk16) + agent loop (wk18) + grounded tools — KB search (wk7-9) and order lookup. Ask about an order (1001/1002) or a policy; it searches, looks up, grounds, and stays safe.

README Try it Source

Showcase — Support assistant

Run

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

Open http://localhost:3000. Gemini: PROVIDER=gemini docker compose up --build.

What's where

backend/ai_openai.py / backend/ai_gemini.py — moderation + the grounded agent loop with two tools.
frontend/app/page.tsx — message box, safe grounded answer.

Stop

docker compose down

Run locally

Download the project as a ZIP and run it with Docker. Brings up a FastAPI backend + Next.js frontend on localhost:3000.

Download support-agent.zip

unzip support-agent.zip
cd support-agent
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

Type some input, pick a provider, and run the same code shown in Source against the live API. Sign-in required.

Input

Provider

Output

The same modules the Run button hits. The whole project (frontend, Dockerfile, compose) is in the ZIP under README.

`backend/ai_openai.py`

"""Showcase 1 (OpenAI): the support assistant — the whole course in one app.

Moderation guard (week 16) + agent loop (week 18) + grounded tools (KB search and
order lookup, weeks 7-9). Ask an order question, a policy question, or something
off-topic and watch it search, look up, ground, or gracefully escalate — safely.
"""
import json

from openai import OpenAI

_client = OpenAI()

_MODEL = "gpt-5.4-nano"

_KB = [
    {"topic": "returns", "text": "Items can be returned within 30 days; opened items get store credit only."},
    {"topic": "shipping", "text": "Standard shipping is free over $50 and takes 3-5 business days."},
    {"topic": "warranty", "text": "Electronics have a 1-year warranty covering defects, not accidental damage."},
]
_ORDERS = {
    "1001": {"status": "shipped", "carrier": "UPS", "eta": "Tuesday"},
    "1002": {"status": "processing", "eta": "ships within 24 hours"},
}


def _search_kb(query: str) -> dict:
    words = [w for w in query.lower().split() if len(w) > 2]
    return {"results": [e for e in _KB if any(w in (e["text"] + e["topic"]).lower() for w in words)][:3]}


def _order_status(order_id: str) -> dict:
    return _ORDERS.get(order_id.strip(), {"error": f"no order {order_id!r}"})


_IMPL = {"search_kb": _search_kb, "order_status": _order_status}
_TOOLS = [
    {"type": "function", "name": "search_kb", "description": "Search help articles (returns, shipping, warranty).",
     "parameters": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}},
    {"type": "function", "name": "order_status", "description": "Look up an order's status by its id (try 1001 or 1002).",
     "parameters": {"type": "object", "properties": {"order_id": {"type": "string"}}, "required": ["order_id"]}},
]

_GOAL = ("You are a customer support assistant. Use search_kb for policy questions and order_status "
         "for order questions. Answer ONLY from tool results; if the tools don't cover it, say you'll "
         "escalate to a human. Be warm and brief.")


def run(message: str) -> str:
    m = message.strip()
    if not m:
        return "Ask about an order (try 1001 or 1002) or a policy (returns, shipping, warranty)."
    if _client.moderations.create(model="omni-moderation-latest", input=m).results[0].flagged:
        return "[message blocked by moderation]"
    input_list = [{"role": "user", "content": m}]
    response = None
    for _ in range(6):
        response = _client.responses.create(model=_MODEL, instructions=_GOAL, input=input_list, tools=_TOOLS)
        calls = [i for i in response.output if i.type == "function_call"]
        if not calls:
            break
        input_list += response.output
        for c in calls:
            result = _IMPL[c.name](**json.loads(c.arguments))
            input_list.append({"type": "function_call_output", "call_id": c.call_id, "output": json.dumps(result)})
    return response.output_text if response else ""

`backend/ai_gemini.py`

"""Showcase 1 (Gemini): the support assistant — the whole course in one app.

Same moderation + agent + grounded-tools composition, on Gemini.
"""
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"

_KB = [
    {"topic": "returns", "text": "Items can be returned within 30 days; opened items get store credit only."},
    {"topic": "shipping", "text": "Standard shipping is free over $50 and takes 3-5 business days."},
    {"topic": "warranty", "text": "Electronics have a 1-year warranty covering defects, not accidental damage."},
]
_ORDERS = {
    "1001": {"status": "shipped", "carrier": "UPS", "eta": "Tuesday"},
    "1002": {"status": "processing", "eta": "ships within 24 hours"},
}


def _search_kb(query: str) -> dict:
    words = [w for w in query.lower().split() if len(w) > 2]
    return {"results": [e for e in _KB if any(w in (e["text"] + e["topic"]).lower() for w in words)][:3]}


def _order_status(order_id: str) -> dict:
    return _ORDERS.get(order_id.strip(), {"error": f"no order {order_id!r}"})


_IMPL = {"search_kb": _search_kb, "order_status": _order_status}

_CONFIG = types.GenerateContentConfig(
    system_instruction=("You are a customer support assistant. Use search_kb for policy questions and "
                        "order_status for order questions. Answer ONLY from tool results; if the tools "
                        "don't cover it, say you'll escalate to a human. Be warm and brief."),
    tools=[types.Tool(function_declarations=[
        types.FunctionDeclaration(name="search_kb", description="Search help articles (returns, shipping, warranty).",
                                  parameters=types.Schema(type=types.Type.OBJECT,
                                      properties={"query": types.Schema(type=types.Type.STRING)}, required=["query"])),
        types.FunctionDeclaration(name="order_status", description="Look up an order's status by id (try 1001 or 1002).",
                                  parameters=types.Schema(type=types.Type.OBJECT,
                                      properties={"order_id": types.Schema(type=types.Type.STRING)}, required=["order_id"])),
    ])],
)


def _flagged(text: str) -> bool:
    r = _client.models.generate_content(
        model=_MODEL, contents=[types.Content(role="user", parts=[types.Part(text=text)])],
        config=types.GenerateContentConfig(system_instruction=(
            "Reply 'flag' if this text is hateful, harassing, sexual, violent, or dangerous; else 'ok'. One word.")),
    )
    return "flag" in (r.text or "").lower()


def run(message: str) -> str:
    m = message.strip()
    if not m:
        return "Ask about an order (try 1001 or 1002) or a policy (returns, shipping, warranty)."
    if _flagged(m):
        return "[message blocked by moderation]"
    contents = [types.Content(role="user", parts=[types.Part(text=m)])]
    response = None
    for _ in range(6):
        response = _client.models.generate_content(model=_MODEL, contents=contents, config=_CONFIG)
        if not response.function_calls:
            break
        contents.append(response.candidates[0].content)
        parts = [types.Part.from_function_response(name=fc.name, response=_IMPL[fc.name](**dict(fc.args)))
                 for fc in response.function_calls]
        contents.append(types.Content(role="user", parts=parts))
    return (response.text or "") if response else ""