zep-python

import asyncio import os from zep_python import ZepClient from zep_python.models import Message, Memory async def main(): # Ensure ZEP_API_URL is set in your environment or provide a default api_url = os.environ.get("ZEP_API_URL", "http://localhost:8000") api_key = os.environ.get("ZEP_API_KEY", "") # Zep API key if authentication is enabled client = ZepClient(api_url=api_url, api_key=api_key) session_id = "my-test-session-123" user_id = "test_user" # Optional try: # Sessions must be explicitly created before adding messages (Zep v2+) await client.add_session(session_id=session_id, user_id=user_id, metadata={"project": "chatbot"}) print(f"Session '{session_id}' created.") except Exception as e: # Session might already exist, or another error occurred if "already exists" in str(e): print(f"Session '{session_id}' already exists.") else: print(f"Error creating session '{session_id}': {e}") # Add messages to the session messages_to_add = [ Message(role="user", content="Hello, my name is Alice."), Message(role="assistant", content="Hi Alice, how can I help you today?") ] await client.add_transcript(session_id=session_id, messages=messages_to_add) print(f"Added {len(messages_to_add)} messages to session '{session_id}'.") # Retrieve memory for the session memory: Memory = await client.get_memory(session_id=session_id) if memory and memory.messages: print(f"\nRetrieved memory for session '{session_id}':") for msg in memory.messages: print(f" {msg.role}: {msg.content}") else: print(f"\nNo memory found or memory is empty for session '{session_id}'.") # Clean up: delete the session (uncomment to enable) # await client.delete_session(session_id=session_id) # print(f"Session '{session_id}' deleted.") # Close the client session await client.close() if __name__ == "__main__": asyncio.run(main())