Building an AI Game Recommender with NVIDIA NOOA and OpenAI
- Ganesh Sharma
- 5 hours ago
- 26 min read
You know that thing where you ask someone “what should I play?” and they just hand you a generic top-10 list? Useless, right? It doesn’t know if you want to sink into a slow story for six hours or blast through something fast on your lunch break. It doesn’t know anything about you.
So here’s what we’re going to build together: a video game recommender that actually listens first. Using NVIDIA’s NOOA agent framework and OpenAI, you just describe your taste in your own words, and three little agents team up behind the scenes to turn that into a real profile, six real games that fit, and even the order to play them in. And the best part, it doesn’t just answer once and walk away. You can keep talking to it, “actually, swap the fantasy stuff for sci-fi”, and it remembers everything you already told it. Stick with me and I’ll walk you through every piece of it.
What We Are Building
Before we touch any code, let me show you the shape of the thing we’re making, three agents working behind one terminal chat:
You describe your taste in games, your own words, no rigid form to fill out
Agent 1 assesses: ProfileAgent checks whether what you said is enough to work with; if it isn’t, it asks you exactly one good follow-up question instead of grilling you with a checklist
Agent 1 profiles: once it has enough, that same agent writes up a structured GamerProfile (pacing, genres, traits, how you feel about multiplayer)
Agent 2 recommends: RecommendationAgent picks exactly six real, specific games that actually fit you
Agent 3 orders: PlayOrderAgent lines those six up from the easiest one to start with to the longest or toughest
You keep talking: after you see the results, say whatever you want, “swap fantasy for sci-fi”, “I actually have less free time than I said”, and the whole thing re-runs using everything you’ve said so far, not just your last message
Tech Stack
Here’s everything we’re reaching for, nothing exotic:
Component | Tool |
Agent framework | NVIDIA NOOA (Agent, strategy, PredictStrategy) |
LLM routing | litellm, via NOOA’s unifiedllm registry |
Model | OpenAI gpt-4o-mini |
Structured output | Pydantic models, validated by NOOA’s PredictStrategy |
UI | Rich (panels, tables, spinners) in a plain terminal loop |
Config | python-dotenv |
Pricing
Don’t worry, this won’t cost you much at all. NOOA itself is free and open-source, so the only thing you’re actually paying for is the OpenAI API call, and we’re using gpt-4o-mini, which is cheap. You’ll set your own rates in .env, defaulting to $0.15 per million input tokens and $0.60 per million output tokens. A normal round through all three agents costs a fraction of a cent. And just so you never have to wonder where your money went, every single call gets logged to stats.json, prompt tokens, completion tokens, how long it took, what it cost, and the terminal itself shows you a running total after every agent’s result and again after each thing you type.
Project Structure
Okay, let’s actually set this up. Create a file named requirements.txt in your project root:
nooa # the NVIDIA NOOA agent framework: Agent, strategy, PredictStrategy, unifiedllm
python-dotenv # loads OPENAI_API_KEY and other config from .env
rich # colorful terminal panels, tables, and spinners
Three packages, that’s it. nooa quietly pulls in litellm underneath, that’s the part actually doing the OpenAI talking. rich is what makes the terminal look nice instead of a wall of plain text, and python-dotenv just makes sure your config is loaded before anything else runs.
Next, create a file named .env in the project root:
OPENAI_API_KEY=your_openai_api_key_here # your own OpenAI secret key, never commit this file
OPENAI_MODEL=gpt-4o-mini # any litellm-recognized OpenAI model name
OPENAI_INPUT_COST_PER_TOKEN=0.00000015 # USD per input token, matches whichever model you set above
OPENAI_OUTPUT_COST_PER_TOKEN=0.00000060 # USD per output token, matches whichever model you set above
Notice I didn’t name those last two variables after gpt-4o-mini specifically. That’s on purpose. If you swap OPENAI_MODEL for something else down the road, you just update the two rate numbers here, you never have to go dig through the code.
Once you’re done, here’s roughly what you should be looking at:
nooa_game_recommender/
├── game_recommender.py # LLM setup, Pydantic models, the three Agent classes
├── app.py # terminal UI, conversational loop, cost display
├── stats_tracker.py # wraps every LLM call and logs it to stats.json
├── requirements.txt # nooa, python-dotenv, rich
├── .env # OPENAI_API_KEY, OPENAI_MODEL, per-token cost rates
└── stats.json # created on first run, accumulates token and cost data
Go ahead and get your environment ready:
python3 -m venv venv # create an isolated Python environment in ./venv
source venv/bin/activate # activate it so pip installs land inside venv, not system-wide
pip install -r requirements.txt # install nooa, python-dotenv, rich
Building the Agents
Alright, this is the fun part. Everything for the agent side lives in one file. Let’s start at the top, imports, the LLM client, and a little metadata tag that rides along on every request we send.
Create a file named game_recommender.py:
import os # read OPENAI_API_KEY / OPENAI_MODEL from the environment
import litellm # exposes enable_preview_features, the gate for forwarding metadata to OpenAI
from dotenv import load_dotenv # load .env before any os.environ.get() call
from pydantic import BaseModel, Field # structured, validated agent outputs
from nooa import Agent, strategy, PredictStrategy # the NVIDIA NOOA agent framework
from nooa.unifiedllm.registry import get_llm_client # builds the shared LLM client every agent uses
from stats_tracker import instrument # logs token usage + cost for every LLM call to stats.json
load_dotenv() # reads .env into the environment before any os.environ.get() call below
MODEL = os.environ.get("OPENAI_MODEL", "gpt-5-mini") # any litellm-recognized OpenAI model name, gpt-5-mini as a fallback
if not os.environ.get("OPENAI_API_KEY"): # fail fast with a clear message instead of a cryptic error deep in litellm
raise RuntimeError( # stop the whole script before any agent tries to call the API
"OPENAI_API_KEY is not set. Add it to a .env file next to this script " # first half of the error message
"(see .env.example)." # second half, points the user at the example file
) # closes the RuntimeError(...) call
# Tags every OpenAI request with who/what/where it came from, visible in the
# OpenAI dashboard's request metadata. litellm only forwards `metadata` to the
# OpenAI API itself when enable_preview_features is on; otherwise it's kept
# for litellm's own logging and never reaches OpenAI.
litellm.enable_preview_features = True # turns on the preview behavior that forwards metadata to OpenAI
OPENAI_CALL_METADATA = { # forwarded to every request this llm client makes
"dev_name": "Ganesh", # who triggered the call, shown in OpenAI's usage dashboard
"project": "codex-test", # groups usage under this project label
"environment": "local", # distinguishes local dev calls from staging or production
"purpose": "testing", # flags these calls as non-production traffic
} # closes the OPENAI_CALL_METADATA dict
llm = get_llm_client(MODEL, metadata=OPENAI_CALL_METADATA) # litellm reads OPENAI_API_KEY from the environment automatically
instrument(llm) # every call below now appends a token/cost record to stats.json
Quick tour: get_llm_client is NOOA’s easy button, hand it any model name litellm understands and it builds you a ready-to-go client, no registry setup needed. Anything extra you pass in, like our metadata here, just rides along on every request underneath. That litellm.enable_preview_features = True line matters more than it looks, litellm only actually forwards that metadata to OpenAI when this flag is on, otherwise it just sits there for litellm’s own internal logging and never shows up on your OpenAI dashboard. And that last line, instrument(llm), is doing something sneaky and wonderful: it wraps the client so every single call any of our agents makes gets logged automatically. You’ll see exactly how in a minute.
Now let’s define the shapes of data our agents hand back and forth to each other.
class GamerProfile(BaseModel): # structured taste profile every downstream agent relies on
summary: str = Field(description="A concise 2-3 sentence description of this player's gaming taste.") # the model writes its own short summary here; this is what shows as the opening paragraph of the profile panel
pacing: str = Field(description="Preferred pacing, e.g. 'slow, story-driven' vs 'fast-paced action'.") # captures how fast-moving the player likes a game to feel, used later when PlayOrderAgent decides what to save for last
preferred_genres: list[str] = Field(description="Genres this player gravitates toward.") # the genres the model inferred the player likes, displayed as a list in the profile panel
traits: list[str] = Field(description="Distinct taste traits, e.g. 'enjoys open-world exploration', 'dislikes heavy multiplayer pressure'.") # specific likes and dislikes pulled from what the player actually said, shown as bullet points
multiplayer_tolerance: str = Field(description="How this player feels about multiplayer or competitive pressure.") # whether the player wants to play alone, with friends, or against strangers
class ProfileResult(BaseModel): # one schema that both judges sufficiency and carries the profile
sufficient: bool = Field(description="True if the player's answer already has enough detail to build a solid gamer profile.") # the app reads this to decide whether to ask a follow-up question or move straight on to recommendations
follow_up_question: str | None = Field( # holds the exact question the app should show the player next
default=None, # empty by default, since most answers already contain enough detail
description="If not sufficient, exactly ONE specific question targeting the biggest gap. Null when sufficient.", # this description text is part of the model's instructions, not something shown to the player
) # closes the Field(...) call for follow_up_question
profile: GamerProfile | None = Field( # the actual profile the rest of the app uses, once the model is confident it has enough to work with
default=None, # stays empty until the model decides the player's answer is detailed enough
description="The structured gamer profile. Fill this in only when sufficient is true; leave it null otherwise.", # this description text is part of the model's instructions, not something shown to the player
) # closes the Field(...) call for profile
class GameRecommendation(BaseModel): # everything the table on screen needs for one recommended game
title: str # the game's real, exact name, as it would appear in a store listing
genre: str # e.g. RPG, action-adventure, puzzle
platform: str # which console, PC, or storefront the player can actually get it on
reason: str = Field(description="Why this game fits, tied directly to one or more traits from the profile.") # printed in the "Why it fits" column, meant to reference the player's own stated traits, not generic praise
class RecommendationSet(BaseModel): # the full batch of picks, produced by RecommendationAgent and passed on to PlayOrderAgent
games: list[GameRecommendation] = Field(min_length=6, max_length=6) # Pydantic rejects the model's output outright if it is not exactly six games
class OrderedGame(BaseModel): # one game placed into the final play order
position: int # its rank in the sequence; 1 means play this one first
title: str # must match a title from RecommendationSet exactly, so the app can line the two lists up
rationale: str = Field(description="Why this game belongs at this position, considering time commitment and difficulty curve.") # printed under the game in the play-order panel, explains why it sits at this specific spot
class PlayOrderPlan(BaseModel): # the final sequenced play order shown to the player
ordered_games: list[OrderedGame] # the six games, in the order the player should actually tackle them
strategy_note: str = Field(description="One or two sentences on the overall ramp-up logic across the whole order.") # a short closing explanation of why the whole sequence makes sense, printed at the bottom of the panel
Now look closely at ProfileResult, this is honestly my favorite bit of the whole design. Most people would write this as two separate calls: one to ask “do I have enough info?” and another to actually build the profile. I didn’t do that. I put both jobs in one schema, sufficient and follow_up_question handle the gatekeeping, and profile only fills in once sufficient is true. Since NOOA’s PredictStrategy is a single-shot call anyway, the model can decide and act in that same response. That’s a whole extra API call saved, every single round.
Okay, now the agents themselves, the part that actually talks to the model.
class ProfileAgent(Agent, llm=llm): # llm=llm binds this agent to the shared, instrumented client
"""You are a video game taste profiler. You study a player's own description of
what they like and dislike in games and translate it into a precise gamer
profile that the recommendation and play-order agents downstream will rely
on as their only source of truth about this player.""" # this whole docstring is the agent's system prompt
@strategy(PredictStrategy()) # single-shot call, output validated against ProfileResult
async def build_profile(self, answers: str) -> ProfileResult: # answers is the full accumulated conversation text
"""Read the player's free-form answers about their gaming preferences.
First decide whether they already have enough detail to build a solid
gamer profile: pacing preference, favorite genres, multiplayer
tolerance, and at least one or two concrete taste traits.
If something important is missing, vague, or unclear, set
sufficient=false, ask exactly one specific follow-up question
targeting the single biggest gap, and leave profile null. Do not ask
about things already covered, and do not nitpick minor details.
If the answer already covers enough ground, set sufficient=true, leave
follow_up_question null, and fill in profile: a structured gamer
profile covering pacing preference, favorite genres, distinct taste
traits, and multiplayer tolerance. Be specific and grounded only in
what the player actually said. Do not invent preferences they never
mentioned.""" # this whole docstring is the method's task instructions to the model
... # no Python body: NOOA calls the LLM and validates its output against ProfileResult
class RecommendationAgent(Agent, llm=llm): # shares the same instrumented llm client as ProfileAgent
"""You are a video game recommender. Given a structured gamer profile, you
pick exactly six real, specific games that match this player's taste.""" # this docstring is the system prompt
@strategy(PredictStrategy()) # single-shot call, output validated against RecommendationSet
async def recommend(self, profile: GamerProfile) -> RecommendationSet: # profile flows in as a live typed argument
"""Pick exactly six specific, real video games that match this gamer
profile. For each game, give its title, genre, platform, and a reason
that explicitly ties back to one or more traits in the profile. Vary the
picks across genres and series unless the profile strongly justifies
similar games.""" # the task instructions the model follows
... # no Python body: filled in by the LLM call under the hood
class PlayOrderAgent(Agent, llm=llm): # shares the same instrumented llm client as the other two agents
"""You are a play-order strategist. Given a set of six recommended games,
you sequence them into the best order for this player to actually play
them in, from a cold start to fully warmed up.""" # this docstring is the system prompt
@strategy(PredictStrategy()) # single-shot call, output validated against PlayOrderPlan
async def order(self, recommendations: RecommendationSet) -> PlayOrderPlan: # recommendations flows in as a live typed argument
"""Arrange these six games into the best play order: the easiest or most
approachable game first, building toward longer or more demanding games
later. Weigh each game's genre, likely time commitment, and difficulty
curve. Give a short rationale for each game's position and a one or two
sentence overall strategy note for the whole sequence.""" # the task instructions the model follows
... # no Python body: filled in by the LLM call under the hood
Here’s the whole trick to NOOA, once it clicks it clicks forever: an agent is just a plain Python class, its docstring is the system prompt, and any method with a ... body and a return type is a job you’re handing to the LLM. @strategy(PredictStrategy()) is what actually guarantees you get back a valid object of that type, and if the model fumbles the first try, it quietly retries with the validation error fed right back in. Notice all three agents share the exact same llm instance, which means that one instrument(llm) call from earlier covers every one of them. Nice, right?
Tracking Cost and Usage
Here’s something I always tell people, however small your project is: track what your API calls are actually costing you, from day one. It’s way easier to build the habit early than to bolt it on later once you’re confused about a surprise bill. This file wraps the LLM client itself, so we set it up once and every call from every agent gets logged automatically, no extra work per agent.
Create a file named stats_tracker.py:
import json # read and write the accumulated stats.json file
import os # read per-token cost rates from the environment
import time # measure wall-clock time around each LLM call
from contextlib import contextmanager # turns agent_context() into a plain with-block
from contextvars import ContextVar # tags each LLM call with the agent that triggered it, safe across async tasks
from datetime import datetime # timestamp every call record written to stats.json
from pathlib import Path # resolve the project root regardless of the working directory
from typing import Any # loose typing for the raw usage dict and call records
from dotenv import load_dotenv # must run before reading cost rates from os.environ below
load_dotenv() # reads .env into the environment before the os.environ[...] lookups below
PROJECT_ROOT = Path(__file__).resolve().parent # the directory containing this file, used to locate stats.json
STATS_FILE = PROJECT_ROOT / "stats.json" # accumulates across every run, never overwritten
# Per-token USD rates for whichever model OPENAI_MODEL is currently set to.
# Live in .env, not here, so switching models or repricing needs no code change.
_INPUT_COST = float(os.environ["OPENAI_INPUT_COST_PER_TOKEN"]) # no default: a missing rate fails loudly at import time
_OUTPUT_COST = float(os.environ["OPENAI_OUTPUT_COST_PER_TOKEN"]) # no default: a missing rate fails loudly at import time
current_agent: ContextVar[str] = ContextVar("current_agent", default="unknown") # which agent is calling right now
_session_calls: list[dict[str, Any]] = [] # calls logged by this process only, separate from stats.json's lifetime totals
@contextmanager # lets agent_context be used as `with agent_context("X"):`
def agent_context(name: str): # name is whichever agent label the caller wants attached to logged calls
"""Tag every LLM call made inside this block with `name` in stats.json."""
token = current_agent.set(name) # set the ContextVar for the duration of the block, remembering the old value
try: # ensures the reset below still runs even if the wrapped code raises
yield # control returns to the caller's with-block body here
finally: # runs on the way out, whether the block succeeded or raised
current_agent.reset(token) # always restore the previous value, even if the block raised
def _messages_to_text(messages: list[dict[str, Any]] | None) -> str: # turns a chat messages list into one string
"""Flatten a chat `messages` list into one readable string for stats.json."""
if not messages: # no messages were passed (or it was None)
return "" # nothing to flatten
parts = [] # accumulates one formatted line per message
for msg in messages: # walk every message in the conversation sent to the model
role = msg.get("role", "?") # e.g. "system", "user", "assistant"
content = msg.get("content", "") # the message text, or a list of content blocks for some providers
if isinstance(content, list): # some providers send content as a list of blocks
content = " ".join(block.get("text", "") for block in content if isinstance(block, dict)) # join block text into one string
parts.append(f"[{role}] {content}") # label each message with its role
return "\n\n".join(parts) # one readable block per message, separated by blank linesdef _log_call(
agent_name: str, # which agent triggered this call, from agent_context()
model: str, # the model string actually used
usage: dict[str, Any], # raw token usage dict from the LLM response
generation_seconds: float, # wall-clock time for just this API call
prompt: str, # flattened prompt text, for the audit trail
response_text: str, # raw response text, for the audit trail
) -> None: # this function only has the side effect of writing to disk; nothing to return
prompt_tokens = usage.get("prompt_tokens", 0) or 0 # input tokens billed for this call
completion_tokens = usage.get("completion_tokens", 0) or 0 # output tokens billed for this call
total_tokens = usage.get("total_tokens", prompt_tokens + completion_tokens) # fall back to the sum if the API omitted it
input_cost = round(prompt_tokens * _INPUT_COST, 7) # USD cost of the input tokens for this call
output_cost = round(completion_tokens * _OUTPUT_COST, 7) # USD cost of the output tokens for this call
record = { # the single record that will be appended to stats.json's "calls" list
"timestamp": datetime.now().isoformat(), # when this specific call was logged
"agent": agent_name, # which agent produced this call
"model": model, # which model was used
"generation_seconds": round(generation_seconds, 3), # how long the API call took in seconds
"prompt": prompt[:2000], # first 2000 chars, keeps stats.json readable
"response": response_text[:2000], # first 2000 chars, keeps stats.json readable
"prompt_tokens": prompt_tokens, # how many tokens the request itself used
"completion_tokens": completion_tokens, # how many tokens the model's answer used
"total_tokens": total_tokens, # prompt and completion tokens added together, for convenience
"input_cost": input_cost, # dollar cost of just the prompt tokens
"output_cost": output_cost, # dollar cost of just the completion tokens
"total_cost": round(input_cost + output_cost, 7), # combined USD cost for this call
} # closes the record dict
try: # guard against a missing or corrupted stats.json
existing = json.loads(STATS_FILE.read_text(encoding="utf-8")) if STATS_FILE.exists() else {"summary": {}, "calls": []} # load prior state, or start empty
except (json.JSONDecodeError, OSError): # the file exists but is unreadable or not valid JSON
existing = {"summary": {}, "calls": []} # start fresh if the file is missing or corrupt
_session_calls.append(record) # also keep it in memory for this process's own running totals
existing["calls"].append(record) # accumulate: never overwrite, always append
calls = existing["calls"] # local alias, avoids repeated dict lookups below
existing["summary"] = { # recomputed from scratch every call, so it can never drift
"timestamp": datetime.now().isoformat(), # when this summary was last recomputed
"total_calls": len(calls), # lifetime count of every logged call
"total_generation_seconds": round(sum(c.get("generation_seconds", 0) for c in calls), 3), # cumulative wall-clock time
"total_prompt_tokens": sum(c["prompt_tokens"] for c in calls), # sum of input tokens across all calls
"total_completion_tokens": sum(c["completion_tokens"] for c in calls), # sum of output tokens across all calls
"total_tokens": sum(c["total_tokens"] for c in calls), # combined input and output tokens
"total_input_cost": round(sum(c.get("input_cost", 0) for c in calls), 6), # cumulative USD cost of input tokens
"total_output_cost": round(sum(c.get("output_cost", 0) for c in calls), 6), # cumulative USD cost of output tokens
"total_cost": round(sum(c["total_cost"] for c in calls), 6), # lifetime cost across every call
} # closes the summary dict
STATS_FILE.write_text(json.dumps(existing, indent=2, ensure_ascii=False), encoding="utf-8") # atomic overwrite of the whole file
Take a breath, that function looks bigger than it is. Our cost rates come only from .env, no fallback hiding in the code, so if you forget to set one it fails loudly right at startup instead of quietly reporting $0 forever, which trust me, is the better failure mode. agent_context is just a little tag you wrap around each agent call so every logged record knows which of the three agents made it, even though they all share one client. And notice the record saves the actual prompt and response text too, truncated to 2000 characters, so stats.json doubles as a lightweight diary of exactly what each agent saw and said, not just what it cost you.
The function reads whatever’s already in stats.json, falls back to an empty structure if it’s missing or broken, appends the new record, then rebuilds the whole summary block from scratch by re-adding everything, that way it can never drift out of sync.
def latest_call() -> dict[str, Any] | None: # returns None only if instrument() has never logged a call yet
"""The most recently logged call, or None if nothing has been logged yet."""
return _session_calls[-1] if _session_calls else None # used to print a cost line right after each agent's panel
def call_count() -> int: # a plain length check, used as a before/after checkpoint
"""Number of calls logged so far. Use as a checkpoint with session_summary(since=...)."""
return len(_session_calls) # a snapshot taken right before a round starts
def session_summary(since: int = 0) -> dict[str, Any]: # since=0 (the default) means "the whole session so far"
"""Aggregate token usage and cost across calls made by this process.
Pass `since=call_count()` taken before a round starts to get that round's
totals only, instead of the whole session's.
"""
calls = _session_calls[since:] # slice from the checkpoint to now, or everything if since=0
return { # one aggregated dict, mirroring the shape of stats.json's "summary" block
"calls": len(calls), # how many separate API requests happened in this window
"prompt_tokens": sum(c["prompt_tokens"] for c in calls), # total tokens sent to the model across those requests
"completion_tokens": sum(c["completion_tokens"] for c in calls), # total tokens the model generated back
"total_tokens": sum(c["total_tokens"] for c in calls), # prompt and completion tokens added together
"total_cost": round(sum(c["total_cost"] for c in calls), 6), # what those requests actually cost, in US dollars
"total_generation_seconds": round(sum(c["generation_seconds"] for c in calls), 3), # how long the model spent actually generating, combined
} # closes the returned summary dict
Three little helpers here, and honestly they’re the reason the CLI can show you numbers without any extra bookkeeping. latest_call() just hands back whatever was logged most recently, so we can print a cost line right under each agent’s panel. call_count() tells you how many calls have happened so far, which we use as a before-and-after checkpoint.
And session_summary(since=...) adds up tokens, cost, and time across only what happened since that checkpoint, so the exact same function gives you a per-input total and a whole-session total, just by changing what you pass in.
def instrument(llm) -> None: # called once, right after the llm client is built
"""Wrap `llm`'s call/acall so every real API request gets logged to stats.json.
Wraps the client instance directly (rather than hooking nooa's harness
metrics callback, which is only wired up when running under the full nooa
actor/session runtime) so this works for a bare script too.
"""
original_call = llm.call # keep a reference to the real synchronous call method before overwriting it
original_acall = llm.acall # keep a reference to the real async call method before overwriting it
def _prompt_text(args: tuple, kwargs: dict) -> str: # recovers the messages list regardless of how it was passed
messages = kwargs.get("messages") or (args[0] if args else None) # messages may be positional or keyword
return _messages_to_text(messages) # flatten to a readable string for stats.json
def _response_text(response: Any) -> str: # extracts the raw text the model actually produced
return (response.assistant_message or {}).get("content", "") or str(response.content) # raw text, even for structured output
def wrapped_call(*args, **kwargs): # replaces llm.call; same signature and return value as the original
start = time.perf_counter() # start the clock right before the real call
response = original_call(*args, **kwargs) # the actual synchronous LLM request
elapsed = time.perf_counter() - start # pure generation time, nothing else
_log_call( # write one record to stats.json for this exact call
current_agent.get(), llm.model, response.usage or {}, elapsed, # who called it, which model, raw usage, timing
_prompt_text(args, kwargs), _response_text(response), # the flattened prompt and raw response text
) # closes the _log_call(...) call
return response # callers never notice the wrapping
async def wrapped_acall(*args, **kwargs): # replaces llm.acall; the async counterpart of wrapped_call
start = time.perf_counter() # start the clock right before the real call
response = await original_acall(*args, **kwargs) # the actual async LLM request
elapsed = time.perf_counter() - start # pure generation time, nothing else
_log_call( # write one record to stats.json for this exact call
current_agent.get(), llm.model, response.usage or {}, elapsed, # who called it, which model, raw usage, timing
_prompt_text(args, kwargs), _response_text(response), # the flattened prompt and raw response text
) # closes the _log_call(...) call
return response # callers never notice the wrapping
llm.call = wrapped_call # replace the instance's call method with the wrapped version
llm.acall = wrapped_acall # replace the instance's acall method with the wrapped version
This is the part I actually want you to remember, more than any other piece of this tutorial: instead of hooking into some internal framework callback that only fires under a full runtime, we just swap out the client’s own call and acall methods for our own wrapped versions that time the request and log it. Dead simple, works in a bare script, no ceremony required.
time.perf_counter() starts right before the real request and stops right after, so generation_seconds is pure model time, it never counts the time you spend sitting there typing your answer. And because all three agents share that one llm instance, you get every single one of them instrumented for free the moment you wrap it once.
Building the CLI
Alright, last big piece. The terminal app has three things going on: one open-ended question instead of a rigid form, a follow-up loop that only bugs you when it actually needs more, and a conversation that keeps going instead of quitting after one answer.
Create a file named app.py:
import asyncio # drives the async main() from a plain script entry point
import sys # sys.exit() on startup errors or Ctrl-C
from rich.console import Console # the shared console every print goes through
from rich.panel import Panel # bordered boxes for the profile and play-order results
from rich.prompt import Prompt # reads a line of input with a styled prompt marker
from rich.table import Table # the recommended-games table
from rich.text import Text # styled multi-run text blocks inside panels
console = Console() # one Console instance shared by every print/status call
try: # game_recommender raises RuntimeError at import time if OPENAI_API_KEY is missing
from game_recommender import ProfileAgent, RecommendationAgent, PlayOrderAgent, GamerProfile # the three agents plus the profile type
except RuntimeError as exc: # catches the missing-API-key error raised while importing game_recommender
console.print(f"\n[bold red]Error:[/] {exc}\n") # show the exact message from game_recommender.py
sys.exit(1) # stop before anything tries to make a real API call
from stats_tracker import agent_context, call_count, latest_call, session_summary # cost/usage helpers used throughout this file
MAX_FOLLOW_UPS = 3 # cap on clarifying questions so a stubbornly vague answer can't loop forever
EXIT_WORDS = {"quit", "exit", "q", "bye", "stop", "done"} # any of these (or a blank line) ends the session
SAMPLE_ANSWER = ( # shown to the player as a worked example, not sent to the model
"I like slow, story-driven games with open-world exploration. I enjoy " # first line of the example answer
"wandering off the main quest. I avoid multiplayer and competitive pressure, " # second line of the example answer
"have maybe 4-6 hours a week to play, and dislike heavy grinding or " # third line of the example answer
"pay-to-win mechanics." # fourth and final line of the example answer
) # closes the SAMPLE_ANSWER string concatenation
Notice there’s no rigid five-question form here, just one open question with a worked example underneath it, so someone using this for the first time has a rough sense of how much to say, without feeling like they’re filling out a job application.
async def gather_profile(answers: str) -> tuple[GamerProfile, str]: # returns the built profile plus the (possibly extended) answers
"""Call ProfileAgent until it has enough to build a profile, asking a follow-up
question in between when it doesn't. Returns the profile and the (possibly
extended) answers text, so later rounds keep the full conversation."""
for _ in range(MAX_FOLLOW_UPS): # ask at most MAX_FOLLOW_UPS clarifying questions
with console.status("[bold cyan]Agent 1 is studying your taste...", spinner="dots"): # spinner shown while the call is in flight
with agent_context("ProfileAgent"): # tags this call as ProfileAgent in stats.json
result = await ProfileAgent().build_profile(answers) # the single combined assess-and-build call
if result.sufficient and result.profile is not None: # the common case: one call, done
return result.profile, answers # hand back the profile immediately, no follow-up needed
if not result.follow_up_question: # model said "not sufficient" but gave no question; bail out
break # fall through to the unconditional final attempt below
console.print(f"\n[green]?[/] {result.follow_up_question}") # show the model's own follow-up question
follow_up = Prompt.ask("Input") # collect the player's answer to that specific question
answers = f"{answers}\n{follow_up}" # append, never replace: memory accumulates here
with console.status("[bold cyan]Agent 1 is studying your taste...", spinner="dots"): # one last unconditional attempt after the loop
with agent_context("ProfileAgent"): # still tagged as ProfileAgent for stats.json
result = await ProfileAgent().build_profile(answers) # one last unconditional attempt after the cap
profile = result.profile or GamerProfile( # rare fallback: still nothing after MAX_FOLLOW_UPS tries
summary=answers[:300], # first 300 chars of the raw conversation as a stand-in summary
pacing="unspecified", # placeholder value, since the model never confirmed one
preferred_genres=[], # empty list, since the model never confirmed any
traits=[], # empty list, since the model never confirmed any
multiplayer_tolerance="unspecified", # placeholder value, since the model never confirmed one
) # closes the fallback GamerProfile(...) call
return profile, answers # hand back whichever profile we ended up with
Since ProfileResult already does both jobs in one call, this loop is usually just one round-trip: a detailed first answer comes back sufficient=True immediately, done. A vague answer gets exactly one good follow-up question per pass, capped at MAX_FOLLOW_UPS, so someone being stubbornly unhelpful can’t trap you in an endless loop. And if, after all that, the model still hasn’t handed over a real profile, we don’t crash, we just build a minimal fallback straight from whatever raw text we have. Always have a plan for the weird edge case.
def print_call_cost(style: str) -> None: # style is a Rich color name matching the agent's own panel color
call = latest_call() # the record _log_call() just appended for this exact call
if call: # guards against printing nothing useful if somehow no call was logged
console.print( # a single styled line summarizing this one call
f"[{style}]Tokens: {call['total_tokens']} " # total tokens used by this call
f"Cost: ${call['total_cost']:.6f} " # USD cost of this call, six decimal places
f"Time taken: {call['generation_seconds']}s[/{style}]" # generation time for this call, closes the style tag
) # closes the console.print(...) call
def print_summary(label: str, summary: dict) -> None: # label distinguishes a per-input line from a per-session line
console.print( # a single styled line summarizing the whole given `summary` dict
f"\n[bold]{label}:[/] Calls: {summary['calls']} " # leading blank line, the label, and the call count
f"Tokens: {summary['total_tokens']} " # total tokens across every call in the summary
f"Cost: [bold green]${summary['total_cost']:.6f}[/] " # total USD cost, highlighted in green
f"Time taken: {summary['total_generation_seconds']}s" # total generation time across every call in the summary
) # closes the console.print(...) call
Little thing, but it matters: every number here has its own label, calls, tokens, cost, time taken, no guessing which figure means what. print_call_cost even matches the color of the agent’s own panel, so the cost line visually belongs to the result sitting right above it.
def render_profile(profile) -> None: # draws Agent 1's result as a bordered panel
body = Text() # accumulates every styled run that makes up the panel's content
body.append(profile.summary + "\n\n", style="white") # the 2-3 sentence taste summary, then a blank line
body.append("Pacing: ", style="bold yellow") # bold label for the pacing field
body.append(profile.pacing + "\n") # the pacing value itself, in the default style
body.append("Preferred genres: ", style="bold yellow") # bold label for the genres field
body.append(", ".join(profile.preferred_genres) + "\n") # genres joined into one comma-separated line
body.append("Multiplayer tolerance: ", style="bold yellow") # bold label for the multiplayer field
body.append(profile.multiplayer_tolerance + "\n") # the multiplayer tolerance value itself
body.append("Traits:\n", style="bold yellow") # bold label introducing the bulleted traits list
for trait in profile.traits: # one bullet line per taste trait
body.append(f" • {trait}\n", style="white") # indented bullet point for this trait
console.print() # blank line for spacing before the panel
console.print(Panel(body, title="[bold]Agent 1: Your Gamer Profile[/]", border_style="cyan", expand=False)) # draw the bordered panel
def render_recommendations(recommendations) -> None: # draws Agent 2's result as a table
table = Table(title="Agent 2: Recommended Games", border_style="green", header_style="bold green") # the table shell with title and colors
table.add_column("Title", style="bold white") # first column: the game's title
table.add_column("Genre", style="magenta") # second column: the game's genre
table.add_column("Platform", style="cyan") # third column: where the game can be played
table.add_column("Why it fits", style="white", max_width=50) # fourth column: the model's reason, wrapped at 50 chars
for game in recommendations.games: # exactly six, guaranteed by RecommendationSet's validation
table.add_row(game.title, game.genre, game.platform, game.reason) # one row per recommended game
console.print() # blank line for spacing before the table
console.print(table) # draw the finished table
def render_play_order(play_order) -> None: # draws Agent 3's result as a bordered panel
body = Text() # accumulates every styled run that makes up the panel's content
for game in sorted(play_order.ordered_games, key=lambda g: g.position): # render in position order, 1 first
body.append(f"{game.position}. ", style="bold yellow") # bold position number, e.g. "1. "
body.append(f"{game.title}\n", style="bold white") # the game's title on its own line
body.append(f" {game.rationale}\n\n", style="dim white") # indented rationale, then a blank line
body.append(play_order.strategy_note, style="italic cyan") # the closing overall strategy note, in italics
console.print() # blank line for spacing before the panel
console.print(Panel(body, title="[bold]Agent 3: Recommended Play Order[/]", border_style="yellow", expand=False)) # draw the bordered panel
Nothing scary in these three, just Rich formatting: build up a styled Text or Table, print a blank line so it doesn’t feel cramped, then print the thing itself. None of this touches the model, it’s purely turning already-validated data into something nice to look at.
async def run_pipeline(answers: str) -> str: # runs all three agents once and returns the (possibly extended) answers
"""Build a profile and recommendations from the full conversation so far.
Returns the (possibly extended) answers text, since gather_profile may have
appended follow-up question answers to it."""
checkpoint = call_count() # remember how many calls existed before this round started
profile, answers = await gather_profile(answers) # Agent 1: assess and build, looping on follow-ups as needed
render_profile(profile) # draw Agent 1's panel
print_call_cost("cyan") # matches Agent 1's panel border color
with console.status("[bold green]Agent 2 is picking six matching games...", spinner="dots"): # spinner while Agent 2 runs
with agent_context("RecommendationAgent"): # tags this call as RecommendationAgent in stats.json
recommendations = await RecommendationAgent().recommend(profile) # Agent 2: pick six matching games
render_recommendations(recommendations) # draw Agent 2's table
print_call_cost("green") # matches Agent 2's table color
with console.status("[bold yellow]Agent 3 is arranging your play order...", spinner="dots"): # spinner while Agent 3 runs
with agent_context("PlayOrderAgent"): # tags this call as PlayOrderAgent in stats.json
play_order = await PlayOrderAgent().order(recommendations) # Agent 3: sequence the six games
render_play_order(play_order) # draw Agent 3's panel
print_call_cost("yellow") # matches Agent 3's panel border color
print_summary("Usage stats for this input", session_summary(since=checkpoint)) # only this round's calls
return answers # handed back to main() so the next round keeps the full conversation
async def main() -> None: # the script's entry point, driven via asyncio.run() below
print_banner() # show the title panel once at startup
answers = collect_answers() # the very first, open-ended question
answers = await run_pipeline(answers) # run all three agents on the first answer
while True: # the conversational loop: keep going until the user quits
console.print() # blank line before the next input prompt
follow_up = Prompt.ask("Input") # read the player's next message
if not follow_up.strip() or follow_up.strip().lower() in EXIT_WORDS: # blank input or an exit word ends the loop
break # leave the while loop and fall through to the closing summary
answers = f"{answers}\n{follow_up}" # append this turn to the running conversation
answers = await run_pipeline(answers) # re-run all three agents on the full accumulated conversation
print_summary("Usage stats for this session", session_summary()) # every call across every round
console.print("\n[bold magenta]Happy gaming![/]\n") # closing message before the process exits
if __name__ == "__main__": # only runs when this file is executed directly, e.g. `python3 app.py`
try:
asyncio.run(main()) # drives the async main() to completion
except KeyboardInterrupt: # Ctrl-C
console.print("\n[dim]Cancelled.[/]") # a plain, quiet message instead of a raw traceback
sys.exit(0) # exit code 0: this is a normal, user-initiated stop
except Exception as exc: # anything unhandled: show it plainly instead of a raw traceback
console.print(f"\n[bold red]Error:[/] {exc}\n") # print the exception message in red
sys.exit(1) # exit code 1: this is an actual failure
Here’s the piece I really want you to notice: answers never gets reset between rounds. Everything you type gets tacked onto it, and that whole growing string is what gets handed to ProfileAgent every single round. That’s the entire trick behind this thing remembering you, there’s no fancy memory database anywhere, we’re just quietly re-sending the whole conversation every time. Type quit, exit, bye, or just hit enter on an empty line, and it wraps up with one final total for the whole session.
Running the Application
Alright, moment of truth. Let’s run it.
source venv/bin/activate
python3 app.py
Try opening with something like the example the app itself shows you:
I like slow, story-driven games with open-world exploration. I enjoy
wandering off the main quest. I avoid multiplayer and competitive
pressure, have maybe 4-6 hours a week to play, and dislike heavy
grinding or pay-to-win mechanics.
stats.json file
Who Can Benefit
Enterprises that need a working template for per-request LLM cost and usage tracking before rolling AI features out at scale
Developers building a foundation before moving to more complex multi-agent systems, where an agent is a class, a docstring is a system prompt, and a typed return is a validated generation call
Gamers who want picks based on how they actually play, not a generic best-of list
Community and Discord bot builders who want to recommend things conversationally instead of through a rigid form
Students learning Python and AI who want a real, working multi-agent app to study and build on, not just a toy example
How Codersarts Can Help
If you want to take this further, Codersarts offers hands-on support at every stage.
For enterprises: Architecture consulting for production agent deployments, including authentication, per-request cost attribution, and integrating agents with existing catalog or inventory systems.
For teams: End-to-end development of recommendation and conversational agents, including support for persistent storage, multi-user sessions, and richer memory beyond a single accumulated transcript.
For learners: Live 1-to-1 sessions with an AI engineer who can walk through NOOA’s agent and strategy model, structured output validation, and how to instrument any LLM client for cost tracking.
Reach out at contact@codersarts.com or visit www.codersarts.com to get started.
Continue Exploring AI Resources
If you found this blog helpful, explore more AI, RAG, and enterprise AI resources from CodersArts AI to see how organizations are applying these systems to real world applications.
Comments