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​