DynamoDBChatMessageHistory does not correctly serialize json messages

[arash2060]

Checked other resources

• I added a very descriptive title to this issue.
• I searched the LangChain.js documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.
• I am sure that this is a bug in LangChain.js rather than my code.
• The bug is not resolved by updating to the latest stable version of LangChain (or the specific integration package).

Example Code

The following code:

// tests/dynamo-history.integration.test.ts // Renamed to indicate integration test

import {afterEach, beforeEach, describe, expect, it} from '@jest/globals';
import {AIMessage, BaseMessage, HumanMessage} from "@langchain/core/messages";
import {v4 as uuidv4} from "uuid";

// AWS SDK v3 imports for direct interaction in tests
import {DynamoDBClient} from "@aws-sdk/client-dynamodb";
import {DeleteCommand, DynamoDBDocumentClient, GetCommand} from "@aws-sdk/lib-dynamodb";
import {DynamoDBChatMessageHistory} from '@langchain/community/stores/message/dynamodb';

// --- Configuration ---
// Ensure these environment variables are set correctly for your test table/region
const TEST_TABLE_NAME = "TEST_TABLE"; // Use a dedicated test table name
const TEST_REGION = process.env.AWS_REGION || "us-east-1";

if (!process.env.DYNAMODB_HISTORY_TABLE_TEST) {
    console.warn(`TEST_TABLE env var not set. Using default: ${TEST_TABLE_NAME}. Ensure this table exists.`);
}

// --- Real AWS SDK Client for Test Setup/Verification/Cleanup ---
// Uses default credential provider chain
const realDdbClient = new DynamoDBClient({ region: TEST_REGION });
const realDocClient = DynamoDBDocumentClient.from(realDdbClient);

// Helper to create history instance pointing to the real table
const createHistory = (sessionId: string) => {
    return new DynamoDBChatMessageHistory({
        tableName: TEST_TABLE_NAME,
        sessionId: sessionId,
        partitionKey: 'pk',
        sortKey: 'sk',
        // Config for the *internal* client used by the history class
        config: {
            region: TEST_REGION,
        },
    });
};

// Define the sort key value the library likely uses or expects
// Adjust if needed based on library specifics or explicit configuration
const ACTUAL_SORT_KEY_VALUE = 'sk';

// Helper function to clean up item after test using the REAL client
const cleanupItem = async (sessionId: string) => {
    try {
        await realDocClient.send(new DeleteCommand({
            TableName: TEST_TABLE_NAME,
            Key: { pk: sessionId, sk: ACTUAL_SORT_KEY_VALUE },
        }));
        // console.log(`Cleaned up item for sessionId: ${sessionId}`);
    } catch (error) {
        console.error(`Error cleaning up sessionId ${sessionId}:`, error);
        // Don't fail test run due to cleanup error, but log it
    }
};

// --- Test Suite ---
describe("DynamoDBChatMessageHistory Integration Test", () => {
    let currentSessionId: string;

    beforeEach(() => {
        // Generate a unique sessionId for each test to ensure isolation
        currentSessionId = uuidv4();
    });

    afterEach(async () => {
        // Clean up the item created during the test
        if (currentSessionId) {
            await cleanupItem(currentSessionId);
        }
    });

    it("should instantiate correctly (no AWS call)", () => {
        const historyStore = createHistory(currentSessionId);
        expect(historyStore).toBeInstanceOf(DynamoDBChatMessageHistory);
    });

    it("should get empty messages if no history exists in DB", async () => {
        const historyStore = createHistory(currentSessionId);
        // Ensure item doesn't exist (handled by afterEach generally)
        const messages = await historyStore.getMessages();
        expect(messages).toEqual([]);
    });

    it("should add messages and store them in DynamoDB", async () => {
        const historyStore = createHistory(currentSessionId);
        const messagesToAdd: BaseMessage[] = [
            new HumanMessage({
                content: [
                    {
                        type: "text",
                        text: `Hi! I'm human` // Uses domString variable
                    }
                ]
            }),
            new AIMessage("Hi there! I'm real!"),
        ];

        // --- Act ---
        await historyStore.addMessages(messagesToAdd);

        // --- Assert ---
        // Verify by fetching directly from DynamoDB
        const { Item } = await realDocClient.send(new GetCommand({
            TableName: TEST_TABLE_NAME,
            Key: { pk: currentSessionId, sk: 'sk' },
        }));

        console.log(Item);

        expect(Item).toBeDefined();
        expect(Item?.pk).toBe(currentSessionId);
        expect(Item?.sk).toBe('sk');
        expect(Item?.messages).toBeDefined();
    });
});

Error Message and Stack Trace (if applicable)

Error: Error adding messages: UnknownError

at DynamoDBChatMessageHistory.addMessages (C:\Users\aaras\Dropbox\startup\website_augmenter\similar_extensions\utterXAmplifyBackend\node_modules\.pnpm\@langchain+community@0.3.40_c3b7e1eec298838f464006b49002acf9\node_modules\@langchain\community\dist\stores\message\dynamodb.cjs:210:19)
at processTicksAndRejections (node:internal/process/task_queues:105:5)
at Object.<anonymous> (C:\Users\aaras\Dropbox\startup\website_augmenter\similar_extensions\utterXAmplifyBackend\tests\dynamo-history-issue.test.ts:101:9)

Description

This works with HumanMessage("I'm human"), the issue is serialization.

System Info

langchain@0.3.21 | MIT | deps: 12 | versions: 316
Typescript bindings for langchain
https://github.com/langchain-ai/langchainjs/tree/main/langchain/

keywords: llm, ai, gpt3, chain, prompt, prompt engineering, chatgpt, machine learning, ml, openai, embeddings, vectorstores

dist
.tarball: https://registry.npmjs.org/langchain/-/langchain-0.3.21.tgz
.shasum: 5f41f9e11f7f087b1883cc579c2f8e11597d6e4f
.integrity: sha512-fpor/V/xJRoLM7s1yQGlUE6aZIe6LLm7ciZcstNbBUYdFpCSigvADsW2cqlBCdbMXJxb5I/z9jznjGnmciJ+aw==
.unpackedSize: 2.9 MB

dependencies:
@langchain/openai: >=0.1.0 <0.6.0 js-yaml: ^4.1.0 openapi-types: ^12.1.3 yaml: ^2.2.1
@langchain/textsplitters: >=0.0.0 <0.2.0 jsonpointer: ^5.0.1 p-retry: 4 zod-to-json-schema: ^3.22.3
js-tiktoken: ^1.0.12 langsmith: >=0.2.8 <0.4.0 uuid: ^10.0.0 zod: ^3.22.4

maintainers:

• nfcampos nuno@boringbits.io
• jacoblee93 jacoblee93@gmail.com
• andrewnguonly andrewnguonly@gmail.com
• benjamincburns benjamin.c.burns@gmail.com
• davidduong david@duong.cz
• hwchase17 hw.chase.17@gmail.com
• basproul braceasproul@gmail.com

dist-tags:
latest: 0.3.21 next: 0.3.2-rc.0 tag-for-publishing-older-releases: 0.2.20

[dosubot]

Hello @arash2060! I'm Dosu, a bot here to help you solve bugs, answer questions, and guide you on your journey to becoming a contributor. While you wait for a human maintainer, feel free to let me know how I can assist you!

To resolve the serialization issue when adding messages using the DynamoDBChatMessageHistory class, ensure that the StoredMessage is correctly transformed into a DynamoDBSerializedChatMessage using the createDynamoDBSerializedChatMessage method. This method converts the message into a format suitable for DynamoDB storage. Here's a code snippet that demonstrates how to transform a StoredMessage:

private createDynamoDBSerializedChatMessage(
  message: StoredMessage
): DynamoDBSerializedChatMessage {
  const {
    type,
    data: { content, role, additional_kwargs },
  } = message;

  const isAdditionalKwargs =
    additional_kwargs && Object.keys(additional_kwargs).length;

  const dynamoSerializedMessage: DynamoDBSerializedChatMessage = {
    M: {
      type: {
        S: type,
      },
      text: {
        S: content,
      },
      additional_kwargs: isAdditionalKwargs
        ? { S: JSON.stringify(additional_kwargs) }
        : { S: "{}" },
    },
  };

  if (role) {
    dynamoSerializedMessage.M.role = { S: role };
  }

  return dynamoSerializedMessage;
}

Ensure that the StoredMessage object has the correct structure and that all fields are properly serialized, especially additional_kwargs, which should be JSON-stringified if it contains any data. This will help avoid serialization issues when storing messages in DynamoDB ^[1].

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}

[arash2060]

I have a working solution which I don't love as I have no confidence in possible edge cases that I'm not capturing.

import {
    AttributeValue,
    DeleteItemCommand,
    DeleteItemCommandInput,
    DynamoDBClient,
    DynamoDBClientConfig,
    GetItemCommand,
    GetItemCommandInput,
    UpdateItemCommand,
    UpdateItemCommandInput,
} from "@aws-sdk/client-dynamodb";

import {BaseListChatMessageHistory} from "@langchain/core/chat_history";
import {
    AIMessage,
    BaseMessage,
    ChatMessage,
    FunctionMessage,
    HumanMessage,
    mapChatMessagesToStoredMessages,
    MessageContent,
    StoredMessage,
    SystemMessage,
    ToolMessage,
} from "@langchain/core/messages";
import type { ToolCall as LangChainToolCall } from "@langchain/core/messages/tool";
// Interface definitions (keep as originally provided in the library)
export interface DynamoDBChatMessageHistoryFields { /* ... */ }
// Interface describing the structure stored in DynamoDB List elements
interface DynamoDBStoredMessageMap {
    type: { S: string };
    text: { S: string }; // Represents the string or stringified content
    role?: { S: string };
    name?: { S: string };
    tool_call_id?: { S: string };
    additional_kwargs?: { S: string }; // Stringified JSON (potentially includes tool_calls)
}


/**
 * Interface defining the fields required to create an instance of
 * `DynamoDBChatMessageHistory`. It includes the DynamoDB table name,
 * session ID, partition key, sort key, message attribute name, and
 * DynamoDB client configuration.
 */
export interface DynamoDBChatMessageHistoryFields {
    tableName: string;
    sessionId: string;
    partitionKey?: string;
    sortKey?: string;
    messageAttributeName?: string;
    config?: DynamoDBClientConfig;
    key?: Record<string, AttributeValue>;
}

/**
 * Interface defining the structure of a chat message as it is stored in
 * DynamoDB.
 */
interface DynamoDBSerializedChatMessage {
    M: {
        type: {
            S: string;
        };
        text: {
            S: string;
        };
        role?: {
            S: string;
        };
        additional_kwargs?: {
            S: string;
        };
    };
}

/**
 * Class providing methods to interact with a DynamoDB table to store and
 * retrieve chat messages. It extends the `BaseListChatMessageHistory`
 * class.
 */
export class DynamoDBChatMessageHistory extends BaseListChatMessageHistory {
    lc_namespace = ["langchain", "stores", "message", "dynamodb"];

    get lc_secrets(): { [key: string]: string } | undefined {
        return {
            "config.credentials.accessKeyId": "AWS_ACCESS_KEY_ID",
            "config.credentials.secretAccessKey": "AWS_SECRETE_ACCESS_KEY",
            "config.credentials.sessionToken": "AWS_SESSION_TOKEN",
        };
    }

    private tableName: string;

    private sessionId: string;

    private client: DynamoDBClient;

    private partitionKey = "id";

    private sortKey?: string;

    private messageAttributeName = "messages";

    private dynamoKey: Record<string, AttributeValue> = {};

    /**
     * Transforms a `StoredMessage` into a `DynamoDBSerializedChatMessage`.
     * Handles string and array content by stringifying non-string content.
     *
     * @param message - The `StoredMessage` to be transformed.
     * @returns The transformed `DynamoDBSerializedChatMessage`.
     */
    private createDynamoDBSerializedChatMessage(
        message: StoredMessage
    ): DynamoDBSerializedChatMessage {
        const {
            type,
            data: { content, role, name, tool_call_id, additional_kwargs },
        } = message;

        const isAdditionalKwargs =
            additional_kwargs && Object.keys(additional_kwargs).length > 0;

        let contentString: string;
        if (typeof content === "string") {
            contentString = content;
        } else {
            try {
                contentString = JSON.stringify(content);
            } catch (e) {
                console.error("Failed to stringify message content:", content, e);
                contentString = "[FAILED TO SERIALIZE CONTENT]";
            }
        }

        const messageMap: DynamoDBStoredMessageMap = {
            type: { S: type },
            text: { S: contentString },
            additional_kwargs: { S: isAdditionalKwargs ? JSON.stringify(additional_kwargs) : "{}" },
        };

        if (role) { messageMap.role = { S: role }; }
        if (name) { messageMap.name = { S: name }; }
        if (tool_call_id) { messageMap.tool_call_id = { S: tool_call_id }; }

        return { M: messageMap };
    }

    constructor({
                    tableName,
                    sessionId,
                    partitionKey,
                    sortKey,
                    messageAttributeName,
                    config,
                    key = {},
                }: DynamoDBChatMessageHistoryFields) {
        super();

        this.tableName = tableName;
        this.sessionId = sessionId;
        this.client = new DynamoDBClient(config ?? {});
        this.partitionKey = partitionKey ?? this.partitionKey;
        this.sortKey = sortKey;
        this.messageAttributeName =
            messageAttributeName ?? this.messageAttributeName;
        this.dynamoKey = key;

        // override dynamoKey with partition key and sort key when key not specified
        if (Object.keys(this.dynamoKey).length === 0) {
            this.dynamoKey[this.partitionKey] = {S: this.sessionId};
            if (this.sortKey) {
                this.dynamoKey[this.sortKey] = {S: this.sortKey};
            }
        }
    }

    /**
     * Retrieves messages, parses potentially stringified content,
     * transforms raw tool calls from kwargs into LangChain ToolCall objects,
     * and inserts corresponding ToolMessages after AIMessages with tool calls.
     */
    async getMessages(): Promise<BaseMessage[]> {
        try {
            const params: GetItemCommandInput = {
                TableName: this.tableName,
                Key: this.dynamoKey,
            };
            const response = await this.client.send(new GetItemCommand(params));
            const items = response.Item
                ? response.Item[this.messageAttributeName]?.L ?? []
                : [];

            // Step 1: Deserialize messages from DynamoDB format first,
            //         including transformation of raw tool calls to LangChain ToolCall[]
            const deserializedMessages: BaseMessage[] = items
                .filter(
                    (item): item is AttributeValue & { M: DynamoDBStoredMessageMap } =>
                        item.M?.type?.S !== undefined
                )
                .map((item): BaseMessage | null => {
                    const messageMap = item.M;
                    const type = messageMap.type.S as string;
                    const contentStr = messageMap.text?.S;
                    const role = messageMap.role?.S;
                    const name = messageMap.name?.S;
                    const tool_call_id = messageMap.tool_call_id?.S;
                    const additional_kwargs_str = messageMap.additional_kwargs?.S;

                    if (contentStr === undefined) return null;

                    // Parse content (handles string or stringified array)
                    let finalContent: MessageContent;
                    try { finalContent = JSON.parse(contentStr); }
                    catch (e) { finalContent = contentStr; }

                    // Parse additional_kwargs and TRANSFORM raw tool calls
                    let finalKwargs: Record<string, unknown> = {};
                    let parsedLangChainToolCalls: LangChainToolCall[] | undefined;

                    if (additional_kwargs_str) {
                        try {
                            finalKwargs = JSON.parse(additional_kwargs_str);
                            const rawToolCalls = finalKwargs?.tool_calls; // Get raw array

                            if (rawToolCalls && Array.isArray(rawToolCalls)) {
                                // Transform raw structures into LangChain ToolCall objects
                                parsedLangChainToolCalls = rawToolCalls.map((rawCall: any): LangChainToolCall | null => {
                                    // Validate needed parts of the raw OpenAI-like structure
                                    if (rawCall.type !== 'function' || !rawCall.function || typeof rawCall.function.name !== 'string' || typeof rawCall.function.arguments !== 'string' || !rawCall.id) {
                                        console.warn("Invalid raw tool call structure found in history, skipping transformation:", rawCall);
                                        return null;
                                    }
                                    let parsedArgs: Record<string, any> = {};
                                    try {
                                        parsedArgs = JSON.parse(rawCall.function.arguments);
                                    } catch (parseError) {
                                        console.error(`Failed to parse tool call arguments string for id ${rawCall.id}:`, rawCall.function.arguments, parseError);
                                        // Store error info in args instead of failing entirely
                                        parsedArgs = { _raw_arguments: rawCall.function.arguments, _parse_error: String(parseError) };
                                    }
                                    // Create object matching LangChain's ToolCall type definition
                                    return {
                                        name: rawCall.function.name,
                                        args: parsedArgs, // The parsed arguments object
                                        id: rawCall.id,
                                        type: "tool_call" // Add standard type property
                                    };
                                }).filter((call): call is LangChainToolCall => call !== null);

                                // Remove raw tool_calls from finalKwargs now that they've been processed
                                delete finalKwargs.tool_calls;
                            }
                        } catch (e) {
                            console.warn(`Failed to parse additional_kwargs JSON: ${additional_kwargs_str}`, e);
                            finalKwargs = {}; // Default to empty if parsing fails
                        }
                    }

                    // Construct specific message types
                    switch (type) {
                        case "human":
                            return new HumanMessage({ content: finalContent, additional_kwargs: finalKwargs });
                        case "ai":
                            return new AIMessage({
                                content: finalContent,
                                tool_calls: parsedLangChainToolCalls, // Assign the TRANSFORMED ToolCalls
                                additional_kwargs: finalKwargs,
                            });
                        case "system":
                            return new SystemMessage({ content: finalContent, additional_kwargs: finalKwargs });
                        case "chat":
                            if (!role) return null;
                            return new ChatMessage({ content: finalContent, role: role, additional_kwargs: finalKwargs });
                        case "function":
                            if (!name || typeof finalContent !== "string") return null;
                            return new FunctionMessage({ content: finalContent, name: name, additional_kwargs: finalKwargs });
                        case "tool":
                            if (!tool_call_id || typeof finalContent !== "string") return null;
                            return new ToolMessage({ content: finalContent, tool_call_id: tool_call_id, additional_kwargs: finalKwargs });
                        default: return null;
                    }
                })
                .filter((msg): msg is BaseMessage => msg !== null);

            // Step 2: Build the final list, interleaving ToolMessages based on the AIMessage's ToolCalls
            const finalMessagesWithToolResults: BaseMessage[] = [];
            for (const message of deserializedMessages) {
                finalMessagesWithToolResults.push(message); // Add the original message

                // Check if it's an AIMessage which now has the correctly typed 'tool_calls' property
                if (message instanceof AIMessage && message.tool_calls && message.tool_calls.length > 0) {
                    // Add a ToolMessage response for each call
                    for (const toolCall of message.tool_calls) { // toolCall is type LangChainToolCall
                        if (!toolCall.id) {
                            console.warn("Skipping tool response generation: ToolCall missing id", toolCall);
                            continue;
                        }

                        // Re-stringify the parsed 'args' object to create the ToolMessage content
                        let toolContent = "";
                        try {
                            // toolCall.args should be the parsed object now
                            toolContent = JSON.stringify(toolCall.args);
                        } catch (e) {
                            console.error(`Failed to re-stringify ToolCall args for id ${toolCall.id}:`, toolCall.args, e);
                            toolContent = `{"error": "failed to stringify arguments during history processing for id ${toolCall.id}"}`;
                        }

                        finalMessagesWithToolResults.push(new ToolMessage({
                            content: toolContent,
                            tool_call_id: toolCall.id,
                        }));
                    }
                }
            }

            return finalMessagesWithToolResults; // Return the list with interleaved ToolMessages

        } catch (error: any) {
            console.error("Error details getting messages:", error);
            throw new Error(`Error getting messages: ${error.message}`);
        }
    }

    /**
     * Adds a new message to the DynamoDB table.
     * @param message The message to be added to the DynamoDB table.
     */
    async addMessage(message: BaseMessage) {
        try {
            const messages = mapChatMessagesToStoredMessages([message]);

            const params: UpdateItemCommandInput = {
                TableName: this.tableName,
                Key: this.dynamoKey,
                ExpressionAttributeNames: {
                    "#m": this.messageAttributeName,
                },
                ExpressionAttributeValues: {
                    ":empty_list": {
                        L: [],
                    },
                    ":m": {
                        L: messages.map(this.createDynamoDBSerializedChatMessage),
                    },
                },
                UpdateExpression:
                    "SET #m = list_append(if_not_exists(#m, :empty_list), :m)",
            };
            await this.client.send(new UpdateItemCommand(params));
            // eslint-disable-next-line @typescript-eslint/no-explicit-any
        } catch (error: any) {
            throw new Error(`Error adding message: ${error.message}`);
        }
    }

    /**
     * Adds new messages to the DynamoDB table.
     * @param messages The messages to be added to the DynamoDB table.
     */
    async addMessages(messages: BaseMessage[]): Promise<void> {
        try {
            const storedMessages = mapChatMessagesToStoredMessages(messages);
            const dynamoMessages = storedMessages.map(
                this.createDynamoDBSerializedChatMessage
            );

            const params: UpdateItemCommandInput = {
                TableName: this.tableName,
                Key: this.dynamoKey,
                ExpressionAttributeNames: {
                    "#m": this.messageAttributeName,
                },
                ExpressionAttributeValues: {
                    ":empty_list": {
                        L: [],
                    },
                    ":m": {
                        L: dynamoMessages,
                    },
                },
                UpdateExpression:
                    "SET #m = list_append(if_not_exists(#m, :empty_list), :m)",
            };
            await this.client.send(new UpdateItemCommand(params));
            // eslint-disable-next-line @typescript-eslint/no-explicit-any
        } catch (error: any) {
            throw new Error(`Error adding messages: ${error.message}`);
        }
    }

    /**
     * Deletes all messages from the DynamoDB table.
     */
    async clear(): Promise<void> {
        try {
            const params: DeleteItemCommandInput = {
                TableName: this.tableName,
                Key: this.dynamoKey,
            };
            await this.client.send(new DeleteItemCommand(params));
            // eslint-disable-next-line @typescript-eslint/no-explicit-any
        } catch (error: any) {
            throw new Error(`Error clearing messages: ${error.message}`);
        }
    }
}