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