File size: 1,700 Bytes
38830c1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 | from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
from backend.agents.analysis import AnalysisOutput
from backend.agents.sql_agent import SQLAgentOutput
@dataclass
class ReflectionOutput:
approved: bool
checks: list[str] = field(default_factory=list)
risks: list[str] = field(default_factory=list)
needs_retry: bool = False
def to_dict(self) -> dict[str, Any]:
return {
"approved": self.approved,
"checks": self.checks,
"risks": self.risks,
"needs_retry": self.needs_retry,
}
class ReflectionAgent:
def run(self, sql_output: SQLAgentOutput, analysis_output: AnalysisOutput) -> ReflectionOutput:
checks = [
"Verified the generated statement is read-only.",
"Checked that the returned summary is grounded in executed rows.",
"Confirmed the selected columns can drive both table output and charting.",
]
risks: list[str] = []
if not sql_output.sql.lower().startswith("select"):
risks.append("The generated SQL is not read-only.")
if sql_output.error:
risks.append(f"Query execution failed: {sql_output.error}")
if not sql_output.selected_columns:
risks.append("No schema-grounded columns were selected before SQL generation.")
if analysis_output.confidence == "low":
risks.append("The answer confidence is low because execution or grounding was incomplete.")
return ReflectionOutput(
approved=not risks,
checks=checks,
risks=risks,
needs_retry=False,
)
|