fenic.core.metrics

Metrics tracking for query execution and model usage.

This module defines classes for tracking various metrics during query execution, including language model usage, embedding model usage, operator performance, and overall query statistics.

Classes:

• LMMetrics –

  Tracks language model usage metrics including token counts and costs.

• OperatorMetrics –

  Metrics for a single operator in the query execution plan.

• PhysicalPlanRepr –

  Tree node representing the physical execution plan, used for pretty printing execution plan.

• QueryMetrics –

  Comprehensive metrics for an executed query.

• RMMetrics –

  Tracks embedding model usage metrics including token counts and costs.

LMMetrics dataclass

LMMetrics(num_uncached_input_tokens: int = 0, num_cached_input_tokens: int = 0, num_output_tokens: int = 0, cost: float = 0.0, num_requests: int = 0)

Tracks language model usage metrics including token counts and costs.

Attributes:

• num_uncached_input_tokens (int) –

  Number of uncached tokens in the prompt/input

• num_cached_input_tokens (int) –

  Number of cached tokens in the prompt/input,

• num_output_tokens (int) –

  Number of tokens in the completion/output

• cost (float) –

  Total cost in USD for the LM API call

OperatorMetrics dataclass

OperatorMetrics(operator_id: str, num_output_rows: int = 0, execution_time_ms: float = 0.0, lm_metrics: LMMetrics = LMMetrics(), rm_metrics: RMMetrics = RMMetrics(), is_cache_hit: bool = False)

Metrics for a single operator in the query execution plan.

Attributes:

• operator_id (str) –

  Unique identifier for the operator

• num_output_rows (int) –

  Number of rows output by this operator

• execution_time_ms (float) –

  Execution time in milliseconds

• lm_metrics (LMMetrics) –

  Language model usage metrics for this operator

• is_cache_hit (bool) –

  Whether results were retrieved from cache

PhysicalPlanRepr dataclass

PhysicalPlanRepr(operator_id: str, children: List[PhysicalPlanRepr] = list())

Tree node representing the physical execution plan, used for pretty printing execution plan.

QueryMetrics dataclass

QueryMetrics(execution_time_ms: float = 0.0, num_output_rows: int = 0, total_lm_metrics: LMMetrics = LMMetrics(), total_rm_metrics: RMMetrics = RMMetrics(), _operator_metrics: Dict[str, OperatorMetrics] = dict(), _plan_repr: PhysicalPlanRepr = lambda: PhysicalPlanRepr(operator_id='empty')())

Comprehensive metrics for an executed query.

Includes overall statistics and detailed metrics for each operator in the execution plan.

Attributes:

• execution_time_ms (float) –

  Total query execution time in milliseconds

• num_output_rows (int) –

  Total number of rows returned by the query

• total_lm_metrics (LMMetrics) –

  Aggregated language model metrics across all operators

Methods:

• get_execution_plan_details –

  Generate a formatted execution plan with detailed metrics.

• get_summary –

  Summarize the query metrics in a single line.

get_execution_plan_details

get_execution_plan_details() -> str

Generate a formatted execution plan with detailed metrics.

Produces a hierarchical representation of the query execution plan, including performance metrics and language model usage for each operator.

Returns:

• str ( str ) –

  A formatted string showing the execution plan with metrics.

Source code in src/fenic/core/metrics.py

def get_execution_plan_details(self) -> str:
    """Generate a formatted execution plan with detailed metrics.

    Produces a hierarchical representation of the query execution plan,
    including performance metrics and language model usage for each operator.

    Returns:
        str: A formatted string showing the execution plan with metrics.
    """

    def _format_node(node: PhysicalPlanRepr, indent: int = 1) -> str:
        op = self._operator_metrics[node.operator_id]
        indent_str = "  " * indent

        details = [
            f"{indent_str}{op.operator_id}",
            f"{indent_str}  Output Rows: {op.num_output_rows:,}",
            f"{indent_str}  Execution Time: {op.execution_time_ms:.2f}ms",
            f"{indent_str}  Cached: {op.is_cache_hit}",
        ]

        if op.lm_metrics.cost > 0:
            details.extend(
                [
                    f"{indent_str}  Language Model Usage: {op.lm_metrics.num_uncached_input_tokens:,} input tokens, {op.lm_metrics.num_cached_input_tokens:,} cached input tokens, {op.lm_metrics.num_output_tokens:,} output tokens",
                    f"{indent_str}  Language Model Cost: ${op.lm_metrics.cost:.6f}",
                ]
            )

        if op.rm_metrics.cost > 0:
            details.extend(
                [
                    f"{indent_str}  Embedding Model Usage: {op.rm_metrics.num_input_tokens:,} input tokens",
                    f"{indent_str}  Embedding Model Cost: ${op.rm_metrics.cost:.6f}",
                ]
            )
        return (
            "\n".join(details)
            + "\n"
            + "".join(_format_node(child, indent + 1) for child in node.children)
        )

    return f"Execution Plan\n{_format_node(self._plan_repr)}"

get_summary

get_summary() -> str

Summarize the query metrics in a single line.

Returns:

• str ( str ) –

  A concise summary of execution time, row count, and LM cost.

Source code in src/fenic/core/metrics.py

def get_summary(self) -> str:
    """Summarize the query metrics in a single line.

    Returns:
        str: A concise summary of execution time, row count, and LM cost.
    """
    return (
        f"Query executed in {self.execution_time_ms:.2f}ms, "
        f"returned {self.num_output_rows:,} rows, "
        f"language model cost: ${self.total_lm_metrics.cost:.6f}, "
        f"embedding model cost: ${self.total_rm_metrics.cost:.6f}"
    )

RMMetrics dataclass

RMMetrics(num_input_tokens: int = 0, num_requests: int = 0, cost: float = 0.0)

Tracks embedding model usage metrics including token counts and costs.

Attributes:

• num_input_tokens (int) –

  Number of tokens to embed

• cost (float) –

  Total cost in USD to embed the tokens