LangGraph and LangSmith Explained: Stateful Graphs, Checkpoints, Human Approval and Tracing

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

class State(TypedDict):                        # the shared state that every node reads and updates
    topic: str
    draft: str
    feedback: str
    tries: int

def write(state: State) -> dict:
    hint = f" Fix this problem from last time: {state['feedback']}" if state["feedback"] else ""
    draft = ask(f"Explain {state['topic']} in ONE short sentence.{hint}", max_tokens=60, temperature=0.0)
    return {"draft": draft, "tries": state["tries"] + 1}

def review(state: State) -> dict:
    ok = "chlorophyll" in state["draft"].lower()           # a simple, checkable requirement
    return {"feedback": "" if ok else "it must mention the word 'chlorophyll'"}

def route(state: State) -> str:                # a conditional edge: decide the next node from the state
    if state["feedback"] == "" or state["tries"] >= 4:
        return END
    return "write"                              # loop back

g = StateGraph(State)
g.add_node("write", write); g.add_node("review", review)
g.add_edge(START, "write"); g.add_edge("write", "review")
g.add_conditional_edges("review", route)
app = g.compile()

for step in app.stream({"topic": "photosynthesis", "draft": "", "feedback": "", "tries": 0}):
    node, update = next(iter(step.items()))
    shown = {k: (v if k != "draft" else v[:60]) for k, v in update.items()}
    print(f"{node:7s} -> {shown}")

class State(TypedDict):
    topic: str
    draft: str
    feedback: str
    tries: int

def route(s: State) -> str:                       # a conditional edge: choose the next node from the state
    return END if s["feedback"] == "" or s["tries"] >= 4 else "write"

g = StateGraph(State)
g.add_node("write", write); g.add_node("review", review)
g.add_edge(START, "write"); g.add_edge("write", "review")
g.add_conditional_edges("review", route)
app = g.compile()

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

class State(TypedDict):
    log: list

calls = {"fetch": 0, "summarise": 0, "publish": 0}
crash_once = {"armed": True}

def fetch(s):                                   # step 1: an expensive step we never want to repeat
    calls["fetch"] += 1
    return {"log": s["log"] + ["fetched 500 records"]}

def summarise(s):                               # step 2: dies the first time, as if the server restarted
    calls["summarise"] += 1
    if crash_once["armed"]:
        crash_once["armed"] = False
        raise RuntimeError("connection lost")
    return {"log": s["log"] + ["summary written"]}

def publish(s):
    calls["publish"] += 1
    return {"log": s["log"] + ["published"]}

g = StateGraph(State)
for name, fn in [("fetch", fetch), ("summarise", summarise), ("publish", publish)]: g.add_node(name, fn)
g.add_edge(START, "fetch"); g.add_edge("fetch", "summarise"); g.add_edge("summarise", "publish"); g.add_edge("publish", END)
app = g.compile(checkpointer=MemorySaver())     # save the state after every node
cfg = {"configurable": {"thread_id": "report-1"}}   # one thread = one run that can be resumed

try:
    app.invoke({"log": []}, cfg)
except RuntimeError as e:
    print("run 1 crashed:", e)
saved = app.get_state(cfg)
print("saved state after the crash:", saved.values["log"], "| next step would be:", saved.next)

result = app.invoke(None, cfg)                  # resume the same thread from the last checkpoint
print("\nrun 2 (resumed):", result["log"])
print("how many times each step actually ran:", calls)

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

class State(TypedDict):
    to: str
    body: str
    sent: bool

def draft(s): return {"body": f"Dear {s['to']}, your mock-test result is ready on the portal."}
def send(s):  return {"sent": True}             # the risky step: pretend this really emails a parent

g = StateGraph(State)
g.add_node("draft", draft); g.add_node("send", send)
g.add_edge(START, "draft"); g.add_edge("draft", "send"); g.add_edge("send", END)
app = g.compile(checkpointer=MemorySaver(), interrupt_before=["send"])    # pause BEFORE the risky node
cfg = {"configurable": {"thread_id": "mail-7"}}

app.invoke({"to": "Mr Khan", "body": "", "sent": False}, cfg)
state = app.get_state(cfg)
print("paused before:", state.next)
print("draft waiting for approval:", state.values["body"])
print("sent so far?", state.values["sent"])

print("\n-- a human reads it and approves --")
app.invoke(None, cfg)                           # resume
print("sent now?", app.get_state(cfg).values["sent"])

from langchain_core.runnables import RunnableLambda

attempts = {"primary": 0}

def primary_model(text):                        # a flaky "model": fails twice, then works
    attempts["primary"] += 1
    if attempts["primary"] <= 2:
        raise TimeoutError("provider timed out")
    return f"primary answer to {text!r}"

def backup_model(text):
    return f"backup answer to {text!r}"

# retry: try again (with growing waits) before giving up
retrying = RunnableLambda(primary_model).with_retry(stop_after_attempt=4, wait_exponential_jitter=False)
print(retrying.invoke("hello"), "| primary was called", attempts["primary"], "times")

# fallback: if the primary keeps failing, use another model
attempts["primary"] = -10                       # now it will keep failing
safe = RunnableLambda(primary_model).with_retry(stop_after_attempt=2, wait_exponential_jitter=False).with_fallbacks([RunnableLambda(backup_model)])
print(safe.invoke("hello"))

import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "<your key>"
os.environ["LANGSMITH_PROJECT"] = "ai-course"

from langsmith import traceable

@traceable                      # also records your own plain Python functions
def my_step(text: str) -> str:
    return ask(text)

import time
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.language_models.llms import LLM
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

class Local(LLM):
    @property
    def _llm_type(self): return "local-qwen"
    def _call(self, prompt, stop=None, run_manager=None, **kw): return ask(prompt, max_tokens=40)

class Trace(BaseCallbackHandler):
    """Records what a tracing service such as LangSmith records: every step, its input, output and time."""
    def __init__(self): self.t0, self.rows = {}, []
    def on_chain_start(self, serialized, inputs, *, run_id, **kw): self.t0[run_id] = time.time(); self.name = (serialized or {}).get("name", "chain")
    def on_llm_start(self, serialized, prompts, *, run_id, **kw): self.t0[run_id] = time.time(); self.rows.append(["llm start", prompts[0][:44], ""])
    def on_llm_end(self, response, *, run_id, **kw):
        self.rows.append(["llm end", response.generations[0][0].text[:44].replace("\n", " "), f"{time.time() - self.t0[run_id]:.2f}s"])
    def on_chain_end(self, outputs, *, run_id, **kw): self.rows.append(["chain end", str(outputs)[:44], f"{time.time() - self.t0[run_id]:.2f}s"])

chain = ChatPromptTemplate.from_template("Define {term} in one short sentence.") | Local() | StrOutputParser()
trace = Trace()
chain.invoke({"term": "epoch"}, config={"callbacks": [trace]})
print(f"{'event':10s} {'content':46s} time")
for kind, content, t in trace.rows:
    print(f"{kind:10s} {content:46s} {t}")