| from typing import List, Optional, Literal |
| from pydantic import BaseModel, Field, field_validator |
|
|
| class ComplexityLevel(BaseModel): |
| level: Literal["simple", "moderate", "complex"] = Field(description="Complexity level of the query") |
| reasoning: str = Field(description="Explanation for the complexity assessment") |
| needs_planning: bool = Field(description="Whether this query requires detailed planning") |
| suggested_approach: str = Field(description="Recommended approach for handling this query") |
|
|
| class CritiqueFeedback(BaseModel): |
| quality_score: int = Field(ge=1, le=10, description="Quality score from 1-10") |
| is_complete: bool = Field(description="Whether the answer is complete") |
| is_accurate: bool = Field(description="Whether the answer appears accurate") |
| missing_elements: List[str] = Field(default_factory=list, description="What's missing from the answer") |
| errors_found: List[str] = Field(default_factory=list, description="Potential errors identified") |
| suggested_improvements: List[str] = Field(default_factory=list, description="Suggestions for improvement") |
| needs_replanning: bool = Field(description="Whether the plan should be revised") |
| replan_instructions: Optional[str] = Field(default=None, description="Instructions for replanning") |
|
|
| TaskType = Literal["info", "calc", "table", "doc_qa", "image_qa", "multi_hop"] |
|
|
| class PlanStep(BaseModel): |
| id: str = Field(description="Unique step identifier (e.g., s1)") |
| goal: str = Field(description="What the step accomplishes and why") |
| tool: Optional[str] = Field(default=None, description="Exact tool name or null when no tool is required") |
| inputs: Optional[str] = Field(default=None, description="Important inputs or references needed for the step") |
| expected_result: str = Field(description="How to confirm the step succeeded") |
| on_fail: str = Field(default="replan", description="Fallback action if the step fails (replan or stop)") |
|
|
| @field_validator("tool", mode="before") |
| @classmethod |
| def normalize_tool(cls, value: Optional[str]) -> Optional[str]: |
| """Ensure blank or null-like values are interpreted as no tool.""" |
|
|
| if value is None: |
| return None |
| if isinstance(value, str): |
| cleaned = value.strip() |
| if not cleaned or cleaned.lower() in {"null", "none"}: |
| return None |
| return cleaned |
| return value |
|
|
| class PlannerPlan(BaseModel): |
| task_type: TaskType |
| summary: str = Field(description="Short explanation of the chosen strategy") |
| assumptions: List[str] = Field(default_factory=list) |
| steps: List[PlanStep] = Field(default_factory=list) |
| answer_guidelines: Optional[str] = Field(default=None, description="Reminders for formatting, citations, etc.") |
|
|
|
|
| class ToolExecution(BaseModel): |
| tool_name: str |
| arguments: str |
| call_id: str |
| |
| class Config: |
| extra = "forbid" |
|
|
| class ExecutionReport(BaseModel): |
| """Structured report for critic evaluation.""" |
| query_summary: str = Field(description="Brief summary of the user's query") |
| approach_used: str = Field(description="What approach/strategy was used") |
| tools_executed: List[ToolExecution] = Field(default_factory=list, description="List of tools used with results") |
| key_findings: List[str] = Field(default_factory=list, description="Main findings or results") |
| data_sources: List[str] = Field(default_factory=list, description="Sources of information used") |
| assumptions_made: List[str] = Field(default_factory=list, description="Any assumptions made during execution") |
| confidence_level: Literal["low", "medium", "high"] = Field(description="Confidence in the answer") |
| limitations: List[str] = Field(default_factory=list, description="Known limitations or caveats") |
| final_answer: str = Field(description="The actual answer to the user's query") |
|
|
| class Config: |
| extra = "forbid" |
