LangChain Quick Start | NVIDIA NeMo Relay

Goal

Run one LangChain agent and write its model and tool lifecycle to a local Agent Trajectory Observability Format (ATOF) file.

Use the event relationships to follow model and tool work through a run and check where the expected activity occurred.

Prerequisites

Before you start, make sure you have:

Python 3.11 or newer
uv
Inference access: an NVIDIA API key for this example, or credentials for your own provider

Install

Create a project and install the NeMo Relay version for these docs with the LangChain and NVIDIA model provider extras. For an existing uv project, run only the uv add command from its directory.

mkdir relay-langchain-quickstart
cd relay-langchain-quickstart
uv init
uv add "nemo-relay[langchain,langchain-nvidia]==0.11.0"

Set your NVIDIA API key for the current shell:

Unix shell

Windows PowerShell

export NVIDIA_API_KEY="your-api-key"

Relay does not impose a model choice. This copyable example selects the public nvidia/nemotron-3.5-lightning-30b-a3b model through the NVIDIA provider. To use another model or provider, replace the example’s model argument and configure that provider’s credentials. The NVIDIA provider reads NVIDIA_API_KEY. In the LangChain string, the first nvidia: selects the provider; the remaining nvidia/... value is the model ID.

Configure

plugins.toml enables Relay’s observability component and writes its events to logs/events.jsonl. In your own project, use the same plugin configuration with the integration code below; the marker task and success checks are only the quick start’s validation example.

Save the following configuration as plugins.toml in your project directory:

plugins.toml

# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
version = 1
[policy]
unknown_component = "error"
unknown_field = "error"
unsupported_value = "error"
[[components]]
kind = "observability"
enabled = true
[components.config]
version = 4
enable_full_payloads = false
[components.config.atof]
enabled = true
[[components.config.atof.sinks]]
type = "file"
output_directory = "logs"
filename = "events.jsonl"
mode = "overwrite"

ATOF output can contain sensitive prompts, responses, tool arguments, and tool results. Protect logs/events.jsonl even when full payload export is disabled.

Run

The example creates a LangChain agent with one marker tool. The require_marker_tool middleware makes this test request a tool result before its final answer; it is test scaffolding rather than required Relay setup. NemoRelayMiddleware observes model and tool execution, and NemoRelayCallbackHandler records the framework lifecycle. Plugin activation loads plugins.toml, the outer scope groups the request, and flushing finishes writing queued events before verification.

In your own agent, keep the Relay middleware, callback, activation, scope, and flush lifecycle. Replace the marker tool and prompt with your application work; the assertions after invocation and event-file checks belong to this example.

Save this file as quickstart.py beside plugins.toml:

quickstart.py

# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
import asyncio
import json
from pathlib import Path
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
import nemo_relay
from nemo_relay.integrations.langchain import NemoRelayCallbackHandler, NemoRelayMiddleware
@tool
def emit_marker() -> str:
    """Return the quick-start validation marker."""
    return "RELAY_QUICKSTART_TOOL_OK"
@wrap_model_call
async def require_marker_tool(request, handler):
    """Require the marker tool once, then allow the model's final response."""
    # Require a real tool call before the final answer; a prompt alone can be ignored.
    tool_used = any(getattr(message, "type", None) == "tool" for message in request.messages)
    return await handler(request.override(tool_choice="none" if tool_used else "emit_marker"))
agent = create_agent(
    model=init_chat_model("nvidia:nvidia/nemotron-3.5-lightning-30b-a3b", temperature=0),
    tools=[emit_marker],
    middleware=[require_marker_tool, NemoRelayMiddleware()],
    system_prompt="Call emit_marker exactly once. Keep the final answer brief.",
)
async def main() -> None:
    """Run the instrumented agent and verify its response and event lifecycle."""
    async with nemo_relay.plugin.activate({}, "plugins.toml"):
        try:
            with nemo_relay.scope.scope("langchain-request", nemo_relay.ScopeType.Agent):
                result = await agent.ainvoke(
                    {
                        "messages": [
                            {
                                "role": "user",
                                "content": (
                                    "Use one available tool to produce "
                                    "RELAY_QUICKSTART_TOOL_OK. Then include "
                                    "RELAY_QUICKSTART_RESPONSE_OK in your final response."
                                ),
                            }
                        ]
                    },
                    config={"callbacks": [NemoRelayCallbackHandler()]},
                )
            tool_outputs = [
                str(message.content) for message in result["messages"] if getattr(message, "type", None) == "tool"
            ]
            if tool_outputs.count("RELAY_QUICKSTART_TOOL_OK") != 1:
                raise RuntimeError("Expected exactly one marker tool result")
            final_message = result["messages"][-1]
            if getattr(final_message, "type", None) != "ai":
                raise RuntimeError("Expected a final assistant message")
            final_response = str(final_message.content)
            if "RELAY_QUICKSTART_RESPONSE_OK" not in final_response:
                raise RuntimeError("Expected final response is missing")
            print("RELAY_QUICKSTART_TOOL_OK")
            print("RELAY_QUICKSTART_RESPONSE_OK")
        finally:
            await nemo_relay.subscribers.flush_async()
    events = [json.loads(line) for line in Path("logs/events.jsonl").read_text().splitlines()]
    if not events or any(
        not isinstance(event, dict) or event.get("atof_version") != "0.1" or event.get("kind") not in {"scope", "mark"}
        for event in events
    ):
        raise RuntimeError("Expected a nonempty ATOF 0.1 event stream")
    starts = {
        event["uuid"]: event for event in events if event["kind"] == "scope" and event.get("scope_category") == "start"
    }
    closed = [
        event
        for event in events
        if event["kind"] == "scope"
        and event.get("scope_category") == "end"
        and event.get("uuid") in starts
        and all(event.get(key) == starts[event["uuid"]].get(key) for key in ("name", "category", "parent_uuid"))
    ]
    if not {"agent", "tool", "llm"} <= {event.get("category") for event in closed}:
        raise RuntimeError("Expected paired agent, tool, and LLM lifecycle records")
    print("RELAY_QUICKSTART_EVENTS_OK")
asyncio.run(main())

Run it from the project directory:

uv run python quickstart.py

Verify

The program checks the tool result, final assistant response, and paired ATOF agent, tool, and LLM records. A successful run prints these markers in order:

RELAY_QUICKSTART_TOOL_OK
RELAY_QUICKSTART_RESPONSE_OK
RELAY_QUICKSTART_EVENTS_OK

The model can use different wording around the response marker. These checks validate a live provider call; they do not make the model deterministic.

Inspect the Recorded Activity

Print one completed model call from the event file:

uv run python -c "
import json
from pathlib import Path
events = [json.loads(line) for line in Path('logs/events.jsonl').read_text().splitlines()]
call = next(event for event in events if event.get('scope_category') == 'end' and event.get('category') == 'llm')
fields = ('name', 'category', 'category_profile', 'uuid', 'parent_uuid', 'timestamp', 'metadata')
print(json.dumps({key: call.get(key) for key in fields}, indent=2))
"

category and name identify the recorded work. The same uuid pairs its start and end records; parent_uuid connects it to its enclosing scope. timestamp records when this event occurred. For model calls, category_profile and metadata can include the model identity and completion status, depending on the integration. These fields let you follow a call through the run without requiring full prompt or response capture.

For More Details

For more details, refer to: