LangGraph Agent Workflow Practical Guide: From Multi-Agent Orchestration to Production Deployment

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

# 1) Define state
class AgentState(TypedDict):
    messages: Annotated[list, "add_messages"]  # Accumulate messages
    next_action: str
    result: str

# 2) Define node functions
def classify_intent(state: AgentState) -> dict:
    """Classify user intent"""
    last_msg = state["messages"][-1].content
    # Classify intent with LLM
    intent = llm.invoke(f"Classify intent: {last_msg}")
    return {"next_action": intent.content}

def handle_question(state: AgentState) -> dict:
    """Handle questions"""
    answer = llm.invoke(state["messages"])
    return {"result": answer.content}

def handle_task(state: AgentState) -> dict:
    """Execute tasks"""
    result = agent_executor.invoke(state["messages"])
    return {"result": result}

# 3) Build graph
graph = StateGraph(AgentState)
graph.add_node("classify", classify_intent)
graph.add_node("question", handle_question)
graph.add_node("task", handle_task)

# 4) Connect edges
graph.add_edge(START, "classify")
graph.add_conditional_edges(
    "classify",
    lambda state: state["next_action"],
    {
        "question": "question",
        "task": "task",
    }
)
graph.add_edge("question", END)
graph.add_edge("task", END)

# 5) Compile & run
app = graph.compile()
result = app.invoke({"messages": [HumanMessage("What is Kubernetes?")]})

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

@tool
def search_docs(query: str) -> str:
    """Search internal documentation."""
    # Vector DB search
    results = vectorstore.similarity_search(query, k=3)
    return "\n".join([r.page_content for r in results])

@tool
def run_query(sql: str) -> str:
    """Execute a read-only SQL query."""
    return db.execute(sql).fetchall()

@tool
def create_ticket(title: str, description: str) -> str:
    """Create a Jira ticket."""
    return jira.create_issue(title=title, description=description)

llm = ChatOpenAI(model="gpt-4o")

# Automatically create a ReAct agent
agent = create_react_agent(
    llm,
    tools=[search_docs, run_query, create_ticket],
    state_modifier="You are a helpful DevOps assistant."
)

# Execute
result = agent.invoke({
    "messages": [HumanMessage("Check if order-service has errors and create a ticket")]
})

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import create_react_agent

class SupervisorState(TypedDict):
    messages: Annotated[list, "add_messages"]
    next_agent: str
    final_answer: str

# Specialist agents
researcher = create_react_agent(llm, tools=[search_web, search_docs])
coder = create_react_agent(llm, tools=[run_code, read_file])
reviewer = create_react_agent(llm, tools=[analyze_code, lint_code])

def supervisor(state: SupervisorState) -> dict:
    """Decide which agent to delegate to"""
    response = llm.invoke([
        SystemMessage("You are a supervisor. Route to: researcher, coder, reviewer, or FINISH"),
        *state["messages"]
    ])
    return {"next_agent": response.content.strip()}

def run_researcher(state):
    result = researcher.invoke({"messages": state["messages"]})
    return {"messages": result["messages"]}

def run_coder(state):
    result = coder.invoke({"messages": state["messages"]})
    return {"messages": result["messages"]}

def run_reviewer(state):
    result = reviewer.invoke({"messages": state["messages"]})
    return {"messages": result["messages"]}

# Build graph
workflow = StateGraph(SupervisorState)
workflow.add_node("supervisor", supervisor)
workflow.add_node("researcher", run_researcher)
workflow.add_node("coder", run_coder)
workflow.add_node("reviewer", run_reviewer)

workflow.add_edge(START, "supervisor")
workflow.add_conditional_edges(
    "supervisor",
    lambda s: s["next_agent"],
    {
        "researcher": "researcher",
        "coder": "coder",
        "reviewer": "reviewer",
        "FINISH": END,
    }
)
# Each agent returns to supervisor after execution
for agent_name in ["researcher", "coder", "reviewer"]:
    workflow.add_edge(agent_name, "supervisor")

app = workflow.compile()

from langgraph.types import interrupt, Command

def sensitive_action(state):
    """Request human approval before sensitive operations"""
    action = state["pending_action"]

    # Pause execution and wait for human approval
    approval = interrupt({
        "question": f"Approve this action?\n{action}",
        "options": ["approve", "reject", "modify"]
    })

    if approval == "approve":
        return execute_action(action)
    elif approval == "reject":
        return {"result": "Action cancelled by user"}
    else:
        return {"result": f"Action modified: {approval}"}

# Compile with checkpointer
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)

# Execute (pauses at interrupt)
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke({"messages": [...]}, config)
# result = {"__interrupt__": {...}}

# Resume after human approval
app.invoke(Command(resume="approve"), config)

from langgraph.graph import StateGraph

class ParallelState(TypedDict):
    query: str
    web_results: str
    db_results: str
    combined: str

def search_web_node(state):
    return {"web_results": search_web(state["query"])}

def search_db_node(state):
    return {"db_results": search_db(state["query"])}

def combine_results(state):
    combined = f"Web: {state['web_results']}\nDB: {state['db_results']}"
    return {"combined": combined}

graph = StateGraph(ParallelState)
graph.add_node("web", search_web_node)
graph.add_node("db", search_db_node)
graph.add_node("combine", combine_results)

# Parallel execution: START -> [web, db] -> combine
graph.add_edge(START, "web")
graph.add_edge(START, "db")
graph.add_edge("web", "combine")
graph.add_edge("db", "combine")
graph.add_edge("combine", END)

app = graph.compile()

# Token-level streaming
async for event in app.astream_events(
    {"messages": [HumanMessage("Explain CQRS")]},
    version="v2"
):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="", flush=True)

# Node-level streaming
for chunk in app.stream({"messages": [HumanMessage("...")]}):
    for node_name, output in chunk.items():
        print(f"[{node_name}]: {output}")

from langgraph.checkpoint.postgres import PostgresSaver

# PostgreSQL checkpointer (production)
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost/langgraph"
)

app = workflow.compile(checkpointer=checkpointer)

# Persistent conversations (identified by thread_id)
config = {"configurable": {"thread_id": "user-456"}}

# First message
app.invoke({"messages": [HumanMessage("Hi")]}, config)

# Second message (previous conversation context preserved)
app.invoke({"messages": [HumanMessage("What did I just say?")]}, config)

{
  "dependencies": ["."],
  "graphs": {
    "agent": "./agent.py:app"
  },
  "env": ".env"
}

# Install LangGraph CLI
pip install langgraph-cli

# Local testing
langgraph dev

# Docker build
langgraph build -t my-agent:latest

# LangGraph Cloud deployment
langgraph deploy --app my-agent