Create-knowledge-find-decode/query_planning/domain.py

"""Framework-free domain values for planning search queries."""
from __future__ import annotations

from dataclasses import dataclass, field
from enum import StrEnum
from typing import Any


class GeneratorKind(StrEnum):
    CARTESIAN = "cartesian"
    MANUAL = "manual"
    TOPIC_TABLE = "topic_table"
    AGENT_PLAN = "agent_plan"
    HISTORY_GAP = "history_gap"


@dataclass(frozen=True)
class GenerationRequest:
    generator_kind: GeneratorKind
    payload: dict[str, Any]
    name: str
    target_platforms: tuple[str, ...] = ("xiaohongshu", "weixin", "douyin")
    max_queries: int | None = None
    metadata: dict[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        if not self.name.strip():
            raise ValueError("generation request name must not be empty")
        if not self.target_platforms:
            raise ValueError("at least one target platform is required")
        if self.max_queries is not None and self.max_queries < 0:
            raise ValueError("max_queries must be non-negative or None")


@dataclass(frozen=True)
class KnowledgeNeedDraft:
    need_key: str
    decision_context: str
    unknown_information: str
    source_ref: dict[str, Any] = field(default_factory=dict)
    priority: int = 0
    metadata: dict[str, Any] = field(default_factory=dict)


@dataclass
class QueryCandidate:
    query_text: str
    axes: dict[str, Any] = field(default_factory=dict)
    priority: int = 0
    original_position: int = 0
    source_refs: list[dict[str, Any]] = field(default_factory=list)
    knowledge_need_keys: list[str] = field(default_factory=list)
    metadata: dict[str, Any] = field(default_factory=dict)
    filter_reason: str | None = None
    dedupe_key: str = ""


@dataclass(frozen=True)
class GeneratorOutput:
    candidates: tuple[QueryCandidate, ...]
    knowledge_needs: tuple[KnowledgeNeedDraft, ...] = ()
    generator_config: dict[str, Any] = field(default_factory=dict)


@dataclass(frozen=True)
class GenerationPlanDraft:
    generator_kind: GeneratorKind
    input_snapshot: dict[str, Any]
    generator_config: dict[str, Any]
    metadata: dict[str, Any]


@dataclass(frozen=True)
class GenerationStats:
    generated_count: int
    normalized_count: int
    unique_count: int
    selected_count: int
    dropped_count: int


@dataclass(frozen=True)
class GenerationResult:
    request: GenerationRequest
    plan: GenerationPlanDraft
    knowledge_needs: tuple[KnowledgeNeedDraft, ...]
    selected_queries: tuple[QueryCandidate, ...]
    stats: GenerationStats