LQConstPlanWriterAgent/core/document_chat/schemas.py

# -*- coding: utf-8 -*-
"""文档对话模块数据模型。"""

from typing import Any, Dict, List, Literal, Optional

from pydantic import BaseModel, Field


class SelectedSection(BaseModel):
    index: str = Field(default="", description="章节编号，例如 2.1")
    title: str = Field(default="", description="章节标题")
    content: str = Field(default="", description="当前编辑器中的章节内容")
    code: str = Field(default="", description="章节代码标识")
    chapter_level_1: str = Field(default="", description="可选的一级章节分类，用于检索范围限定")
    chapter_level_2: str = Field(default="", description="可选的二级章节分类，用于检索范围限定")


class DocumentContext(BaseModel):
    before: str = Field(default="", description="前文上下文片段")
    after: str = Field(default="", description="后文上下文片段")
    siblings: List[Dict[str, Any]] = Field(default_factory=list)
    references: List[Dict[str, Any]] = Field(default_factory=list)
    retrieval_filters: Dict[str, Any] = Field(default_factory=dict)


class DocumentChatRequest(BaseModel):
    user_id: str
    message: str = Field(..., min_length=1, description="用户消息内容")
    selected_section: Optional[SelectedSection] = Field(default=None, description="选中的章节；为空时表示通用问题")
    conversation_id: Optional[str] = None
    task_id: Optional[str] = None
    project_info: Dict[str, Any] = Field(default_factory=dict)
    document_context: DocumentContext = Field(default_factory=DocumentContext)
    conversation_history: List[Dict[str, Any]] = Field(default_factory=list)
    response_mode: Literal["json", "sse"] = "json"

    class Config:
        extra = "forbid"


class IntentResult(BaseModel):
    intent: Literal["document_modify", "document_answer", "clarify", "unsupported"]
    confidence: float = Field(default=0.0, ge=0.0, le=1.0)
    skill_name: Optional[str] = None
    operation: str = ""
    target_scope: str = "selected_section"
    normalized_instruction: str = ""
    needs_clarification: bool = False
    clarification_question: str = ""
    reason: str = ""
    warnings: List[str] = Field(default_factory=list)


class DocumentChatSkillInput(BaseModel):
    user_id: str
    user_message: str
    selected_section: Optional[SelectedSection] = None
    intent_result: IntentResult
    conversation_id: Optional[str] = None
    task_id: Optional[str] = None
    project_info: Dict[str, Any] = Field(default_factory=dict)
    document_context: DocumentContext = Field(default_factory=DocumentContext)
    conversation_history: List[Dict[str, Any]] = Field(default_factory=list)


class DocumentChatSkillOutput(BaseModel):
    skill_name: str
    response_type: Literal["answer", "proposal", "clarify", "unsupported", "general_answer", "error"]
    answer: Optional[str] = None
    old_content: Optional[str] = None
    proposed_content: Optional[str] = None
    change_summary: List[str] = Field(default_factory=list)
    references: List[Dict[str, Any]] = Field(default_factory=list)
    warnings: List[str] = Field(default_factory=list)


class DiffItem(BaseModel):
    type: Literal["equal", "insert", "delete", "replace", "full_content"]
    old_text: str = ""
    new_text: str = ""


class DiffResult(BaseModel):
    old_content_hash: str
    new_content_hash: str
    diff: List[DiffItem] = Field(default_factory=list)
    diff_granularity: Literal["line", "full_content"] = "line"


class DocumentChatData(BaseModel):
    callback_task_id: str
    response_type: Literal["answer", "proposal", "clarify", "unsupported", "general_answer", "error"]
    intent_result: Optional[Dict[str, Any]] = None
    answer: Optional[str] = None
    proposed_content: Optional[str] = None
    old_content_hash: Optional[str] = None
    new_content_hash: Optional[str] = None
    diff: List[Dict[str, Any]] = Field(default_factory=list)
    diff_granularity: Optional[str] = None
    change_summary: List[str] = Field(default_factory=list)
    references: List[Dict[str, Any]] = Field(default_factory=list)
    retrieval_status: Optional[str] = None
    retrieval_metrics: Dict[str, Any] = Field(default_factory=dict)
    warnings: List[str] = Field(default_factory=list)
    selected_section: Dict[str, Any] = Field(default_factory=dict)
    error_message: Optional[str] = None


class DocumentChatResponse(BaseModel):
    code: int
    message: str
    data: Optional[DocumentChatData] = None


def model_to_dict(value: Any) -> Dict[str, Any]:
    """将 Pydantic v1/v2 模型转为字典。"""
    if value is None:
        return {}
    if isinstance(value, dict):
        return value
    if hasattr(value, "model_dump"):
        return value.model_dump()
    if hasattr(value, "dict"):
        return value.dict()
    return dict(value)