| Name | nexamem JSON |
| Version |
0.4.0
JSON |
| download |
| home_page | None |
| Summary | NexaMem/AIMemory - Complete enterprise-grade Redis-backed conversational memory with built-in governance, Azure integration, content processing, and audit capabilities. Everything included - no extras needed! |
| upload_time | 2025-07-14 15:30:59 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.8 |
| license | MIT License
Copyright (c) Microsoft Corporation.
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE |
| keywords |
aimemory
nexamem
chatbot
conversation
memory
llm
ai
redis
azure
governance
policy
audit
compliance
pii
content-processing
enterprise
async
channels
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# NexaMem / AIMemory
A Python library providing governed, Redis-backed conversational memory for AI components. Designed for easy integration with AI agents, orchestrators, and chatbot systems.
[](https://github.com/microsoft/nexamem/actions/workflows/tests.yml)
[](https://github.com/microsoft/nexamem/actions/workflows/python-publish-to-test.yml)
[](https://www.python.org/downloads/)
[](LICENSE)
## 📖 Quick Navigation
**New to NexaMem?** → [Quick Start - AIMemory](#quick-start---aimemory-recommended)
**Want Azure Redis?** → [Azure Redis Ready](#-azure-redis-ready) | [Production Configuration](#production-with-azure-redis)
**YAML Channels?** → [Channel Configuration](#channel-configuration-channelsyaml) | [Schema Validation](#-yaml-schema-validation)
**Need Help?** → [Azure Redis Integration Guide](AZURE_REDIS.md) | [Migration Guide](LEGACY_API.md)
**Testing & Development?** → [Testing & Development](#-testing--development)
## Features
### AIMemory API (Recommended) - v0.3
- **Redis-backed channels** with user-defined configuration
- **Sync and async client classes** for high-performance applications
- **YAML channel configuration** with TTL, encryption, and quota settings
- **Policy enforcement** for TTL, PII detection, and quota management
- **Scope-based access control** with environment isolation
- **Dynamic channel registration** for runtime flexibility
- **Checkpoint operations** for workflow state management
- **Content processing chain** with PII detection/redaction
- **Audit logging** to Redis Stream for compliance
- **Base metrics capturing** for monitoring and observability
- **Azure Redis integration** with enterprise security and performance
## Installation
```sh
pip install nexamem
# Development/testing
pip install -i https://test.pypi.org/simple/ nexamem
```
### Or from GitHub
```sh
pip install git+https://github.com/microsoft/nexamem.git
uv add git+https://github.com/microsoft/nexamem.git
```
## 🔥 Azure Redis Ready
NexaMem includes **first-class support for Azure Cache for Redis** with enterprise security features:
- **🚀 Quick Setup**: Connect with just hostname + access key
- **🔐 Enterprise Auth**: Full Azure Entra ID (Azure AD) integration
- **⚡ High Performance**: Sub-millisecond latency, automatic scaling
- **🛡️ Security**: VNet integration, encryption at rest/transit, compliance-ready
- **📊 Advanced Querying**: Pattern-based message search and analytics
**Ready to use Azure Redis?** See our [Quick Start with Azure Redis](#quick-start-with-azure-redis) section below or check the comprehensive [Azure Redis Integration Guide](AZURE_REDIS.md) for advanced configurations.
## Quick Start - AIMemory (Recommended)
The new AIMemory API provides enterprise-grade conversational memory with Redis backing and full YAML channel configuration support:
```python
from nexamem import AIMemory, AIMemoryConfig, StorageConfig, MemoryScope
# Initialize with in-memory storage for testing (or Redis for production)
config = AIMemoryConfig(
default_scope="test_session",
storage=StorageConfig.memory(), # For testing, use Redis for production
channels_yaml="channels.yaml", # Optional: YAML channel configuration
strict_yaml_validation=True # Optional: Enable strict YAML validation
)
memory = AIMemory(config)
# Create a memory scope (defines access control)
scope = MemoryScope(
agent_id="investment_agent",
user_id="user_12345",
session_id="session_abc123",
env="prod" # Environment isolation
)
# Write to a channel
message_uuid = memory.write(
scope=scope,
channel="working",
content="User asked about portfolio diversification",
auto_pii=True # Auto-detect PII
)
# Read from a channel
messages, metadata = memory.read(
scope=scope,
channel="working",
max_msgs=10,
since_sec=3600 # Last hour only
)
print(f"Retrieved {len(messages)} messages")
# Validate configuration (optional but recommended)
from nexamem.channels import validate_yaml_schema
try:
validate_yaml_schema("channels.yaml")
print("✅ Channel configuration is valid!")
except Exception as e:
print(f"❌ Configuration error: {e}")
```
### Production with Azure Redis
For production applications, use Azure Cache for Redis with the new AIMemory API:
```python
from nexamem import AIMemory, AIMemoryConfig, StorageConfig
# Production configuration with Azure Redis
config = AIMemoryConfig(
default_scope="production_session",
storage=StorageConfig.azure_redis(
hostname="your-cache.redis.cache.windows.net",
access_key="your_primary_access_key",
port=6380,
ssl=True
),
channels_yaml="channels.yaml",
strict_yaml_validation=True
)
memory = AIMemory(config)
# Use the same write/read operations as before
scope = MemoryScope(
agent_id="prod_agent",
user_id="customer_12345",
session_id="session_abc123",
env="prod"
)
message_uuid = memory.write(
scope=scope,
channel="working",
content="Production message with Azure Redis backing",
auto_pii=True
)
messages, metadata = memory.read(scope=scope, channel="working")
print(f"Retrieved {len(messages)} messages from Azure Redis")
```
> **💡 Pro Tip**: AIMemory includes strict YAML schema validation by default to catch configuration errors early. See the [Schema Validation](#-yaml-schema-validation) section for details.
### Channel Configuration (channels.yaml)
AIMemory uses **strict YAML schema validation** to ensure configuration integrity and security. Each channel must follow specific rules and constraints.
#### Schema Requirements
**Channel Names:**
- Must be `snake_case`: start with lowercase letter, followed by letters/numbers/underscores
- Maximum 50 characters
- Cannot use reserved names: `admin`, `system`, `internal`, `redis`, `audit`
- Pattern: `^[a-z][a-z0-9_]*$`
**Required Fields:**
- `ttl_sec`: Time-to-live in seconds (60 to 604,800 = 1 minute to 7 days)
**Optional Fields:**
- `encrypt`: Boolean (default: `false`) - enables client-side encryption
- `quota_bytes`: Integer (1 to 1,000,000,000) - daily quota per (agent, user, channel)
#### Valid Configuration Example
```yaml
channels:
working: # ✅ Valid: snake_case name
ttl_sec: 14400 # ✅ Valid: 4 hours (within 1 min - 7 days range)
encrypt: true # ✅ Valid: boolean value
quota_bytes: 1000000 # ✅ Valid: 1MB daily quota
procedure: # ✅ Valid: workflow checkpoints
ttl_sec: 604800 # ✅ Valid: 7 days (maximum allowed)
encrypt: true # ✅ Valid: encryption for sensitive workflows
routing: # ✅ Valid: classification hints
ttl_sec: 86400 # ✅ Valid: 24 hours
encrypt: false # ✅ Valid: no encryption needed for routing
```
#### Schema Validation Control
```python
# Strict validation (default - recommended)
memory = AIMemory(
channels_yaml="channels.yaml",
strict_yaml_validation=True # Default
)
# Backward compatibility mode (legacy configurations)
memory = AIMemory(
channels_yaml="legacy_channels.yaml",
strict_yaml_validation=False
)
# Standalone validation
from nexamem.channels import validate_yaml_schema, YamlSchemaError
try:
validate_yaml_schema("channels.yaml")
print("✅ Configuration is valid!")
except YamlSchemaError as e:
print(f"❌ Validation failed: {e}")
```
## 🔐 YAML Schema Validation
AIMemory enforces **strict schema validation** to prevent configuration errors and ensure security compliance.
### Validation Features
- **Channel Name Validation**: Enforces `snake_case` naming conventions
- **Field Type Checking**: Validates data types for all configuration fields
- **Range Validation**: Ensures TTL and quota values are within safe limits
- **Security Enforcement**: Blocks reserved names and unknown fields
- **Clear Error Messages**: Provides specific, actionable error descriptions
### Validation Modes
```python
# Strict validation (recommended for production)
memory = AIMemory(
channels_yaml="channels.yaml",
strict_yaml_validation=True # Default
)
# Legacy mode (for backward compatibility)
memory = AIMemory(
channels_yaml="legacy_channels.yaml",
strict_yaml_validation=False
)
```
### Schema Validation Tools
```python
from nexamem.channels import validate_yaml_schema, generate_yaml_schema_docs
# Validate configuration file
try:
validate_yaml_schema("channels.yaml")
print("✅ Configuration is valid")
except YamlSchemaError as e:
print(f"❌ Validation failed: {e}")
# Generate schema documentation
print(generate_yaml_schema_docs())
```
### Common Validation Errors
| Error Type | Example | Fix |
|------------|---------|-----|
| Invalid Name | `Invalid-Channel` | Use `invalid_channel` |
| TTL Range | `ttl_sec: 999999` | Use value ≤ 604800 (7 days) |
| Reserved Name | `admin:` | Use `admin_channel` or similar |
| Unknown Field | `custom_field: value` | Remove or use allowed fields |
| Wrong Type | `encrypt: "yes"` | Use `encrypt: true` |
#### Common Validation Errors
```yaml
# ❌ INVALID Examples:
channels:
Invalid-Name: # ❌ Hyphens not allowed
user_123: # ❌ Cannot start with number
admin: # ❌ Reserved name
working:
ttl_sec: 999999 # ❌ Exceeds 7 days (604,800 seconds)
encrypt: "yes" # ❌ Must be boolean (true/false)
quota_bytes: -100 # ❌ Must be positive
unknown_field: 1 # ❌ Unknown field not allowed
```
For detailed schema documentation, see [YAML_SCHEMA_VALIDATION.md](YAML_SCHEMA_VALIDATION.md).
### Enhanced Features
```python
# Dynamic channel registration
memory.register_channel(
name="debug_logs",
ttl_sec=3600, # 1 hour
encrypt=False,
quota_bytes=500000 # 500KB
)
# Checkpoint operations for workflow state
memory.checkpoint(scope, "workflow_state", {
"current_step": "risk_assessment",
"completed_steps": ["onboarding", "kyc"],
"progress": 0.6
})
# Atomic updates
def advance_step(current_state):
return {"step": current_state["step"] + 1}
updated = memory.checkpoint_atomic(scope, "counter", advance_step)
# Metrics and audit
metrics_data = memory.metrics.get_all_metrics()
audit_records = memory.audit.get_audit_records(count=10)
```
### Policy Enforcement
AIMemory automatically enforces policies:
```python
# PII protection - fails if PII sent to non-encrypted channel
try:
memory.write(
scope=scope,
channel="routing", # encrypt=false
content="Customer SSN: 123-45-6789",
pii=True
)
except EncryptionRequired:
print("PII blocked from non-encrypted channel")
# Quota enforcement
try:
large_content = "x" * 2000000 # 2MB
memory.write(scope=scope, channel="working", content=large_content)
except QuotaExceeded:
print("Daily quota exceeded")
# TTL validation
try:
memory.write(
scope=scope,
channel="working",
content="Test",
ttl_override=999999 # Exceeds channel limit
)
except TTLViolation:
print("TTL override exceeds channel limit")
```
### Dynamic Channel Management
Register channels at runtime:
```python
# Register new channel
memory.register_channel(
name="debug_logs",
ttl_sec=3600, # 1 hour
encrypt=False,
quota_bytes=500000 # 500KB
)
# List all channels
channels = memory.list_channels()
for name, config in channels.items():
print(f"{name}: TTL={config.ttl_sec}s, Encrypt={config.encrypt}")
```
### Checkpoint Operations
Manage workflow state:
```python
# Set checkpoint
memory.checkpoint(scope, "workflow_state", {
"current_step": "risk_assessment",
"completed_steps": ["onboarding", "kyc"],
"progress": 0.6
})
# Atomic update
def advance_step(current_state):
if current_state is None:
return {"step": 1}
return {"step": current_state["step"] + 1}
updated = memory.checkpoint_atomic(scope, "counter", advance_step)
print(f"Step: {updated['step']}")
```
### Async API
For high-throughput applications:
```python
import asyncio
from nexamem import AsyncAIMemory, AIMemoryConfig, StorageConfig
async def main():
config = AIMemoryConfig(
default_scope="async_session",
storage=StorageConfig.memory(), # For testing, use Redis for production
channels_yaml="channels.yaml" # Optional: YAML channel configuration
)
async_memory = AsyncAIMemory(config)
scope = MemoryScope(
agent_id="async_agent",
user_id="user_456"
)
# Async operations
message_uuid = await async_memory.write(
scope=scope,
channel="working",
content="Async message"
)
messages, metadata = await async_memory.read(
scope=scope,
channel="working"
)
await async_memory.close() # Important!
asyncio.run(main())
```
## 🧪 Testing & Development
### Running Tests
NexaMem includes a comprehensive test suite covering all major functionality. Tests are organized to provide both unit and integration coverage.
#### Prerequisites
Make sure you're using the project's virtual environment:
```bash
# On Windows (recommended)
.venv/Scripts/python.exe -m pytest
# Or activate the virtual environment first
.venv\Scripts\activate
python -m pytest
# On Unix/Linux/macOS
source .venv/bin/activate
python -m pytest
```
### Legacy ChatHistory API (Deprecated)
> ⚠️ **Note**: The legacy ChatHistory API is deprecated and will be removed in a future version. Please migrate to the new [AIMemory API](#quick-start---aimemory-recommended) for new projects.
For existing applications still using the legacy API, see [LEGACY_API.md](LEGACY_API.md) for documentation and migration guidance.
# On Unix/Linux/macOS
source .venv/bin/activate
python -m pytest
```
## License
This project is licensed under the MIT License.
Raw data
{
"_id": null,
"home_page": null,
"name": "nexamem",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": "Chad Adams <chadada@microsoft.com>",
"keywords": "aimemory, nexamem, chatbot, conversation, memory, llm, ai, redis, azure, governance, policy, audit, compliance, pii, content-processing, enterprise, async, channels",
"author": null,
"author_email": "Chad Adams <chadada@microsoft.com>",
"download_url": "https://files.pythonhosted.org/packages/0f/10/66871e92d7896aa67961d7e86b5b7f821b675bc2cab95e8224dd1ca3d676/nexamem-0.4.0.tar.gz",
"platform": null,
"description": "# NexaMem / AIMemory\n\nA Python library providing governed, Redis-backed conversational memory for AI components. Designed for easy integration with AI agents, orchestrators, and chatbot systems.\n\n[](https://github.com/microsoft/nexamem/actions/workflows/tests.yml)\n[](https://github.com/microsoft/nexamem/actions/workflows/python-publish-to-test.yml)\n[](https://www.python.org/downloads/)\n[](LICENSE)\n\n## \ud83d\udcd6 Quick Navigation\n\n**New to NexaMem?** \u2192 [Quick Start - AIMemory](#quick-start---aimemory-recommended) \n**Want Azure Redis?** \u2192 [Azure Redis Ready](#-azure-redis-ready) | [Production Configuration](#production-with-azure-redis) \n**YAML Channels?** \u2192 [Channel Configuration](#channel-configuration-channelsyaml) | [Schema Validation](#-yaml-schema-validation) \n**Need Help?** \u2192 [Azure Redis Integration Guide](AZURE_REDIS.md) | [Migration Guide](LEGACY_API.md) \n**Testing & Development?** \u2192 [Testing & Development](#-testing--development)\n\n## Features\n\n### AIMemory API (Recommended) - v0.3\n- **Redis-backed channels** with user-defined configuration\n- **Sync and async client classes** for high-performance applications \n- **YAML channel configuration** with TTL, encryption, and quota settings\n- **Policy enforcement** for TTL, PII detection, and quota management\n- **Scope-based access control** with environment isolation\n- **Dynamic channel registration** for runtime flexibility\n- **Checkpoint operations** for workflow state management\n- **Content processing chain** with PII detection/redaction\n- **Audit logging** to Redis Stream for compliance\n- **Base metrics capturing** for monitoring and observability\n- **Azure Redis integration** with enterprise security and performance\n\n## Installation\n\n```sh\npip install nexamem\n\n# Development/testing\npip install -i https://test.pypi.org/simple/ nexamem\n```\n\n### Or from GitHub\n\n```sh\npip install git+https://github.com/microsoft/nexamem.git\nuv add git+https://github.com/microsoft/nexamem.git \n```\n\n## \ud83d\udd25 Azure Redis Ready\n\nNexaMem includes **first-class support for Azure Cache for Redis** with enterprise security features:\n\n- **\ud83d\ude80 Quick Setup**: Connect with just hostname + access key \n- **\ud83d\udd10 Enterprise Auth**: Full Azure Entra ID (Azure AD) integration\n- **\u26a1 High Performance**: Sub-millisecond latency, automatic scaling\n- **\ud83d\udee1\ufe0f Security**: VNet integration, encryption at rest/transit, compliance-ready\n- **\ud83d\udcca Advanced Querying**: Pattern-based message search and analytics\n\n**Ready to use Azure Redis?** See our [Quick Start with Azure Redis](#quick-start-with-azure-redis) section below or check the comprehensive [Azure Redis Integration Guide](AZURE_REDIS.md) for advanced configurations.\n\n## Quick Start - AIMemory (Recommended)\n\nThe new AIMemory API provides enterprise-grade conversational memory with Redis backing and full YAML channel configuration support:\n\n```python\nfrom nexamem import AIMemory, AIMemoryConfig, StorageConfig, MemoryScope\n\n# Initialize with in-memory storage for testing (or Redis for production)\nconfig = AIMemoryConfig(\n default_scope=\"test_session\",\n storage=StorageConfig.memory(), # For testing, use Redis for production\n channels_yaml=\"channels.yaml\", # Optional: YAML channel configuration\n strict_yaml_validation=True # Optional: Enable strict YAML validation\n)\nmemory = AIMemory(config)\n\n# Create a memory scope (defines access control)\nscope = MemoryScope(\n agent_id=\"investment_agent\",\n user_id=\"user_12345\",\n session_id=\"session_abc123\",\n env=\"prod\" # Environment isolation\n)\n\n# Write to a channel\nmessage_uuid = memory.write(\n scope=scope,\n channel=\"working\",\n content=\"User asked about portfolio diversification\",\n auto_pii=True # Auto-detect PII\n)\n\n# Read from a channel\nmessages, metadata = memory.read(\n scope=scope,\n channel=\"working\",\n max_msgs=10,\n since_sec=3600 # Last hour only\n)\n\nprint(f\"Retrieved {len(messages)} messages\")\n\n# Validate configuration (optional but recommended)\nfrom nexamem.channels import validate_yaml_schema\n\ntry:\n validate_yaml_schema(\"channels.yaml\")\n print(\"\u2705 Channel configuration is valid!\")\nexcept Exception as e:\n print(f\"\u274c Configuration error: {e}\")\n```\n\n### Production with Azure Redis\n\nFor production applications, use Azure Cache for Redis with the new AIMemory API:\n\n```python\nfrom nexamem import AIMemory, AIMemoryConfig, StorageConfig\n\n# Production configuration with Azure Redis\nconfig = AIMemoryConfig(\n default_scope=\"production_session\",\n storage=StorageConfig.azure_redis(\n hostname=\"your-cache.redis.cache.windows.net\",\n access_key=\"your_primary_access_key\",\n port=6380,\n ssl=True\n ),\n channels_yaml=\"channels.yaml\",\n strict_yaml_validation=True\n)\n\nmemory = AIMemory(config)\n\n# Use the same write/read operations as before\nscope = MemoryScope(\n agent_id=\"prod_agent\",\n user_id=\"customer_12345\",\n session_id=\"session_abc123\",\n env=\"prod\"\n)\n\nmessage_uuid = memory.write(\n scope=scope,\n channel=\"working\",\n content=\"Production message with Azure Redis backing\",\n auto_pii=True\n)\n\nmessages, metadata = memory.read(scope=scope, channel=\"working\")\nprint(f\"Retrieved {len(messages)} messages from Azure Redis\")\n```\n\n> **\ud83d\udca1 Pro Tip**: AIMemory includes strict YAML schema validation by default to catch configuration errors early. See the [Schema Validation](#-yaml-schema-validation) section for details.\n\n### Channel Configuration (channels.yaml)\n\nAIMemory uses **strict YAML schema validation** to ensure configuration integrity and security. Each channel must follow specific rules and constraints.\n\n#### Schema Requirements\n\n**Channel Names:**\n- Must be `snake_case`: start with lowercase letter, followed by letters/numbers/underscores\n- Maximum 50 characters\n- Cannot use reserved names: `admin`, `system`, `internal`, `redis`, `audit`\n- Pattern: `^[a-z][a-z0-9_]*$`\n\n**Required Fields:**\n- `ttl_sec`: Time-to-live in seconds (60 to 604,800 = 1 minute to 7 days)\n\n**Optional Fields:**\n- `encrypt`: Boolean (default: `false`) - enables client-side encryption\n- `quota_bytes`: Integer (1 to 1,000,000,000) - daily quota per (agent, user, channel)\n\n#### Valid Configuration Example\n\n```yaml\nchannels:\n working: # \u2705 Valid: snake_case name\n ttl_sec: 14400 # \u2705 Valid: 4 hours (within 1 min - 7 days range)\n encrypt: true # \u2705 Valid: boolean value\n quota_bytes: 1000000 # \u2705 Valid: 1MB daily quota\n\n procedure: # \u2705 Valid: workflow checkpoints\n ttl_sec: 604800 # \u2705 Valid: 7 days (maximum allowed)\n encrypt: true # \u2705 Valid: encryption for sensitive workflows\n\n routing: # \u2705 Valid: classification hints\n ttl_sec: 86400 # \u2705 Valid: 24 hours\n encrypt: false # \u2705 Valid: no encryption needed for routing\n```\n\n#### Schema Validation Control\n\n```python\n# Strict validation (default - recommended)\nmemory = AIMemory(\n channels_yaml=\"channels.yaml\",\n strict_yaml_validation=True # Default\n)\n\n# Backward compatibility mode (legacy configurations)\nmemory = AIMemory(\n channels_yaml=\"legacy_channels.yaml\",\n strict_yaml_validation=False\n)\n\n# Standalone validation\nfrom nexamem.channels import validate_yaml_schema, YamlSchemaError\n\ntry:\n validate_yaml_schema(\"channels.yaml\")\n print(\"\u2705 Configuration is valid!\")\nexcept YamlSchemaError as e:\n print(f\"\u274c Validation failed: {e}\")\n```\n\n## \ud83d\udd10 YAML Schema Validation\n\nAIMemory enforces **strict schema validation** to prevent configuration errors and ensure security compliance.\n\n### Validation Features\n\n- **Channel Name Validation**: Enforces `snake_case` naming conventions\n- **Field Type Checking**: Validates data types for all configuration fields \n- **Range Validation**: Ensures TTL and quota values are within safe limits\n- **Security Enforcement**: Blocks reserved names and unknown fields\n- **Clear Error Messages**: Provides specific, actionable error descriptions\n\n### Validation Modes\n\n```python\n# Strict validation (recommended for production)\nmemory = AIMemory(\n channels_yaml=\"channels.yaml\",\n strict_yaml_validation=True # Default\n)\n\n# Legacy mode (for backward compatibility)\nmemory = AIMemory(\n channels_yaml=\"legacy_channels.yaml\", \n strict_yaml_validation=False\n)\n```\n\n### Schema Validation Tools\n\n```python\nfrom nexamem.channels import validate_yaml_schema, generate_yaml_schema_docs\n\n# Validate configuration file\ntry:\n validate_yaml_schema(\"channels.yaml\")\n print(\"\u2705 Configuration is valid\")\nexcept YamlSchemaError as e:\n print(f\"\u274c Validation failed: {e}\")\n\n# Generate schema documentation\nprint(generate_yaml_schema_docs())\n```\n\n### Common Validation Errors\n\n| Error Type | Example | Fix |\n|------------|---------|-----|\n| Invalid Name | `Invalid-Channel` | Use `invalid_channel` |\n| TTL Range | `ttl_sec: 999999` | Use value \u2264 604800 (7 days) |\n| Reserved Name | `admin:` | Use `admin_channel` or similar |\n| Unknown Field | `custom_field: value` | Remove or use allowed fields |\n| Wrong Type | `encrypt: \"yes\"` | Use `encrypt: true` |\n\n#### Common Validation Errors\n\n```yaml\n# \u274c INVALID Examples:\nchannels:\n Invalid-Name: # \u274c Hyphens not allowed\n user_123: # \u274c Cannot start with number\n admin: # \u274c Reserved name\n working:\n ttl_sec: 999999 # \u274c Exceeds 7 days (604,800 seconds)\n encrypt: \"yes\" # \u274c Must be boolean (true/false)\n quota_bytes: -100 # \u274c Must be positive\n unknown_field: 1 # \u274c Unknown field not allowed\n```\n\nFor detailed schema documentation, see [YAML_SCHEMA_VALIDATION.md](YAML_SCHEMA_VALIDATION.md).\n\n### Enhanced Features\n\n```python\n# Dynamic channel registration\nmemory.register_channel(\n name=\"debug_logs\",\n ttl_sec=3600, # 1 hour\n encrypt=False,\n quota_bytes=500000 # 500KB\n)\n\n# Checkpoint operations for workflow state\nmemory.checkpoint(scope, \"workflow_state\", {\n \"current_step\": \"risk_assessment\",\n \"completed_steps\": [\"onboarding\", \"kyc\"],\n \"progress\": 0.6\n})\n\n# Atomic updates\ndef advance_step(current_state):\n return {\"step\": current_state[\"step\"] + 1}\n\nupdated = memory.checkpoint_atomic(scope, \"counter\", advance_step)\n\n# Metrics and audit\nmetrics_data = memory.metrics.get_all_metrics()\naudit_records = memory.audit.get_audit_records(count=10)\n```\n\n### Policy Enforcement\n\nAIMemory automatically enforces policies:\n\n```python\n# PII protection - fails if PII sent to non-encrypted channel\ntry:\n memory.write(\n scope=scope,\n channel=\"routing\", # encrypt=false\n content=\"Customer SSN: 123-45-6789\",\n pii=True\n )\nexcept EncryptionRequired:\n print(\"PII blocked from non-encrypted channel\")\n\n# Quota enforcement\ntry:\n large_content = \"x\" * 2000000 # 2MB\n memory.write(scope=scope, channel=\"working\", content=large_content)\nexcept QuotaExceeded:\n print(\"Daily quota exceeded\")\n\n# TTL validation\ntry:\n memory.write(\n scope=scope,\n channel=\"working\",\n content=\"Test\",\n ttl_override=999999 # Exceeds channel limit\n )\nexcept TTLViolation:\n print(\"TTL override exceeds channel limit\")\n```\n\n### Dynamic Channel Management\n\nRegister channels at runtime:\n\n```python\n# Register new channel\nmemory.register_channel(\n name=\"debug_logs\",\n ttl_sec=3600, # 1 hour\n encrypt=False,\n quota_bytes=500000 # 500KB\n)\n\n# List all channels\nchannels = memory.list_channels()\nfor name, config in channels.items():\n print(f\"{name}: TTL={config.ttl_sec}s, Encrypt={config.encrypt}\")\n```\n\n### Checkpoint Operations\n\nManage workflow state:\n\n```python\n# Set checkpoint\nmemory.checkpoint(scope, \"workflow_state\", {\n \"current_step\": \"risk_assessment\",\n \"completed_steps\": [\"onboarding\", \"kyc\"],\n \"progress\": 0.6\n})\n\n# Atomic update\ndef advance_step(current_state):\n if current_state is None:\n return {\"step\": 1}\n return {\"step\": current_state[\"step\"] + 1}\n\nupdated = memory.checkpoint_atomic(scope, \"counter\", advance_step)\nprint(f\"Step: {updated['step']}\")\n```\n\n### Async API\n\nFor high-throughput applications:\n\n```python\nimport asyncio\nfrom nexamem import AsyncAIMemory, AIMemoryConfig, StorageConfig\n\nasync def main():\n config = AIMemoryConfig(\n default_scope=\"async_session\",\n storage=StorageConfig.memory(), # For testing, use Redis for production\n channels_yaml=\"channels.yaml\" # Optional: YAML channel configuration\n )\n \n async_memory = AsyncAIMemory(config)\n \n scope = MemoryScope(\n agent_id=\"async_agent\",\n user_id=\"user_456\"\n )\n \n # Async operations\n message_uuid = await async_memory.write(\n scope=scope,\n channel=\"working\", \n content=\"Async message\"\n )\n \n messages, metadata = await async_memory.read(\n scope=scope,\n channel=\"working\"\n )\n \n await async_memory.close() # Important!\n\nasyncio.run(main())\n```\n\n## \ud83e\uddea Testing & Development\n\n### Running Tests\n\nNexaMem includes a comprehensive test suite covering all major functionality. Tests are organized to provide both unit and integration coverage.\n\n#### Prerequisites\n\nMake sure you're using the project's virtual environment:\n\n```bash\n# On Windows (recommended)\n.venv/Scripts/python.exe -m pytest\n\n# Or activate the virtual environment first\n.venv\\Scripts\\activate\npython -m pytest\n\n# On Unix/Linux/macOS\nsource .venv/bin/activate\npython -m pytest\n```\n\n### Legacy ChatHistory API (Deprecated)\n\n> \u26a0\ufe0f **Note**: The legacy ChatHistory API is deprecated and will be removed in a future version. Please migrate to the new [AIMemory API](#quick-start---aimemory-recommended) for new projects.\n\nFor existing applications still using the legacy API, see [LEGACY_API.md](LEGACY_API.md) for documentation and migration guidance.\n\n# On Unix/Linux/macOS\nsource .venv/bin/activate\npython -m pytest\n```\n\n## License\n\nThis project is licensed under the MIT License.\n",
"bugtrack_url": null,
"license": "MIT License\n \n Copyright (c) Microsoft Corporation.\n \n Permission is hereby granted, free of charge, to any person obtaining a copy\n of this software and associated documentation files (the \"Software\"), to deal\n in the Software without restriction, including without limitation the rights\n to use, copy, modify, merge, publish, distribute, sublicense, and/or sell\n copies of the Software, and to permit persons to whom the Software is\n furnished to do so, subject to the following conditions:\n \n The above copyright notice and this permission notice shall be included in all\n copies or substantial portions of the Software.\n \n THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR\n IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,\n FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE\n AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER\n LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,\n OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE\n SOFTWARE",
"summary": "NexaMem/AIMemory - Complete enterprise-grade Redis-backed conversational memory with built-in governance, Azure integration, content processing, and audit capabilities. Everything included - no extras needed!",
"version": "0.4.0",
"project_urls": {
"Bug Tracker": "https://github.com/microsoft/nexamem/issues",
"Changelog": "https://github.com/microsoft/nexamem/releases",
"Discussions": "https://github.com/microsoft/nexamem/discussions",
"Documentation": "https://github.com/microsoft/nexamem/blob/main/README.md",
"Homepage": "https://github.com/microsoft/nexamem",
"Repository": "https://github.com/microsoft/nexamem.git"
},
"split_keywords": [
"aimemory",
" nexamem",
" chatbot",
" conversation",
" memory",
" llm",
" ai",
" redis",
" azure",
" governance",
" policy",
" audit",
" compliance",
" pii",
" content-processing",
" enterprise",
" async",
" channels"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "d8fdec30df7ba75af0c882fddf371334d832ae035193399fca3f427bc2aa5a6c",
"md5": "87c999e244461991b459dd05d0c00927",
"sha256": "ccb51a65df295e75dec1042555ee105896c8b36abd084e5588c627efdc9543c4"
},
"downloads": -1,
"filename": "nexamem-0.4.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "87c999e244461991b459dd05d0c00927",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 56774,
"upload_time": "2025-07-14T15:30:57",
"upload_time_iso_8601": "2025-07-14T15:30:57.754143Z",
"url": "https://files.pythonhosted.org/packages/d8/fd/ec30df7ba75af0c882fddf371334d832ae035193399fca3f427bc2aa5a6c/nexamem-0.4.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "0f1066871e92d7896aa67961d7e86b5b7f821b675bc2cab95e8224dd1ca3d676",
"md5": "be9d6c4e3687e31f839b29f1dab396d9",
"sha256": "ead7e7f6bea69dd7ca85a1e51808d3ea2e3567b440b08b89fcfe5b7193c693d2"
},
"downloads": -1,
"filename": "nexamem-0.4.0.tar.gz",
"has_sig": false,
"md5_digest": "be9d6c4e3687e31f839b29f1dab396d9",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 59950,
"upload_time": "2025-07-14T15:30:59",
"upload_time_iso_8601": "2025-07-14T15:30:59.248981Z",
"url": "https://files.pythonhosted.org/packages/0f/10/66871e92d7896aa67961d7e86b5b7f821b675bc2cab95e8224dd1ca3d676/nexamem-0.4.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-14 15:30:59",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "microsoft",
"github_project": "nexamem",
"github_not_found": true,
"lcname": "nexamem"
}
