# langgraph/core.py from typing import TypedDict, Annotated, List, Optional from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_core.messages import BaseMessage, HumanMessage, AIMessage class AgentState(TypedDict): messages: Annotated[List[BaseMessage], add_messages] user_id: str session_id: str context: dict current_step: str retry_count: int metadata: dict def create_basic_graph(): workflow = StateGraph(AgentState) # Add nodes  workflow.add_node("process_input", process_input_node) workflow.add_node("reason", reason_node) workflow.add_node("act", act_node) workflow.add_node("respond", respond_node) # Define edges  workflow.add_edge(START, "process_input") workflow.add_edge("process_input", "reason") workflow.add_conditional_edges("reason", route_after_reason, { "act": "act", "respond": "respond" }) workflow.add_edge("act", "reason") workflow.add_edge("respond", END) return workflow.compile() async def process_input_node(state: AgentState): # Preprocess, validate, enrich  return {"current_step": "input_processed"} async def reason_node(state: AgentState): # LLM reasoning  return {"current_step": "reasoned"} def route_after_reason(state: AgentState) -> str: # Decide next action  return "act" if needs_action(state) else "respond" async def act_node(state: AgentState): # Tool execution  return {"current_step": "acted"} async def respond_node(state: AgentState): # Generate final response  return {"current_step": "responded"} 

# langgraph/persistence.py from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.checkpoint.postgres import PostgresSaver from langgraph.checkpoint.redis import RedisSaver import sqlite3 # SQLite (development) sqlite_conn = sqlite3.connect("checkpoints.db", check_same_thread=False) sqlite_checkpointer = SqliteSaver(sqlite_conn) # PostgreSQL (production) postgres_checkpointer = PostgresSaver.from_conn_string( "postgresql://user:pass@localhost/langgraph" ) # Redis (high-throughput) redis_checkpointer = RedisSaver.from_conn_string( "redis://localhost:6379" ) # Usage with checkpointer app = workflow.compile(checkpointer=sqlite_checkpointer) # Run with thread_id for persistence config = {"configurable": {"thread_id": "user-123-session-456"}} result = await app.ainvoke({"messages": [HumanMessage(content="Hello")]}, config) # Resume later with same thread_id result2 = await app.ainvoke({"messages": [HumanMessage(content="Continue")]}, config) 

# langgraph/routing.py from typing import Literal def route_by_intent(state: AgentState) -> Literal["research", "code", "write", "error"]: last_msg = state["messages"][-1].content.lower() if any(kw in last_msg for kw in ["research", "find", "search", "lookup"]): return "research" elif any(kw in last_msg for kw in ["code", "implement", "build", "function"]): return "code" elif any(kw in last_msg for kw in ["write", "document", "explain", "summarize"]): return "write" return "error" def route_by_confidence(state: AgentState) -> Literal["proceed", "clarify", "escalate"]: confidence = state.get("confidence", 1.0) if confidence > 0.8: return "proceed" elif confidence > 0.5: return "clarify" return "escalate" # Multi-condition routing def complex_router(state: AgentState) -> dict: """Return multiple routing decisions.""" return { "next_node": "process", "parallel_nodes": ["validate", "enrich"], "skip_nodes": [] if state["retry_count"] < 3 else ["optional_step"] } 

# langgraph/parallel.py from langgraph.graph import StateGraph from typing: Annotated, List import operator class ParallelState(TypedDict): messages: Annotated[List[BaseMessage], add_messages] research_results: Annotated[List[dict], operator.add] # Accumulate  code_results: Annotated[List[dict], operator.add] write_results: Annotated[List[dict], operator.add] async def research_node(state: ParallelState): # Run multiple searches in parallel  import asyncio async def search(query): return {"query": query, "results": await web_search(query)} queries = extract_queries(state["messages"][-1].content) results = await asyncio.gather(*[search(q) for q in queries]) return {"research_results": results} async def code_node(state: ParallelState): # Parallel code generation for different components  tasks = extract_code_tasks(state["messages"][-1].content) results = await asyncio.gather(*[generate_code(t) for t in tasks]) return {"code_results": results} # Fan-out/fan-in pattern workflow = StateGraph(ParallelState) workflow.add_node("research", research_node) workflow.add_node("code", code_node) workflow.add_node("synthesize", synthesize_node) workflow.add_edge(START, "research") workflow.add_edge(START, "code") # Both start from START workflow.add_edge("research", "synthesize") workflow.add_edge("code", "synthesize") workflow.add_edge("synthesize", END) 

# langgraph/human_in_loop.py from langgraph.types import interrupt, Command from dataclasses import dataclass @dataclass class HumanReview: approved: bool feedback: str modifications: dict async def human_review_node(state: AgentState): # Present to human for review  review_request = { "task": state["messages"][-1].content, "proposed_action": state.get("proposed_action"), "context": state.get("context") } # This pauses execution and waits for human input  human_response = interrupt(review_request) # Resume with human decision  review = HumanReview(**human_response) if review.approved: return {"human_approved": True, "modifications": review.modifications} else: return { "human_approved": False, "feedback": review.feedback, "retry_count": state["retry_count"] + 1 } def route_after_review(state: AgentState) -> str: if state.get("human_approved"): return "execute" elif state["retry_count"] >= 3: return "escalate" return "revise" # Usage app = workflow.compile(checkpointer=checkpointer, interrupt_before=["human_review"]) # First invocation - pauses at human_review result = await app.ainvoke(input_data, config) # Check if interrupted if "__interrupt__" in result: interrupt_info = result["__interrupt__"] # Present to human, get response  human_decision = await get_human_decision(interrupt_info) # Resume with human input  result = await app.ainvoke(Command(resume=human_decision), config) 

# langgraph/subgraphs.py from langgraph.graph import StateGraph # Subgraph for research research_graph = StateGraph(AgentState) research_graph.add_node("search", search_node) research_graph.add_node("summarize", summarize_node) research_graph.add_edge(START, "search") research_graph.add_edge("search", "summarize") research_graph.add_edge("summarize", END) research_app = research_graph.compile() # Subgraph for code generation code_graph = StateGraph(AgentState) code_graph.add_node("plan", plan_code_node) code_graph.add_node("implement", implement_node) code_graph.add_node("test", test_node) code_graph.add_edge(START, "plan") code_graph.add_edge("plan", "implement") code_graph.add_edge("implement", "test") code_graph.add_edge("test", END) code_app = code_graph.compile() # Main graph using subgraphs main_graph = StateGraph(AgentState) main_graph.add_node("research", research_app) main_graph.add_node("code", code_app) main_graph.add_node("integrate", integrate_node) main_graph.add_edge(START, "research") main_graph.add_edge("research", "code") main_graph.add_edge("code", "integrate") main_graph.add_edge("integrate", END) # Compile with shared checkpointer app = main_graph.compile(checkpointer=checkpointer) 

# langgraph/streaming.py from langgraph.graph import StateGraph from typing: AsyncGenerator async def stream_graph(app, input_data: dict, config: dict) -> AsyncGenerator[dict, None]: """Stream graph execution events.""" async for event in app.astream_events(input_data, config, version="v2"): event_type = event["event"] if event_type == "on_chain_start": yield {"type": "node_start", "node": event["name"]} elif event_type == "on_chain_end": yield {"type": "node_end", "node": event["name"], "output": event["data"]["output"]} elif event_type == "on_chat_model_stream": yield {"type": "token", "content": event["data"]["chunk"].content} elif event_type == "on_tool_start": yield {"type": "tool_start", "tool": event["name"], "input": event["data"]["input"]} elif event_type == "on_tool_end": yield {"type": "tool_end", "tool": event["name"], "output": event["data"]["Source：https://dev.to/ipsphil_digital/building-production-ai-applications-with-langgraph-state-graphs-deployment-f51
My favorites