Most teams experimenting with large language models start with the same question:
Can we make this work?
A developer builds a prototype. A product manager tests a few prompts. The results look useful. Soon the team wants to connect the model to documents, internal tools, customer data, support tickets, reports, or dashboards.
That is usually when the harder questions appear:
- How much will this cost when more users start using it?
- Which workflows are generating the most token usage?
- Are we sending sensitive data into the model?
- Which prompts are risky?
- Are we tracking failures, retries, and expensive requests?
- Can we explain where the data went?
Many AI prototypes fail not because the model is weak, but because the team has no visibility around cost, risk, and data movement.
In this article, we will build a simple Python-based LLM cost and risk visibility tool. The goal is not to create a full governance platform. The goal is to show how a small team can start monitoring AI usage in a practical way.
We will create a lightweight workflow that:
- loads sample LLM request logs;
- estimates token-based cost;
- flags risky prompts;
- groups usage by user and workflow;
- identifies expensive or sensitive requests;
- produces a simple summary report.
This pattern can be extended later into a dashboard, database table, Streamlit app, or internal monitoring tool.
Why cost and risk visibility matters
LLM usage can grow quietly.
One user testing a chatbot may not cost much. But once an AI feature is used across a team, costs can increase through:
- long prompts;
- large uploaded documents;
- repeated retries;
- agent loops;
- unnecessary model calls;
- verbose responses;
- background automations;
- multiple tools calling the model without tracking.
Risk grows in a similar way.
At the prototype stage, users may paste test data. In production, they may paste customer details, financial information, confidential documents, employee data, or business-sensitive material.
If the application does not log what is happening, the team cannot manage it.
A simple monitoring layer can help answer:
- Which workflows use the most tokens?
- Which users or teams generate the highest cost?
- Which prompts contain sensitive terms?
- Which model calls failed or retried?
- Which requests need review?
Sample dataset
For this example, we will use a small CSV file called llm_requests.csv.
Each row represents one LLM request.
request_id,user_id,workflow,model,prompt,response,status
1,u001,customer_support,gpt-4,"Summarise this customer complaint: My card number is 4111 1111 1111 1111 and I was charged twice.","The customer reports a duplicate charge.",success
2,u002,sales_email,gpt-4,"Write a follow-up email for a SaaS lead interested in pricing.","Here is a professional follow-up email...",success
3,u003,hr_policy,gpt-4,"Can you review this employee performance note and suggest improvements?","The note can be rewritten as follows...",success
4,u001,customer_support,gpt-3.5,"Classify this ticket as billing, technical, or account access.","Billing",success
5,u004,legal_review,gpt-4,"Review this contract clause and identify risks related to liability.","The clause may create liability exposure...",success
6,u005,analytics,gpt-3.5,"Generate SQL to calculate monthly active users from the events table.","SELECT DATE_TRUNC('month', event_time)...",success
7,u006,finance,gpt-4,"Analyse this invoice and bank account number 12345678 for payment validation.","The invoice appears valid...",success
In a real system, these logs could come from:
- application logs;
- API gateway logs;
- database tables;
- prompt logging middleware;
- SaaS AI tool exports;
- internal workflow automation systems.
For this tutorial, the CSV keeps things simple.
Step 1: Load the data
import pandas as pd
df = pd.read_csv("llm_requests.csv")
print(df.head())
We should also check the basic structure.
print(df.info())
print(df["workflow"].value_counts())
This gives us a starting point: who is using the system, which workflows are active, and what kind of model calls are being made.
Step 2: Estimate token usage
In production, the best option is to use actual token counts returned by the model provider.
For this example, we will use a rough approximation:
- 1 token is around 4 characters in English text.
- Total tokens = prompt tokens + response tokens.
This is not exact, but it is good enough for a basic monitoring prototype.
def estimate_tokens(text):
if pd.isna(text):
return 0
return max(1, round(len(str(text)) / 4))
df["prompt_tokens"] = df["prompt"].apply(estimate_tokens)
df["response_tokens"] = df["response"].apply(estimate_tokens)
df["total_tokens"] = df["prompt_tokens"] + df["response_tokens"]
print(df[["request_id", "workflow", "model", "total_tokens"]])
In a real application, replace this approximation with the token usage returned by the API.
Step 3: Add model pricing
Different models have different costs. We can create a simple pricing table.
The numbers below are only example prices. You should replace them with current pricing from your model provider.
pricing = {
"gpt-4": {
"input_per_1k": 0.03,
"output_per_1k": 0.06
},
"gpt-3.5": {
"input_per_1k": 0.0015,
"output_per_1k": 0.002
}
}
Now we can estimate cost per request.
def estimate_cost(row):
model = row["model"]
if model not in pricing:
return 0
input_cost = (row["prompt_tokens"] / 1000) * pricing[model]["input_per_1k"]
output_cost = (row["response_tokens"] / 1000) * pricing[model]["output_per_1k"]
return input_cost + output_cost
df["estimated_cost_usd"] = df.apply(estimate_cost, axis=1)
print(df[["request_id", "workflow", "model", "total_tokens", "estimated_cost_usd"]])
This gives us an estimated cost for each LLM request.
Step 4: Summarise cost by workflow
Cost is more useful when grouped by business workflow.
workflow_cost = (
df.groupby("workflow")
.agg(
requests=("request_id", "count"),
total_tokens=("total_tokens", "sum"),
estimated_cost_usd=("estimated_cost_usd", "sum")
)
.reset_index()
.sort_values("estimated_cost_usd", ascending=False)
)
print(workflow_cost)
This helps answer:
- Which workflow is most expensive?
- Which workflows may need prompt optimisation?
- Which use cases are growing fastest?
A workflow with high cost is not automatically bad. It may be valuable. But without this visibility, the team cannot make informed decisions.
Step 5: Summarise cost by user
User-level monitoring can help detect unusual behaviour.
user_cost = (
df.groupby("user_id")
.agg(
requests=("request_id", "count"),
total_tokens=("total_tokens", "sum"),
estimated_cost_usd=("estimated_cost_usd", "sum")
)
.reset_index()
.sort_values("estimated_cost_usd", ascending=False)
)
print(user_cost)
This can be useful for internal AI tools where different teams use the same system.
For example, if one user or team generates unusually high usage, it may indicate:
- a genuine high-value use case;
- a prompt that is too long;
- repeated retries;
- misuse;
- an automation loop;
- lack of user training.
Step 6: Flag risky prompts
Now we can add a simple risk scanner.
This is not a full data-loss-prevention system. It is a basic first layer that flags prompts containing sensitive terms or patterns.
import re
risk_patterns = {
"payment_card": r"b(?:d[ -]*?){13,16}b",
"bank_account": r"bd{8}b",
"confidential_terms": r"b(confidential|private|secret|internal only)b",
"employee_data": r"b(employee|performance note|salary|disciplinary)b",
"legal_content": r"b(contract|liability|clause|legal)b"
}
Now apply the patterns to each prompt.
def detect_risks(text):
if pd.isna(text):
return []
detected = []
text = str(text).lower()
for risk_name, pattern in risk_patterns.items():
if re.search(pattern, text, flags=re.IGNORECASE):
detected.append(risk_name)
return detected
df["risk_flags"] = df["prompt"].apply(detect_risks)
df["risk_count"] = df["risk_flags"].apply(len)
df["has_risk"] = df["risk_count"] > 0
print(df[["request_id", "workflow", "prompt", "risk_flags"]])
This gives us a basic view of which prompts may need review.
Step 7: Create a risk summary
We can now summarise risk by workflow.
risk_summary = (
df.groupby("workflow")
.agg(
total_requests=("request_id", "count"),
risky_requests=("has_risk", "sum"),
estimated_cost_usd=("estimated_cost_usd", "sum")
)
.reset_index()
)
risk_summary["risk_rate"] = (
risk_summary["risky_requests"] / risk_summary["total_requests"]
)
risk_summary = risk_summary.sort_values("risk_rate", ascending=False)
print(risk_summary)
This helps identify which workflows are most likely to involve sensitive or high-risk content.
For example:
- HR workflows may contain employee data.
- Finance workflows may contain bank details.
- Legal workflows may contain contracts.
- Customer support workflows may contain personal information.
The goal is not to block every request. The goal is to understand which workflows need stronger controls.
Step 8: Add a simple priority score
A useful monitoring tool should help teams decide what to review first.
We can create a simple priority score based on:
- risk count;
- estimated cost;
- model used;
- workflow type.
high_risk_workflows = ["finance", "legal_review", "hr_policy"]
def calculate_priority(row):
score = 0
# Risk flags
score += row["risk_count"] * 3
# Expensive model
if row["model"] == "gpt-4":
score += 2
# High-risk workflow
if row["workflow"] in high_risk_workflows:
score += 3
# Higher token usage
if row["total_tokens"] > 100:
score += 1
return score
df["priority_score"] = df.apply(calculate_priority, axis=1)
review_queue = df.sort_values("priority_score", ascending=False)
print(review_queue[[
"request_id",
"workflow",
"model",
"total_tokens",
"estimated_cost_usd",
"risk_flags",
"priority_score"
]])
This creates a lightweight review queue.
Requests with higher scores may need:
- manual review;
- prompt rewriting;
- workflow restrictions;
- user training;
- model downgrade;
- stronger data controls.
Step 9: Generate a simple report
Now we can generate a short summary.
total_requests = len(df)
total_cost = df["estimated_cost_usd"].sum()
risky_requests = df["has_risk"].sum()
risk_rate = risky_requests / total_requests
print("LLM Usage Summary")
print("-----------------")
print(f"Total requests: {total_requests}")
print(f"Estimated cost: ${total_cost:.4f}")
print(f"Risky requests: {risky_requests}")
print(f"Risk rate: {risk_rate:.1%}")
print("nTop workflows by cost:")
print(workflow_cost.head())
print("nHighest priority requests:")
print(review_queue[[
"request_id",
"workflow",
"model",
"risk_flags",
"priority_score"
]].head())
This kind of report can be run daily or weekly.
It can also be exported to CSV.
workflow_cost.to_csv("workflow_cost_summary.csv", index=False)
risk_summary.to_csv("workflow_risk_summary.csv", index=False)
review_queue.to_csv("llm_review_queue.csv", index=False)
Step 10: Visualise cost by workflow
A simple bar chart can make the result easier to understand.
import matplotlib.pyplot as plt
plt.figure(figsize=(10, 6))
plt.bar(workflow_cost["workflow"], workflow_cost["estimated_cost_usd"])
plt.xlabel("Workflow")
plt.ylabel("Estimated cost in USD")
plt.title("Estimated LLM cost by workflow")
plt.xticks(rotation=45, ha="right")
plt.tight_layout()
plt.show()
You can also visualise risky requests.
plt.figure(figsize=(10, 6))
plt.bar(risk_summary["workflow"], risk_summary["risky_requests"])
plt.xlabel("Workflow")
plt.ylabel("Risky requests")
plt.title("Risky LLM requests by workflow")
plt.xticks(rotation=45, ha="right")
plt.tight_layout()
plt.show()
These charts are simple, but they are enough to start a conversation with product, data, security, or finance teams.
What this prototype shows
This small project gives us a basic visibility layer around LLM usage.
It helps answer:
- Which workflows use the most tokens?
- Which workflows cost the most?
- Which users generate the most usage?
- Which prompts contain risky content?
- Which requests should be reviewed first?
This is not a complete AI governance system, but it is a practical starting point.
Many organisations do not need a complex platform on day one. They need a simple way to see what is happening.
How to improve this further
This prototype can be extended in many ways.
Use real token counts
Instead of estimating tokens by character length, collect actual token usage from the model provider.
Add user and team metadata
Join request logs with team, department, or cost-centre data.
Track success and failure rates
Add fields such as:
- error type;
- retry count;
- latency;
- timeout;
- fallback model;
- user feedback.
Add data classification
Use a more advanced scanner to detect:
- personally identifiable information;
- financial data;
- health data;
- legal data;
- customer records;
- source code;
- credentials.
Build a Streamlit dashboard
A simple Streamlit interface could show:
- total cost;
- cost by workflow;
- cost by user;
- risky prompts;
- review queue;
- model usage;
- trend over time.
Store logs in a database
Instead of using CSV files, store logs in PostgreSQL, BigQuery, Snowflake, or another analytics database.
Add governance actions
For high-risk requests, the system could:
- flag for review;
- block the request;
- redact sensitive values;
- require approval;
- route to a safer model;
- warn the user before submission.
Final thoughts
AI systems do not become production-ready just because the model works.
They need visibility.
Teams need to understand how data moves, which workflows create risk, where costs are coming from, and which requests require human review.
A simple Python monitoring layer can provide that first level of visibility. It does not need to be perfect. It just needs to make invisible problems visible.
For teams adopting LLMs, that is often the difference between an exciting prototype and a system that can actually be trusted in production.