LangGraph: Build Production-Ready LLM Agents That Don't Break

pip install langgraph langchain-openai

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    search_results: str
    final_answer: str

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

def search_node(state: AgentState) -> dict:
    # Simulate web search tool call
    query = state["messages"][-1].content
    results = f"Search results for: {query}  -  [simulated]"
    return {"search_results": results}

def synthesize_node(state: AgentState) -> dict:
    context = state["search_results"]
    prompt = f"Based on: {context}
Answer the original question concisely."
    response = llm.invoke([HumanMessage(content=prompt)])
    return {"final_answer": response.content, "messages": [response]}

def router(state: AgentState) -> str:
    if state.get("search_results"):
        return "synthesize"
    return "search"

graph = StateGraph(AgentState)
graph.add_node("search", search_node)
graph.add_node("synthesize", synthesize_node)
graph.set_entry_point("search")
graph.add_conditional_edges("search", router, {"synthesize": "synthesize"})
graph.add_edge("synthesize", END)

app = graph.compile()
result = app.invoke({"messages": [HumanMessage(content="What is PagedAttention?")]})
print(result["final_answer"])

from langgraph.checkpoint.sqlite import SqliteSaver

memory = SqliteSaver.from_conn_string(":memory:")
app = graph.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "session-123"}}
result = app.invoke({"messages": [HumanMessage(content="Start research")]}, config)

for event in app.stream({"messages": [HumanMessage(content="Explain HNSW")]}, config):
    for key, value in event.items():
        print(f"Node: {key} → {value}")