# state.py from typing import TypedDict, List, Annotated from langgraph.graph import add_messages from pydantic import BaseModel class AgentState(TypedDict): # Messages with automatic reducer (appends)  messages: Annotated[List[BaseModel], add_messages] # Custom fields  user_id: str session_id: str tool_call_count: int current_plan: str retrieved_docs: List[dict] error_count: int 

# nodes.py from langgraph.graph import END from langchain_core.messages import AIMessage, ToolMessage from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) llm_with_tools = llm.bind_tools(tools) async def agent_node(state: AgentState): """Main reasoning node.""" response = await llm_with_tools.ainvoke(state["messages"]) return { "messages": [response], "tool_call_count": state.get("tool_call_count", 0) + len(response.tool_calls) if response.tool_calls else 0 } async def tool_node(state: AgentState): """Execute tool calls.""" last_message = state["messages"][-1] tool_outputs = [] for tool_call in last_message.tool_calls: tool = get_tool_by_name(tool_call["name"]) try: result = await tool.ainvoke(tool_call["args"]) tool_outputs.append(ToolMessage( content=str(result), tool_call_id=tool_call["id"] )) except Exception as e: tool_outputs.append(ToolMessage( content=f"Error: {str(e)}", tool_call_id=tool_call["id"] )) return {"messages": tool_outputs} def should_continue(state: AgentState) -> str: """Route: continue to tools or end.""" last_message = state["messages"][-1] if last_message.tool_calls: return "tools" return END 

# graph.py from langgraph.graph import StateGraph, START def build_agent_graph(): workflow = StateGraph(AgentState) # Add nodes  workflow.add_node("agent", agent_node) workflow.add_node("tools", tool_node) workflow.add_node("validate", validation_node) # Edges  workflow.add_edge(START, "agent") workflow.add_conditional_edges( "agent", should_continue, {"tools": "tools", END: END} ) workflow.add_edge("tools", "agent") # Compile with checkpointing  return workflow.compile( checkpointer=MemorySaver(), # Or PostgresSaver for production  interrupt_before=["tools"] # Human-in-the-loop  ) 

# research_agent.py from typing import TypedDict, List, Annotated, Literal from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.checkpoint.sqlite import SqliteSaver from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage from langchain_core.tools import tool from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field import json import sqlite3 from datetime import datetime # ============ STATE ============ class ResearchState(TypedDict): messages: Annotated[List[BaseMessage], add_messages] research_topic: str sub_questions: List[str] gathered_info: List[dict] current_step: str iteration: int max_iterations: int final_report: str # ============ TOOLS ============ @tool def web_search(query: str, num_results: int = 5) -> str: """Search the web for information.""" # Implementation using your preferred search API  # Return formatted results  pass @tool def save_research_note(topic: str, content: str, source: str) -> str: """Save a research finding.""" conn = sqlite3.connect("research.db") conn.execute(""" CREATE TABLE IF NOT EXISTS notes ( id INTEGER PRIMARY KEY, topic TEXT, content TEXT, source TEXT, created_at TEXT ) """) conn.execute( "INSERT INTO notes (topic, content, source, created_at) VALUES (?, ?, ?, ?)", (topic, content, source, datetime.utcnow().isoformat()) ) conn.commit() conn.close() return f"Saved note for '{topic}'" tools = [web_search, save_research_note] # ============ NODES ============ llm = ChatOpenAI(model="gpt-4o", temperature=0) planner_llm = llm.bind_tools(tools) researcher_llm = llm.bind_tools([web_search]) writer_llm = llm.bind_tools([save_research_note]) async def planner(state: ResearchState): """Plan the research approach.""" topic = state["research_topic"] prompt = f"""Break down this research topic into 3-5 specific sub-questions: Topic: {topic} Return as JSON: {{"sub_questions": ["q1", "q2", ...]}}""" response = await planner_llm.ainvoke([HumanMessage(content=prompt)]) plan = json.loads(response.content) return { "sub_questions": plan["sub_questions"], "current_step": "researching", "iteration": 0, "gathered_info": [] } async def researcher(state: ResearchState): """Research each sub-question.""" if state["iteration"] >= len(state["sub_questions"]): return {"current_step": "writing"} question = state["sub_questions"][state["iteration"]] prompt = f"""Research this question thoroughly using web search: Question: {question} Search for multiple sources and synthesize findings.""" response = await researcher_llm.ainvoke([ HumanMessage(content=prompt) ]) # Execute tool calls  tool_results = [] for tool_call in response.tool_calls: if tool_call["name"] == "web_search": result = await web_search.ainvoke(tool_call["args"]) tool_results.append({ "question": question, "result": result, "source": "web_search" }) return { "gathered_info": state["gathered_info"] + tool_results, "iteration": state["iteration"] + 1 } async def writer(state: ResearchState): """Compile final report.""" prompt = f"""Write a comprehensive research report on: {state['research_topic']} Sub-questions and findings: {json.dumps(state['gathered_info'], indent=2)} Structure: 1. Executive Summary 2. Detailed Findings (per sub-question) 3. Key Insights 4. Sources 5. Limitations""" response = await writer_llm.ainvoke([HumanMessage(content=prompt)]) # Save final report  await save_research_note.ainvoke({ "topic": state["research_topic"], "content": response.content, "source": "final_report" }) return { "final_report": response.content, "current_step": "complete" } def route_research(state: ResearchState) -> Literal["researcher", "writer"]: return state["current_step"] # ============ GRAPH ============ def build_research_agent(): workflow = StateGraph(ResearchState) workflow.add_node("planner", planner) workflow.add_node("researcher", researcher) workflow.add_node("writer", writer) workflow.add_edge(START, "planner") workflow.add_edge("planner", "researcher") workflow.add_conditional_edges( "researcher", route_research, {"researcher": "researcher", "writer": "writer"} ) workflow.add_edge("writer", END) # Persistence  checkpointer = SqliteSaver.from_conn_string("sqlite:///research_agent.db") return workflow.compile(checkpointer=checkpointer) # ============ USAGE ============ async def run_research(topic: str, thread_id: str): agent = build_research_agent() config = {"configurable": {"thread_id": thread_id}} initial_state = { "messages": [HumanMessage(content=f"Research: {topic}")], "research_topic": topic, "max_iterations": 5 } # Stream for progress  async for chunk in agent.astream(initial_state, config): for node, update in chunk.items(): print(f"[{node}] {update.get('current_step', '')}") if update.get("final_report"): print("\n=== FINAL REPORT ===") print(update["final_report"]) # Run # asyncio.run(run_research("Impact of AI on software development jobs", "thread-123")) 

# multi_agent.py from langgraph.graph import StateGraph from typing import Literal class MultiAgentState(TypedDict): messages: Annotated[List[BaseMessage], add_messages] next_agent: str task: str results: dict def supervisor(state: MultiAgentState) -> Literal["researcher", "coder", "reviewer", END]: """Route to appropriate specialist.""" last_msg = state["messages"][-1].content.lower() if "research" in last_msg or "find" in last_msg: return "researcher" elif "code" in last_msg or "implement" in last_msg: return "coder" elif "review" in last_msg or "check" in last_msg: return "reviewer" return END # Each specialist is a subgraph researcher_graph = build_research_subgraph() coder_graph = build_coder_subgraph() reviewer_graph = build_reviewer_subgraph() workflow = StateGraph(MultiAgentState) workflow.add_node("supervisor", supervisor_node) workflow.add_node("researcher", researcher_graph) workflow.add_node("coder", coder_graph) workflow.add_node("reviewer", reviewer_graph) workflow.add_edge(START, "supervisor") workflow.add_conditional_edges("supervisor", supervisor) workflow.add_edge("researcher", "supervisor") workflow.add_edge("coder", "supervisor") workflow.add_edge("reviewer", "supervisor") 

# human_in_loop.py from langgraph.types import interrupt, Command async def sensitive_action(state: AgentState): """Action requiring human approval.""" action = state.get("pending_action") # Interrupt and wait for human  human_response = interrupt({ "action": action, "question": f"Approve {action['name']} with params {action['params']}?" }) if human_response.get("approved"): result = await execute_action(action) return {"messages": [ToolMessage(content=result, tool_call_id=action["id"])]} else: return {"messages": [Source：https://dev.to/ipsphil_digital/building-ai-agents-with-langgraph-state-tools-multi-agent-workflows-3a7a
My favorites