Skip to navigation
Supported Integrations

NeMo Relay LangGraph Integration

View as Markdown

Use the nemo_relay.integrations.langgraph package to add NeMo Relay observability to LangGraph workflows through public LangGraph APIs.

Setup

Install the LangGraph integration extra in your application environment.

uv add "nemo-relay[langgraph]"

Installing the langgraph extra also installs the LangChain integration dependencies.

Usage Example

from typing_extensions import TypedDict
import nemo_relay
from langgraph.graph import END, START, StateGraph
from nemo_relay.integrations.langgraph import NemoRelayCallbackHandler
class State(TypedDict):
    value: int
def increment(state: State) -> State:
    return {"value": state["value"] + 1}
builder = StateGraph(State)
builder.add_node("increment", increment)
builder.add_edge(START, "increment")
builder.add_edge("increment", END)
graph = builder.compile()
with nemo_relay.scope.scope("langgraph-request", nemo_relay.ScopeType.Agent):
    result = graph.invoke(
        {"value": 1},
        config={"callbacks": [NemoRelayCallbackHandler()]},
    )
print(result)

Managed ToolNode Calls

For a standalone LangGraph ToolNode, construct the node with create_tool_node. It routes each model-requested tool call through Relay’s managed execution pipeline while preserving LangGraph state, store, and runtime argument injection.

from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph
from nemo_relay.integrations.langgraph import (
    NemoRelayCallbackHandler,
    create_tool_node,
)
@tool
def get_weather(location: str) -> str:
    """Return the weather for a location."""
    return f"Sunny in {location}"
builder = StateGraph(MessagesState)
builder.add_node("tools", create_tool_node([get_weather]))
builder.add_edge(START, "tools")
builder.add_edge("tools", END)
graph = builder.compile()
with nemo_relay.scope.scope("langgraph-request", nemo_relay.ScopeType.Agent):
    result = graph.invoke(
        {
            "messages": [
                AIMessage(
                    content="",
                    tool_calls=[{"name": "get_weather", "args": {"location": "Boston"}, "id": "call-1"}],
                )
            ]
        },
        config={"callbacks": [NemoRelayCallbackHandler()]},
    )

NemoRelayCallbackHandler records the graph lifecycle and provides its scopes. create_tool_node provides managed tool execution. Use both for complete standalone LangGraph instrumentation. For custom ToolNode wrapper composition, construct ToolNode directly and pass wrap_tool_call and awrap_tool_call from this integration.

Managed ToolNode calls use the JSON format described in Normalized Tool Results. Relay chooses the codec automatically. It keeps message fields, artifacts, graph state, and command destinations. It also handles lists of messages and commands.

When ToolNode handles an error and returns a ToolMessage with status="error", ATIF adds extra.is_error: true. Relay also sets the tool span status to ERROR. The message still returns to LangGraph. Graph interrupts still pause the workflow as usual.

Middleware and Wrapper Ordering

LangChain agents use the same middleware order when they run in LangGraph or through the LangGraph API. Both integrations use NemoRelayMiddleware. Place it before your middleware to record tool results after your wrappers return. Refer to LangChain middleware ordering for details.

configure_graph adds callbacks that create scopes for graph runs. It does not change the middleware order. For Deep Agents, use the middleware position option.

A standalone ToolNode uses tool call wrappers. It does not take an agent middleware list. create_tool_node sets the wrappers for you. It rejects custom wrap_tool_call and awrap_tool_call arguments.

To combine your wrappers with Relay, create a ToolNode directly. Add your result changes inside Relay’s wrapper. The following example uses get_weather from the earlier example. It adds a status line before Relay records the result:

from langgraph.prebuilt import ToolNode
from nemo_relay.integrations.langgraph import awrap_tool_call, wrap_tool_call
def observed_tool_call(request, execute):
    def execute_with_status(inner_request):
        result = execute(inner_request)
        return result.model_copy(update={"content": result.content + "\nStatus: complete"})
    return wrap_tool_call(request, execute_with_status)
async def observed_async_tool_call(request, execute):
    async def execute_with_status(inner_request):
        result = await execute(inner_request)
        return result.model_copy(update={"content": result.content + "\nStatus: complete"})
    return await awrap_tool_call(request, execute_with_status)
tool_node = ToolNode(
    [get_weather],
    wrap_tool_call=observed_tool_call,
    awrap_tool_call=observed_async_tool_call,
)

Relay records the result with the status line. If you add the line after wrap_tool_call or awrap_tool_call returns, the graph still receives it. Relay’s tool event and ATIF observation do not include it.

This example uses a ToolMessage with string content. If your tool returns a command or content blocks, keep that structure when you change the result. Changes in later graph nodes are outside the recorded tool call.

Served LangChain Agents

When you export a LangChain agent for langgraph dev or langgraph up, add the Relay callback handler to the compiled graph before exporting it. Request callbacks are not part of the serialized LangGraph API request.

from langchain.agents import create_agent
from nemo_relay.integrations.langgraph import NemoRelayMiddleware, configure_graph
agent = create_agent(
    model="nvidia:nvidia/nemotron-3-nano-omni-30b-a3b-reasoning",
    tools=[],
    middleware=[NemoRelayMiddleware()],
)
# Ensure the callables are configured with NeMo Relay scope events
agent = configure_graph(agent)

configure_graph stores a NemoRelayCallbackHandler in the graph’s base configuration and returns the same compiled graph, so the LangGraph server can load it. The callback handler creates the graph and node scopes. The middleware routes model and tool calls through Relay under those scopes.

The handler closes each graph scope on the stack where it started. This works even when the server runs the completion callback in a different context. Successful runs emit an end event with status OK. Failed runs emit an end event with status ERROR.

For tracked runs, Graph Interrupt and Graph Resume marks belong to the graph scope identified by their run ID, including when the server delivers them in a different context. When the run ID is absent or untracked, the marks use the current scope.

For LangChain agents inside a LangGraph workflow, use NemoRelayMiddleware from this package the same way as the LangChain integration and pass the LangGraph config into the nested agent call:

from langchain.agents import create_agent
from langchain_core.runnables import RunnableConfig
from nemo_relay.integrations.langgraph import NemoRelayMiddleware
agent = create_agent(
    model="nvidia:nvidia/nemotron-3-nano-omni-30b-a3b-reasoning",
    tools=[],
    middleware=[NemoRelayMiddleware()],
)
def agent_node(state: dict, config: RunnableConfig) -> dict:
    return agent.invoke({"messages": state["messages"]}, config=config)

Install the NVIDIA LangChain provider if you want to run the nested agent example as written:

uv add "nemo-relay[langgraph,langchain-nvidia]"

Verify the Integration

The integration works correctly when:

  • graph.invoke(...) returns the incremented state from the example.
  • The langgraph-request scope contains the LangGraph run.
  • Nested LangChain agents inherit the same callback config when you pass the LangGraph config through.

Observability

Refer to Observability for details on exporting NeMo Relay observability data to third-party systems.