Building LangGraph Agents

LangGraph is a library for building stateful, multi-step AI agent applications using a graph-based architecture. AgentVisor™ uses LangGraph as its primary framework for stateful agents because it provides built-in support for state management, conditional branching, and human-in-the-loop patterns that are essential for production AI agents.

This guide covers building LangGraph agents for AgentVisor.

Basic Structure

Every LangGraph agent needs:

State definition: TypedDict describing agent state
Nodes: Functions that process state
Edges: Connections between nodes
Checkpointer: For state persistence

Minimal Example

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import AIMessage
from agentvisor.langgraph import AgentVisorCheckpointer


class AgentState(TypedDict):
    messages: Annotated[list, add_messages]


def respond(state: AgentState) -> dict:
    return {"messages": [AIMessage(content="Hello!")]}


builder = StateGraph(AgentState)
builder.add_node("respond", respond)
builder.set_entry_point("respond")
builder.add_edge("respond", END)

graph = builder.compile(checkpointer=AgentVisorCheckpointer())

State Design

Using Annotations

The add_messages annotation automatically appends new messages:

class AgentState(TypedDict):
    # Messages accumulate across runs
    messages: Annotated[list, add_messages]

    # Other fields are replaced
    step_count: int
    current_task: str

Custom Reducers

For custom accumulation logic:

from langgraph.graph import add_messages

def add_to_list(existing: list, new: list) -> list:
    """Custom reducer that deduplicates."""
    seen = set(existing)
    return existing + [x for x in new if x not in seen]

class AgentState(TypedDict):
    items: Annotated[list, add_to_list]

Node Types

Processing Nodes

Standard nodes that transform state:

def process(state: AgentState) -> dict:
    messages = state.get("messages", [])
    # Process and return updates
    return {
        "messages": [AIMessage(content="Processed")],
        "step_count": state.get("step_count", 0) + 1
    }

Tool Nodes

Nodes that call external tools:

from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """Search the web for information."""
    # HTTP requests go through AgentVisor proxy
    import requests
    response = requests.get(f"https://api.example.com/search?q={query}")
    return response.json()["results"]


def use_tools(state: AgentState) -> dict:
    # Get tool calls from last message
    last_msg = state["messages"][-1]
    if hasattr(last_msg, "tool_calls"):
        results = []
        for call in last_msg.tool_calls:
            result = search.invoke(call["args"])
            results.append({"tool": call["name"], "result": result})
        return {"tool_results": results}
    return {}

LLM Nodes

Nodes that call language models:

from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-sonnet-4-20250514")

def call_llm(state: AgentState) -> dict:
    messages = state.get("messages", [])
    response = llm.invoke(messages)
    return {"messages": [response]}

Edge Types

Direct Edges

Always go from A to B:

builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", END)

Conditional Edges

Route based on state:

def should_continue(state: AgentState) -> str:
    if state.get("step_count", 0) >= 3:
        return "end"
    return "continue"

builder.add_conditional_edges(
    "process",
    should_continue,
    {
        "continue": "process",
        "end": END
    }
)

Tool Decision Edges

Route based on tool calls:

def route_tools(state: AgentState) -> str:
    last_msg = state["messages"][-1]
    if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
        return "tools"
    return "respond"

builder.add_conditional_edges(
    "call_llm",
    route_tools,
    {
        "tools": "use_tools",
        "respond": END
    }
)

Complete Example

A ReAct-style agent with tool use:

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langchain_core.messages import AIMessage, ToolMessage
from agentvisor.langgraph import AgentVisorCheckpointer


class AgentState(TypedDict):
    messages: Annotated[list, add_messages]


@tool
def calculator(expression: str) -> str:
    """Evaluate a math expression."""
    try:
        result = eval(expression)  # In production, use a safe evaluator
        return str(result)
    except Exception as e:
        return f"Error: {e}"


# Bind tools to LLM
llm = ChatAnthropic(model="claude-sonnet-4-20250514")
llm_with_tools = llm.bind_tools([calculator])


def call_model(state: AgentState) -> dict:
    messages = state.get("messages", [])
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}


def call_tools(state: AgentState) -> dict:
    messages = state.get("messages", [])
    last_msg = messages[-1]

    tool_results = []
    for call in last_msg.tool_calls:
        if call["name"] == "calculator":
            result = calculator.invoke(call["args"])
            tool_results.append(
                ToolMessage(content=result, tool_call_id=call["id"])
            )

    return {"messages": tool_results}


def should_use_tools(state: AgentState) -> str:
    last_msg = state["messages"][-1]
    if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
        return "tools"
    return "end"


# Build graph
builder = StateGraph(AgentState)
builder.add_node("model", call_model)
builder.add_node("tools", call_tools)
builder.set_entry_point("model")

builder.add_conditional_edges(
    "model",
    should_use_tools,
    {"tools": "tools", "end": END}
)
builder.add_edge("tools", "model")

graph = builder.compile(checkpointer=AgentVisorCheckpointer())

Project Configuration

langgraph.json

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

Multiple graphs:

{
  "dependencies": ["."],
  "graphs": {
    "chat": "./agent.py:chat_graph",
    "research": "./agent.py:research_graph"
  }
}

requirements.txt

langgraph>=0.2.0
langchain-core>=0.3.0
langchain-anthropic>=0.2.0

Testing Locally

# Start development services (from the scaffolded project's compose.yml)
docker compose up -d

# Run agent
agentvisor serve ./my-agent

# Test
curl -sX POST http://localhost:8090/threads | jq
THREAD=$(curl -sX POST http://localhost:8090/threads | jq -r '.thread_id')
curl -sX POST "http://localhost:8090/threads/$THREAD/runs?wait=30s" \
    -H "Content-Type: application/json" \
    -d '{"input": {"messages": [{"role": "user", "content": "What is 2+2?"}]}}' | jq

Best Practices

Keep Nodes Small

Each node should do one thing:

# Good
def parse_input(state): ...
def call_api(state): ...
def format_output(state): ...

# Bad
def do_everything(state): ...

Handle Errors

Return error state rather than raising:

def safe_api_call(state: AgentState) -> dict:
    try:
        result = call_external_api()
        return {"result": result, "error": None}
    except Exception as e:
        return {"result": None, "error": str(e)}

Use Type Hints

Type hints help catch errors early:

from typing import TypedDict, Annotated, Optional

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    result: Optional[str]
    error: Optional[str]

Streaming: Real-time SSE streaming for LLM responses and agent output
Human-in-the-Loop: Interrupt patterns for approval gates and user input
Policy Configuration: Authorization policies for LangGraph tool calls

Basic Structure​

Minimal Example​

State Design​

Using Annotations​

Custom Reducers​

Node Types​

Processing Nodes​

Tool Nodes​

LLM Nodes​

Edge Types​

Direct Edges​

Conditional Edges​

Tool Decision Edges​

Complete Example​

Project Configuration​

langgraph.json​

requirements.txt​

Testing Locally​

Best Practices​

Keep Nodes Small​

Handle Errors​

Use Type Hints​

Related Guides​