LangGraph TypedDict & Pydantic State Schema Errors

from typing import Annotated, TypedDict, Optional
from operator import add
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages


class State(TypedDict):
    # Accumulating field — messages APPEND across nodes (add_messages handles IDs)
    messages: Annotated[list[AnyMessage], add_messages]

    # Accumulating list of arbitrary items — use operator.add
    citations: Annotated[list[str], add]

    # Overwritten field — plain type, no reducer
    question: str

    # Optional field — use total=False on the class, or Optional[T]
    classification: Optional[str]


# ✅ Node returns a full update
def classifier(state: State) -> dict:
    q = state["question"]
    label = classify(q)
    return {"classification": label}          # overwrites


# ✅ Node appends to messages via the add_messages reducer
def responder(state: State) -> dict:
    reply = llm.invoke(state["messages"])
    return {"messages": [reply]}              # appended, not replaced


# ✅ Node appends citations
def cite(state: State) -> dict:
    return {"citations": ["doc-42", "doc-99"]}   # merged with prior citations


# Initial state provides every REQUIRED key
initial = {
    "messages": [],
    "citations": [],
    "question": "What is LangGraph?",
    "classification": None,
}
result = app.invoke(initial)

from typing import Annotated, Optional, Literal
from pydantic import BaseModel, Field, field_validator
from langchain_core.messages import AnyMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages


class State(BaseModel):
    # Reducer-bearing fields work the same way with Annotated
    messages: Annotated[list[AnyMessage], add_messages] = Field(default_factory=list)

    # Validated fields
    question: str = Field(min_length=1, max_length=2000)
    temperature: float = Field(ge=0.0, le=2.0, default=0.7)
    mode: Literal["chat", "search", "math"] = "chat"

    # Optional with default — the RIGHT way to say "may be missing"
    classification: Optional[str] = None
    retry_count: int = Field(ge=0, default=0)

    @field_validator("question")
    @classmethod
    def strip_whitespace(cls, v: str) -> str:
        v = v.strip()
        if not v:
            raise ValueError("question cannot be empty after stripping whitespace")
        return v


def classifier(state: State) -> dict:
    # state is a BaseModel instance — dot access works
    label = classify(state.question)
    return {"classification": label}


graph = StateGraph(State)                     # <-- pass the BaseModel class
graph.add_node("classifier", classifier)
graph.add_edge(START, "classifier")
graph.add_edge("classifier", END)

app = graph.compile()

# Invoke with a dict — Pydantic validates on the way in
result = app.invoke({"question": "  what is LangGraph?  "})
# Passing invalid input raises ValidationError before any node runs:
# result = app.invoke({"question": "", "temperature": 5.0})   # -> ValidationError

from typing import Annotated, TypedDict
from operator import add
from langgraph.graph.message import add_messages


# ❌ BUG — no reducer, messages get overwritten every node
class BadState(TypedDict):
    messages: list         # <-- just list, no Annotated
    tool_calls: list


# In this graph, node_b sees only its own messages — node_a's are LOST
def node_a(state):
    return {"messages": [("system", "a's message")]}

def node_b(state):
    # state["messages"] is JUST [("system", "a's message")] here,
    # then node_b overwrites it entirely
    return {"messages": [("system", "b's message")]}
# Final state: messages = [("system", "b's message")]


# ✅ FIX — annotate messages with add_messages (special-case reducer for chat)
class GoodState(TypedDict):
    messages: Annotated[list, add_messages]     # appends, handles message IDs
    tool_calls: Annotated[list, add]            # plain concat for other lists

# Same nodes as above, but with add_messages reducer:
# Final state: messages = [
#     ("system", "a's message"),
#     ("system", "b's message"),
# ]


# ✅ Custom reducer — merge dicts by key
def merge_dict(current: dict, update: dict) -> dict:
    return {**(current or {}), **(update or {})}

class StateWithDict(TypedDict):
    metadata: Annotated[dict, merge_dict]


# Guard: reducers receive None on first call — always default the base
def safe_concat(current: list | None, update: list) -> list:
    return (current or []) + (update or [])

class StateWithSafeReducer(TypedDict):
    events: Annotated[list, safe_concat]