LangGraph (stateful agents)

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, BaseMessage
import operator

llm = ChatOpenAI(model='gpt-4o-mini', temperature=0)

# 1) State — a typed dict that flows through the graph
class State(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]    # accumulated
    question: str
    answer: str
    confidence: float

# Annotated[..., operator.add] means returned values are APPENDED, not replaced.
# Other reducers: operator.or_ for sets, custom functions for dicts.

# 2) Nodes — functions that take state, return (partial) state
def classify(state: State) -> dict:
    q = state['question']
    out = llm.invoke([
        SystemMessage(content='Classify the question as "factual" or "opinion".'),
        HumanMessage(content=q),
    ])
    return { 'messages': [out], 'confidence': 1.0 if 'factual' in out.content.lower() else 0.5 }

def answer_factual(state: State) -> dict:
    out = llm.invoke([HumanMessage(content=f'Answer factually: ${state["question"]}')])
    return { 'messages': [out], 'answer': out.content }

def answer_opinion(state: State) -> dict:
    out = llm.invoke([HumanMessage(content=f'Give a nuanced opinion: ${state["question"]}')])
    return { 'messages': [out], 'answer': out.content }

# 3) Routing — conditional edges based on state
def route(state: State) -> str:
    return 'factual' if state['confidence'] > 0.8 else 'opinion'

# 4) Build the graph
graph = StateGraph(State)
graph.add_node('classify',  classify)
graph.add_node('factual',   answer_factual)
graph.add_node('opinion',   answer_opinion)

graph.add_edge(START, 'classify')
graph.add_conditional_edges('classify', route, {'factual': 'factual', 'opinion': 'opinion'})
graph.add_edge('factual', END)
graph.add_edge('opinion', END)

app = graph.compile()

# 5) Run it
result = app.invoke({
    'question': 'When was the eiffel tower built?',
    'messages': [], 'answer': '', 'confidence': 0.0,
})
print(result['answer'])

# 6) Streaming
for event in app.stream({'question': 'Why is the sky blue?', 'messages': [], 'answer': '', 'confidence': 0.0}):
    print(event)
# Yields each node's output as it produces — great for streaming progress to the UI.

# 7) Cycles + loops — try until a quality threshold is met
class DraftState(TypedDict):
    topic:      str
    draft:      str
    feedback:   str
    iterations: int

def write(state: DraftState) -> dict:
    msg = f'Write a one-paragraph essay on ${state["topic"]}'
    if state.get('feedback'):
        msg += f'\\n\\nRevise based on: ${state["feedback"]}'
    out = llm.invoke([HumanMessage(content=msg)])
    return { 'draft': out.content, 'iterations': state['iterations'] + 1 }

def critique(state: DraftState) -> dict:
    if state['iterations'] >= 3:
        return { 'feedback': 'DONE' }
    out = llm.invoke([HumanMessage(content=f'Critique this draft, or say DONE if good:\\n${state["draft"]}')])
    return { 'feedback': out.content }

def should_loop(state: DraftState) -> str:
    return END if 'DONE' in state['feedback'].upper() else 'write'

g = StateGraph(DraftState)
g.add_node('write',    write)
g.add_node('critique', critique)
g.add_edge(START, 'write')
g.add_edge('write', 'critique')
g.add_conditional_edges('critique', should_loop, {'write': 'write', END: END})

draft_app = g.compile()
final = draft_app.invoke({'topic': 'sustainable cities', 'draft': '', 'feedback': '', 'iterations': 0})
print(final['draft'])

# 8) Tool use — agents with proper planning
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

@tool
def get_weather(city: str) -> str:
    'Return current weather for a city.'
    return f'It is 22°C and sunny in {city}.'

@tool
def search_web(query: str) -> str:
    'Search the web.'
    return 'Search results placeholder.'

agent = create_react_agent(llm, tools=[get_weather, search_web])
result = agent.invoke({'messages': [HumanMessage(content='What is the weather in Sydney?')]})
print(result['messages'][-1].content)

# 9) Checkpointing — pause and resume
checkpointer = MemorySaver()
app_with_memory = graph.compile(checkpointer=checkpointer)

config = { 'configurable': { 'thread_id': 'session-42' } }
app_with_memory.invoke({'question': 'Tell me about Sydney', 'messages': [], 'answer': '', 'confidence': 0.0}, config)

# Same thread_id → resumes with prior state
app_with_memory.invoke({'question': 'What about its harbour?', 'messages': [], 'answer': '', 'confidence': 0.0}, config)

# For production, use a persistent checkpointer (Postgres, Redis):
# from langgraph.checkpoint.postgres import PostgresSaver

# 10) Human-in-the-loop — interrupt before / after a node
graph.compile(checkpointer=checkpointer, interrupt_before=['answer_factual'])

# UI side: receive the state, render approval UI, then call:
app_with_memory.update_state(config, { 'answer': 'Approved' }, as_node='answer_factual')
app_with_memory.invoke(None, config)        # continues from the checkpoint

# 11) Subgraphs — encapsulate complex sub-workflows
def build_research_subgraph() -> StateGraph:
    g = StateGraph(State)
    g.add_node('search', lambda s: { 'messages': [llm.invoke([HumanMessage(content='Search…')])] })
    g.add_node('summarise', lambda s: { 'messages': [llm.invoke([HumanMessage(content='Summarise the results…')])] })
    g.add_edge(START, 'search')
    g.add_edge('search', 'summarise')
    g.add_edge('summarise', END)
    return g.compile()

main = StateGraph(State)
main.add_node('research', build_research_subgraph())
main.add_node('answer',   answer_factual)
main.add_edge(START, 'research')
main.add_edge('research', 'answer')
main.add_edge('answer', END)

# 12) Parallel branches — fan out / fan in
from langgraph.graph import StateGraph, START, END

class FanState(TypedDict):
    question: str
    results:  Annotated[list[str], operator.add]

def search_web(state):  return { 'results': ['web result'] }
def search_db(state):   return { 'results': ['db result'] }
def merge(state):       return { 'answer': '\\n'.join(state['results']) }

g = StateGraph(FanState)
g.add_node('web', search_web)
g.add_node('db',  search_db)
g.add_node('merge', merge)
g.add_edge(START, 'web')
g.add_edge(START, 'db')
g.add_edge('web',   'merge')
g.add_edge('db',    'merge')
g.add_edge('merge', END)
# 'merge' runs once both upstream nodes finish; results auto-accumulated via operator.add.

# 13) Visualising the graph
print(graph.get_graph().draw_mermaid())     # ASCII / Mermaid
# Or render with graphviz / save to a PNG for docs

# 14) Observability — LangSmith integration
import os
os.environ['LANGCHAIN_TRACING_V2'] = 'true'
os.environ['LANGCHAIN_PROJECT'] = 'my-graph'
# Every node, every LLM call, every retry traced in the LangSmith UI.

# 15) Patterns to reach for
# • Plan → Execute → Reflect → Loop until done (Tree of Thoughts, ReAct, Reflexion)
# • Reviewer + Executor — second LLM checks the first's work before returning
# • Multi-agent — supervisor routes to specialists (researcher, coder, summariser)
# • Async/sync transitions — node returns a coroutine for long-running work
# • Streaming user updates while planning the next step

# 16) Common bugs
# • Mutating state inside a node — return a NEW dict instead
# • Forgetting Annotated reducers — list overwrites instead of appending
# • Infinite loops — always have a termination check + an iteration count guard
# • Conditional routes pointing to nonexistent nodes — runtime error
# • Tool calls inside a node that returns nothing — agent can't see the result; emit a Message
# • MemorySaver in production — fine for demos; switch to PostgresSaver for real apps
# • Forgetting thread_id when using checkpoints — every call looks like a fresh conversation

from langgraph.graph import StateGraph, END
g = StateGraph(dict)
g.add_node('decide', decide)
g.add_node('do_work', do_work)
g.add_edge('decide', 'do_work')
g.add_edge('do_work', END)
g.set_entry_point('decide')
app = g.compile()