Message Exporter

Message Exporter#

The MessageExporter utility extracts conversation messages from Redis checkpoints into structured dictionaries. This is useful for analytics, debugging, auditing, and exporting conversation data to external systems.

Basic Usage#

Create a MessageExporter with a checkpoint saver, then export messages from a thread:

from langgraph.checkpoint.redis import RedisSaver
from langgraph.checkpoint.redis.message_exporter import MessageExporter

with RedisSaver.from_conn_string("redis://localhost:6379") as saver:
    saver.setup()

    # Run your graph to produce checkpoints...
    graph = builder.compile(checkpointer=saver)
    config = {"configurable": {"thread_id": "my-thread"}}
    graph.invoke({"messages": [("human", "What is Redis?")]}, config)

    # Export messages from the latest checkpoint
    exporter = MessageExporter(saver)
    messages = exporter.export("my-thread")

    for msg in messages:
        print(f"{msg['role']}: {msg['content']}")

Each message is a dictionary with at least role and content keys:

{
    "role": "human",
    "content": "What is Redis?",
    "type": "HumanMessage",
    "id": "msg-123",
    "metadata": {
        "name": None,
        "tool_calls": None,
        "additional_kwargs": {},
    },
}

Exporting a Specific Checkpoint#

Pass a checkpoint_id to export messages from a specific point in time rather than the latest:

messages = exporter.export("my-thread", checkpoint_id="1ef4f797-8335-6428-8001-8a1503f9b875")

Exporting an Entire Thread#

export_thread iterates over all checkpoints in a thread, deduplicates messages by ID, and annotates each message with checkpoint metadata:

result = exporter.export_thread("my-thread")

print(f"Thread: {result['thread_id']}")
print(f"Exported at: {result['export_timestamp']}")

for msg in result["messages"]:
    print(f"[{msg['checkpoint_ts']}] {msg['role']}: {msg['content']}")

The returned dictionary has this structure:

{
    "thread_id": "my-thread",
    "messages": [
        {
            "role": "human",
            "content": "What is Redis?",
            "type": "HumanMessage",
            "id": "msg-123",
            "checkpoint_id": "1ef4f797-...",
            "checkpoint_ts": "2024-07-31T20:14:19.804150+00:00",
            "metadata": {...},
        },
        # ...
    ],
    "export_timestamp": "2024-08-01T10:30:00.000000+00:00",
}

Messages are deduplicated by their id field, so the same message appearing in multiple checkpoints is only included once.

Recipes#

The exporter uses a recipe to convert raw message objects into structured dictionaries. The default LangChainRecipe handles:

LangChain message objects (HumanMessage, AIMessage, ToolMessage, etc.)
Serialized LangChain format (dicts with lc and type keys)
Simple dict messages (dicts with role and content keys)
Plain strings (mapped to role: "unknown")

Custom Recipes#

Implement the MessageRecipe protocol to support custom message formats:

from typing import Any, Dict, Optional
from langgraph.checkpoint.redis.message_exporter import MessageExporter, MessageRecipe


class MyCustomRecipe:
    """Extract messages from a custom format."""

    def extract(self, message: Any) -> Optional[Dict[str, Any]]:
        if isinstance(message, dict) and "sender" in message:
            return {
                "role": message["sender"],
                "content": message.get("text", ""),
                "timestamp": message.get("ts"),
            }
        return None


exporter = MessageExporter(saver, recipe=MyCustomRecipe())
messages = exporter.export("my-thread")

The extract method should return a dictionary with at least role and content keys, or None if the message format is not recognized.

Example: Export to JSON#

import json
from langgraph.checkpoint.redis import RedisSaver
from langgraph.checkpoint.redis.message_exporter import MessageExporter

with RedisSaver.from_conn_string("redis://localhost:6379") as saver:
    saver.setup()

    exporter = MessageExporter(saver)
    result = exporter.export_thread("my-thread")

    with open("conversation.json", "w") as f:
        json.dump(result, f, indent=2)

API Summary#

Method	Description
`MessageExporter(saver, recipe=None)`	Create an exporter with a checkpoint saver and optional recipe
`export(thread_id, checkpoint_id=None)`	Export messages from a single checkpoint (latest if no ID given)
`export_thread(thread_id)`	Export all messages across all checkpoints in a thread, deduplicated

Class	Description
`LangChainRecipe`	Default recipe that handles LangChain message objects and serialized formats
`MessageRecipe`	Protocol for custom recipes — implement `extract(message) -> dict or None`

Version