### Example Shell Setup for Environment Variables Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/configuration.md Provides an example of how to set environment variables in a shell or CI/CD pipeline for use with the MemU SDK. These variables can then be accessed in your Python application. ```bash # In your shell or CI/CD pipeline export MEMU_API_KEY="sk_live_abc123def456" export MEMU_BASE_URL="https://api.memu.so" export MEMU_TIMEOUT="60" export MEMU_MAX_RETRIES="3" # Then in Python python your_app.py ``` -------------------------------- ### Install MemU Python SDK Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/quick-start.md Clone the repository, navigate to the directory, and install the SDK in editable mode. Requires Python 3.10+. ```bash git clone https://github.com/NevaMind-AI/memU-sdk-py.git cd memU-sdk-py pip install -e . ``` -------------------------------- ### Install MemU SDK with Development Dependencies Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/README.md Installs the MemU SDK in editable mode with development dependencies included. ```bash # Install with development dependencies pip install -e ".[dev]" ``` -------------------------------- ### Minimal Sync Example Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/quick-start.md Initialize the synchronous MemU client, memorize a conversation, and retrieve memories. Ensure to close the client when done. ```python from memu_sdk import MemUClient # Initialize client client = MemUClient(api_key="sk_live_YOUR_KEY") try: # Memorize result = client.memorize_sync( conversation_text="User: I like pasta\nAssistant: Noted!", user_id="user_123", agent_id="agent_456" ) # Retrieve memories = client.retrieve_sync( query="User preferences", user_id="user_123", agent_id="agent_456" ) print(f"Found {len(memories.items)} memories") finally: client.close_sync() ``` -------------------------------- ### Minimal Async Example Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/quick-start.md Initialize the async MemU client, memorize a conversation, and retrieve memories. Uses a context manager for client lifecycle. ```python import asyncio from memu_sdk import MemUClient async def main(): # Initialize client async with MemUClient(api_key="sk_live_YOUR_KEY") as client: # Memorize a conversation result = await client.memorize( conversation_text="User: I like Italian food\nAssistant: That's great!", user_id="user_123", agent_id="agent_456" ) print(f"Task: {result.task_id}") # Retrieve memories memories = await client.retrieve( query="What food does the user like?", user_id="user_123", agent_id="agent_456" ) print(f"Found {len(memories.items)} items") asyncio.run(main()) ``` -------------------------------- ### Get Task Status Example Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/api-reference/models.md Demonstrates how to retrieve the status of an asynchronous task using the client and how to interpret the returned TaskStatus object. ```python from memu_sdk.models import TaskStatusEnum status = await client.get_task_status("task_abc123") print(f"Task: {status.task_id}") print(f"Status: {status.status}") if status.progress is not None: print(f"Progress: {status.progress}%") if status.status == TaskStatusEnum.SUCCESS and status.result: items = status.result.get("items", []) print(f"Result: {len(items)} items extracted") elif status.status == TaskStatusEnum.FAILED: print(f"Error: {status.message}") # Check timestamps if status.updated_at: print(f"Last updated: {status.updated_at}") ``` -------------------------------- ### Memorize Operation Examples Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/types.md Illustrates how to use the memorize function asynchronously with and without waiting for completion, and demonstrates synchronous usage. ```python # Async usage result = await client.memorize( conversation_text="...", user_id="user_123", agent_id="agent_456", wait_for_completion=False ) # Task submitted asynchronously print(f"Task ID: {result.task_id}") # Async with waiting result = await client.memorize( conversation_text="...", user_id="user_123", agent_id="agent_456", wait_for_completion=True ) # Results are populated print(f"Items: {len(result.items)}") print(f"Categories: {len(result.categories)}") # Sync usage result = client.memorize_sync( conversation_text="...", user_id="user_123", agent_id="agent_456", wait_for_completion=True ) ``` -------------------------------- ### List and Retrieve Categories Example Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/api-reference/models.md Demonstrates how to list all categories for a user and how to retrieve categories with relevance scores. The score field is only populated in retrieve results. ```python # List all categories categories = await client.list_categories(user_id="user_123") for cat in categories: print(f"Category: {cat.name}") print(f" Items: {cat.item_count}") print(f" Summary: {cat.summary}") # In retrieve results with scoring result = await client.retrieve( query="interests", user_id="user_123", agent_id="agent_456" ) for cat in result.categories: print(f"{cat.name} (relevance: {cat.score})") # Access category content full_content = categories[0].content # Aggregated content string ``` -------------------------------- ### MemU SDK Error Handling Example Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/README.md Demonstrates how to handle specific MemU SDK exceptions for authentication, rate limiting, validation, and general API errors. ```python from memu_sdk.client import ( MemUClientError, MemUAuthenticationError, MemURateLimitError, MemUNotFoundError, MemUValidationError, ) try: result = await client.memorize( conversation_text="Hello", user_id="user_123", agent_id="agent_456" ) except MemUAuthenticationError: print("Invalid API key") except MemURateLimitError as e: print(f"Rate limited. Retry after {e.retry_after} seconds") except MemUValidationError as e: print(f"Validation error: {e.response}") except MemUClientError as e: print(f"API error: {e.message}") ``` -------------------------------- ### Synchronous Usage Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/api-reference/memory-operations.md Provides examples of synchronous memory operations including memorizing, retrieving, and listing categories. ```APIDOC ## Synchronous Memory Operations ### Description Demonstrates how to perform memory operations synchronously using the MemUClient. ### Methods * **`memorize_sync`**: Memorizes a given conversation. * **`retrieve_sync`**: Retrieves memories based on a query. * **`list_categories_sync`**: Synchronously lists categories for a user. ### Parameters for `memorize_sync` * **conversation_text** (string) - Required - The conversation to memorize. * **user_id** (string) - Required - The ID of the user. * **agent_id** (string) - Required - The ID of the agent. * **wait_for_completion** (boolean) - Optional - Whether to wait for the operation to complete. ### Parameters for `retrieve_sync` * **query** (string) - Required - The query to search for memories. * **user_id** (string) - Required - The ID of the user. * **agent_id** (string) - Required - The ID of the agent. ### Parameters for `list_categories_sync` * **user_id** (string) - Required - The ID of the user. ### Request Example ```python from memu_sdk import MemUClient client = MemUClient(api_key="sk_live_...") try: # Memorize result = client.memorize_sync( conversation_text="User: I prefer coffee\nAssistant: Good to know!", user_id="user_123", agent_id="agent_456", wait_for_completion=True ) print(f"Memorized {len(result.items)} items") # Retrieve memories = client.retrieve_sync( query="What does the user like to drink?", user_id="user_123", agent_id="agent_456" ) print(f"Found {len(memories.items)} memories") # List categories categories = client.list_categories_sync(user_id="user_123") print(f"User has {len(categories)} categories") finally: client.close_sync() ``` ``` -------------------------------- ### Example Usage of RetrieveResult Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/types.md Demonstrates how to use the retrieve() method and access the fields of the returned RetrieveResult object, including categories, items, and suggested follow-up queries. ```python result = await client.retrieve( query="What are the user's food preferences?", user_id="user_123", agent_id="agent_456" ) # Access ranked results print(f"Categories: {len(result.categories)}") for cat in result.categories: print(f" {cat.name} (score: {cat.score})") print(f"Items: {len(result.items)}") for item in result.items: print(f" [{item.memory_type}] {item.summary} (score: {item.score})") # Follow-up suggestion if result.next_step_query: print(f"Suggested next query: {result.next_step_query}") # Original resources for res in result.resources: print(f" {res.modality}: {res.caption}") ``` -------------------------------- ### TaskStatus Result Structure Example Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/api-reference/models.md Illustrates the expected structure of the 'result' field within a TaskStatus object when a task is successfully completed. ```python { "items": [MemoryItem, ...], "categories": [MemoryCategory, ...], "resource": MemoryResource, } ``` -------------------------------- ### User and Agent Scoping Examples Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/quick-start.md Demonstrates how memories are scoped to specific user-agent pairs. Operations with different `user_id` or `agent_id` will access separate memory stores. ```python # User 123 with Agent A await client.memorize(..., user_id="user_123", agent_id="agent_A") # User 123 with Agent B (separate memory store) await client.memorize(..., user_id="user_123", agent_id="agent_B") # User 456 with Agent A (separate memory store) await client.memorize(..., user_id="user_456", agent_id="agent_A") ``` -------------------------------- ### Checking Task Status with TaskStatusEnum Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/types.md Example of how to retrieve a task's status using the SDK and compare it against the TaskStatusEnum. This pattern is useful for handling different task outcomes. ```python from memu_sdk.models import TaskStatusEnum status = await client.get_task_status("task_123") if status.status == TaskStatusEnum.SUCCESS: print("Task succeeded") elif status.status == TaskStatusEnum.FAILED: print(f"Task failed: {status.message}") elif status.status == TaskStatusEnum.PROCESSING: print(f"Still processing: {status.progress}%") ``` -------------------------------- ### Comprehensive Error Handling Example Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/errors.md Implement a robust error handling strategy for MemU API calls, covering authentication, rate limits, validation, not found errors, general client errors, and value errors. This function ensures graceful failure and provides informative messages. ```python import asyncio from memu_sdk import MemUClient from memu_sdk.client import ( MemUClientError, MemUAuthenticationError, MemURateLimitError, MemUNotFoundError, MemUValidationError, ) async def safe_memorize(api_key: str, conversation: str, user_id: str, agent_id: str): """Memorize with comprehensive error handling.""" try: async with MemUClient(api_key=api_key) as client: result = await client.memorize( conversation_text=conversation, user_id=user_id, agent_id=agent_id, wait_for_completion=True, timeout=60.0 ) print(f"✅ Success: {len(result.items)} items memorized") return result except MemUAuthenticationError as e: print("❌ Authentication failed. Invalid API key.") print(f" {e.message}") return None except MemURateLimitError as e: print("⏳ Rate limit exceeded.") if e.retry_after: print(f" Retry after {e.retry_after} seconds") return None except MemUValidationError as e: print("⚠️ Validation error. Check your request parameters.") print(f" {e.message}") if e.response: print(f" Details: {e.response}") return None except MemUNotFoundError as e: print("❌ Resource not found.") print(f" {e.message}") return None except MemUClientError as e: print("❌ API error occurred.") print(f" Status: {e.status_code}") print(f" Message: {e.message}") if e.response: print(f" Response: {e.response}") return None except ValueError as e: print("❌ Invalid arguments.") print(f" {str(e)}") return None # Usage if __name__ == "__main__": asyncio.run(safe_memorize( api_key="sk_live_...", conversation="User: Hello\nAssistant: Hi there!", user_id="user_123", agent_id="agent_456" )) ``` -------------------------------- ### Implement Continuous Memory Updates Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/quick-start.md Periodically memorize new conversations to keep the memory up-to-date. This example shows a basic loop that updates memories hourly. ```python # Periodically update user memories async def remember_conversation(client, user_id, agent_id, text): await client.memorize( conversation_text=text, user_id=user_id, agent_id=agent_id, wait_for_completion=True ) # In your event loop while True: new_conversation = await get_next_conversation() await remember_conversation(client, "user_123", "agent_456", new_conversation) await asyncio.sleep(3600) # Update hourly ``` -------------------------------- ### Initialize MemUClient with Configuration Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/configuration.md Instantiate the MemUClient with essential configuration parameters. Ensure your API key is valid and the base URL is correctly set for your environment. ```python from memu_sdk import MemUClient client = MemUClient( api_key="sk_live_...", base_url="https://api.memu.so", timeout=60.0, max_retries=3 ) ``` -------------------------------- ### Initialize MemU Client with API Key from Environment Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/quick-start.md Initialize the MemU client by retrieving the API key from the environment variable `MEMU_API_KEY`. This is a recommended practice for security. ```python import os from memu_sdk import MemUClient client = MemUClient(api_key=os.environ["MEMU_API_KEY"]) ``` -------------------------------- ### get_task_status() Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/README.md Gets the status of an asynchronous memorization task. This is an asynchronous method. ```APIDOC ## get_task_status() ### Description Get the status of an asynchronous memorization task. This is an asynchronous method. ### Method `async def get_task_status(task_id: str) -> TaskStatus` ### Parameters: - `task_id` (str) - Required - The ID of the task to check. ### Returns: - `TaskStatus` - An object containing the status of the task. ``` -------------------------------- ### Initialize MemUClient using Environment Variables Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/configuration.md Shows how to manually read configuration values from environment variables to initialize the MemUClient. This approach is useful for managing secrets and configurations in different environments. ```python import os from memu_sdk import MemUClient # Manual environment variable handling api_key = os.environ.get("MEMU_API_KEY") if not api_key: raise ValueError("MEMU_API_KEY environment variable not set") client = MemUClient(api_key=api_key) ``` -------------------------------- ### Get Task Status Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/api-reference/memory-operations.md Retrieves the status and results of an asynchronous memorization task using its task ID. ```APIDOC ## GET /api/v3/memory/memorize/status/{task_id} ### Description Retrieves the status and results of an asynchronous memorization task. ### Method GET ### Endpoint /api/v3/memory/memorize/status/{task_id} ### Parameters #### Path Parameters - **task_id** (string) - Required - Task ID obtained from the memorize response. ### Response #### Success Response (200 OK) - **task_id** (string) - The ID of the task. - **status** (string) - Current status (PENDING, PROCESSING, COMPLETED, SUCCESS, FAILED). - **progress** (integer) - Completion progress (0-100). - **message** (string) - Optional error message if the task failed. - **result** (object) - The outcome of the task, including items, categories, and resource details. - **created_at** (string) - ISO 8601 datetime of task creation. - **updated_at** (string) - ISO 8601 datetime of last task update. #### Response Example ```json { "task_id": "task-abc-123", "status": "COMPLETED", "progress": 100, "result": { "items": [ { "id": "item-001", "summary": "User asked about weather.", "content": "What's the weather like today?", "memory_type": "query", "category_id": "cat-weather", "category_name": "Weather" } ], "categories": [ { "id": "cat-weather", "name": "Weather", "summary": "Information related to weather forecasts and conditions.", "item_count": 1 } ], "resource": {} }, "created_at": "2023-10-27T10:00:00Z", "updated_at": "2023-10-27T10:05:00Z" } ``` ### Error Handling - **401 Unauthorized**: Invalid API key. - **404 Not Found**: Task ID not found. - **429 Too Many Requests**: Rate limited (client auto-retries). - **5xx**: Server error (client auto-retries). ``` -------------------------------- ### Basic Async Usage of MemUClient Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/README.md Demonstrates how to initialize an async MemUClient, memorize a conversation, and retrieve memories. Ensure you have an API key and have run asyncio.run(main()). ```python import asyncio from memu_sdk import MemUClient async def main(): # Initialize the client async with MemUClient(api_key="your_api_key") as client: # Memorize a conversation result = await client.memorize( conversation=[ {"role": "user", "content": "I love Italian food, especially pasta."}, {"role": "assistant", "content": "That's great! What's your favorite dish?"}, {"role": "user", "content": "Carbonara is my absolute favorite!"} ], user_id="user_123", agent_id="my_assistant", wait_for_completion=True ) print(f"Task ID: {result.task_id}") # Retrieve memories memories = await client.retrieve( query="What food does the user like?", user_id="user_123", agent_id="my_assistant" ) print(f"Found {len(memories.items)} relevant memories") for item in memories.items: print(f" - [{item.memory_type}] {item.content}") asyncio.run(main()) ``` -------------------------------- ### Initialize MemUClient with Custom Settings Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/configuration.md Instantiate the MemUClient with API key, base URL, timeout, and max retries. Environment variables are used for defaults. ```python import os from memu_sdk import MemUClient client = MemUClient( api_key=os.environ.get("MEMU_API_KEY", ""), base_url=os.environ.get("MEMU_BASE_URL", "https://api.memu.so"), timeout=float(os.environ.get("MEMU_TIMEOUT", "60.0")), max_retries=int(os.environ.get("MEMU_MAX_RETRIES", "3")) ) ``` -------------------------------- ### Async Context Manager for Client Initialization Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/INDEX.md Initializes and cleans up the MemUClient using an async context manager. This is the recommended approach for ensuring proper resource management. ```python # Recommended: automatic cleanup async with MemUClient(api_key="...") as client: result = await client.memorize(...) # Manual cleanup client = MemUClient(api_key="...") result = await client.memorize(...) await client.close() ``` -------------------------------- ### Handle MemUValidationError Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/errors.md Catch MemUValidationError to manage request validation failures. This example demonstrates how to print the error message and any detailed response from the API. ```python from memu_sdk.client import MemUValidationError try: result = await client.memorize( conversation_text="", # Empty text user_id="", # Empty user_id agent_id="agent_456" ) except MemUValidationError as e: print("Validation error. Check your request:") print(e.message) if e.response: print(f"Details: {e.response}") ``` -------------------------------- ### MemUClient Initialization Parameters Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/README.md Illustrates the parameters available when initializing the MemUClient, including API key, base URL, timeout, and maximum retries. ```python MemUClient( api_key: str, *, base_url: str = "https://api.memu.so", timeout: float = 60.0, max_retries: int = 3 ) ``` -------------------------------- ### Synchronously Get Task Status Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/api-reference/memu-client.md This synchronous wrapper retrieves the status of a specific task using its ID. It is equivalent to the asynchronous `get_task_status()` method. ```python status = client.get_task_status_sync("task_abc123") ``` -------------------------------- ### Configure MemUClient with Custom Base URL Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/configuration.md Shows how to initialize the MemUClient using a custom base URL for different deployment environments. The SDK automatically handles trailing slashes. ```python # Default (recommended) client = MemUClient(api_key="sk_live_...", base_url="https://api.memu.so") # Custom deployment client = MemUClient(api_key="sk_live_...", base_url="https://custom.example.com") # Staging client = MemUClient(api_key="sk_test_...", base_url="https://staging.memu.so") ``` -------------------------------- ### Get Task Status Method Signature Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/README.md Presents the asynchronous `get_task_status` method, which takes a `task_id` to retrieve the status of an asynchronous memorization task. ```python async def get_task_status(task_id: str) -> TaskStatus ``` -------------------------------- ### MemUClient Initialization Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/README.md Initializes the MemUClient with your API key and optional configuration parameters. ```APIDOC ## MemUClient ```python MemUClient( api_key: str, *, base_url: str = "https://api.memu.so", timeout: float = 60.0, max_retries: int = 3 ) ``` ### Parameters: - `api_key` (str) - Required - Your MemU API key - `base_url` (str) - Optional - API base URL (default: https://api.memu.so) - `timeout` (float) - Optional - Request timeout in seconds (default: 60.0) - `max_retries` (int) - Optional - Maximum retry attempts for failed requests (default: 3) ``` -------------------------------- ### Valid API Key Initialization Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/configuration.md Demonstrates the correct way to initialize the MemUClient with a valid API key. Invalid or empty API keys will raise a ValueError. ```python # ✅ Valid client = MemUClient(api_key="sk_live_abc123") # ❌ Invalid - raises ValueError try: client = MemUClient(api_key="") except ValueError: print("API key is required") ``` -------------------------------- ### Initialize MemUClient Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/api-reference/memu-client.md Instantiate the MemUClient with your API key. You can optionally provide a custom base URL, request timeout, and the maximum number of retries for failed requests. ```python from memu_sdk import MemUClient # Initialize client with default settings client = MemUClient(api_key="sk_live_...") # Initialize with custom configuration client = MemUClient( api_key="sk_live_...", base_url="https://api.example.com", timeout=30.0, max_retries=5 ) ``` -------------------------------- ### Clone MemU SDK Repository Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/README.md Instructions to clone the MemU SDK Python repository from GitHub. ```bash # Clone the repository git clone https://github.com/NevaMind-AI/memU-sdk-py.git cd memU-sdk-py ``` -------------------------------- ### List and Retrieve Memory Categories Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/types.md Demonstrates how to list memory categories for a user and how to interpret relevance scores in retrieve results. ```python categories = await client.list_categories(user_id="user_123") for cat in categories: print(f"{cat.name}") print(f" Items: {cat.item_count}") print(f" Summary: {cat.summary}") # In retrieve results, score indicates relevance result = await client.retrieve( query="preferences", user_id="user_123", agent_id="agent_456" ) for cat in result.categories: print(f"{cat.name} (relevance: {cat.score})") ``` -------------------------------- ### Get Task Status API Response Body Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/api-reference/memory-operations.md The response body for retrieving the status of a memorization task. Includes progress, status, and results upon completion. ```json { "task_id": "string", "status": "PENDING|PROCESSING|COMPLETED|SUCCESS|FAILED", "progress": 0-100, "message": "string (optional, error message if failed)", "result": { "items": [ { "id": "string", "summary": "string", "content": "string", "memory_type": "string", "category_id": "string", "category_name": "string" } ], "categories": [ { "id": "string", "name": "string", "summary": "string", "item_count": "integer" } ], "resource": { } }, "created_at": "ISO 8601 datetime", "updated_at": "ISO 8601 datetime" } ``` -------------------------------- ### Configure MemU Client for Development/Testing Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/configuration.md Use this configuration for development and testing environments. It specifies a test API key and a staging base URL with a shorter timeout and fewer retries for faster feedback. ```python from memu_sdk import MemUClient client = MemUClient( api_key="sk_test_...", # Test API key base_url="https://staging.memu.so", timeout=30.0, # Shorter timeout max_retries=1 # Fewer retries for faster failures ) ``` -------------------------------- ### Configure MemU Client for Production Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/configuration.md This configuration is for production environments. It uses an environment variable for the API key, the production base URL, and standard timeout and retry values. ```python import os from memu_sdk import MemUClient client = MemUClient( api_key=os.environ["MEMU_API_KEY"], base_url="https://api.memu.so", timeout=60.0, # Reasonable timeout max_retries=3 # Standard retries ) ``` -------------------------------- ### Run MemU SDK Integration Tests Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/README.md Executes integration tests for the MemU SDK, requiring an API key to be set as an environment variable. ```bash # Run integration tests (requires API key) MEMU_API_KEY=your_key python tests/test_integration.py ``` -------------------------------- ### Using Async Context Manager for Cleanup Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/configuration.md Employ the async context manager (`async with`) for automatic client resource management, ensuring `client.close()` is called even if errors occur. ```python # Explicit cleanup client = MemUClient(api_key="sk_live_...") try: result = await client.memorize(...) finally: await client.close() # Automatic cleanup (recommended) async with MemUClient(api_key="sk_live_...") as client: result = await client.memorize(...) # client.close() called automatically ``` -------------------------------- ### Format MemU SDK Code Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/README.md Applies code formatting to the entire MemU SDK project using Ruff. ```bash # Format code ruff format . ``` -------------------------------- ### Accessing MemoryResource Details Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/types.md Demonstrates how to access details of a MemoryResource after it has been created or retrieved. This is useful for inspecting the properties of stored data. ```python result = await client.memorize( conversation_text="...", user_id="user_123", agent_id="agent_456" ) if result.resource: print(f"Resource ID: {result.resource.id}") print(f"Type: {result.resource.modality}") print(f"Created: {result.resource.created_at}") ``` -------------------------------- ### Use MemUClient as Async Context Manager Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/api-reference/memu-client.md Utilize the MemUClient within an `async with` block to ensure proper resource management and cleanup. This is the recommended way to use the client for asynchronous operations. ```python async with MemUClient(api_key="your_key") as client: result = await client.memorize( conversation_text="Hello!", user_id="user_123", agent_id="agent_456" ) ``` -------------------------------- ### Configure MemU Client for Unstable Network Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/configuration.md For environments with unstable networks, this configuration increases the timeout and the number of retries to improve reliability. ```python from memu_sdk import MemUClient client = MemUClient( api_key="sk_live_...", timeout=120.0, # Longer timeout max_retries=5 # More aggressive retries ) ``` -------------------------------- ### Synchronous Usage of MemUClient Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/README.md Shows how to use the synchronous methods of MemUClient for memorizing and retrieving memories without async/await. Remember to call client.close_sync() when done. ```python from memu_sdk import MemUClient # Initialize the client client = MemUClient(api_key="your_api_key") # Memorize a conversation (sync) result = client.memorize_sync( conversation_text="User: I love pasta\nAssistant: Great choice!", user_id="user_123", agent_id="my_assistant" ) # Retrieve memories (sync) memories = client.retrieve_sync( query="What are the user's preferences?", user_id="user_123", agent_id="my_assistant" ) # Clean up client.close_sync() ``` -------------------------------- ### Perform Semantic Search Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/quick-start.md Retrieve memory items based on a query and sort them by relevance score in descending order. Requires an initialized client. ```python result = await client.retrieve( query="What are the user's interests?", user_id="user_123", agent_id="agent_456" ) # Items are ranked by relevance for item in sorted(result.items, key=lambda x: x.score, reverse=True): print(f"{item.summary}") ``` -------------------------------- ### Configure MemU Client for Performance-Sensitive Applications Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/configuration.md This configuration is suitable for performance-sensitive applications. It uses a very short timeout and minimal retries to ensure quick failures. ```python from memu_sdk import MemUClient client = MemUClient( api_key="sk_live_...", timeout=10.0, # Fast timeout max_retries=1 # Quick failure detection ) ``` -------------------------------- ### Access MemoryResource Details Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/api-reference/models.md Demonstrates how to access and print details of a MemoryResource object obtained after a memorization operation. Shows how to access custom metadata. ```python from memu_sdk import MemUClient result = await client.memorize( conversation_text="...", user_id="user_123", agent_id="agent_456" ) resource = result.resource if resource: print(f"ID: {resource.id}") print(f"Type: {resource.modality}") print(f"Created: {resource.created_at.isoformat()}") print(f"Custom field: {resource.metadata.get('custom_key')}") ``` -------------------------------- ### Set MemU API Key Environment Variable Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/quick-start.md Securely manage your API key by exporting it as an environment variable. Avoid hardcoding sensitive credentials directly in your code. ```bash export MEMU_API_KEY="sk_live_abc123..." ``` -------------------------------- ### Configure MemUClient with Custom Max Retries Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/configuration.md Demonstrates how to configure the maximum number of retries for transient errors. Setting `max_retries` to 0 effectively disables automatic retries. ```python # Disable retries client = MemUClient(api_key="sk_live_...", max_retries=0) # More aggressive retries for unstable networks client = MemUClient(api_key="sk_live_...", max_retries=5) # Default (recommended) client = MemUClient(api_key="sk_live_...", max_retries=3) ``` -------------------------------- ### Retrieve Method Signature Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/README.md Shows the parameters for the asynchronous `retrieve` method, requiring a query and user/agent IDs to fetch relevant memories. ```python async def retrieve( query: str | list[dict], *, user_id: str, # Required agent_id: str, # Required ) -> RetrieveResult ``` -------------------------------- ### Configure MemUClient with Custom Timeout Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/configuration.md Illustrates setting custom timeout values for MemUClient requests. Adjust the timeout based on network conditions or application latency requirements. ```python # Fast timeout (good for latency-sensitive applications) client = MemUClient(api_key="sk_live_...", timeout=10.0) # Longer timeout (good for large uploads or slow networks) client = MemUClient(api_key="sk_live_...", timeout=300.0) # Very short timeout (for quick checks) client = MemUClient(api_key="sk_live_...", timeout=5.0) ``` -------------------------------- ### Configure MemUClient with Increased Retries and Timeout Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/errors.md Adjust the `max_retries` and `timeout` parameters when initializing `MemUClient` to handle unstable network conditions more effectively. The default values are 3 retries and a 60-second timeout. ```python client = MemUClient( api_key="sk_live_...", max_retries=5, # Default is 3 timeout=120.0 # Default is 60.0 ) ``` -------------------------------- ### Run MemU SDK Unit Tests Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/README.md Executes the unit tests for the MemU SDK using pytest. ```bash # Run unit tests pytest tests/test_sdk.py ``` -------------------------------- ### Retrieve Memories with Text Query (Async) Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/api-reference/memory-operations.md Retrieves memories based on a simple text query. This is the most basic way to search through stored memories. Results include summary, relevance score, and category. ```python async def main(): async with MemUClient(api_key="sk_live_...") as client: # Simple text query result = await client.retrieve( query="What are my favorite books?", user_id="user_123", agent_id="agent_456" ) print(f"Found {len(result.items)} relevant memories:") for item in result.items: print(f" [{item.memory_type}] {item.summary}") print(f" Relevance: {item.score:.2f}") print(f" Category: {item.category_name}") asyncio.run(main()) ``` -------------------------------- ### Using Async and Sync Client Interfaces Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/INDEX.md The MemUClient supports both asynchronous and synchronous operations. Async methods have corresponding `_sync` equivalents for synchronous usage. ```python # Async async with MemUClient(api_key="...") as client: result = await client.memorize(...) # Sync client = MemUClient(api_key="...") result = client.memorize_sync(...) ``` -------------------------------- ### Pydantic v2 Model Creation and Validation Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/api-reference/models.md Demonstrates creating Pydantic v2 models and handling validation errors for missing required fields. ```python # Model creation item = MemoryItem(summary="Test", memory_type="preference") # Validation error on invalid input try: status = TaskStatus(task_id="123") # Missing required 'status' field except ValueError as e: print(f"Validation error: {e}") ``` -------------------------------- ### Handle ValueError for Empty API Key Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/errors.md Catch `ValueError` during `MemUClient` initialization if an empty or `None` API key is provided. This indicates an invalid client configuration. ```python try: client = MemUClient(api_key="") # Empty key except ValueError as e: print(f"Invalid client configuration: {e}") ``` -------------------------------- ### Retrieve Memories with Conversation Context (Async) Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/api-reference/memory-operations.md Retrieves memories using a conversation history as context for the query. This allows for more nuanced and context-aware memory retrieval. Results may include category relevance and suggested next steps. ```python async def main(): async with MemUClient(api_key="sk_live_...") as client: # Query with conversation context result = await client.retrieve( query=[ {"role": "user", "content": "What do you know about me?"}, {"role": "assistant", "content": "Let me check your memories..."}, {"role": "user", "content": "Tell me specifically about my preferences"} ], user_id="user_123", agent_id="agent_456" ) # Results with relevance scores for cat in result.categories: print(f"Category: {cat.name} (relevance: {cat.score})") # Follow-up suggestion if result.next_step_query: print(f"Suggested next query: {result.next_step_query}") asyncio.run(main()) ``` -------------------------------- ### List Categories (Async) Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/api-reference/memory-operations.md Asynchronously retrieves a list of categories for a given user and agent. It iterates through the categories and prints their names, item counts, and summaries. ```python async def main(): async with MemUClient(api_key="sk_live_...") as client: categories = await client.list_categories( user_id="user_123", agent_id="agent_456" ) for cat in categories: print(f"{cat.name}") print(f" Items: {cat.item_count}") print(f" Summary: {cat.summary}") print() asyncio.run(main()) ``` -------------------------------- ### Iterating Through MemoryItems Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/types.md Shows how to retrieve and process multiple MemoryItems from a retrieval query. This pattern is common for analyzing user preferences, skills, or facts stored in MemU. ```python result = await client.retrieve( query="What are the user's preferences?", user_id="user_123", agent_id="agent_456" ) for item in result.items: print(f"Type: {item.memory_type}") print(f"Category: {item.category_name}") print(f"Summary: {item.summary}") print(f"Relevance: {item.score}") ``` -------------------------------- ### List Categories Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/api-reference/memory-operations.md Lists all memory categories associated with a user. ```APIDOC ## GET /api/v3/memory/categories ### Description Lists all memory categories for a user. ### Method GET ### Endpoint /api/v3/memory/categories ### Parameters (No specific parameters mentioned for this endpoint in the source, assuming user context is derived from authentication) ### Response #### Success Response (200 OK) - **categories** (array) - A list of category objects, each containing id, name, summary, and item_count. #### Response Example ```json { "categories": [ { "id": "cat-weather", "name": "Weather", "summary": "Information related to weather forecasts and conditions.", "item_count": 5 }, { "id": "cat-tasks", "name": "Tasks", "summary": "User-defined tasks and to-dos.", "item_count": 3 } ] } ``` ### Error Handling (Error handling details not explicitly provided for this endpoint in the source, but common errors like 401 Unauthorized and 5xx Server Error would likely apply.) ``` -------------------------------- ### Synchronous Wrapper Methods Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/README.md Lists the synchronous wrapper methods available for each corresponding asynchronous method in the MemUClient, including `close_sync()`. ```python All async methods have synchronous wrappers: - memorize_sync() → wraps memorize() - retrieve_sync() → wraps retrieve() - list_categories_sync() → wraps list_categories() - get_task_status_sync() → wraps get_task_status() - close_sync() → wraps close() ``` -------------------------------- ### Continuous Memory Updates (Async) Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/api-reference/memory-operations.md Asynchronously updates user memories by memorizing new conversations. It then verifies the update by listing the user's categories. ```python async def update_user_memory(client, user_id: str, agent_id: str, new_conversation: str): """Keep user memories current by periodically memorizing new conversations.""" # Memorize new conversation result = await client.memorize( conversation_text=new_conversation, user_id=user_id, agent_id=agent_id, wait_for_completion=True ) # Verify update categories = await client.list_categories(user_id=user_id, agent_id=agent_id) print(f"Updated. User now has {len(categories)} categories") return result ``` -------------------------------- ### list_categories() Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/README.md Lists all memory categories for a user. This is an asynchronous method. ```APIDOC ## list_categories() ### Description List all memory categories for a user. This is an asynchronous method. ### Method `async def list_categories( *, user_id: str, # Required agent_id: str | None = None, ) -> list[MemoryCategory]` ### Parameters: - `user_id` (str) - Required - The unique identifier for the user. - `agent_id` (str | None) - Optional - The unique identifier for the agent. If None, categories for all agents of the user are returned. ### Returns: - `list[MemoryCategory]` - A list of memory category objects. ``` -------------------------------- ### Synchronous Memory Operations Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/api-reference/memory-operations.md Demonstrates synchronous usage of the MemUClient for memorizing, retrieving, and listing categories. It includes error handling with a finally block to ensure the client is closed. ```python from memu_sdk import MemUClient # Initialize client = MemUClient(api_key="sk_live_...") try: # Memorize result = client.memorize_sync( conversation_text="User: I prefer coffee\nAssistant: Good to know!", user_id="user_123", agent_id="agent_456", wait_for_completion=True ) print(f"Memorized {len(result.items)} items") # Retrieve memories = client.retrieve_sync( query="What does the user like to drink?", user_id="user_123", agent_id="agent_456" ) print(f"Found {len(memories.items)} memories") # List categories categories = client.list_categories_sync(user_id="user_123") print(f"User has {len(categories)} categories") finally: client.close_sync() ``` -------------------------------- ### Synchronous Memory Retrieval Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/api-reference/memu-client.md This is a synchronous wrapper for the `retrieve` method, suitable for use in non-async applications. It abstracts away the asyncio execution. Provide the query, user ID, and agent ID. ```python result = client.retrieve_sync( query="What foods does the user like?", user_id="user_123", agent_id="agent_456" ) ``` -------------------------------- ### Lint MemU SDK Code Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/README.md Checks the MemU SDK code for linting errors using Ruff. ```bash # Lint ruff check . ``` -------------------------------- ### Model Behavior Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/api-reference/models.md Details on the general behavior of Memu SDK models, including Pydantic v2 features, handling of extra fields, nullable fields, and datetime parsing. ```APIDOC ## Model Behavior ### Pydantic v2 Features All models use Pydantic v2. ```python # Model creation item = MemoryItem(summary="Test", memory_type="preference") # Validation error on invalid input try: status = TaskStatus(task_id="123") # Missing required 'status' field except ValueError as e: print(f"Validation error: {e}") # Serialization dict_form = item.model_dump() json_form = item.model_dump_json() # JSON parsing item_dict = {"summary": "Test", "memory_type": "preference"} item = MemoryItem(**item_dict) ``` ### Extra Fields All models allow extra fields. ```python item = MemoryItem( summary="Test", custom_field="preserved", another_custom="also preserved" ) print(item.custom_field) # "preserved" ``` ### Nullable Fields Fields with `| None` default to None. ```python # All valid item1 = MemoryItem() # All optional fields are None item2 = MemoryItem(summary=None, content=None) item3 = MemoryItem(summary="Test") # content is None ``` ### Datetime Handling Datetime fields are automatically parsed. ```python # String input is automatically parsed item = MemoryItem(created_at="2024-01-15T10:30:00Z") print(item.created_at) # datetime object # Datetime objects work directly from datetime import datetime item = MemoryItem(created_at=datetime.now()) ``` ``` -------------------------------- ### Synchronously List Memory Categories Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/api-reference/memu-client.md Use this synchronous wrapper to list memory categories for a given user. It mirrors the functionality of the asynchronous `list_categories()` method. ```python categories = client.list_categories_sync(user_id="user_123") ``` -------------------------------- ### list_categories Source: https://github.com/nevamind-ai/memu-sdk-py/blob/main/_autodocs/api-reference/memu-client.md Lists all available memory categories for a specific user, optionally scoped to an agent. Useful for organizing and filtering memories. ```APIDOC ## list_categories ### Description List all memory categories for a user. This can be scoped to a specific agent. ### Method `list_categories` (async) ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None #### Method Parameters - **user_id** (str) - Required - User identifier - **agent_id** (str | None) - Optional - Agent identifier for scoping (defaults to None) ### Request Example ```python # List all user categories categories = await client.list_categories(user_id="user_123") # List categories for a specific agent categories = await client.list_categories( user_id="user_123", agent_id="agent_456" ) ``` ### Response #### Success Response - **list[MemoryCategory]** - A list of memory category objects, each containing name, item count, and summary. #### Response Example ```json [ { "name": "food", "item_count": 5, "summary": "User's food preferences." } ] ``` ### Raises - `MemUClientError` for API errors ```