LangGraph Integration

LangGraph provides powerful graph-based orchestration for building complex, multi-step AI agents. CopilotKit’s LangGraphAgent seamlessly integrates LangGraph workflows into your application.

Installation

pip install copilotkit

LangGraph and LangChain dependencies are included automatically.

Quick Start

from copilotkit import LangGraphAgent
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import HumanMessage, SystemMessage

# Define your graph
def agent_node(state):
    # Your agent logic here
    return {"messages": state["messages"]}

# Build the graph
builder = StateGraph(dict)
builder.add_node("agent", agent_node)
builder.add_edge(START, "agent")
builder.add_edge("agent", END)
graph = builder.compile()

# Create the agent
agent = LangGraphAgent(
    name="my_agent",
    description="A helpful assistant",
    graph=graph
)

Creating a LangGraph Agent

Basic Configuration

name

str

required

The unique identifier for your agent. Must contain only alphanumeric characters, underscores, and hyphens.

graph

CompiledStateGraph

required

The compiled LangGraph graph to execute.

description

str

Description of the agent’s purpose. Used for dynamic routing when multiple agents are available.

langgraph_config

RunnableConfig

LangChain/LangGraph configuration to use with the agent.

copilotkit_config

CopilotKitConfig

Advanced CopilotKit configuration for customizing message conversion and state merging.

Complete Example

from copilotkit import LangGraphAgent
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

# Define state schema
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    copilotkit: dict

# Initialize LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

# Define agent node
def call_model(state: AgentState):
    messages = state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}

# Build graph
builder = StateGraph(AgentState)
builder.add_node("agent", call_model)
builder.add_edge(START, "agent")
builder.add_edge("agent", END)

# Add memory for conversation history
memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

# Create CopilotKit agent
agent = LangGraphAgent(
    name="chat_agent",
    description="A conversational AI assistant",
    graph=graph,
    langgraph_config={
        "configurable": {
            "thread_id": "default"
        }
    }
)

Working with Tools

Defining Tools for LangGraph

from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode
from langgraph.graph import StateGraph, START, END

@tool
def search_database(query: str) -> str:
    """Search the database for relevant information."""
    # Your search implementation
    return f"Results for: {query}"

@tool
def send_email(to: str, subject: str, body: str) -> dict:
    """Send an email to a recipient."""
    # Your email implementation
    return {"success": True, "message_id": "123"}

# Create tools list
tools = [search_database, send_email]

# Bind tools to LLM
llm_with_tools = ChatOpenAI(model="gpt-4o").bind_tools(tools)

# Define agent node
def call_agent(state):
    messages = state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

# Create tool node
tool_node = ToolNode(tools)

# Build graph with tools
builder = StateGraph(AgentState)
builder.add_node("agent", call_agent)
builder.add_node("tools", tool_node)
builder.add_edge(START, "agent")

# Add conditional routing for tool calls
def should_continue(state):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

builder.add_conditional_edges("agent", should_continue)
builder.add_edge("tools", "agent")

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

Advanced State Management

Custom State Schema

from typing import TypedDict, List
from pydantic import BaseModel

class Task(BaseModel):
    id: str
    title: str
    completed: bool

class CustomState(TypedDict):
    messages: Annotated[list, add_messages]
    copilotkit: dict
    tasks: List[Task]
    user_preferences: dict

def task_agent(state: CustomState):
    # Access custom state fields
    tasks = state.get("tasks", [])
    preferences = state.get("user_preferences", {})
    
    # Your logic here
    messages = state["messages"]
    response = llm.invoke(messages)
    
    return {
        "messages": [response],
        "tasks": tasks  # Update tasks if needed
    }

State Schema Filtering

Control which state fields are exposed to the frontend:

from langgraph.graph import StateGraph
from pydantic import BaseModel

class InputState(TypedDict):
    messages: Annotated[list, add_messages]
    copilotkit: dict
    user_id: str  # Only accepted as input

class OutputState(TypedDict):
    messages: Annotated[list, add_messages]
    tasks: List[Task]  # Only returned as output
    # Note: user_id is not in output

graph = StateGraph(
    state_schema=AgentState,
    input=InputState,
    output=OutputState
)

Multi-Agent Workflows

Sequential Agent Chain

from langgraph.graph import StateGraph, START, END

def research_agent(state):
    # Research phase
    return {"messages": [...], "research_data": data}

def writing_agent(state):
    # Use research data to write
    research = state.get("research_data")
    return {"messages": [...], "draft": draft}

def review_agent(state):
    # Review and finalize
    draft = state.get("draft")
    return {"messages": [...], "final": final}

builder = StateGraph(AgentState)
builder.add_node("research", research_agent)
builder.add_node("write", writing_agent)
builder.add_node("review", review_agent)

builder.add_edge(START, "research")
builder.add_edge("research", "write")
builder.add_edge("write", "review")
builder.add_edge("review", END)

graph = builder.compile()

Parallel Agent Execution

from langgraph.graph import StateGraph, START, END

def data_agent(state):
    return {"data_results": [...]}

def image_agent(state):
    return {"image_results": [...]}

def synthesis_agent(state):
    # Combine results from parallel agents
    data = state.get("data_results")
    images = state.get("image_results")
    return {"messages": [combined_response]}

builder = StateGraph(AgentState)
builder.add_node("data", data_agent)
builder.add_node("images", image_agent)
builder.add_node("synthesis", synthesis_agent)

builder.add_edge(START, "data")
builder.add_edge(START, "images")
builder.add_edge("data", "synthesis")
builder.add_edge("images", "synthesis")
builder.add_edge("synthesis", END)

graph = builder.compile()

Human-in-the-Loop with Interrupts

Creating Breakpoints

from langgraph.checkpoint.memory import MemorySaver

def approval_needed(state):
    # Agent requests approval
    return {"messages": [...], "pending_action": action}

builder = StateGraph(AgentState)
builder.add_node("agent", agent_node)
builder.add_node("approval", approval_needed)
builder.add_node("execute", execute_action)

builder.add_edge(START, "agent")
builder.add_edge("agent", "approval")
builder.add_edge("approval", "execute")
builder.add_edge("execute", END)

# Compile with interrupt before approval
graph = builder.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["approval"]
)

agent = LangGraphAgent(
    name="approval_agent",
    graph=graph
)

Responding to Interrupts

When an interrupt occurs, CopilotKit emits an interrupt event to the frontend:

// Frontend handling
useCopilotChat({
  onInterrupt: (interrupt) => {
    // Show approval UI
    const approved = await showApprovalDialog(interrupt.value);
    
    // Resume with response
    return { approved };
  }
});

Custom Message Conversion

Advanced CopilotKit Configuration

from copilotkit import LangGraphAgent

def custom_merge_state(*, state, messages, actions, agent_name):
    """Custom state merging logic"""
    # Your custom merge logic
    return {
        **state,
        "messages": messages,
        "custom_field": "value"
    }

def custom_convert_messages(messages):
    """Custom message conversion"""
    def converter(copilotkit_messages):
        # Convert CopilotKit messages to LangChain format
        langchain_messages = []
        for msg in copilotkit_messages:
            # Your conversion logic
            langchain_messages.append(...)
        return langchain_messages
    return converter

agent = LangGraphAgent(
    name="custom_agent",
    graph=graph,
    copilotkit_config={
        "merge_state": custom_merge_state,
        "convert_messages": custom_convert_messages
    }
)

Streaming Intermediate State

Stream state updates during long-running operations:

from langgraph.types import Send

def emit_progress(state):
    """Emit intermediate state updates"""
    # Add metadata to enable streaming
    return {
        "messages": [response],
        "progress": 50
    }

# In your graph definition
builder.add_node(
    "process",
    emit_progress,
    metadata={"copilotkit:emit-intermediate-state": [
        {
            "state_key": "progress",
            "tool": "process_data",
            "tool_argument": "progress"
        }
    ]}
)

Using with FastAPI

from fastapi import FastAPI
from copilotkit import LangGraphAgent
from copilotkit.integrations.fastapi import add_fastapi_endpoint

app = FastAPI()

agent = LangGraphAgent(
    name="my_agent",
    graph=graph
)

add_fastapi_endpoint(
    app,
    agents=[agent],
    endpoint="/copilotkit"
)

Memory and Persistence

In-Memory Checkpointing

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

PostgreSQL Checkpointing

from langgraph.checkpoint.postgres import PostgresSaver

# Configure PostgreSQL checkpointer
checkpointer = PostgresSaver.from_conn_string(
    conn_string="postgresql://user:pass@localhost/db"
)

graph = builder.compile(checkpointer=checkpointer)

SQLite Checkpointing

from langgraph.checkpoint.sqlite import SqliteSaver

checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
graph = builder.compile(checkpointer=checkpointer)

Error Handling

def safe_agent_node(state):
    try:
        # Your agent logic
        response = llm.invoke(state["messages"])
        return {"messages": [response]}
    except Exception as e:
        # Handle errors gracefully
        error_message = AIMessage(
            content=f"I encountered an error: {str(e)}"
        )
        return {"messages": [error_message]}

builder.add_node("agent", safe_agent_node)

Complete Production Example

from copilotkit import LangGraphAgent
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.tools import tool
from typing import TypedDict, Annotated, List
from langgraph.graph.message import add_messages

# Define tools
@tool
def search_knowledge_base(query: str) -> str:
    """Search internal knowledge base."""
    return f"Results for: {query}"

@tool  
def create_ticket(title: str, description: str) -> dict:
    """Create a support ticket."""
    return {"ticket_id": "T-12345", "status": "created"}

tools = [search_knowledge_base, create_ticket]

# Define state
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    copilotkit: dict

# Initialize components
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
llm_with_tools = llm.bind_tools(tools)
tool_node = ToolNode(tools)

# Build graph
def call_agent(state: AgentState):
    messages = state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

def should_continue(state):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

builder = StateGraph(AgentState)
builder.add_node("agent", call_agent)
builder.add_node("tools", tool_node)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue)
builder.add_edge("tools", "agent")

# Compile with persistence
checkpointer = PostgresSaver.from_conn_string(
    conn_string=os.environ["DATABASE_URL"]
)
graph = builder.compile(checkpointer=checkpointer)

# Create agent
agent = LangGraphAgent(
    name="support_agent",
    description="A customer support agent with knowledge base access",
    graph=graph,
    langgraph_config={
        "configurable": {
            "thread_id": "default"
        }
    }
)

API Reference

See the LangGraphAgent source code for complete implementation details.

Next Steps

BuiltInAgent

Use the built-in agent for simple cases

CrewAI Integration

Build multi-agent systems with CrewAI

Custom Agents

Create custom agent implementations

LangGraph Docs

Official LangGraph documentation

​Installation

​Quick Start

​Creating a LangGraph Agent

​Basic Configuration

​Complete Example

​Working with Tools

​Defining Tools for LangGraph

​Advanced State Management

​Custom State Schema

​State Schema Filtering

​Multi-Agent Workflows

​Sequential Agent Chain

​Parallel Agent Execution

​Human-in-the-Loop with Interrupts

​Creating Breakpoints

​Responding to Interrupts

​Custom Message Conversion

​Advanced CopilotKit Configuration

​Streaming Intermediate State

​Using with FastAPI

​Memory and Persistence

​In-Memory Checkpointing

​PostgreSQL Checkpointing

​SQLite Checkpointing

​Error Handling

​Complete Production Example

​API Reference

​Next Steps