Exporting Archival Memories

1
#!/usr/bin/env python3
2
"""
3
Utility script to export all archival memories (passages) from a Letta agent.
4
5
Usage:
6
    python export_agent_memories.py <agent_id> [--output <file>] [--limit <limit>]
7
8
Example:
9
    python export_agent_memories.py agent-123e4567-e89b-42d3-8456-426614174000 --output memories.json
10
"""
11
12
import argparse
13
import json
14
import os
15
import sys
16
from typing import Any, Dict, List
17
18
from letta_client import Letta
19
20
21
def export_agent_memories(
22
    client: Letta,
23
    agent_id: str,
24
    page_limit: int = 100,
25
) -> List[Dict[str, Any]]:
26
    """
27
    Export all archival memories from an agent by paginating through all results.
28
29
    Args:
30
        client: Initialized Letta client
31
        agent_id: The agent ID in format 'agent-<uuid4>'
32
        page_limit: Number of results per page (default 100)
33
34
    Returns:
35
        List of passage dictionaries with embedding and embedding_config removed
36
    """
37
    all_passages = []
38
    after_cursor = None
39
    page_num = 1
40
41
    print(f"Exporting archival memories for agent: {agent_id}")
42
    print(f"Using pagination with limit: {page_limit}")
43
    print("-" * 60)
44
45
    while True:
46
        # Fetch next page
47
        print(f"Fetching page {page_num}...", end=" ", flush=True)
48
49
        try:
50
            passages = client.agents.passages.list(
51
                agent_id=agent_id,
52
                after=after_cursor,
53
                limit=page_limit,
54
                ascending=True  # Get oldest to newest
55
            )
56
        except Exception as e:
57
            print(f"\nError fetching memories: {e}")
58
            raise
59
60
        if not passages:
61
            print("(no more results)")
62
            break
63
64
        print(f"got {len(passages)} passages")
65
66
        # Convert to dict and remove embedding fields
67
        for passage in passages:
68
            passage_dict = passage.model_dump() if hasattr(passage, 'model_dump') else passage.dict()
69
            passage_dict.pop("embedding", None)
70
            passage_dict.pop("embedding_config", None)
71
            all_passages.append(passage_dict)
72
73
        # Check if we got fewer results than the limit (last page)
74
        if len(passages) < page_limit:
75
            break
76
77
        # Set cursor for next page (use the ID of the last passage)
78
        after_cursor = passages[-1].id if hasattr(passages[-1], 'id') else passages[-1]['id']
79
        page_num += 1
80
81
    print("-" * 60)
82
    print(f"Total passages exported: {len(all_passages)}")
83
84
    return all_passages
85
86
87
def main():
88
    parser = argparse.ArgumentParser(
89
        description="Export archival memories from a Letta agent"
90
    )
91
    parser.add_argument(
92
        "agent_id",
93
        help="Agent ID in format 'agent-<uuid4>'"
94
    )
95
    parser.add_argument(
96
        "--output",
97
        "-o",
98
        help="Output JSON file path (default: <agent_id>_memories.json)"
99
    )
100
    parser.add_argument(
101
        "--limit",
102
        "-l",
103
        type=int,
104
        default=100,
105
        help="Number of results per page (default: 100)"
106
    )
107
108
    args = parser.parse_args()
109
110
    # Check for API key
111
    api_key = os.getenv("LETTA_API_KEY")
112
    if not api_key:
113
        print("Error: LETTA_API_KEY environment variable not set", file=sys.stderr)
114
        print("Please export LETTA_API_KEY with your API key", file=sys.stderr)
115
        return 1
116
117
    # Determine output file
118
    output_file = args.output or f"{args.agent_id}_memories.json"
119
120
    try:
121
        # Initialize client
122
        client = Letta(token=api_key)
123
124
        # Export memories
125
        passages = export_agent_memories(
126
            client=client,
127
            agent_id=args.agent_id,
128
            page_limit=args.limit
129
        )
130
131
        # Write to file
132
        with open(output_file, "w") as f:
133
            json.dump(passages, f, indent=2, default=str)
134
135
        print(f"\nMemories exported successfully to: {output_file}")
136
        return 0
137
138
    except Exception as e:
139
        print(f"\nError: {e}", file=sys.stderr)
140
        return 1
141
142
143
if __name__ == "__main__":
144
    sys.exit(main())