Shizu0n's picture
refactor: model scope for sql query only
ad5be9b
from dataclasses import dataclass, field
@dataclass(frozen=True)
class ConversationState:
active_schema: str = ""
last_intent: str | None = None
debug: dict = field(default_factory=dict)
@classmethod
def from_value(cls, value=None, *, active_schema=""):
if isinstance(value, cls):
if active_schema and active_schema != value.active_schema:
return value.with_active_schema(active_schema)
return value
if not isinstance(value, dict):
return cls(active_schema=(active_schema or "").strip())
state_active_schema = (value.get("active_schema") or active_schema or "").strip()
return cls(
active_schema=state_active_schema,
last_intent=value.get("last_intent"),
debug=dict(value.get("debug") or {}),
)
def to_dict(self):
return {
"active_schema": self.active_schema,
"last_intent": self.last_intent,
"debug": dict(self.debug or {}),
}
def with_active_schema(self, schema):
return ConversationState(
active_schema=(schema or "").strip(),
last_intent=self.last_intent,
debug=dict(self.debug or {}),
)
def with_intent(self, intent_result):
debug = dict(self.debug or {})
debug["intent"] = getattr(intent_result, "intent", None)
debug["confidence"] = getattr(intent_result, "confidence", None)
debug["reason"] = getattr(intent_result, "reason", None)
return ConversationState(
active_schema=self.active_schema,
last_intent=getattr(intent_result, "intent", None),
debug=debug,
)
def default_state(active_schema=""):
return ConversationState(active_schema=(active_schema or "").strip()).to_dict()