tessl/npm-langchain--langgraph

Low-level orchestration framework for building stateful, multi-actor applications with LLMs

Overview

Eval results

Files

Remote Graph

Name: tessl/npm-langchain--langgraph
Author: tessl

Client implementation for calling remote LangGraph APIs deployed on LangGraph Cloud or LangSmith. RemoteGraph behaves identically to CompiledStateGraph and can be used as a node in another graph.

RemoteGraph

Client for remote graph execution via HTTP API.

class RemoteGraph<
  InputType = any,
  OutputType = any
> extends Runnable<InputType, OutputType> implements PregelInterface {
  graphId: string;
  client: Client;
  syncClient: Client;

  constructor(params: RemoteGraphParams);

  invoke(
    input: InputType,
    options?: PregelOptions
  ): Promise<OutputType>;

  stream(
    input: InputType,
    options?: PregelOptions
  ): IterableReadableStream;

  streamEvents(
    input: InputType,
    options: PregelOptions & { version: "v1" | "v2" }
  ): IterableReadableStream<StreamEvent>;

  getState(
    config: LangGraphRunnableConfig,
    options?: GetStateOptions
  ): Promise<StateSnapshot>;

  getStateHistory(
    config: LangGraphRunnableConfig,
    options?: CheckpointListOptions
  ): AsyncIterableIterator<StateSnapshot>;

  updateState(
    config: LangGraphRunnableConfig,
    values: UpdateType | Command,
    asNode?: string
  ): Promise<RunnableConfig>;

  getGraph(
    config?: RunnableConfig & { xray?: boolean | number }
  ): DrawableGraph;

  getSubgraphs(
    namespace?: string,
    recurse?: boolean
  ): IterableIterator<[string, Pregel]>;

  getSubgraphsAsync(
    namespace?: string,
    recurse?: boolean
  ): AsyncIterableIterator<[string, Pregel]>;

  withConfig(config: Partial<PregelOptions>): this;
}

RemoteGraphParams

Configuration for RemoteGraph client.

interface RemoteGraphParams {
  graphId: string;
  client?: Client;
  url?: string;
  apiKey?: string;
  headers?: Record<string, string>;
  streamResumable?: boolean;
  checkpointer?: BaseCheckpointSaver | false;
  store?: BaseStore;
  cache?: BaseCache;
  interruptBefore?: string[] | All;
  interruptAfter?: string[] | All;
  name?: string;
  description?: string;
}

Parameters

graphId - Unique identifier for the remote graph
client - LangGraph SDK Client instance (optional, created from other params if not provided)
url - Base URL of the remote LangGraph API
apiKey - API key for authentication
headers - Additional HTTP headers
streamResumable - Whether streaming can be resumed after interruption
checkpointer - Local checkpoint saver (for offline operation)
store - Local store for long-term memory
cache - Local cache for results
interruptBefore - Nodes to interrupt before (overrides remote config)
interruptAfter - Nodes to interrupt after (overrides remote config)
name - Graph name
description - Graph description

Usage Examples

Basic Remote Graph

import { RemoteGraph } from "@langchain/langgraph/remote";

const remoteGraph = new RemoteGraph({
  graphId: "my-deployed-graph",
  url: "https://api.langgraph.cloud",
  apiKey: process.env.LANGGRAPH_API_KEY
});

const result = await remoteGraph.invoke({
  messages: [{ role: "user", content: "Hello!" }]
});

With LangGraph SDK Client

import { Client } from "@langchain/langgraph-sdk";
import { RemoteGraph } from "@langchain/langgraph/remote";

const client = new Client({
  apiUrl: "https://api.langgraph.cloud",
  apiKey: process.env.LANGGRAPH_API_KEY
});

const remoteGraph = new RemoteGraph({
  graphId: "my-graph",
  client
});

Remote Graph with State Management

const remoteGraph = new RemoteGraph({
  graphId: "stateful-graph",
  url: process.env.LANGGRAPH_URL,
  apiKey: process.env.LANGGRAPH_API_KEY
});

// First invocation
await remoteGraph.invoke(input, {
  configurable: { thread_id: "thread-123" }
});

// Continue from checkpoint
await remoteGraph.invoke(null, {
  configurable: { thread_id: "thread-123" }
});

Streaming from Remote Graph

for await (const chunk of await remoteGraph.stream(input, {
  streamMode: "values",
  configurable: { thread_id: "thread-1" }
})) {
  console.log(chunk);
}

Multiple Stream Modes

for await (const chunk of await remoteGraph.stream(input, {
  streamMode: ["values", "updates", "debug"],
  configurable: { thread_id: "thread-1" }
})) {
  const [namespace, mode, data] = chunk;
  console.log(`Mode: ${mode}`, data);
}

Getting Remote State

const state = await remoteGraph.getState({
  configurable: { thread_id: "thread-1" }
});

console.log(state.values); // Current state
console.log(state.next); // Next nodes to execute
console.log(state.tasks); // Pending tasks

State History

for await (const snapshot of remoteGraph.getStateHistory({
  configurable: { thread_id: "thread-1" }
}, {
  limit: 10
})) {
  console.log(snapshot.values);
  console.log(snapshot.metadata);
}

Updating Remote State

await remoteGraph.updateState(
  { configurable: { thread_id: "thread-1" } },
  { messages: [{ role: "user", content: "Override" }] },
  "node_name" // Act as if this node made the update
);

Remote Graph as Subgraph

import { StateGraph, Annotation } from "@langchain/langgraph";
import { RemoteGraph } from "@langchain/langgraph/remote";

const remoteResearchAgent = new RemoteGraph({
  graphId: "research-agent",
  url: process.env.LANGGRAPH_URL,
  apiKey: process.env.LANGGRAPH_API_KEY
});

const State = Annotation.Root({
  messages: Annotation<BaseMessage[]>({
    reducer: (a, b) => a.concat(b),
    default: () => []
  }),
  researchData: Annotation<any>
});

const mainGraph = new StateGraph(State)
  .addNode("local_processing", localNode)
  .addNode("remote_research", remoteResearchAgent) // Use as node
  .addNode("synthesis", synthesisNode)
  .addEdge("__start__", "local_processing")
  .addEdge("local_processing", "remote_research")
  .addEdge("remote_research", "synthesis")
  .addEdge("synthesis", "__end__")
  .compile();

await mainGraph.invoke(input);

Custom Headers

const remoteGraph = new RemoteGraph({
  graphId: "my-graph",
  url: "https://api.langgraph.cloud",
  apiKey: process.env.LANGGRAPH_API_KEY,
  headers: {
    "X-Custom-Header": "value",
    "Authorization": `Bearer ${process.env.CUSTOM_TOKEN}`
  }
});

Interrupts with Remote Graphs

const remoteGraph = new RemoteGraph({
  graphId: "interruptible-graph",
  url: process.env.LANGGRAPH_URL,
  apiKey: process.env.LANGGRAPH_API_KEY,
  interruptBefore: ["human_review"]
});

// First call - will interrupt
const result = await remoteGraph.invoke(input, {
  configurable: { thread_id: "thread-1" }
});

if (isInterrupted(result)) {
  console.log("Waiting for human input");

  // Resume with Command
  await remoteGraph.invoke(
    new Command({ resume: humanInput }),
    { configurable: { thread_id: "thread-1" } }
  );
}

Error Handling

import { RemoteException } from "@langchain/langgraph";

try {
  const result = await remoteGraph.invoke(input);
} catch (error) {
  if (error instanceof RemoteException) {
    console.error("Remote graph error:", error.message);
  } else {
    console.error("Network or other error:", error);
  }
}

Stream Events

for await (const event of remoteGraph.streamEvents(input, {
  version: "v2",
  streamMode: "values",
  configurable: { thread_id: "thread-1" }
})) {
  console.log(event);
}

Visualization

const graph = remoteGraph.getGraph();
console.log(graph.nodes); // Nodes in remote graph
console.log(graph.edges); // Edges in remote graph

// Generate visualization
const mermaid = graph.drawMermaid();
console.log(mermaid);

Subgraph Iteration

// Iterate through subgraphs
for (const [namespace, subgraph] of remoteGraph.getSubgraphs()) {
  console.log(`Subgraph: ${namespace}`);
}

// Async iteration
for await (const [namespace, subgraph] of remoteGraph.getSubgraphsAsync(undefined, true)) {
  console.log(`Subgraph: ${namespace}`);
}

Deployment Patterns

Production Deployment

// Environment-based configuration
const remoteGraph = new RemoteGraph({
  graphId: process.env.GRAPH_ID!,
  url: process.env.LANGGRAPH_API_URL!,
  apiKey: process.env.LANGGRAPH_API_KEY!,
  headers: {
    "X-Environment": process.env.NODE_ENV || "production"
  }
});

// Retry wrapper
async function invokeWithRetry(input: any, maxRetries = 3) {
  for (let i = 0; i < maxRetries; i++) {
    try {
      return await remoteGraph.invoke(input);
    } catch (error) {
      if (i === maxRetries - 1) throw error;
      await new Promise(resolve => setTimeout(resolve, 1000 * (i + 1)));
    }
  }
}

Multi-Tenant Setup

function createTenantGraph(tenantId: string) {
  return new RemoteGraph({
    graphId: "multi-tenant-graph",
    url: process.env.LANGGRAPH_URL!,
    apiKey: process.env.LANGGRAPH_API_KEY!,
    headers: {
      "X-Tenant-ID": tenantId
    }
  });
}

const tenantGraph = createTenantGraph("tenant-123");
await tenantGraph.invoke(input, {
  configurable: { thread_id: `${tenantId}-${sessionId}` }
});

Load Balancing

const endpoints = [
  "https://api1.langgraph.cloud",
  "https://api2.langgraph.cloud",
  "https://api3.langgraph.cloud"
];

function getBalancedGraph() {
  const url = endpoints[Math.floor(Math.random() * endpoints.length)];
  return new RemoteGraph({
    graphId: "balanced-graph",
    url,
    apiKey: process.env.LANGGRAPH_API_KEY
  });
}

Client Object

LangGraph SDK Client for direct API access.

class Client {
  constructor(config: {
    apiUrl: string;
    apiKey?: string;
    headers?: Record<string, string>;
  });

  threads: ThreadsClient;
  runs: RunsClient;
  crons: CronsClient;
  assistants: AssistantsClient;
}

Use the Client directly for advanced operations beyond RemoteGraph capabilities.

import { Client } from "@langchain/langgraph-sdk";

const client = new Client({
  apiUrl: "https://api.langgraph.cloud",
  apiKey: process.env.LANGGRAPH_API_KEY
});

// List assistants
const assistants = await client.assistants.list();

// Get thread state
const threadState = await client.threads.getState(threadId);

// List runs
const runs = await client.runs.list(threadId);

Install with Tessl CLI

npx tessl i tessl/npm-langchain--langgraph