first-aid-rag-assistant / src / rag / metrics.py
metrics.py
Raw
import logging
import time
from dataclasses import dataclass, field
from datetime import datetime
from types import SimpleNamespace

from config import calculate_cost
from rag.rag_usage import RAGWithUsage

logger = logging.getLogger(__name__)


@dataclass
class LLMCallRecord:
    model: str
    prompt: str
    instructions: str
    answer: str
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    response_time: float
    cost: float
    timestamp: datetime = field(default_factory=datetime.now)


class RAGWithMetrics(RAGWithUsage):

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.last_call: LLMCallRecord = None

    def llm(self, prompt):
        start_time = time.time()
        # Call parent's llm() which handles _call_llm + usage tracking
        answer = super().llm(prompt)
        response_time = time.time() - start_time

        # last_usage is set by super().llm(); normalise missing usage
        usage = self.last_usage
        if usage is None:
            usage = SimpleNamespace(
                prompt_tokens=0, completion_tokens=0, total_tokens=0
            )

        cost = calculate_cost(self.model, usage)
        call_record = LLMCallRecord(
            model=self.model,
            prompt=prompt,
            instructions=self.instructions,
            answer=answer,
            prompt_tokens=usage.prompt_tokens,
            completion_tokens=usage.completion_tokens,
            total_tokens=usage.total_tokens,
            response_time=response_time,
            cost=cost,
        )
        logger.info("LLM call: %s", call_record)
        self.last_call = call_record
        return answer