LangGraph with_structured_output Inside Graph Nodes

from typing import Annotated, Literal, Optional
from pydantic import BaseModel, Field, field_validator


# ❌ TOO STRICT — model often fails at least one field
class StrictAnswer(BaseModel):
    answer: str = Field(min_length=10)
    confidence: float = Field(ge=0.0, le=1.0)
    citations: list[str] = Field(min_length=1, max_length=5)
    reasoning: str = Field(min_length=50)
    tags: list[Literal["factual", "opinion", "computed"]] = Field(min_length=1)


# ✅ FORGIVING — most fields optional or with defaults; model succeeds
class Answer(BaseModel):
    answer: str = Field(description="The concise final answer to the user")
    confidence: float = Field(
        default=0.5,
        ge=0.0, le=1.0,
        description="Confidence in the answer, 0-1",
    )
    citations: list[str] = Field(
        default_factory=list,
        description="Source URLs supporting the answer, if any",
    )
    reasoning: Optional[str] = Field(
        default=None,
        description="Brief reasoning trace (optional)",
    )

    @field_validator("confidence", mode="before")
    @classmethod
    def coerce_conf(cls, v):
        # Model sometimes returns 87 instead of 0.87 — coerce
        if isinstance(v, (int, float)) and v > 1.0:
            return v / 100.0
        return v


# ✅ Use it in a graph node
from langgraph.graph import StateGraph, START, END, MessagesState

def answer_node(state):
    structured_llm = model.with_structured_output(Answer, method="json_schema")
    result: Answer = structured_llm.invoke(state["messages"])
    # Put the structured value in state under an explicit key
    return {"answer": result.model_dump()}


# ✅ State knows about the field
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]
    answer: dict | None      # populated by answer_node


# ✅ include_raw=True — get both the parsed result AND the original AIMessage
def answer_node_with_raw(state):
    wrapper = model.with_structured_output(Answer, method="json_schema", include_raw=True)
    result = wrapper.invoke(state["messages"])
    # result is a dict: {"raw": AIMessage, "parsed": Answer or None, "parsing_error": Exception or None}
    if result["parsed"] is None:
        return {"answer": None, "messages": [result["raw"]]}
    return {"answer": result["parsed"].model_dump(), "messages": [result["raw"]]}

from typing import Annotated, TypedDict, Literal
from pydantic import BaseModel, Field
from langchain_core.messages import AIMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode


class Report(BaseModel):
    summary: str = Field(description="1-3 sentence summary")
    key_findings: list[str] = Field(default_factory=list, description="Bullet-point findings")
    sources_used: list[str] = Field(default_factory=list, description="Tool call names used")


class State(TypedDict):
    messages: Annotated[list, add_messages]
    report: dict | None


# Node 1 — model with tools, loops until no more tool calls
def researcher(state):
    llm_with_tools = model.bind_tools([search, fetch_url])
    reply = llm_with_tools.invoke(state["messages"])
    return {"messages": [reply]}


def route_researcher(state) -> Literal["tools", "structurer"]:
    last = state["messages"][-1]
    if isinstance(last, AIMessage) and last.tool_calls:
        return "tools"
    return "structurer"


# Node 2 — final structuring, NO tools, just structured output
def structurer(state):
    # A separate model call — no bind_tools, just with_structured_output
    structured = model.with_structured_output(Report, method="json_schema")

    # Feed it a fresh prompt that asks for the report shape
    prompt = [
        SystemMessage(content=(
            "Read the conversation above and produce a structured report. "
            "Do NOT call any tools. Just return the Report schema."
        )),
        *state["messages"],
    ]
    report: Report = structured.invoke(prompt)
    return {"report": report.model_dump()}


graph = StateGraph(State)
graph.add_node("researcher", researcher)
graph.add_node("tools", ToolNode([search, fetch_url]))
graph.add_node("structurer", structurer)

graph.add_edge(START, "researcher")
graph.add_conditional_edges("researcher", route_researcher)
graph.add_edge("tools", "researcher")     # loop back with tool results
graph.add_edge("structurer", END)         # terminal

app = graph.compile()

result = app.invoke({
    "messages": [("user", "Research LangGraph's checkpoint model and summarize.")],
    "report": None,
})
print(result["report"])

from typing import Literal, Union
from pydantic import BaseModel, Field, Discriminator, Tag


# ✅ Nested schema — json_schema handles this cleaner than function_calling
class Address(BaseModel):
    street: str
    city: str
    zip: str


class Contact(BaseModel):
    name: str
    email: str
    address: Address              # nested
    phones: list[str] = Field(default_factory=list, max_length=3)


# ✅ Discriminated union — tool_calling flattens; json_schema respects the discriminator
class SearchAction(BaseModel):
    kind: Literal["search"] = "search"
    query: str

class ClickAction(BaseModel):
    kind: Literal["click"] = "click"
    selector: str

class TypeAction(BaseModel):
    kind: Literal["type"] = "type"
    selector: str
    text: str


class BrowserPlan(BaseModel):
    goal: str
    actions: list[SearchAction | ClickAction | TypeAction] = Field(min_length=1)


# ✅ Use json_schema on Claude / GPT / Gemini
structured = model.with_structured_output(
    BrowserPlan,
    method="json_schema",         # <-- explicit; cleaner for nested/union
    strict=True,                  # OpenAI: enforces schema; Claude ignores
)

def planner(state):
    plan: BrowserPlan = structured.invoke(state["messages"])
    return {"plan": plan.model_dump()}


# ✅ include_raw=True is a good default in production for auditability
production_structured = model.with_structured_output(
    Contact,
    method="json_schema",
    include_raw=True,
)

def robust_extract(state):
    result = production_structured.invoke(state["messages"])
    if result["parsed"] is None:
        # Log the raw output and the parsing error, don't crash
        import logging
        logging.warning(
            "structured output failed: err=%s raw=%r",
            result["parsing_error"],
            result["raw"].content[:500],
        )
        return {"contact": None}
    return {"contact": result["parsed"].model_dump()}


# ✅ Fallback chain — try json_schema, fall back to function_calling
def robust_structure(model, schema):
    primary = model.with_structured_output(schema, method="json_schema", include_raw=True)
    fallback = model.with_structured_output(schema, method="function_calling", include_raw=True)

    def _extract(messages):
        r = primary.invoke(messages)
        if r["parsed"] is not None:
            return r["parsed"]
        r = fallback.invoke(messages)
        return r["parsed"]     # may still be None

    return _extract