LangGraph

pip install langgraph

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

# 1. Define the state schema
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]  # Chat messages (auto-appended)
    next_step: str                           # Custom state field

# 2. Create the graph
graph = StateGraph(AgentState)

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model="gpt-4o-mini")

# Define node functions
def chatbot(state: AgentState):
    """Call the LLM with current messages."""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# Add nodes to the graph
graph.add_node("chatbot", chatbot)

# Add edges
graph.add_edge(START, "chatbot")    # Start → chatbot
graph.add_edge("chatbot", END)       # chatbot → End

# Compile the graph into a runnable
app = graph.compile()

# Run it
result = app.invoke({
    "messages": [HumanMessage(content="Hello!")]
})
print(result["messages"][-1].content)

def should_use_tools(state: AgentState) -> str:
    """Decide whether to use tools or end."""
    last_message = state["messages"][-1]
    # If the LLM made tool calls, route to tools node
    if last_message.tool_calls:
        return "tools"
    # Otherwise, end the conversation
    return "end"

# Add conditional edge from chatbot
graph.add_conditional_edges(
    "chatbot",
    should_use_tools,
    {
        "tools": "tool_node",  # Route to tool execution
        "end": END,             # Route to end
    },
)

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

# State
class State(TypedDict):
    messages: Annotated[list, add_messages]

# Tools
@tool
def search(query: str) -> str:
    """Search the web for information."""
    return f"Results for '{query}': LangGraph is a framework by LangChain..."

@tool
def calculate(expr: str) -> str:
    """Calculate a math expression."""
    return str(eval(expr))

tools = [search, calculate]

# LLM with tools bound
llm = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)

# Nodes
def agent(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

tool_node = ToolNode(tools)

# Router
def should_continue(state: State) -> str:
    if state["messages"][-1].tool_calls:
        return "tools"
    return "end"

# Build graph
graph = StateGraph(State)
graph.add_node("agent", agent)
graph.add_node("tools", tool_node)
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, {
    "tools": "tools",
    "end": END,
})
graph.add_edge("tools", "agent")  # Loop back after tool execution

# Compile and run
app = graph.compile()
result = app.invoke({
    "messages": [("human", "Search for LangGraph and calculate 42 * 17")]
})
print(result["messages"][-1].content)

from langgraph.checkpoint.memory import MemorySaver

# Compile with checkpointer and interrupt
checkpointer = MemorySaver()
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["tools"],  # Pause before tool execution
)

# Run until interrupt
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke(
    {"messages": [("human", "Delete all files in /tmp")]},
    config=config,
)
# Agent pauses here before executing the tool

# Review the pending tool call, then resume
print("Pending action:", result["messages"][-1].tool_calls)

# Resume execution (approve)
result = app.invoke(None, config=config)

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver

# In-memory (development)
checkpointer = MemorySaver()

# SQLite (persists across restarts)
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")

# Compile with checkpointer
app = graph.compile(checkpointer=checkpointer)

# Each thread_id maintains its own conversation
config = {"configurable": {"thread_id": "user-alice"}}
app.invoke({"messages": [("human", "Hi, I'm Alice")]}, config)

# Later, same thread remembers the conversation
app.invoke({"messages": [("human", "What's my name?")]}, config)

# Supervisor pattern: one agent routes to specialists
def supervisor(state):
    """Decide which specialist agent should handle the task."""
    response = supervisor_llm.invoke(state["messages"])
    return {"messages": [response], "next": "researcher"}

def researcher(state):
    """Research agent with search tools."""
    ...

def writer(state):
    """Writing agent that drafts content."""
    ...

# Build multi-agent graph
graph = StateGraph(State)
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)

graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", route_to_agent)
graph.add_edge("researcher", "supervisor")
graph.add_edge("writer", "supervisor")