my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

my-project-public — gitlawb

#1	"""
#2	Zep memory provider importer.
#3
#4	Zep is a cloud-hosted enterprise memory platform with a temporal
#5	knowledge graph (Neo4j-based). Extraction requires iterating:
#6	users → sessions/threads → memory.get() per session.
#7
#8	Zep has NO bulk export API. Data must be extracted per-session.
#9
#10	Extraction method:
#11	1. List all users via client.user.list_ordered()
#12	2. For each user: list sessions
#13	3. For each session: client.memory.get(session_id) → messages + facts + summary
#14	"""
#15
#16	import json
#17	from datetime import datetime
#18	from typing import List, Dict, Optional, Any
#19
#20	from mnemosyne.core.importers.base import BaseImporter, ImporterResult
#21
#22
#23	class ZepImporter(BaseImporter):
#24	"""Import memories from Zep into Mnemosyne.
#25
#26	Usage:
#27	importer = ZepImporter(
#28	api_key="sk-xxx", # Zep API key
#29	user_id="alice", # optional: filter by user
#30	base_url=None, # for self-hosted (rare)
#31	max_sessions=None, # limit sessions to extract
#32	)
#33	result = importer.run(mnemosyne_instance)
#34	"""
#35
#36	provider_name = "zep"
#37
#38	def __init__(self, api_key: str = None, user_id: str = None,
#39	base_url: str = None, max_sessions: int = None,
#40	**kwargs):
#41	super().__init__(**kwargs)
#42	self.api_key = api_key
#43	self.user_id = user_id
#44	self.base_url = base_url
#45	self.max_sessions = max_sessions
#46
#47	def extract(self) -> List[Dict]:
#48	"""Extract memories from Zep via session-by-session iteration."""
#49	try:
#50	return self._extract_via_sdk()
#51	except ImportError:
#52	pass
#53	try:
#54	return self._extract_via_rest()
#55	except Exception:
#56	pass
#57	raise RuntimeError(
#58	"Could not extract from Zep. Install the SDK: pip install zep-cloud"
#59	)
#60
#61	def _extract_via_sdk(self) -> List[Dict]:
#62	"""Extract using Zep Python SDK."""
#63	from zep_cloud.client import Zep
#64	client = Zep(api_key=self.api_key)
#65
#66	all_memories = []
#67
#68	# Step 1: List users
#69	users = []
#70	if self.user_id:
#71	users = [self.user_id]
#72	else:
#73	page = 1
#74	while True:
#75	resp = client.user.list_ordered(page_size=50, page_number=page)
#76	batch = resp.get("users", resp.get("results", []))
#77	if not batch:
#78	break
#79	users.extend(u.get("user_id", u.get("id", "")) for u in batch)
#80	if resp.get("next") is None:
#81	break
#82	page += 1
#83
#84	# Step 2: For each user, list sessions and extract
#85	session_count = 0
#86	for uid in users:
#87	if self.max_sessions and session_count >= self.max_sessions:
#88	break
#89
#90	try:
#91	sessions = client.user.get_sessions(uid)
#92	except Exception:
#93	sessions = []
#94
#95	for session in sessions:
#96	if self.max_sessions and session_count >= self.max_sessions:
#97	break
#98
#99	sid = session.get("session_id", session.get("uuid", ""))
#100	if not sid:
#101	continue
#102
#103	try:
#104	mem = client.memory.get(sid)
#105	all_memories.extend(
#106	self._parse_session_data(mem, uid, sid)
#107	)
#108	session_count += 1
#109	except Exception:
#110	continue
#111
#112	return all_memories
#113
#114	def _extract_via_rest(self) -> List[Dict]:
#115	"""Extract using Zep REST API directly."""
#116	import urllib.request
#117
#118	base = self.base_url or "https://api.getzep.com"
#119	headers = {"Authorization": f"Bearer {self.api_key}"}
#120
#121	all_memories = []
#122
#123	# List users
#124	users = []
#125	if self.user_id:
#126	users = [self.user_id]
#127	else:
#128	req = urllib.request.Request(
#129	f"{base}/api/v2/users?page_size=50",
#130	headers=headers,
#131	)
#132	with urllib.request.urlopen(req, timeout=30) as resp:
#133	data = json.loads(resp.read().decode())
#134	users = [u["user_id"] for u in data.get("users", [])]
#135
#136	for uid in users:
#137	req = urllib.request.Request(
#138	f"{base}/api/v2/users/{uid}/sessions",
#139	headers=headers,
#140	)
#141	with urllib.request.urlopen(req, timeout=30) as resp:
#142	sessions = json.loads(resp.read().decode())
#143
#144	for session in sessions:
#145	sid = session.get("session_id", "")
#146	if not sid:
#147	continue
#148	req = urllib.request.Request(
#149	f"{base}/api/v2/sessions/{sid}/memory",
#150	headers=headers,
#151	)
#152	with urllib.request.urlopen(req, timeout=30) as resp:
#153	mem = json.loads(resp.read().decode())
#154	all_memories.extend(
#155	self._parse_session_data(mem, uid, sid)
#156	)
#157
#158	return all_memories
#159
#160	def _parse_session_data(self, mem: dict, user_id: str,
#161	session_id: str) -> List[Dict]:
#162	"""Parse a single Zep session's memory data."""
#163	items = []
#164
#165	# 1. Messages
#166	messages = mem.get("messages", [])
#167	for msg in messages:
#168	content = msg.get("content", msg.get("text", ""))
#169	if not content:
#170	continue
#171	role = msg.get("role_type", msg.get("role", "user"))
#172	ts = msg.get("created_at")
#173	items.append({
#174	"content": content,
#175	"source": "zep_message",
#176	"user_id": user_id,
#177	"session_id": session_id,
#178	"role": role,
#179	"timestamp": ts,
#180	"metadata": msg.get("metadata", {}),
#181	})
#182
#183	# 2. Summary
#184	summary = mem.get("summary", "")
#185	if summary:
#186	items.append({
#187	"content": summary,
#188	"source": "zep_summary",
#189	"user_id": user_id,
#190	"session_id": session_id,
#191	"role": "system",
#192	"timestamp": None,
#193	"metadata": {"zep_type": "summary"},
#194	})
#195
#196	# 3. Relevant facts (edges)
#197	facts = mem.get("relevant_facts", mem.get("facts", []))
#198	for fact in facts:
#199	fact_text = fact.get("fact", fact.get("text", ""))
#200	if not fact_text:
#201	continue
#202	items.append({
#203	"content": fact_text,
#204	"source": "zep_fact",
#205	"user_id": user_id,
#206	"session_id": session_id,
#207	"role": "system",
#208	"timestamp": fact.get("created_at"),
#209	"metadata": {
#210	"zep_type": "fact",
#211	"zep_rating": fact.get("rating"),
#212	},
#213	})
#214
#215	return items
#216
#217	def transform(self, raw_data: List[Dict]) -> List[Dict]:
#218	"""Transform Zep data to Mnemosyne format."""
#219	memories = []
#220	for item in raw_data:
#221	content = item.get("content", "")
#222	if not content:
#223	continue
#224
#225	# Author: use Zep user_id
#226	uid = item.get("user_id", "")
#227	author_id = f"zep_user:{uid}" if uid else None
#228
#229	# Importance heuristic
#230	source = item.get("source", "zep_import")
#231	if source == "zep_summary":
#232	importance = 0.8
#233	elif source == "zep_fact":
#234	importance = 0.7
#235	else:
#236	importance = 0.4
#237
#238	# Build metadata
#239	meta = item.get("metadata", {}) or {}
#240	meta["_zep_session_id"] = item.get("session_id", "")
#241	meta["_zep_role"] = item.get("role", "")
#242
#243	ts = item.get("timestamp")
#244	if ts:
#245	meta["_timestamp"] = ts
#246
#247	memories.append({
#248	"content": content,
#249	"source": source,
#250	"importance": importance,
#251	"metadata": meta,
#252	"valid_until": None,
#253	"scope": "session",
#254	"_author_id": author_id,
#255	"_author_type": "human" if uid else "system",
#256	"_channel_id": uid, # user_id as channel for grouping
#257	"_timestamp": ts,
#258	})
#259
#260	return memories
#261
#262	def run(self, mnemosyne, dry_run=False, session_id=None, channel_id=None):
#263	"""Override run to handle identity-aware import."""
#264	result = ImporterResult(provider=self.provider_name,
#265	started_at=datetime.now().isoformat())
#266
#267	try:
#268	raw_data = self.extract()
#269	result.total = len(raw_data)
#270	if result.total == 0:
#271	result.errors.append("No memories found in Zep")
#272	return result
#273	if not self.validate(raw_data):
#274	result.errors.append("Validation failed")
#275	return result
#276
#277	memories = self.transform(raw_data)
#278	if dry_run:
#279	result.imported = len(memories)
#280	return result
#281
#282	for mem_dict in memories:
#283	try:
#284	author_id = mem_dict.pop("_author_id", None)
#285	author_type = mem_dict.pop("_author_type", None)
#286	chan = mem_dict.pop("_channel_id", None) or channel_id
#287	ts = mem_dict.pop("_timestamp", None)
#288	meta = mem_dict.get("metadata", {})
#289	if ts:
#290	meta["imported_at_original"] = ts
#291
#292	mid = mnemosyne.remember(
#293	content=mem_dict["content"],
#294	source=mem_dict.get("source", self.provider_name),
#295	importance=mem_dict.get("importance", 0.5),
#296	metadata=meta,
#297	valid_until=mem_dict.get("valid_until"),
#298	scope=mem_dict.get("scope", "session"),
#299	)
#300	if author_id or author_type or chan:
#301	try:
#302	mnemosyne.beam.conn.execute("""
#303	UPDATE working_memory
#304	SET author_id = COALESCE(author_id, ?),
#305	author_type = COALESCE(author_type, ?),
#306	channel_id = COALESCE(channel_id, ?)
#307	WHERE id = ?
#308	""", (author_id, author_type, chan, mid))
#309	mnemosyne.beam.conn.commit()
#310	except Exception:
#311	pass
#312	result.memory_ids.append(mid)
#313	result.imported += 1
#314	except Exception as e:
#315	result.failed += 1
#316	result.errors.append(f"Failed: {str(e)[:100]}")
#317
#318	except Exception as e:
#319	result.errors.append(f"Zep import failed: {e}")
#320
#321	result.finished_at = datetime.now().isoformat()
#322	return result
#323

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public