LangGraph Entry Point, END Node & Compile-Time Errors

from typing import TypedDict
from langgraph.graph import StateGraph, START, END   # <-- import the sentinels


class State(TypedDict):
    messages: list


def first_node(state):
    return {"messages": state["messages"] + ["hi from first"]}


def last_node(state):
    return {"messages": state["messages"] + ["hi from last"]}


graph = StateGraph(State)
graph.add_node("first", first_node)
graph.add_node("last", last_node)

# THE fix — one line wires the graph's entry
graph.add_edge(START, "first")
graph.add_edge("first", "last")
graph.add_edge("last", END)

app = graph.compile()
result = app.invoke({"messages": []})
print(result["messages"])
# ['hi from first', 'hi from last']

from langgraph.graph import StateGraph, START, END


def check_reachability(builder: StateGraph) -> None:
    """Walk the graph from START; raise if any node isn't reachable."""
    # LangGraph exposes internal edges via builder.edges and builder.branches
    all_nodes = set(builder.nodes.keys())
    reachable = set()
    frontier = [START]

    edges = set(builder.edges)           # {(src, tgt)}
    branches = builder.branches or {}    # {src: {branch_name: [tgts]}}

    while frontier:
        current = frontier.pop()
        if current in reachable:
            continue
        reachable.add(current)
        # Static edges
        for src, tgt in edges:
            if src == current and tgt not in reachable:
                frontier.append(tgt)
        # Conditional edges
        if current in branches:
            for _label, targets in branches[current].items():
                for t in (targets if isinstance(targets, list) else [targets]):
                    if t not in reachable:
                        frontier.append(t)

    unreachable = all_nodes - reachable
    if unreachable:
        raise ValueError(
            f"Unreachable nodes: {sorted(unreachable)}. "
            "Add an edge from a reachable node, or remove the dead nodes."
        )


# Use it before compile()
graph = StateGraph(State)
# ... add_node / add_edge calls ...
check_reachability(graph)     # fails loudly with a clear list
app = graph.compile()

from langgraph.graph import StateGraph, START, END


def build_app():
    """Fresh graph every call — safe to hot-reload."""
    graph = StateGraph(State)
    graph.add_node("retrieve", retrieve)
    graph.add_node("generate", generate)
    graph.add_edge(START, "retrieve")
    graph.add_edge("retrieve", "generate")
    graph.add_edge("generate", END)
    return graph.compile()


# ✅ In a notebook or script — always call the factory
app = build_app()
result = app.invoke({"messages": []})

# ✅ In FastAPI — build once at startup, not per request
from fastapi import FastAPI
api = FastAPI()

@api.on_event("startup")
def startup():
    api.state.graph = build_app()

@api.post("/chat")
def chat(payload: dict):
    return api.state.graph.invoke(payload)


# ❌ ANTI-PATTERN — reusing the same builder
# graph = StateGraph(State)
# ... build ...
# app1 = graph.compile()
# graph.add_node("new", new_fn)      # mutation after compile
# app2 = graph.compile()             # behavior undefined across versions