Langgraph Supervisor and Agents (multi-agent) with memory/session

[KeenanFernandes2000]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

const supervisorChain = formattedPrompt
  .pipe(
    llm.bind({
      tools: [toolDef],
      tool_choice: { type: "function", function: { name: "route" } },
    })
  )
  .pipe(new JsonOutputToolsParser())
  // select the first one
  .pipe((x) => x[0].args);

interface AgentStateChannels {
  messages: {
    value: (x: BaseMessage[], y: BaseMessage[]) => BaseMessage[];
    default: () => BaseMessage[];
  };
  next: string;
}

// This defines the agent state
const agentStateChannels: AgentStateChannels = {
  messages: {
    value: (x: BaseMessage[], y: BaseMessage[]) => x.concat(y),
    default: () => [],
  },
  next: "initialValueForNext", // Replace 'initialValueForNext' with your initial value if needed
};

const researcherAgent = await createAgent(
  llm,
  [searchTool],
  "You are a web researcher. You may use the searchTool to search the web for important information, so the blogger in your team can create a blog."
);

const researcherNode = async (state: any, config: any) => {
  const result = await researcherAgent.invoke(state, config);
  return {
    messages: [
      new HumanMessage({ content: result.output, name: "researcher" }),
    ],
  };
};

const consolidatorAgent = await createAgent(
  llm,
  [placeholderTool],
  "You are consolidattion agent, using all the content provide from the blogger and image generator, consolidate everything into a final out that can be displayed with headings and paragraphs and images in HTML"
);

const consolidatorNode = async (state: any, config: any) => {
  const result = await consolidatorAgent.invoke(state, config);
  return {
    messages: [
      new HumanMessage({ content: result.output, name: "consolidator" }),
    ],
  };
};

const streamResults = graph.stream(
    {
      messages: [
        new HumanMessage({
          content: `Write a blog about ${topic}, and provide an image.`,
        }),
      ],
    },
    { recursionLimit: 100 }
  );

  for await (const output of await streamResults) {
    if (!output?.__end__) {
      console.log(output);
      console.log("----");
    }
  }

Description

I have used the example at https://github.com/langchain-ai/langgraphjs/blob/main/examples/multi_agent/agent_supervisor.ipynb, to create my own version of a supervisor and agent connection and this is running on an express server, however when simultaneous calls are made everything seems to break. The code provided is a small snippet of the whole thing.

Is there a way to make Langgraph session based?

System Info

node: v18.18.0
langchain@0.1.30
@langchain/langgraph@0.0.12
typescript@5.4.3

[dosubot[bot]]

To handle simultaneous calls and manage sessions effectively in your Langgraph setup running on an Express server, consider the following approach:

1. Isolate Agent Instances: For each request, create or use a separate agent instance to avoid state conflicts across concurrent requests.

2. Memory Management: Utilize GenerativeAgentMemory and GenerativeAgentMemoryChain for managing the memory of generative agents. These classes are designed for concurrent memory operations and can help in adding, reflecting, and generating insights based on the agent's experiences.

3. Context Management: Implement AsyncLocalStorage from Node.js to maintain a unique context for each request. This ensures isolated execution contexts for concurrent operations.

4. External Session Store with Redis: Use Redis as an external session store to manage sessions across multiple requests. This can be achieved by integrating BufferMemory and RedisChatMessageHistory from LangChain, which allows maintaining state across multiple agents and requests. Ensure to use the ioredis client for Redis connections, including support for Sentinel connections for high availability.

Here's a simplified code snippet demonstrating how to integrate these concepts into an Express server:

import express from 'express';
import { AsyncLocalStorage } from 'async_hooks';
import { Redis } from "ioredis";
import { BufferMemory, RedisChatMessageHistory } from "@langchain/community/stores/message/ioredis";
import { ConversationChain } from "langchain/chains";

const asyncLocalStorage = new AsyncLocalStorage();
const app = express();
const port = 3000;

const client = new Redis({
  // Configuration for Redis with Sentinel support
});

app.get('/query', async (req, res) => {
  asyncLocalStorage.run(new Map(), async () => {
    const memory = new BufferMemory({
      chatHistory: new RedisChatMessageHistory({
        sessionId: new Date().toISOString(), // Unique identifier for the conversation
        client,
      }),
    });

    const model = new ChatOpenAI({ /* Model configuration */ });
    const chain = new ConversationChain({ llm: model, memory });

    const input = req.query.input;
    try {
      const response = await chain.invoke({ input });
      res.json({ output: response.text });
    } catch (error) {
      res.status(500).send("Error processing your request");
    }
  });
});

app.listen(port, () => {
  console.log(`Server running at http://localhost:${port}`);
});

This setup ensures that each request is handled in an isolated context with its own memory management, leveraging Redis for session persistence across requests. Adjust the Redis configuration and model settings as needed for your specific use case.

Details

Sources

• examples/src/agents/concurrent_mrkl.ts
• langchain/src/experimental/generative_agents/generative_agent_memory.ts
• examples/src/memory/redis.ts
• examples/src/memory/redis-advanced.ts
• examples/src/memory/redis-sentinel.ts
• langchain-core/src/utils/async_caller.ts
• examples/src/agents/agent_cancellation.ts
• libs/langchain-google-common/src/utils/failed_handler.ts
• langchain-core/src/singletons/index.ts

About Dosu
This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

To continue the conversation, mention @dosu.

[KeenanFernandes2000]

@dosu

how would i apply sessions for the supervisor chain and for the models they are agents so how would you include sessions there as well

[dosubot]

To integrate sessions for both the supervisor chain and the agents in your Langgraph setup, especially when running on an Express server, you'll need to focus on context and session management. This involves using AsyncLocalStorage for context management across asynchronous operations and Redis for efficient, scalable session storage. Here's how you can approach this:

1. Context Management with AsyncLocalStorage: This Node.js module allows you to maintain a consistent context across asynchronous calls, which is crucial for tracking session information throughout the lifecycle of each request. By wrapping each request in its own context, you ensure that session data is isolated and managed correctly for each user interaction.

2. Session Storage Using Redis: Redis serves as a powerful in-memory data store that can efficiently manage session data. By integrating Redis into your application, you can store session identifiers and related data, ensuring quick access and persistence across requests. This is particularly useful for maintaining state in stateless HTTP transactions.

3. Middleware for Session Initialization and Retrieval: Implementing middleware in your Express application that checks for a session identifier (e.g., in request headers) at the start of each request is essential. This middleware should either initialize a new session or retrieve an existing session from Redis, using the session ID. The session data can then be stored in the context created by AsyncLocalStorage, making it accessible throughout the request's processing.

4. Applying Sessions to Supervisor Chain and Agents: For the supervisor chain and agents, the session context managed by AsyncLocalStorage and the session data stored in Redis should be leveraged to maintain state and manage interactions. This means that any stateful operation or decision-making process that depends on the user's session should access the session data from the context or Redis as needed.

Here's a simplified example to illustrate how you might set up session management in your Express server, integrating it with Langgraph's supervisor chain and agents:

const express = require('express');
const { AsyncLocalStorage } = require('async_hooks');
const Redis = require('ioredis');
const { AsyncLocalStorageProviderSingleton } = require('langchain-core/src/singletons');

const redisClient = new Redis(); // Configure as needed
const asyncLocalStorage = new AsyncLocalStorage();

// Initialize Langchain AsyncLocalStorage
AsyncLocalStorageProviderSingleton.initializeGlobalInstance(asyncLocalStorage);

const app = express();

// Middleware for session management
app.use(async (req, res, next) => {
  const sessionID = req.headers['x-session-id']; // Assuming session ID is in the header
  asyncLocalStorage.run(new Map(), async () => {
    if (sessionID) {
      const sessionData = await redisClient.get(sessionID);
      if (sessionData) {
        asyncLocalStorage.getStore().set('sessionData', JSON.parse(sessionData));
      }
    }
    next();
  });
});

// Example route using session data
app.get('/data', (req, res) => {
  const sessionData = asyncLocalStorage.getStore().get('sessionData');
  if (sessionData) {
    res.json({ data: 'some data', sessionData });
  } else {
    res.status(401).send('No session');
  }
});

app.listen(3000, () => {
  console.log('Server running on port 3000');
});

This setup ensures that each request is processed with its own session context, leveraging Redis for session persistence. Adapt this example based on your specific requirements for the supervisor chain and agents in your Langgraph application.

Details

About Dosu
This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

To continue the conversation, mention @dosu.

[KeenanFernandes2000]

@dosu

const supervisorChain = formattedPrompt
.pipe(
llm.bind({
tools: [toolDef],
tool_choice: { type: "function", function: { name: "route" } },
})
)
.pipe(new JsonOutputToolsParser())
// select the first one
.pipe((x) => x[0].args);

interface AgentStateChannels {
messages: {
value: (x: BaseMessage[], y: BaseMessage[]) => BaseMessage[];
default: () => BaseMessage[];
};
next: string;
}

how do i give this session memory

[dosubot]

To integrate session memory into your supervisor chain and agents setup, particularly for handling concurrent requests in an Express server environment, you'll need to focus on managing conversation history and context for each session. This involves utilizing BufferMemory from langchainjs to maintain a record of interactions within a session. Here's how you can achieve this:

1. Initialize BufferMemory for Session Management: For each session, you'll use BufferMemory to store and retrieve the conversation history. This ensures that each interaction within a session can be considered in the context of previous exchanges, allowing for more coherent and relevant responses from your agents.

2. Incorporate Session Memory into Supervisor Chain and Agents: When setting up your supervisor chain and agents, ensure that the conversation history managed by BufferMemory is accessible. This means that before processing a request, you should load the session's conversation history into the memory used by your chain and agents.

3. Session Management in Express Server: In your Express server setup, manage sessions by generating or retrieving a unique session identifier for each request. This identifier is crucial for associating the correct conversation history with each session. Use this identifier to load and update the conversation history in BufferMemory before and after processing requests.

4. Process Requests with Context: With the session's conversation history loaded into BufferMemory, process the incoming requests through your supervisor chain and agents. This setup ensures that the processing takes into account the context of the session, allowing for responses that are consistent and relevant to the ongoing conversation.

Here's a simplified code example to illustrate how you might set up session memory management in your Express server:

const express = require('express');
const { BufferMemory } = require('langchain/memory');
const app = express();
app.use(express.json());

// Assume initialization of your language model and conversation chain here

app.post('/chat', async (req, res) => {
  const userInput = req.body.input;
  const sessionId = req.body.sessionId; // Assume this is provided in the request

  // Initialize or retrieve BufferMemory for the session
  let sessionMemory = new BufferMemory(); // Simplified; in practice, you'd retrieve or initialize based on sessionId

  // Load session's conversation history into BufferMemory
  // This step would involve fetching the history from a persistent store if available

  // Process the request through your conversation chain with the session memory
  const response = await chain.invoke({ input: userInput, memory: sessionMemory });

  // Save the updated conversation history back to the persistent store

  res.json(response);
});

app.listen(3000, () => {
  console.log('Server running on port 3000');
});

This example demonstrates the basic flow of managing session memory for processing requests in a context-aware manner. Remember to adapt the memory management and session handling logic to fit the specifics of your application and the requirements of your Langgraph setup.

Details

About Dosu
This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

To continue the conversation, mention @dosu.

[KeenanFernandes2000]

UPDATE:
I have seen the updated documentation; it looks like the checkpointers and memory saver work in the latest version of LangGraph however it changes this current code a lot