How to Monitor LLM API Usage, Costs, and Performance Metrics Across Providers

The lmforge platform tracks cross-provider LLM expenses and performance through a three-layer architecture that aggregates provider pricing metadata, per-step token costs, and application-wide analytics via the AnalysisService.

Effective LLM API usage monitoring requires granular visibility into token consumption, latency, and spend across disparate providers. The haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents repository implements a comprehensive telemetry system that captures pricing metadata at the provider level, calculates costs per inference step, and aggregates metrics at the application layer. This guide explains how to leverage these components to monitor LLM API usage costs and performance metrics across providers in real time.

Provider-Level Pricing Metadata

Every language model provider in the system supplies standardized pricing metadata through an abstract interface. The BaseLanguageModel class in api/internal/core/language_model/entities/model_entity.py defines the get_pricing() method (lines 80–86), which reads a metadata dictionary containing pricing.input, pricing.output, and pricing.unit fields.

This design ensures that whether you are using OpenAI, DeepSeek, Wenxin, or other supported providers, the cost structure is accessible through a uniform API. The unit price typically represents the cost per 1,000 tokens, allowing the system to normalize calculations across different pricing models.

Per-Step Cost Calculation

Each agent execution—such as the Chat agent implementations for various providers—invokes self.llm.get_pricing() during the inference lifecycle. The agent multiplies the input and output token counts by their respective unit prices to derive a total_price for that specific step.

According to the source code in api/internal/model/conversation.py (lines 84–100), these calculated values are persisted immediately to the relational Message table alongside the conversation content. The table stores both total_token_count and total_price for every LLM interaction, creating an immutable audit trail of consumption at the message level.

Application-Wide Aggregation and Caching

For macro-level visibility, the AnalysisService (api/internal/service/analysis_service.py) performs aggregation across all messages belonging to a specific application. The service queries the Message table to sum total_price into cost consumption metrics and calculates the token output rate by dividing total token count by total latency.

The implementation (lines 22–78, 95–127, and 144–166) also generates period-over-period percentage-change ("pop") values and 7-day time-series trends. To optimize performance for frequently accessed dashboards, results are cached in Redis for one day using the service's redis_client.

HTTP Endpoint Exposure

The aggregated analytics are exposed through a REST endpoint registered in api/internal/router/router.py:

router.add_url_rule(
    "/analysis/<uuid:app_id>", view_func=self.analysis_handler.get_app_analysis
)

The AnalysisHandler (api/internal/handler/analysis_handler.py, lines 15–18) forwards requests to the service layer and returns a JSON payload containing cost consumption, token rates, and trend data.

Retrieving Metrics via the API

You can fetch real-time analytics for any application using the public endpoint. The following Python client example retrieves cost consumption, token output rate, and 7-day message trends:

import requests

APP_ID = "d2c1a9e4-3f6b-4f2a-9d2e-5b8c7e9f1a2b"   # Replace with your app UUID

BASE_URL = "https://your‑domain.com/api"

resp = requests.get(
    f"{BASE_URL}/analysis/{APP_ID}", 
    cookies={"session": "YOUR_SESSION_COOKIE"}
)
data = resp.json()["data"]

print("Cost consumption (CNY):", data["cost_consumption"]["data"])
print("Token output rate (tokens/s):", data["token_output_rate"]["data"])
print("7‑day trend – total messages:", data["total_messages_trend"]["y_axis"])

Direct Database Queries and Manual Calculations

Because cost data resides in the relational Message table, you can bypass the API and query directly via SQLAlchemy for custom reporting:

from internal.model.conversation import Message
from internal.service.app_service import AppService
from sqlalchemy import func

app = AppService().get_app(app_id, current_user)
cost = (
    db.session.query(func.sum(Message.total_price))
    .filter(Message.app_id == app.id)
    .scalar()
)
print(f"Total cost for app: ¥{float(cost):.2f}")

To inspect pricing metadata for a specific provider programmatically:

from internal.service.language_model_service import LanguageModelService

service = LanguageModelService()
model_info = service.get_language_model("openai", "gpt‑3.5‑turbo")
pricing = model_info["pricing"]          # {"input": 0.0005, "output": 0.0015, "unit": 1000}

print(f"OpenAI GPT‑3.5 pricing: {pricing}")

For custom cost calculations outside the standard agent flow:

input_tokens, output_tokens = 120, 340

# Pricing: CNY per 1k tokens

input_price, output_price, unit = 0.0005, 0.0015, 1000

total_price = ((input_tokens / unit) * input_price +
               (output_tokens / unit) * output_price)
print(f"Step cost = ¥{total_price:.4f}")

Key Implementation Files

The monitoring system spans several critical components:

• api/internal/core/language_model/entities/model_entity.py – Defines BaseLanguageModel.get_pricing() to read provider-specific price information.

• api/internal/model/conversation.py – Persists total_token_count and total_price for every LLM step in the Message table.

• api/internal/core/agent/agents/react_agent.py (and similar agent files) – Implements per-step cost calculation logic by multiplying token counts against pricing metadata.

• api/internal/service/analysis_service.py – Aggregates cost consumption, token output rates, and trends; manages Redis caching.

• api/internal/handler/analysis_handler.py – HTTP handler exposing aggregated data via /analysis/<app_id>.

• api/internal/router/router.py – Maps the analysis endpoint URL to the handler.

• api/config/default_config.py – Configures the Redis client used by the analysis service for caching.

Summary

• Provider metadata is standardized through BaseLanguageModel.get_pricing() in model_entity.py, exposing pricing.input, pricing.output, and pricing.unit fields.
• Per-step costs are calculated by agents and stored in the Message table alongside token counts, creating a granular audit trail.
• Application analytics are aggregated by AnalysisService, which computes total spend, token output rates, and 7-day trends, caching results in Redis for 24 hours.
• Access patterns include the REST endpoint /analysis/<uuid:app_id> for dashboard data and direct SQLAlchemy queries for custom reporting.

Frequently Asked Questions

How does the platform normalize pricing across different LLM providers?

The BaseLanguageModel abstract class enforces a uniform get_pricing() interface that returns a dictionary with input, output, and unit keys. This allows agents to calculate costs identically whether the provider charges per 1,000 tokens or per 1,000,000 tokens, as the unit field contextualizes the arithmetic.

Where is the cost data physically stored for each conversation?

Every inference step persists cost data to the Message table defined in api/internal/model/conversation.py. The table contains total_token_count and total_price columns that record the actual consumption and calculated expense for that specific message.

How long are aggregated analytics cached, and can this be configured?

The AnalysisService caches aggregated results in Redis for one day (24 hours) by default. This duration is controlled through the Redis client configuration in api/config/default_config.py and the caching logic within the service layer.

Can I calculate costs for a batch of messages without using the built-in agent?

Yes. You can retrieve raw pricing via LanguageModelService.get_language_model(), then apply the formula ((input_tokens / unit) * input_price) + ((output_tokens / unit) * output_price). Alternatively, query the Message table directly using SQLAlchemy to sum existing total_price values for any custom date range or application subset.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}