# langgraph/basics.py from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, List import operator class AgentState(TypedDict): messages: Annotated[List[str], operator.add] # Auto-append  query: str answer: str iterations: int def agent_node(state: AgentState) -> AgentState: # Process and return updated state  return {"answer": "processed", "iterations": state["iterations"] + 1} graph = StateGraph(AgentState) graph.add_node("agent", agent_node) graph.set_entry_point("agent") graph.add_edge("agent", END) app = graph.compile() result = app.invoke({"query": "Hello", "messages": [], "iterations": 0}) 

# langgraph/state.py from typing import TypedDict, Annotated, List, Optional, Dict, Any import operator from pydantic import BaseModel, Field class AgentState(TypedDict): # Core fields  query: str messages: Annotated[List[Dict], operator.add] # Auto-merge lists  # Retrieval  retrieved_docs: Annotated[List[Dict], operator.add] retrieval_scores: Annotated[List[float], operator.add] # Generation  answer: str citations: Annotated[List[str], operator.add] # Control  iterations: int max_iterations: int should_continue: bool # Human-in-the-loop  human_feedback: Optional[str] requires_human: bool # Multi-agent  active_agent: str agent_outputs: Annotated[Dict[str, Any], operator.add] # Pydantic for validation class ValidatedState(BaseModel): query: str = Field(min_length=1) messages: List[Dict] = [] iterations: int = Field(default=0, ge=0, le=10) class Config: arbitrary_types_allowed = True 

# langgraph/reducers.py from typing import Annotated, List, Dict, Any import operator from functools import reduce # Custom reducers for complex merging def merge_dicts(existing: Dict, new: Dict) -> Dict: """Deep merge dictionaries.""" result = existing.copy() for k, v in new.items(): if k in result and isinstance(result[k], dict) and isinstance(v, dict): result[k] = merge_dicts(result[k], v) else: result[k] = v return result def merge_lists_unique(existing: List, new: List) -> List: """Merge lists, deduplicate by ID if dicts.""" seen = set() result = [] for item in existing + new: key = item.get("id") if isinstance(item, dict) else item if key not in seen: seen.add(key) result.append(item) return result # Usage in state class StateWithReducers(TypedDict): config: Annotated[Dict, merge_dicts] documents: Annotated[List[Dict], merge_lists_unique] metrics: Annotated[Dict[str, float], lambda old, new: {**old, **new}] 

# langgraph/agent_loop.py from langgraph.graph import StateGraph, END from langgraph.checkpoint.sqlite import SqliteSaver from langchain_core.messages import HumanMessage, AIMessage from langchain_openai import ChatOpenAI from typing import TypedDict, Annotated, List import operator class AgentState(TypedDict): messages: Annotated[List, operator.add] query: str answer: str steps: int llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) def reason(state: AgentState) -> AgentState: """LLM reasoning step.""" messages = state["messages"] response = llm.invoke(messages) return { "messages": [AIMessage(content=response.content)], "steps": state["steps"] + 1 } def should_continue(state: AgentState) -> str: """Decide whether to continue or end.""" if state["steps"] >= 5: return "end" last_msg = state["messages"][-1] if "FINAL ANSWER" in last_msg.content: return "end" return "continue" graph = StateGraph(AgentState) graph.add_node("reason", reason) graph.set_entry_point("reason") graph.add_conditional_edges( "reason", should_continue, {"continue": "reason", "end": END} ) # With checkpointing checkpointer = SqliteSaver.from_conn_string("checkpoints.db") app = graph.compile(checkpointer=checkpointer) # Run with thread ID for persistence config = {"configurable": {"thread_id": "user-123"}} result = app.invoke( {"query": "Research AI agents", "messages": [HumanMessage(content="Research AI agents")], "steps": 0}, config=config ) 

# langgraph/tool_agent.py from langgraph.prebuilt import ToolNode from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, List import operator import requests @tool def search_web(query: str) -> str: """Search the web for information.""" # Implement with SerpAPI, Tavily, etc.  return f"Search results for: {query}" @tool def calculate(expression: str) -> str: """Evaluate mathematical expression.""" try: return str(eval(expression)) except: return "Invalid expression" @tool def get_weather(city: str) -> str: """Get current weather for a city.""" # Implement with weather API  return f"Weather in {city}: Sunny, 22°C" tools = [search_web, calculate, get_weather] tool_node = ToolNode(tools) llm = ChatOpenAI(model="gpt-4o").bind_tools(tools) class AgentState(TypedDict): messages: Annotated[List, operator.add] def agent(state: AgentState) -> AgentState: response = llm.invoke(state["messages"]) return {"messages": [response]} def should_use_tools(state: AgentState) -> str: last_message = state["messages"][-1] if last_message.tool_calls: return "tools" return "end" graph = StateGraph(AgentState) graph.add_node("agent", agent) graph.add_node("tools", tool_node) graph.set_entry_point("agent") graph.add_conditional_edges("agent", should_use_tools, {"tools": "tools", "end": END}) graph.add_edge("tools", "agent") app = graph.compile() # Run result = app.invoke({ "messages": [HumanMessage(content="What's the weather in Tokyo?")] }) 

# langgraph/rag_flow.py from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, List, Literal import operator class RAGState(TypedDict): query: str messages: Annotated[List, operator.add] retrieved_docs: Annotated[List[Dict], operator.add] answer: str grade: Literal["relevant", "irrelevant", "hallucinated"] iterations: int def retrieve(state: RAGState) -> RAGState: """Retrieve documents for query.""" docs = retriever.invoke(state["query"]) return {"retrieved_docs": docs} def grade_documents(state: RAGState) -> RAGState: """Grade retrieved document relevance.""" query = state["query"] docs = state["retrieved_docs"] grades = [] for doc in docs: score = grader.invoke({"query": query, "document": doc}) grades.append(score) relevant_docs = [d for d, g in zip(docs, grades) if g == "relevant"] return { "retrieved_docs": relevant_docs, "grade": "relevant" if relevant_docs else "irrelevant" } def generate(state: RAGState) -> RAGState: """Generate answer from retrieved docs.""" context = "\n".join([d["content"] for d in state["retrieved_docs"]]) answer = generator.invoke({"query": state["query"], "context": context}) return {"answer": answer} def check_hallucination(state: RAGState) -> RAGState: """Verify answer is grounded in context.""" score = hallucination_checker.invoke({ "answer": state["answer"], "context": state["retrieved_docs"] }) return {"grade": score} def rewrite_query(state: RAGState) -> RAGState: """Rewrite query for better retrieval.""" new_query = query_rewriter.invoke({"query": state["query"]}) return {"query": new_query, "iterations": state["iterations"] + 1} graph = StateGraph(RAGState) graph.add_node("retrieve", retrieve) graph.add_node("grade_docs", grade_documents) graph.add_node("generate", generate) graph.add_node("check_hallucination", check_hallucination) graph.add_node("rewrite_query", rewrite_query) graph.set_entry_point("retrieve") graph.add_edge("retrieve", "grade_docs") graph.add_conditional_edges( "grade_docs", lambda s: s["grade"], {"relevant": "generate", "irrelevant": "rewrite_query"} ) graph.add_edge("generate", "check_hallucination") graph.add_conditional_edges( "check_hallucination", lambda s: s["grade"] if s["iterations"] < 3 else "max_retries", {"relevant": END, "hallucinated": "generate", "max_retries": END} ) graph.add_edge("rewrite_query", "retrieve") app = graph.compile() 

# langgraph/human_in_loop.py from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.graph import StateGraph, END from typing import TypedDict, Annota...