| 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() |