Skip to main content

@pressw/threads-langgraph

LangGraph Cloud adapter for @pressw/threads, providing managed thread storage with built-in scalability and assistant integration.

Overview

@pressw/threads-langgraph connects your thread management to LangGraph Cloud, offering:

  • ☁️ Managed Storage - No database to maintain
  • 🚀 Built-in Scalability - Handles growth automatically
  • 🤖 Assistant Integration - Native LangGraph assistant support
  • 🔄 Real-time Sync - Cloud-based thread synchronization
  • 📊 Analytics Ready - Built-in metrics and monitoring
  • 🔐 Enterprise Security - SOC2 compliant infrastructure

Installation

npm install @pressw/threads @pressw/threads-langgraph @langchain/langgraph-sdk

Quick Start

1. Get Your LangGraph Cloud Credentials

First, sign up for LangGraph Cloud and create an assistant to get your credentials.

2. Create the Adapter

import { LangGraphAdapter } from '@pressw/threads-langgraph';

const adapter = new LangGraphAdapter({
apiUrl: 'https://api.langsmith.com',
apiKey: process.env.LANGGRAPH_API_KEY!,
assistantId: 'your-assistant-id',
});

3. Use with Thread Client

import { createThreadUtilityClient } from '@pressw/threads';

const threadClient = createThreadUtilityClient({
adapter,
getUserContext: async (request) => {
// Your auth logic
const session = await getSession(request);
return {
userId: session.userId,
organizationId: session.organizationId,
};
},
});

// Create a thread
const thread = await threadClient.createThread(request, {
title: 'Customer Support Chat',
metadata: {
source: 'web',
priority: 'high',
},
});

Configuration

Basic Configuration

const adapter = new LangGraphAdapter({
apiUrl: 'https://api.langsmith.com',
apiKey: process.env.LANGGRAPH_API_KEY!,
assistantId: 'your-assistant-id',
});

Advanced Configuration

const adapter = new LangGraphAdapter({
// Required
apiUrl: 'https://api.langsmith.com',
apiKey: process.env.LANGGRAPH_API_KEY!,
assistantId: 'your-assistant-id',

// Optional
defaultThreadConfig: {
// Default configuration for new threads
configurable: {
model_name: 'gpt-4',
temperature: 0.7,
},
},

// HTTP client options
httpOptions: {
timeout: 30000, // 30 seconds
retries: 3,
},

// Custom headers
headers: {
'X-Custom-Header': 'value',
},
});

LangGraph Cloud Integration

Working with Assistants

The adapter seamlessly integrates with LangGraph assistants:

// Create a thread connected to your assistant
const thread = await threadClient.createThread(request, {
title: 'AI Assistant Chat',
metadata: {
assistantVersion: '1.0',
features: ['rag', 'tools'],
},
});

// The thread ID can be used with LangGraph SDK
import { Client } from '@langchain/langgraph-sdk';

const langGraphClient = new Client({
apiUrl: 'https://api.langsmith.com',
apiKey: process.env.LANGGRAPH_API_KEY!,
});

// Use the same thread ID
const run = await langGraphClient.runs.create(
thread.id, // Thread ID from our adapter
'your-assistant-id',
{ input: { message: 'Hello!' } },
);

Thread State Management

LangGraph threads maintain state that can be accessed:

// Get thread with state
const thread = await threadClient.getThread(request, threadId);

// Access LangGraph state through metadata
const state = thread.metadata?.state;
const values = thread.metadata?.values;

Multi-tenant Support

The adapter supports multi-tenancy through metadata:

const threadClient = createThreadUtilityClient({
adapter,
getUserContext: async (request) => ({
userId: session.userId,
organizationId: session.organizationId,
tenantId: session.tenantId, // Stored in metadata
}),
});

// Threads are automatically filtered by tenant
const { threads } = await threadClient.listThreads(request);

Features

Automatic Thread Management

The adapter handles all thread lifecycle operations:

// Create
const thread = await threadClient.createThread(request, {
title: 'Support Thread',
});

// Update
await threadClient.updateThread(request, thread.id, {
title: 'Resolved: Support Thread',
metadata: {
...thread.metadata,
status: 'resolved',
},
});

// List with filtering
const { threads } = await threadClient.listThreads(request, {
limit: 20,
offset: 0,
search: 'billing', // Searches in title and metadata
});

// Delete
await threadClient.deleteThread(request, thread.id);

Metadata Storage

Store custom data with threads:

interface CustomMetadata {
department: string;
priority: 'low' | 'medium' | 'high';
tags: string[];
customerId: string;
resolution?: {
resolvedAt: string;
resolvedBy: string;
solution: string;
};
}

const thread = await threadClient.createThread(request, {
title: 'Technical Issue',
metadata: {
department: 'engineering',
priority: 'high',
tags: ['bug', 'production'],
customerId: 'cust-123',
} satisfies CustomMetadata,
});

Search Capabilities

Search across threads:

// Search in titles and metadata
const { threads } = await threadClient.listThreads(request, {
search: 'payment error',
orderBy: 'createdAt',
orderDirection: 'desc',
});

// Filter by user and organization automatically
// based on getUserContext

Real-time Updates

Threads are stored in the cloud and updates are immediately available:

// Update in one location
await threadClient.updateThread(request, threadId, {
metadata: { status: 'in-progress' },
});

// Immediately available elsewhere
const updated = await threadClient.getThread(request, threadId);
console.log(updated.metadata.status); // 'in-progress'

Usage Patterns

Customer Support System

// Create support thread
const supportThread = await threadClient.createThread(request, {
title: `Support: ${customerIssue.subject}`,
metadata: {
customerId: customer.id,
issueType: customerIssue.type,
priority: calculatePriority(customerIssue),
assignedAgent: null,
status: 'open',
sla: {
responseTime: '2h',
resolutionTime: '24h',
},
},
});

// Assign to agent
await threadClient.updateThread(request, supportThread.id, {
metadata: {
...supportThread.metadata,
assignedAgent: agentId,
status: 'assigned',
assignedAt: new Date().toISOString(),
},
});

AI Assistant Conversations

// Create AI conversation thread
const aiThread = await threadClient.createThread(request, {
title: 'AI Assistant Session',
metadata: {
model: 'gpt-4',
systemPrompt: 'You are a helpful assistant...',
maxTokens: 2000,
temperature: 0.7,
tools: ['web_search', 'calculator'],
},
});

// Track conversation metrics
await threadClient.updateThread(request, aiThread.id, {
metadata: {
...aiThread.metadata,
metrics: {
messageCount: 10,
tokensUsed: 1500,
toolCalls: 3,
duration: 300, // seconds
},
},
});

Project Collaboration

// Create project thread
const projectThread = await threadClient.createThread(request, {
title: project.name,
metadata: {
projectId: project.id,
team: project.teamMembers,
milestone: project.currentMilestone,
dueDate: project.deadline,
labels: project.labels,
visibility: 'team', // or 'private', 'organization'
},
});

Error Handling

The adapter provides detailed error information:

try {
await threadClient.createThread(request, data);
} catch (error) {
if (error.code === 'LANGGRAPH_AUTH_ERROR') {
// Invalid API key or unauthorized
} else if (error.code === 'LANGGRAPH_NOT_FOUND') {
// Assistant not found
} else if (error.code === 'RATE_LIMIT_EXCEEDED') {
// Too many requests
} else {
// Other errors
}
}

Performance Considerations

Caching

The adapter doesn't cache by default. Implement caching at the application level:

import { LRUCache } from 'lru-cache';

const threadCache = new LRUCache<string, Thread>({
max: 500,
ttl: 1000 * 60 * 5, // 5 minutes
});

// Wrap thread client methods
async function getCachedThread(threadId: string) {
const cached = threadCache.get(threadId);
if (cached) return cached;

const thread = await threadClient.getThread(request, threadId);
if (thread) threadCache.set(threadId, thread);
return thread;
}

Batch Operations

For bulk operations, consider batching:

// Instead of individual creates
for (const data of threadsData) {
await threadClient.createThread(request, data);
}

// Consider parallel execution
await Promise.all(threadsData.map((data) => threadClient.createThread(request, data)));

Rate Limiting

Respect LangGraph Cloud rate limits:

import pLimit from 'p-limit';

// Limit concurrent requests
const limit = pLimit(5);

const threads = await Promise.all(
threadIds.map((id) => limit(() => threadClient.getThread(request, id))),
);

Migration Guide

From Local Database

// Export from local database
const localThreads = await db.select().from(threads);

// Import to LangGraph Cloud
for (const thread of localThreads) {
await adapter.create({
model: 'thread',
data: {
id: thread.id, // Preserve IDs if needed
title: thread.title,
userId: thread.userId,
organizationId: thread.organizationId,
metadata: {
...thread.metadata,
migratedFrom: 'local-db',
migratedAt: new Date().toISOString(),
},
createdAt: thread.createdAt,
updatedAt: thread.updatedAt,
},
});
}

From Other Cloud Services

// Map from other service format
const mappedThread = {
title: otherService.thread.name,
userId: otherService.thread.ownerId,
metadata: {
originalId: otherService.thread.id,
originalService: 'other-cloud',
...otherService.thread.properties,
},
};

await threadClient.createThread(request, mappedThread);

Best Practices

1. Use Metadata Effectively

// Good: Structured metadata
metadata: {
category: 'support',
subcategory: 'billing',
tags: ['urgent', 'customer'],
internal: {
notes: 'Escalated by manager',
previousTickets: ['T-123', 'T-456'],
},
}

// Avoid: Unstructured data
metadata: {
data: 'support billing urgent', // Hard to query
}

2. Handle Network Failures

import { retry } from '@lifeomic/attempt';

async function resilientCreateThread(data: CreateThreadData) {
return retry(async () => threadClient.createThread(request, data), {
maxAttempts: 3,
delay: 1000,
factor: 2,
handleError: (error, context) => {
console.log(`Attempt ${context.attemptNum} failed:`, error);
},
});
}

3. Monitor Usage

// Track API usage
let apiCalls = 0;

const monitoredAdapter = new Proxy(adapter, {
get(target, prop) {
if (typeof target[prop] === 'function') {
return async (...args) => {
apiCalls++;
console.log(`API call: ${prop}, Total: ${apiCalls}`);
return target[prop](...args);
};
}
return target[prop];
},
});

Limitations

  • Search: Limited to basic text search in title and metadata
  • Filtering: No complex query operators like SQL
  • Bulk Operations: No native batch API
  • Transactions: No transaction support
  • Real-time: No WebSocket/subscription support

Next Steps