performance-optimization-for-langgraph

Performance Optimization for LangGraph

Optimize LangGraph workflows through node-level caching, async patterns, and cost monitoring.

Note: Examples in this skill assume a compiled graph instance graph. See the Node-Level Caching section for a complete example of graph creation.

Node-Level Caching with CachePolicy

LangGraph provides node-level caching to avoid redundant computation. Use CachePolicy to configure caching per node.

from langgraph.cache.memory import InMemoryCache
from langgraph.types import CachePolicy
from langgraph.graph import StateGraph
from langchain_anthropic import ChatAnthropic

# Initialize LLM instance (reuse for connection pooling)
llm = ChatAnthropic(model="claude-sonnet-4-5-20250929", max_tokens=1024)

def expensive_llm_node(state: dict) -> dict:
    """Node that makes expensive LLM calls."""
    response = llm.invoke(state["query"])
    return {"response": response}

# Create graph builder
builder = StateGraph(dict)

# Add node with cache policy (TTL in seconds)
builder.add_node(
    "llm_node",
    expensive_llm_node,
    cache_policy=CachePolicy(ttl=3600)  # Cache for 1 hour
)

# Compile graph with cache
graph = builder.compile(cache=InMemoryCache())

# Repeated invocations with same input return cached results
result1 = graph.invoke({"query": "What is Python?"})  # Calls LLM
result2 = graph.invoke({"query": "What is Python?"})  # Returns cached

Custom Cache Keys:

import hashlib
import json

def custom_key_func(input_data: dict) -> str:
    """Generate cache key from specific fields."""
    key_data = {"query": input_data.get("query")}
    serialized = json.dumps(key_data, sort_keys=True)
    return hashlib.md5(serialized.encode(), usedforsecurity=False).hexdigest()

builder.add_node(
    "llm_node",
    expensive_llm_node,
    cache_policy=CachePolicy(ttl=3600, key_func=custom_key_func)
)

Async for Throughput

import asyncio

# Sequential (slow)
def process_queries_sync(queries):
    results = []
    for query in queries:
        result = graph.invoke({"messages": [query]})
        results.append(result)
    return results

# Parallel (fast) with error handling
async def process_queries_async(queries: list[str]) -> list[dict]:
    """Process queries in parallel with error handling."""
    tasks = [graph.ainvoke({"messages": [q]}) for q in queries]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    # Filter out exceptions or handle them as needed
    return [r for r in results if not isinstance(r, Exception)]