LangChain & LangGraph Basics

python -m venv .venv && source .venv/bin/activate
pip install -U langgraph
# Optional for LLM examples (verify package names at build):
# pip install -U langchain langchain-openai

from typing import TypedDict
 
class TicketState(TypedDict):
    text: str
    label: str
    response: str

def classify(state: TicketState) -> dict:
    text = state["text"].lower()
    label = "billing" if "invoice" in text or "refund" in text else "general"
    return {"label": label}
 
def draft_reply(state: TicketState) -> dict:
    if state["label"] == "billing":
        body = "We received your billing request and will review it."
    else:
        body = "Thanks for writing in. A teammate will follow up."
    return {"response": body}

from langgraph.graph import StateGraph, START, END
 
builder = StateGraph(TicketState)
builder.add_node("classify", classify)
builder.add_node("draft_reply", draft_reply)
builder.add_edge(START, "classify")
builder.add_edge("classify", "draft_reply")
builder.add_edge("draft_reply", END)
graph = builder.compile()
 
result = graph.invoke({
    "text": "Please refund invoice #4421",
    "label": "",
    "response": "",
})
print(result["label"], result["response"])

def route_label(state: TicketState) -> str:
    return "billing_path" if state["label"] == "billing" else "general_path"
 
def billing_path(state: TicketState) -> dict:
    return {"response": "Billing queue: " + state["text"][:80]}
 
def general_path(state: TicketState) -> dict:
    return {"response": "General queue: " + state["text"][:80]}
 
builder = StateGraph(TicketState)
builder.add_node("classify", classify)
builder.add_node("billing_path", billing_path)
builder.add_node("general_path", general_path)
builder.add_edge(START, "classify")
builder.add_conditional_edges(
    "classify",
    route_label,
    {"billing_path": "billing_path", "general_path": "general_path"},
)
builder.add_edge("billing_path", END)
builder.add_edge("general_path", END)
graph = builder.compile()
 
print(graph.invoke({"text": "refund please", "label": "", "response": ""})["response"])

for chunk in graph.stream(
    {"text": "Where is my invoice?", "label": "", "response": ""},
    stream_mode="updates",
    version="v2",  # LangGraph >= 1.1; omit or use v1 if needed (verify at build)
):
    # v2 chunks: {"type": "updates", "ns": (), "data": {...}}
    if isinstance(chunk, dict) and chunk.get("type") == "updates":
        print(chunk["data"])
    else:
        print(chunk)  # v1 raw updates dict

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
 
class ChatState(TypedDict):
    messages: Annotated[list, add_messages]
 
def greeter(state: ChatState) -> dict:
    return {"messages": [{"role": "assistant", "content": "Hello!"}]}
 
g = (
    StateGraph(ChatState)
    .add_node("greeter", greeter)
    .add_edge(START, "greeter")
    .add_edge("greeter", END)
    .compile()
)
out = g.invoke({"messages": [{"role": "user", "content": "hi"}]})
print(len(out["messages"]))  # user + assistant

from langgraph.checkpoint.memory import InMemorySaver
 
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "demo-1"}}
 
graph.invoke(
    {"text": "refund invoice", "label": "", "response": ""},
    config=config,
)
snapshot = graph.get_state(config)
print(snapshot.values.get("label"), snapshot.next)

from typing import Annotated, Literal, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
 
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
 
def call_model(state: AgentState) -> dict:
    # Swap for a real chat model + bind_tools in production (verify APIs at build)
    last = state["messages"][-1]["content"]
    if "2+2" in last.replace(" ", ""):
        return {"messages": [{
            "role": "assistant",
            "content": "",
            "tool_calls": [{
                "id": "1",
                "name": "add",
                "args": {"a": 2, "b": 2},
            }],
        }]}
    return {"messages": [{"role": "assistant", "content": "Done."}]}
 
def call_tools(state: AgentState) -> dict:
    # Demo tool execution only
    return {"messages": [{
        "role": "tool",
        "content": "4",
        "tool_call_id": "1",
        "name": "add",
    }]}
 
def should_continue(state: AgentState) -> Literal["tools", "__end__"]:
    last = state["messages"][-1]
    if last.get("tool_calls"):
        return "tools"
    return "__end__"
 
agent = StateGraph(AgentState)
agent.add_node("agent", call_model)
agent.add_node("tools", call_tools)
agent.add_edge(START, "agent")
agent.add_conditional_edges("agent", should_continue, {"tools": "tools", "__end__": END})
agent.add_edge("tools", "agent")
app = agent.compile()
print(app.invoke({"messages": [{"role": "user", "content": "What is 2+2?"}]})["messages"][-1])