LangGraph

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class AppState(TypedDict):
    user_input: str
    response: str

def generate_response(state: AppState) -> AppState:
    # In production, call your LLM here
    return {**state, "response": f"Echo: {state['user_input']}"}

graph_builder = StateGraph(AppState)
graph_builder.add_node("generate", generate_response)
graph_builder.add_edge(START, "generate")
graph_builder.add_edge("generate", END)

graph = graph_builder.compile()

result = graph.invoke({"user_input": "Hello LangGraph", "response": ""})
print(result)
# {'user_input': 'Hello LangGraph', 'response': 'Echo: Hello LangGraph'}

from typing import TypedDict, List
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
from langgraph.graph import StateGraph, START, END

class ChatState(TypedDict):
    messages: List[BaseMessage]

def assistant_node(state: ChatState) -> ChatState:
    last_user = state["messages"][-1].content
    reply = AIMessage(content=f"I received: {last_user}")
    return {"messages": state["messages"] + [reply]}

builder = StateGraph(ChatState)
builder.add_node("assistant", assistant_node)
builder.add_edge(START, "assistant")
builder.add_edge("assistant", END)
app = builder.compile()

initial = {"messages": [HumanMessage(content="Summarize quarterly revenue trends")]} 
out = app.invoke(initial)
print(out["messages"][-1].content)

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI

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

# 1. Define state
class AgentState(TypedDict):
    messages: list
    next: str

# 2. Define nodes (functions)
def call_llm(state: AgentState) -> AgentState:
    response = llm.invoke(state["messages"])
    return {"messages": state["messages"] + [response]}

def should_continue(state: AgentState) -> str:
    last_msg = state["messages"][-1]
    if last_msg.tool_calls:
        return "tools"
    return END

# 3. Build graph
graph = StateGraph(AgentState)
graph.add_node("agent", call_llm)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue)

app = graph.compile()

# 4. Run
result = app.invoke({"messages": [("human", "Hello!")]})

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END

class TicketState(TypedDict):
    text: str
    intent: Literal["faq", "escalate"]
    output: str

def classify_intent(state: TicketState):
    t = state["text"].lower()
    intent = "escalate" if "refund" in t or "angry" in t else "faq"
    return {**state, "intent": intent}

def faq_node(state: TicketState):
    return {**state, "output": "Here are help center steps for your issue."}

def escalate_node(state: TicketState):
    return {**state, "output": "I've escalated this to a human support agent."}

def route_by_intent(state: TicketState):
    return state["intent"]

builder = StateGraph(TicketState)
builder.add_node("classify", classify_intent)
builder.add_node("faq", faq_node)
builder.add_node("escalate", escalate_node)

builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route_by_intent, {
    "faq": "faq",
    "escalate": "escalate"
})
builder.add_edge("faq", END)
builder.add_edge("escalate", END)

app = builder.compile()
print(app.invoke({"text": "I want a refund now", "intent": "faq", "output": ""}))

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class DraftState(TypedDict):
    topic: str
    draft: str
    score: int
    max_iters: int
    iters: int

def write_draft(state: DraftState):
    new_text = f"Draft v{state['iters'] + 1} on {state['topic']}"
    return {**state, "draft": new_text, "iters": state["iters"] + 1}

def review_draft(state: DraftState):
    # Placeholder scoring logic
    score = min(10, len(state["draft"]) // 4)
    return {**state, "score": score}

def should_continue(state: DraftState):
    if state["score"] >= 8 or state["iters"] >= state["max_iters"]:
        return "end"
    return "rewrite"

builder = StateGraph(DraftState)
builder.add_node("write", write_draft)
builder.add_node("review", review_draft)

builder.add_edge(START, "write")
builder.add_edge("write", "review")
builder.add_conditional_edges("review", should_continue, {
    "rewrite": "write",
    "end": END
})

app = builder.compile()
print(app.invoke({"topic": "SOC 2 controls", "draft": "", "score": 0, "max_iters": 3, "iters": 0}))

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.postgres import PostgresSaver

# In-memory (dev/testing)
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

# PostgreSQL (production)
with PostgresSaver.from_conn_string(DATABASE_URL) as saver:
    app = graph.compile(checkpointer=saver)

# Use thread_id for conversation memory
config = {"configurable": {"thread_id": "user-123"}}

# First message
app.invoke({"messages": [("human", "My name is Alice")]}, config=config)

# Second message — remembers context
app.invoke({"messages": [("human", "What is my name?")]}, config=config)
# Output: "Your name is Alice"

# Get state history
for state in app.get_state_history(config):
    print(state)

from typing import TypedDict, Optional
import requests
from langgraph.graph import StateGraph, START, END

class InventoryState(TypedDict):
    sku: str
    inventory: Optional[int]
    answer: str

def fetch_inventory(state: InventoryState):
    # Example only: replace with your real API endpoint/auth
    # resp = requests.get(f"https://api.example.com/inventory/{state['sku']}", timeout=10)
    # qty = resp.json()["available"]
    qty = 42
    return {**state, "inventory": qty}

def respond(state: InventoryState):
    return {**state, "answer": f"SKU {state['sku']} has {state['inventory']} units available."}

builder = StateGraph(InventoryState)
builder.add_node("fetch_inventory", fetch_inventory)
builder.add_node("respond", respond)
builder.add_edge(START, "fetch_inventory")
builder.add_edge("fetch_inventory", "respond")
builder.add_edge("respond", END)

app = builder.compile()
print(app.invoke({"sku": "LAPTOP-15", "inventory": None, "answer": ""})["answer"])

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class ToolState(TypedDict):
    query: str
    tool_ok: bool
    result: str

def run_tool(state: ToolState):
    try:
        # Simulate flaky tool
        raise RuntimeError("Tool temporarily unavailable")
    except Exception:
        return {**state, "tool_ok": False, "result": ""}

def fallback_llm(state: ToolState):
    return {**state, "result": f"Fallback answer for: {state['query']}"}

def finalize_tool(state: ToolState):
    return {**state, "result": "Tool answer: 2024 revenue was $12.4M"}

def route_after_tool(state: ToolState):
    return "ok" if state["tool_ok"] else "fallback"

builder = StateGraph(ToolState)
builder.add_node("run_tool", run_tool)
builder.add_node("fallback", fallback_llm)
builder.add_node("finalize", finalize_tool)

builder.add_edge(START, "run_tool")
builder.add_conditional_edges("run_tool", route_after_tool, {
    "ok": "finalize",
    "fallback": "fallback"
})
builder.add_edge("fallback", END)
builder.add_edge("finalize", END)

app = builder.compile()
print(app.invoke({"query": "What is FY revenue?", "tool_ok": True, "result": ""})["result"])

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

class SessionState(TypedDict):
    user_id: str
    messages: list[str]


def append_message(state: SessionState):
    return {**state, "messages": state["messages"] + ["assistant: got it"]}

builder = StateGraph(SessionState)
builder.add_node("append", append_message)
builder.add_edge(START, "append")
builder.add_edge("append", END)

checkpointer = MemorySaver()
app = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "customer-123"}}
app.invoke({"user_id": "u1", "messages": ["user: hello"]}, config=config)
# Later turn can continue same thread_id

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class StreamState(TypedDict):
    task: str
    status: str

def plan(state: StreamState):
    return {**state, "status": "planned"}

def execute(state: StreamState):
    return {**state, "status": "executed"}

builder = StateGraph(StreamState)
builder.add_node("plan", plan)
builder.add_node("execute", execute)
builder.add_edge(START, "plan")
builder.add_edge("plan", "execute")
builder.add_edge("execute", END)
app = builder.compile()

for event in app.stream({"task": "generate report", "status": "new"}):
    print(event)
# Useful for websocket/SSE updates in web apps