app/agents/state.py

"""AgentState - LangGraph 跨节点状态.

使用 TypedDict 而不是 Pydantic, 因为 LangGraph 内置 add_messages reducer
需要 Annotated[Sequence[BaseMessage], add_messages] 这样的类型签名.
"""
from __future__ import annotations

from typing import Annotated, Any, Literal, Sequence, TypedDict

from langchain_core.documents import Document
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

from app.services.vector_store import RetrievalHit


RouteDecision = Literal["direct", "retrieve", "multi_step"]


class AgentState(TypedDict, total=False):
    """LangGraph 跨节点状态. total=False 让所有字段可选, 各 node 自行填充."""

    # ===== 基础 =====
    messages: Annotated[Sequence[BaseMessage], add_messages]
    session_id: str
    user_id: str
    locale: str  # "zh" | "en"

    # ===== 路由 =====
    route_decision: RouteDecision
    query_rewritten: str  # 改写后的查询
    plan: list[str]  # 多步拆解的子任务

    # ===== 检索 =====
    retrieved: list[RetrievalHit]
    reranked: list[RetrievalHit]
    retrieved_doc_ids: list[str]  # 命中 doc 列表 (用于 SSE retrieval 事件)

    # ===== 引用 =====
    citations: list[dict[str, Any]]  # [{doc_id, page, snippet, score, source}]

    # ===== 工具 =====
    tool_calls: list[dict[str, Any]]  # 工具调用历史
    tool_results: list[dict[str, Any]]  # 工具结果摘要

    # ===== CRAG 自校正 =====
    iteration: int
    max_iterations: int
    relevance_score: float  # top-1 rerank score, 0-1
    relevance_verdict: Literal["relevant", "ambiguous", "irrelevant"]
    needs_more_retrieval: bool
    crag_finished: bool

    # ===== 元数据 =====
    elapsed_ms: int
    final_answer: str
    error: str | None


# ========== State helpers ==========
def empty_state_for(session_id: str, user_id: str = "default", locale: str = "zh") -> AgentState:
    from app.config import settings

    return AgentState(
        messages=[],
        session_id=session_id,
        user_id=user_id,
        locale=locale,
        route_decision="retrieve",
        query_rewritten="",
        plan=[],
        retrieved=[],
        reranked=[],
        retrieved_doc_ids=[],
        citations=[],
        tool_calls=[],
        tool_results=[],
        iteration=0,
        max_iterations=settings.crag_max_iterations,
        relevance_score=0.0,
        relevance_verdict="ambiguous",
        needs_more_retrieval=False,
        crag_finished=False,
        elapsed_ms=0,
        final_answer="",
        error=None,
    )


def messages_to_lc(messages: list[dict[str, Any]] | list[BaseMessage]) -> list[BaseMessage]:
    """业务侧 dict 消息列表 (来自 SQLite) 转为 LangChain BaseMessage 列表."""
    from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage

    out: list[BaseMessage] = []
    for m in messages:
        if hasattr(m, "type"):
            out.append(m)  # 已经是 BaseMessage
            continue
        role = m.get("role", "user")
        content = m.get("content", "")
        if role == "system":
            out.append(SystemMessage(content=content))
        elif role == "user":
            out.append(HumanMessage(content=content))
        elif role == "assistant":
            extra = {}
            if m.get("tool_calls"):
                extra["tool_calls"] = m["tool_calls"]
            out.append(AIMessage(content=content, **extra))
        elif role == "tool":
            out.append(ToolMessage(
                content=content,
                tool_call_id=m.get("tool_call_id", ""),
            ))
    return out