Files
group_fqcd_jr/app/core/nl2sql_contracts.py

54 lines
1.7 KiB
Python

from typing import Any
from pydantic import BaseModel, ConfigDict, Field, field_validator
class FinancialNL2SQLInput(BaseModel):
"""金融 NL2SQL 公共工具入参;身份和权限只信任 RequestContext。"""
model_config = ConfigDict(extra="forbid", frozen=True)
question: str = Field(min_length=1, max_length=500)
confirmation: str | None = Field(default=None, max_length=200)
dry_run: bool = False
limit: int = Field(default=50, ge=1, le=200)
@field_validator("question")
@classmethod
def question_must_not_be_blank(cls, value: str) -> str:
if not value.strip():
raise ValueError("question must not be blank")
return value.strip()
class FinancialQueryPlan(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True)
intent: str
domains: tuple[str, ...]
tables: tuple[str, ...]
metrics: tuple[str, ...] = ()
dimensions: tuple[str, ...] = ()
filters: tuple[dict[str, Any], ...] = ()
time_mode: str = "none"
time_column: str | None = None
start: str | None = None
end: str | None = None
limit: int = Field(default=50, ge=1, le=200)
confidence: float = Field(default=0.0, ge=0, le=1)
needs_confirmation: bool = False
confirmation_question: str | None = None
unsupported_reason: str | None = None
class FinancialNL2SQLResult(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True)
status: str
message: str
data: dict[str, Any] = Field(default_factory=dict)
query_plan: dict[str, Any] = Field(default_factory=dict)
sql: str | None = None
parameters: dict[str, Any] = Field(default_factory=dict)
audit: dict[str, Any] = Field(default_factory=dict)