LangGraph Interview Questions

from langgraph.graph import StateGraph, END

workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_function)
workflow.add_edge("agent", END)
app = workflow.compile()

from typing import TypedDict, Annotated
import operator
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]
    next_step: str
    iteration: int

def agent_node(state: AgentState) -> dict:
    # Return partial state update
    return {
        "messages": [AIMessage(content="Response")],
        "iteration": state["iteration"] + 1
    }

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = workflow.compile(
    checkpointer=memory,
    interrupt_before=["human_review"]
)

# Run until interruption
thread = {"configurable": {"thread_id": "1"}}
for event in graph.stream(inputs, thread):
    pass

# Get current state
state = graph.get_state(thread)

# Human reviews and approves
# Resume execution
graph.stream(None, thread)

members = ["researcher", "coder", "writer"]

def supervisor(state):
    # LLM decides next worker
    response = llm.invoke("Who should act next?")
    return {"next": response.next_agent}

workflow = StateGraph(State)
workflow.add_node("supervisor", supervisor)
for member in members:
    workflow.add_node(member, worker_functions[member])

# Conditional edges based on supervisor decision
workflow.add_conditional_edges("supervisor", route_function)

def generate(state):
    response = llm.invoke(state["messages"])
    return {"draft": response}

def reflect(state):
    critique = llm.invoke(f"Critique: {state['draft']}")
    return {"feedback": critique}

def should_continue(state):
    if state["iteration"] > 3 or state["feedback"].approved:
        return "end"
    return "reflect"

workflow.add_edge("generate", "reflect")
workflow.add_conditional_edges("reflect", should_continue)

def tool_node(state):
    try:
        result = execute_tool(state["tool_call"])
        return {"result": result, "error": None}
    except Exception as e:
        return {"result": None, "error": str(e)}

def route_after_tool(state):
    if state["error"]:
        return "handle_error"
    return "continue"

workflow.add_conditional_edges("tool", route_after_tool)

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.postgres import PostgresSaver

# In-memory (development)
memory = MemorySaver()

# Postgres (production)
postgres = PostgresSaver.from_conn_string(conn_string)

graph = workflow.compile(checkpointer=memory)

# State persists across invocations
thread = {"configurable": {"thread_id": "user-123"}}

# Define subgraph
research_graph = StateGraph(ResearchState)
research_graph.add_node("search", search_node)
research_graph.add_node("summarize", summarize_node)
research_compiled = research_graph.compile()

# Use in parent graph
main_graph = StateGraph(MainState)
main_graph.add_node("research", research_compiled)
main_graph.add_node("write", write_node)

def routing_function(state: AgentState) -> str:
    if state["needs_research"]:
        return "research"
    elif state["needs_coding"]:
        return "code"
    else:
        return END

workflow.add_conditional_edges(
    "decision",
    routing_function,
    {
        "research": "research_node",
        "code": "code_node",
        END: END
    }
)

# Visualize structure
graph.get_graph().draw_mermaid()

# Save as image
graph.get_graph().draw_mermaid_png(output_path="graph.png")

# Debug with LangSmith
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"

# Time travel debugging
history = list(graph.get_state_history(thread))

from langgraph.prebuilt import create_react_agent

agent = create_react_agent(llm, tools)

# Agent loop:
# 1. Reason: What should I do?
# 2. Act: Call a tool
# 3. Observe: Get result
# 4. Repeat until done

# Plan: Create high-level steps
plan = planner.invoke("Research topic X")
# → ["Search web", "Read articles", "Summarize"]

# Execute: Run each step
for step in plan:
    result = executor.invoke(step)

from langgraph.checkpoint.postgres_aio import AsyncPostgresSaver

# Async checkpoint for non-blocking
checkpointer = AsyncPostgresSaver.from_conn_string(conn_str)

graph = workflow.compile(checkpointer=checkpointer)

# Run with timeout
import asyncio
result = await asyncio.wait_for(
    graph.ainvoke(inputs, config),
    timeout=300  # 5 min timeout
)

from langchain_core.messages import HumanMessage, AIMessage, ToolMessage

class State(TypedDict):
    messages: Annotated[list, operator.add]

def agent_node(state):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# Messages accumulate through the graph

def tool_node(state):
    tool_call = state["messages"][-1].tool_calls[0]

    # Validate before execution
    if tool_call["name"] == "delete_data":
        if not is_admin(state["user"]):
            return {"messages": [ToolMessage(
                content="Unauthorized",
                tool_call_id=tool_call["id"]
            )]}

    # Execute validated tool
    result = tools[tool_call["name"]].invoke(tool_call["args"])
    return {"messages": [ToolMessage(content=result, ...)]}

from langgraph.constants import Send

def router(state):
    # Send to multiple workers in parallel
    return [
        Send("worker", {"task": task})
        for task in state["tasks"]
    ]

workflow.add_conditional_edges("router", router)

def tool_node_with_retry(state):
    for attempt in range(3):
        try:
            result = execute_tool(state)
            return {"result": result, "error": None}
        except Exception as e:
            if attempt == 2:
                return {"result": None, "error": str(e)}
            time.sleep(2 ** attempt)  # Exponential backoff

from langgraph.types import Command

def decision_node(state):
    if state["should_skip"]:
        return Command(goto="end", update={"skipped": True})
    return Command(goto="next", update={"processed": True})

# Stream node outputs
for event in graph.stream(inputs, stream_mode="values"):
    print(event)

# Stream updates only
for update in graph.stream(inputs, stream_mode="updates"):
    print(update)

# Stream with LLM tokens
async for event in graph.astream_events(inputs, version="v2"):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="")

from langgraph.types import interrupt

def review_node(state):
    if state["confidence"] < 0.8:
        # Dynamically request human review
        human_input = interrupt("Please review this output")
        return {"approved": human_input == "approve"}
    return {"approved": True}

def should_continue(state):
    if state["iteration"] >= 5:
        return "end"  # Prevent infinite loops
    if state["task_complete"]:
        return "end"
    return "agent"  # Continue loop

workflow.add_conditional_edges("check", should_continue, {
    "agent": "agent",
    "end": END
})

from typing import Annotated
import operator

class State(TypedDict):
    # List: append new items
    messages: Annotated[list, operator.add]

    # Counter: sum values
    count: Annotated[int, operator.add]

    # Last value: replace
    current: str

import pytest

def test_agent_workflow():
    # Mock LLM
    mock_llm = FakeLLM(responses=["Use search tool", "Final answer"])

    graph = create_agent_graph(llm=mock_llm)

    result = graph.invoke({"question": "What is X?"})

    assert result["answer"] is not None
    assert "search" in result["tools_used"]

# Each thread has its own state history
config_user1 = {"configurable": {"thread_id": "user-123"}}
config_user2 = {"configurable": {"thread_id": "user-456"}}

# Different users, different states
graph.invoke(input1, config_user1)
graph.invoke(input2, config_user2)

# Get history for specific thread
history = list(graph.get_state_history(config_user1))

# LangServe
from fastapi import FastAPI
from langserve import add_routes

app = FastAPI()
add_routes(app, compiled_graph, path="/agent")

# Endpoints: /agent/invoke, /agent/stream

from langgraph.constants import Send

def map_node(state):
    return [Send("worker", {"item": item}) for item in state["items"]]

def reduce_node(state):
    return {"result": aggregate(state["partial_results"])}

workflow.add_conditional_edges("mapper", map_node)
workflow.add_edge("worker", "reducer")

CEO Agent
├── Research Manager
│   ├── Web Researcher
│   └── Document Analyst
└── Writing Manager
    ├── Drafter
    └── Editor

def routing_function(state):
    if state["needs_research"]:
        return "research_agent"
    elif state["needs_coding"]:
        return "coding_agent"
    return END

workflow.add_conditional_edges("supervisor", routing_function, {
    "research_agent": "research_agent",
    "coding_agent": "coding_agent",
    END: END
})

def prune_messages(state):
    if len(state["messages"]) > 50:
        return {"messages": state["messages"][-20:]}
    return {}

# A → B, A → C runs B and C in parallel
workflow.add_edge("A", "B")
workflow.add_edge("A", "C")
workflow.add_edge("B", "D")
workflow.add_edge("C", "D")
# B and C run concurrently

import asyncio

async def invoke_with_timeout(graph, inputs, config, timeout=300):
    try:
        return await asyncio.wait_for(
            graph.ainvoke(inputs, config),
            timeout=timeout
        )
    except asyncio.TimeoutError:
        return {"error": "Timeout exceeded"}

async def event_handler(event):
    thread = {"configurable": {"thread_id": event.user_id}}

    # Resume or start new conversation
    await graph.ainvoke(
        {"message": event.content},
        config=thread
    )

# Connect to event bus (Kafka, Redis Streams, etc.)

from typing import Optional

class StateV2(TypedDict):
    messages: list
    new_field: Optional[str]  # New in v2

def migrate_state(old_state):
    return {**old_state, "new_field": None}

def invoke_for_tenant(tenant_id, user_id, input):
    config = {
        "configurable": {
            "thread_id": f"{tenant_id}:{user_id}"
        }
    }
    return graph.invoke(input, config=config)

from langsmith import traceable

@traceable
def agent_node(state):
    # Automatically traced
    return {"result": process(state)}

# Or use callbacks
from langchain.callbacks import LangChainTracer

tracer = LangChainTracer(project_name="my-project")
graph.invoke(input, config={"callbacks": [tracer]})

# Compose graphs
research_graph = build_research_graph()
writing_graph = build_writing_graph()

main_graph = StateGraph(State)
main_graph.add_node("research", research_graph)
main_graph.add_node("write", writing_graph)
main_graph.add_edge("research", "write")

# Use optimistic concurrency
state = graph.get_state(thread)

# Check version before update
if state.config["configurable"].get("checkpoint_id"):
    # Include checkpoint_id to prevent conflicts
    pass

class State(TypedDict):
    # Short-term: in state
    messages: list

    # Long-term: external store
    user_preferences: dict  # Loaded from DB

def load_memory(state):
    prefs = db.get(f"user:{state['user_id']}")
    return {"user_preferences": prefs}

def should_recover(state):
    if state["retry_count"] < 3:
        return "retry"
    return "human_escalate"

from ratelimit import limits, sleep_and_retry

@sleep_and_retry
@limits(calls=10, period=60)
def call_llm(prompt):
    return llm.invoke(prompt)

# Or use semaphore
semaphore = asyncio.Semaphore(5)

async def limited_invoke(input):
    async with semaphore:
        return await graph.ainvoke(input)

def evaluate_agent(test_cases):
    results = []
    for case in test_cases:
        output = graph.invoke(case["input"])
        results.append({
            "correct": output == case["expected"],
            "steps": count_steps(output)
        })
    return results

class State(TypedDict):
    messages: list  # Can include image/audio content
    images: list[bytes]

def vision_node(state):
    images = state["images"]
    response = vision_llm.invoke([
        HumanMessage(content=[
            {"type": "text", "text": "Describe:"},
            {"type": "image_url", "image_url": images[0]}
        ])
    ])
    return {"messages": [response]}

def agent_with_degradation(state):
    try:
        return full_capability_response(state)
    except LLMError:
        return limited_response(state)
    except ToolError:
        return {"messages": ["Tool unavailable, trying alternative..."]}

def secure_tool_node(state):
    if not user_has_permission(state["user"], state["tool"]):
        raise PermissionError("Not authorized")

@dataclass
class AgentConfig:
    max_iterations: int = 10
    temperature: float = 0.7
    tools_enabled: list = field(default_factory=list)

config = AgentConfig.from_env()  # Or from config file
graph = build_graph(config)

from typing import TypedDict, Annotated, Sequence
import operator
from langchain_core.messages import BaseMessage
from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]

def agent(state):
    # Agent logic here
    return {"messages": ["Agent response"]}

workflow = StateGraph(AgentState)
workflow.add_node("agent", agent)
workflow.set_entry_point("agent")
workflow.add_edge("agent", END)

app = workflow.compile()

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.output_parsers.openai_functions import JsonOutputFunctionsParser

members = ["researcher", "coder"]
system_prompt = (
    "You are a supervisor tasked with managing a conversation between the"
    " following workers: {members}. Given the following user request,"
    " respond with the worker to act next. Each worker will perform a"
    " task and respond with their results and status. When finished,"
    " respond with FINISH."
)
options = ["FINISH"] + members
function_def = {
    "name": "route",
    "description": "Select the next role.",
    "parameters": {
        "title": "routeSchema",
        "type": "object",
        "properties": {
            "next": {
                "title": "Next",
                "anyOf": [
                    {"enum": options},
                ],
            }
        },
        "required": ["next"],
    },
}
prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    ("user", "{messages}"),
    ("system", "Given the conversation above, who should act next? or should we FINISH?"),
])

supervisor_chain = (
    prompt
    | ChatOpenAI(model="gpt-4-turbo").bind_functions(functions=[function_def], function_call="route")
    | JsonOutputFunctionsParser()
)

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = workflow.compile(checkpointer=memory, interrupt_before=["human_review"])

# Run until interruption
thread = {"configurable": {"thread_id": "1"}}
for event in graph.stream(inputs, thread):
    pass

# Review and continue
full_state = graph.get_state(thread)
# ... human reviews state ...
graph.stream(None, thread) # Resume execution