LangGraph Basics: the Engine behind Agents

def assistant(state: MessagesState):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

b = StateGraph(MessagesState)
b.add_node("assistant", assistant)
b.add_node("tools", ToolNode(tools))
b.add_edge(START, "assistant")
b.add_conditional_edges("assistant", tools_condition)   # tool call? → "tools" else → END
b.add_edge("tools", "assistant")                        # loop back
agent_graph = b.compile()

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import InMemorySaver
from langchain.tools import tool
from standins import ScriptedModel, call

class State(TypedDict):
    name: str
    greeting: str
def say_hello(state: State): return {"greeting": f"Hello {state['name']}!"}
def add_emoji(state: State): return {"greeting": state["greeting"] + " 👋"}
builder = StateGraph(State)
builder.add_node("say_hello", say_hello); builder.add_node("add_emoji", add_emoji)
builder.add_edge(START, "say_hello"); builder.add_edge("say_hello", "add_emoji"); builder.add_edge("add_emoji", END)
print("1 simple   :", builder.compile().invoke({"name": "Ajeet"}))

class Result(TypedDict):
    marks: int
    message: str
b = StateGraph(Result)
b.add_node("check", lambda s: s); b.add_node("passed", lambda s: {"message": "🎉 Pass"}); b.add_node("failed", lambda s: {"message": "📚 Fail, try again"})
b.add_edge(START, "check")
b.add_conditional_edges("check", lambda s: "passed" if s["marks"] >= 35 else "failed", {"passed": "passed", "failed": "failed"})
b.add_edge("passed", END); b.add_edge("failed", END)
g = b.compile()
print("2 routing  :", g.invoke({"marks": 72}), g.invoke({"marks": 20}))

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b
@tool
def get_weather(city: str) -> str:
    """Get weather of a city."""
    return f"{city}: 29°C, sunny"
tools = [multiply, get_weather]

def make_graph(llm, **compile_kw):
    def assistant(state: MessagesState): return {"messages": [llm.invoke(state["messages"])]}
    b = StateGraph(MessagesState)
    b.add_node("assistant", assistant); b.add_node("tools", ToolNode(tools))
    b.add_edge(START, "assistant")
    b.add_conditional_edges("assistant", tools_condition)      # tool call? -> "tools" else -> END
    b.add_edge("tools", "assistant")                           # loop back
    return b.compile(**compile_kw)

from langchain.messages import AIMessage
llm = ScriptedModel([AIMessage("", tool_calls=[{"name": "multiply", "args": {"a": 23, "b": 17}, "id": "m"}, {"name": "get_weather", "args": {"city": "Pune"}, "id": "w"}]),
                     "23 × 17 = 391, and Pune is 29°C and sunny."])
out = make_graph(llm).invoke({"messages": [{"role": "user", "content": "What is 23 * 17, and weather in Pune?"}]})
print("3 agent    :", " → ".join(m.type + (f"[{', '.join(c['name'] for c in m.tool_calls)}]" if getattr(m, "tool_calls", None) else "") for m in out["messages"]))
print("             final:", out["messages"][-1].content)

mem = make_graph(ScriptedModel([call("multiply", {"a": 6, "b": 7}), "Nice to meet you Ajeet! 6 × 7 = 42.",
                                call("multiply", {"a": 42, "b": 2}), "Your name is Ajeet, and 42 × 2 = 84."]), checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "1"}}
mem.invoke({"messages": [{"role": "user", "content": "My name is Ajeet. Multiply 6 and 7."}]}, cfg)
r = mem.invoke({"messages": [{"role": "user", "content": "What is my name? Multiply the last answer by 2."}]}, cfg)
print("4 memory   :", r["messages"][-1].content, f"({len(r['messages'])} messages in thread 1)")

hitl = make_graph(ScriptedModel([call("multiply", {"a": 12, "b": 12}), "12 × 12 = 144."]), checkpointer=InMemorySaver(), interrupt_before=["tools"])
cfg = {"configurable": {"thread_id": "approve-1"}}
hitl.invoke({"messages": [{"role": "user", "content": "Multiply 12 and 12"}]}, cfg)
print("5 paused   : next =", hitl.get_state(cfg).next, "| waiting tool call:", hitl.get_state(cfg).values["messages"][-1].tool_calls[0]["args"])
print("  resumed  :", hitl.invoke(None, cfg)["messages"][-1].content)