LangGraph Projects

# Template 1: Basic StateGraph with LLM node
from typing import TypedDict
from langgraph.graph import StateGraph, END

class State(TypedDict):
    messages: list

def call_model(state: State):
    # call your LLM here
    return {"messages": state["messages"] + [response]}

graph = StateGraph(State)
graph.add_node("agent", call_model)
graph.set_entry_point("agent")
graph.add_edge("agent", END)
app = graph.compile()

# Template 2: Tools + should_continue router
from langgraph.prebuilt import ToolNode

tools = [search_tool, calc_tool]
tool_node = ToolNode(tools)

def should_continue(state):
    last = state["messages"][-1]
    if last.tool_calls:
        return "tools"
    return "end"

graph.add_node("agent", call_model)
graph.add_node("tools", tool_node)
graph.add_conditional_edges(
    "agent", should_continue,
    {"tools": "tools", "end": END}
)
graph.add_edge("tools", "agent")

# Template 3: Retrieve → Generate subgraph
def retrieve(state):
    docs = vectorstore.similarity_search(state["query"])
    return {"context": docs}

def generate(state):
    prompt = f"Context: {state['context']}\nQ: {state['query']}"
    return {"answer": llm.invoke(prompt)}

g = StateGraph(State)
g.add_node("retrieve", retrieve)
g.add_node("generate", generate)
g.set_entry_point("retrieve")
g.add_edge("retrieve", "generate")
g.add_edge("generate", END)
rag_app = g.compile()

# Template 4: Pause for human approval
from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
app = graph.compile(
    checkpointer=memory,
    interrupt_before=["execute_action"]
)

# Run until interrupt
for event in app.stream(inputs, config):
    print(event)

# Human reviews state, then resume
app.invoke(None, config)  # continue

# Template 5: Parallel specialist agents
def researcher(state): ...
def writer(state): ...
def critic(state): ...

graph.add_node("research", researcher)
graph.add_node("write", writer)
graph.add_node("critique", critic)

# Sequential pipeline (or use Send for parallel)
graph.add_edge("research", "write")
graph.add_edge("write", "critique")
graph.add_conditional_edges(
    "critique", route_after_critique
)

# Template 6: Checkpointer + stream modes
from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string(
    "checkpoints.db"
) as checkpointer:
    app = graph.compile(
        checkpointer=checkpointer
    )
    config = {"configurable": {
        "thread_id": "user-42"
    }}
    for chunk in app.stream(
        inputs, config,
        stream_mode="updates"
    ):
        print(chunk)