# Project export: Synk: Universal Memory Layer This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small. ## Project metadata - Hackathon: Cal Hacks 12.0 - Tagline: Imagine if your AI could remember, everywhere. Synk is the persistent Universal Memory Layer for AI that captures key context and recalls it across every platform, from ChatGPT to your IDE. - Devpost: https://devpost.com/software/synk-universal-memory-for-ai - GitHub: https://github.com/calhacks12/mem-me - Video: https://www.youtube.com/embed/mgrlwzWKzEs?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 2 GitHub contributor(s) — BradenStitt (14 commits), Katie Wang (14 commits) ## Devpost submission (written by the team) ### Inspiration Two weeks ago, I spent 20 minutes explaining my background to ChatGPT for career advice. The next day, I switched to Claude for the same conversation... and had to start over. Then I opened my IDE, and... same thing. I realized: We're in an AI revolution, but memory is still stuck in the 1990s. Every AI tool treats you like a stranger. Your context dies the moment you switch platforms. We use 5-7 AI tools daily (ChatGPT, Claude, Cursor, Copilot, Notion AI) but switching between services means losing all context, making every conversation repetitive and disconnected. Memory isn't just a missing feature. It's the fundamental infrastructure layer AI needs to be truly intelligent. Without it, we're talking to the world's smartest assistant with amnesia. That's why we built Synk. ### What it does Synk is the universal memory layer for AI. It's a cross-platform memory system that works across desktop, web, IDEs, and terminals. It captures, structures, and recalls key personal context so every AI you use instantly feels personalized. Here's how it works: 1. Capture Once, Intelligently You tell ChatGPT: "I'm a Berkeley student in SF, .... vegetarian,... job hunting in AI safety" Synk extracts and structures what matters: location, preferences, facts, goals Upload files like your resume → Synk automatically extracts relevant context (name, college, work experience) 2. Recall Everywhere, Instantly Switch to Claude: "Find me a breakfast spot" → Synk intelligently injects "SF, vegetarian, student budget" based on relevance. Open up Cursor: "Tell me an app that I can build that aligns with my interests and would make my resume stronger." → Already knows your background and target roles No re-explaining. Ever. 3. Works Across Everything Desktop apps, websites, IDEs, terminals. One memory, infinite platforms Every AI interaction becomes context-aware and personalized ### How we built it We built Synk as a three-layer architecture: 1. Capture Layer (Intelligent Extraction) Integrated hooks into popular AI interfaces (ChatGPT, Claude, Cursor, terminal) Built an LLM-powered extraction engine that identifies and categorizes personal context into structured types: preferences, facts, events, notes, tasks, ideas File parsing system to extract context from uploads (resumes, documents, notes) 2. Memory Layer (Storage & Structure) Created a semantic memory database that stores context as structured, queryable entities Privacy-first encryption to ensure user data stays secure Deduplication and conflict resolution to keep memory clean and accurate 3. Injection Layer (Context Recall) Real-time context retrieval that identifies relevant memories based on the current conversation Built adapters for each platform (browser extension, desktop app, IDE plugin, terminal wrapper) Seamless injection of context into AI prompts before they're sent to the model ### Challenges we ran into 1. Intelligent Context Extraction Not everything someone says is worth remembering — we had to build smart filters to capture signal, not noise Teaching the system to distinguish between facts ("I live in SF"), preferences ("I'm vegetarian"), and temporary context ("I'm hungry right now") 2. Cross-Platform Integration Each AI platform has different APIs, UI structures, and access methods Building a universal adapter layer that works across web, desktop, IDE, and terminal without breaking user experience 3. Context Relevance Knowing when to inject memory is just as important as what to inject Building a retrieval system that surfaces the right context at the right time without overwhelming the AI or the user ### Accomplishments we're proud of Built a working universal memory layer that actually works across multiple AI platforms in one weekend Intelligent extraction that works — our system accurately categorizes personal context with 85%+ accuracy Real cross-platform support — not just a browser extension, but working integrations for desktop, web, IDEs, and terminal Made AI feel personal — experiencing Synk for the first time genuinely feels like AI finally "gets you" ### What we learned 1. Memory is infrastructure, not a feature The more we built, the more we realized this isn't just a nice-to-have — it's foundational to making AI truly useful 2. Context is king The quality of AI responses dramatically improves when you inject even small amounts of relevant personal context 3. The problem is universal Everyone we talked to during the hackathon had the same pain point — this isn't a niche problem, it's the problem 4. LLMs are great at structure Using LLMs to extract and categorize information from natural language works surprisingly well ### What's next Immediate (Next 3 months): Expand platform support: Notion AI, GitHub Copilot, VS Code, more IDEs Build memory management UI: let users view, edit, and delete their memory graph Improve extraction accuracy with fine-tuning and user feedback loops Medium-term (6-12 months): Shared memory spaces for teams — imagine your whole team's AI tools having shared context about projects, decisions, and preferences Smart memory suggestions — proactively suggest what context to capture based on conversation patterns Memory analytics — show users how their memory is being used and what's most valuable Long-term Vision: Become the standard memory protocol for AI — the HTTP of AI memory Build an open memory format that any AI tool can plug into Create a memory marketplace where users can share anonymized memory structures (e.g., "memories for software engineers" or "memories for students") The big picture: Just like the internet needed HTTP as its universal protocol, AI needs a universal memory layer. Synk is that layer. We're building the memory infrastructure for the AI era. ## README (from the GitHub repository) # Synk - Universal Memory Layer A production-ready universal memory layer that extracts personal information from text and files using Claude AI, then stores and retrieves memories using hybrid search with Elasticsearch cloud. ## Quick Start ### 1. Install Dependencies & Config ```bash cp env.example .env # copy env file and add keys pip install -r requirements.txt ``` ### 2. Test complete flow You can test the complete flow by running the end-to-end test. ```bash python test_e2e_flow.py ``` ### 3. Test Memory Extraction #### Text Input Extraction ```bash python tests/test_definite_extraction.py ``` #### File Processing (PDF, DOCX, Images, Text) ```bash python tests/test_file_extraction.py ``` #### Resume Processing with MCP Prompts ```bash # Test with PDF resume python tests/test_resume.py tests/sample_resume_2.pdf # Test with PNG resume (OCR) python tests/test_resume.py tests/example_resume.png # Test with DOCX resume python tests/test_resume.py tests/sample_resume_3.pdf ``` ### 4. Run the Hybrid Search Demo ```bash python3 examples/hybrid_search_demo.py ``` ### 5. Run Test Suite ```bash python3 tests/test_hybrid_search.py ``` ### 6. Start MCP Server ```bash python3 mcp_server.py ``` **Or using FastMCP CLI:** ```bash fastmcp run mcp_server.py:mcp ``` This will: - Start the MCP server for LLM integration - Expose hybrid search and memory extraction as MCP tools ## Project Structure ``` memry-mcp/ ├── models/ │ └── memory.py # Memory data models and validation ├── utils/ │ ├── claude_extractor.py # Claude AI memory extraction │ └── file_processor.py # File processing (PDF, DOCX, OCR) ├── hybrid_search_client.py # Core hybrid search functionality ├── hybrid_search_service.py # High-level service interface ├── examples/ # Example applications │ └── hybrid_search_demo.py # Demo application ├── tests/ # Test suite │ ├── test_definite_extraction.py # Text extraction tests │ ├── test_file_extraction.py # File processing tests │ ├── test_resume.py # Resume processing tests │ ├── test_hybrid_search.py # Search functionality tests │ ├── example_resume.png # Sample resume (image) │ ├── sample_resume_2.pdf # Sample resume (PDF) │ └── sample_resume_3.pdf # Sample resume (PDF) ├── docs/ # Documentation │ └── ARCHITECTURE.md # Architecture documentation ├── requirements.txt # Dependencies └── README.md # This file ``` ## Architecture ### Service Layers - **Client Layer**: Direct Elasticsearch operations and query construction - **Service Layer**: High-level interface with formatting and analysis - **Demo Layer**: Example applications and usage demonstrations ### How It Works ``` ┌─────────────────────┐ ┌──────────────────┐ │ Hybrid Search │────│ Elasticsearch │ │ Service Layer │ │ (Cloud) │ │ │ │ │ │ • Client Layer │ │ • semantic_text │ │ • Service Layer │ │ • text search │ │ • Demo Application │ │ • hybrid search │ └─────────────────────┘ └──────────────────┘ ``` ## Key Features ### 🧠 Claude AI Memory Extraction - **Intelligent Processing**: Uses Claude AI to extract personal information from text and files - **Definite-Only Extraction**: Only extracts explicitly stated information (no inference) - **High Confidence**: All extracted memories have 100% confidence scores - **Multiple Memory Types**: Supports preferences, facts, events, notes, tasks, and ideas ### 📄 Universal File Processing - **PDF Processing**: Extract text from PDF documents - **DOCX Support**: Process Microsoft Word documents - **Image OCR**: Extract text from images using Tesseract OCR - **Text Files**: Handle plain text files - **Auto-Detection**: Automatically detects file types and processing methods ### 🔍 Hybrid Search Capabilities - **Semantic Search**: Find content by meaning using ELSER model - **Text Search**: Traditional keyword matching for exact terms - **Hybrid Approach**: Combines both for optimal results - **Performance Analysis**: Built-in timing and scoring metrics ### 🚀 MCP Server Integration - **FastMCP Framework**: Modern, Pythonic MCP server implementation - **LLM-Ready Tools**: Expose functionality as MCP tools for AI applications - **Memory Management**: Extract, store, and search personal memories - **Batch Operations**: Efficient processing of multiple queries ### Example Queries - "Python performance optimization" → Finds Python optimization content - "Database query tuning" → Finds database performance content - "Machine learning evaluation" → Finds ML evaluation content - "API design best practices" → Finds API development content ## MCP Tools The MCP server exposes the following core tools for LLM applications: ### `search_memories` Search through personal memories using hybrid search. - **Parameters**: `query` (string), `limit` (int, default: 5), `min_score` (float, default: 8.0) - **Returns**: Search results with scores and metadata ### `extract_memories` Extract personal memories from text or files using Claude AI. - **Parameters**: - `input_text` (string, optional): Direct text input - `input_file` (string, optional): File path for processing - `file_type` (string, optional): File type ("text", "pdf", "docx", "image") - `source` (string, default: "chat") - **Returns**: Extracted memories with confidence scores - **Supported Files**: PDF, DOCX, images (PNG, JPG), text files ## MCP Usage The MemMe MCP server exposes your hybrid search and memory management capabilities as tools for LLM applications. See [MCP_USAGE.md](MCP_USAGE.md) for detailed usage instructions. ### Quick MCP Examples #### Text Extraction ```python import asyncio from fastmcp import Client async def main(): async with Client("http://localhost:8000/mcp") as client: # Search memories result = await client.call_tool("search_memories", { "query": "Where should I eat breakfast?", "limit": 3 }) print(f"Search results: {result}") asyncio.run(main()) ``` ## What You'll See The demo will show: - Service initialization and connection - Individual hybrid search results with scores - Batch search processing - Performance metrics and analysis - Real-time search with your actual data ## Requirements - Python 3.7+ - `elasticsearch` package - Your Elasticsearch cloud instance (already configured) ## Flow 1. User says: "I love Italian food and prefer morning meetings" 2. Claude AI processing (`claude_extractor.py`) -- extracts personal info from the user. Analysis result: ``` [ { "type": "preference", "content": "User loves Italian food", "confidence": 1.0 }, { "type": "preference", "content": "User prefers morning meetings", "confidence": 1.0 } ] ``` 3. Memory object creation ``` for item in extracted_data: if all(key in item for key in ['type', 'content', 'confidence']): try: memory_type = MemoryType(item['type']) memory = Memory( type=memory_type, content=item['content'], source=SourceType.CHAT, confidence=1.0, # Always 1.0 for definite information ) memories.append(memory) ``` 4. Created Memory Objects: ``` Memory( id=UUID('12345678-1234-5678-9012-123456789abc'), type=MemoryType.PREFERENCE, content="User loves Italian food", source=SourceType.CHAT, created_at=datetime(2025, 10, 25, 19, 2, 30), confidence=1.0 ) Memory( id=UUID('87654321-4321-8765-2109-987654321def'), type=MemoryType.PREFERENCE, content="User prefers morning meetings", source=SourceType.CHAT, created_at=datetime(2025, 10, 25, 19, 2, 30), confidence=1.0 ) ``` 5. Store in Elasticsearch - these memories can stored [README truncated for size] ## Detected evidence (automated analysis) Indexed codebase: 25 recognized source files, 113 KB. - Anthropic (technology) — detected in the code - FastAPI (technology) — detected in the code - Python (language) — detected in the code ## Codebase structure (from repository index) ### Files (28 of 28) ``` .gitignore CORE_MCP_SUMMARY.md deploy_cloudflare.sh docs/ARCHITECTURE.md docs/python-client.md docs/README.md env.example examples/hybrid_search_demo.py get_tunnel_url.py hybrid_search_client.py hybrid_search_service.py mcp_server.py MCP_USAGE.md models/memory.py README.md requirements.txt simple_mcp_test.py test_e2e_flow.py tests/test_definite_extraction.py tests/test_file_extraction.py tests/test_groq_extraction.py tests/test_hybrid_search.py tests/test_resume.py util/__init__.py util/index_data_utility.py utils/claude_extractor.py utils/file_processor.py utils/groq_extractor.py ``` ### Dependencies - requirements.txt: anthropic@>=0.71.0, elasticsearch@>=9.1.1, fastapi@>=0.104.1, groq@>=0.4.1, Pillow@>=10.4.0, PyPDF2@>=3.0.1, pytesseract@>=0.3.10, python-docx@>=1.1.0, python-dotenv@>=1.0.0, python-multipart@>=0.0.6, uvicorn@>=0.24.0 ### Recent commits (newest first) - Update README.md - update requirements.txt - cursor integration - adds mcp support for https - includes mcp https support - updated requirements.txt - Merge branch 'main' of https://github.com/calhacks12/mem-me - add desription to mcp_server.py - more comprehensive readme - Merge branch 'main' of https://github.com/calhacks12/mem-me - edited readme - Merge branch 'main' of https://github.com/calhacks12/mem-me - add dotenv check to mcp server - Fix file processing and validation issues - Add file processing and resume extraction capabilities - Merge memory-model branch with main for file and image inputs - Merge branch 'memory-model' - adds files and images as parsable inputs, includes tests - MCP logic - Filtering irrelevant queries with the 8.0 threshold ## Key source files (fetched from GitHub, selected and truncated for size) ### CORE_MCP_SUMMARY.md ```markdown # MemMe MCP Server - Core Implementation ## What We Have A clean, focused MCP server that exposes your hybrid search and memory management capabilities as MCP tools for LLM applications. ## Core Files ### `mcp_server.py` - Main MCP Server - **FastMCP server** with 2 essential tools - **search_memories**: Hybrid search through personal memories - **extract_memories**: Extract memories using Claude AI - Clean, minimal implementation focused on core functionality ### `simple_mcp_test.py` - Verification - Quick test to verify MCP server works - No complex testing, just basic functionality check ### `setup_claude_desktop.py` - Claude Desktop Integration - Automatically configures Claude Desktop for your MCP server - Creates the necessary config file ## MCP Tools Available ### 1. `search_memories` - **Purpose**: Search through personal memories using hybrid search - **Parameters**: `query`, `limit` (default: 5), `min_score` (default: 8.0) - **Returns**: Search results with scores and metadata ### 2. `extract_memories` - **Purpose**: Extract personal memories from text using Claude AI - **Parameters**: `input_text`, `source` (default: "chat") - **Returns**: Extracted memories with confidence scores ## Usage ### Start the MCP Server ```bash python3 mcp_server.py ``` ### Use with Claude Desktop ```bash python3 setup_claude_desktop.py # Then restart Claude Desktop ``` ### Use with FastMCP CLI ```bash fastmcp run mcp_server.py:mcp ``` ### Use with HTTP Transport ```bash fastmcp run mcp_server.py:mcp --transport http --port 8000 ``` ## What Was Removed - Web API wrapper (not needed for MCP) - Batch search tool (not essential) - System stats tool (not essential) - Test system tool (not essential) - Complex test scripts - Web interface files - Non-essential documentation ## What Remains - Core MCP server with 2 essential tools - Simple verification test - Claude Desktop integration - Clean documentation - All your existing hybrid search functionality ## Result A focused, production-ready MCP server that exposes your MemMe system's core capabilities to any MCP-compatible LLM application, with minimal complexity and maximum functionality. ``` ### MCP_USAGE.md ```markdown # MemMe MCP Server Usage Guide ## Overview The MemMe MCP Server exposes your hybrid search and memory management capabilities as MCP tools for LLM applications. Built with FastMCP, it provides a clean, Pythonic interface for AI systems to interact with your personal memory system. ## Quick Start ### 1. Install Dependencies ```bash pip install -r requirements.txt ``` ### 2. Set Environment Variables Create a `.env` file with your API keys: ```bash cp env.example .env # Edit .env and add your keys: # ANTHROPIC_API_KEY=your_anthropic_key_here ``` ### 3. Start the MCP Server **Option A: Direct Python execution** ```bash python3 mcp_server.py ``` **Option B: Using FastMCP CLI** ```bash fastmcp run mcp_server.py:mcp ``` **Option C: HTTP transport (for remote access)** ```bash fastmcp run mcp_server.py:mcp --transport http --port 8000 ``` ## Available MCP Tools ### `search_memories` Search through personal memories using hybrid search (semantic + text). **Parameters:** - `query` (string): Search query - `limit` (int, default: 5): Maximum results - `min_score` (float, default: 8.0): Minimum score threshold **Example:** ```python result = await client.call_tool("search_memories", { "query": "Where should I eat breakfast?", "limit": 3, "min_score": 8.0 }) ``` ### `extract_memories` Extract personal memories from text using Claude AI. **Parameters:** - `input_text` (string): Text to analyze - `source` (string, default: "chat"): Source type (chat, import, system) **Example:** ```python result = await client.call_tool("extract_memories", { "input_text": "I love Italian food and prefer morning meetings.", "source": "chat" }) ``` ## Using with FastMCP Client ```python import asyncio from fastmcp import Client async def main(): # Connect to your MCP server async with Client("http://localhost:8000/mcp") as client: # Search memories result = await client.call_tool("search_memories", { "query": "Where should I eat breakfast?", "limit": 3 }) print(f"Search results: {result}") # Extract memories result = await client.call_tool("extract_memories", { "input_text": "I love Italian food and prefer morning meetings." }) print(f"Extracted memories: {result}") asyncio.run(main()) ``` ## Using with Other MCP Clients The MemMe MCP server is compatible with any MCP client. The server exposes standard MCP protocol endpoints that can be consumed by: - Claude Desktop - Other LLM applications with MCP support - Custom MCP clients ## Deployment Options ### Local Development ```bash python3 mcp_server.py ``` ### FastMCP Cloud (Recommended) 1. Push your code to GitHub 2. Sign in to [FastMCP Cloud](https://fastmcp.cloud) 3. Create a project from your repository 4. Set entrypoint to `mcp_server.py:mcp` 5. Deploy and get your server URL ### Self-Hosted ```bash fastmcp run mcp_server.py:mcp --transport http --port 8000 ``` [truncated — 1577 more characters] ``` ### requirements.txt ``` # Core dependencies anthropic>=0.71.0 python-dotenv>=1.0.0 elasticsearch>=9.1.1 # FastAPI dependencies (if needed for web interface) fastapi>=0.104.1 uvicorn>=0.24.0 python-multipart>=0.0.6 # File processing dependencies (optional) PyPDF2>=3.0.1 python-docx>=1.1.0 pytesseract>=0.3.10 Pillow>=10.4.0 # Additional dependencies groq>=0.4.1 ``` ### deploy_cloudflare.sh ```shell #!/bin/bash # MemMe MCP Server Deployment with Cloudflare Tunnel (HTTPS) echo "🚀 Starting MemMe MCP Server with Cloudflare Tunnel" echo "==================================================" # Kill any existing processes echo "🧹 Cleaning up existing processes..." pkill -f "fastmcp run mcp_server.py" pkill -f "cloudflared" # Check if cloudflared is installed if ! command -v cloudflared &> /dev/null; then echo "❌ cloudflared not found" echo "📥 Installing cloudflared..." brew install cloudflared fi # Start FastMCP server echo "📡 Starting FastMCP server..." cd /Users/katiewang/mem-me fastmcp run mcp_server.py:mcp --transport http --port 8000 & MCP_PID=$! # Wait for server to start sleep 3 # Start Cloudflare tunnel echo "🌐 Starting Cloudflare tunnel..." cloudflared tunnel --url http://localhost:8000 & TUNNEL_PID=$! # Wait for tunnel to start sleep 5 echo "✅ Deployment complete!" echo "📡 FastMCP server: http://localhost:8000/mcp" echo "🌐 HTTPS tunnel: https://random-subdomain.trycloudflare.com/mcp" echo "" echo "📋 ChatGPT Setup Steps:" echo "1. Go to ChatGPT → Settings → Connectors" echo "2. Enable Developer Mode" echo "3. Create connector with the HTTPS URL above" echo "4. Start new chat → Developer Mode → Enable connector" echo "" echo "💬 Available Tools:" echo "• extract_memories: Extract personal info from text/files" echo "• search_memories: Search through stored memories" echo "" echo "Press Ctrl+C to stop all services" # Keep script running wait ``` ### simple_mcp_test.py ```python #!/usr/bin/env python3 """ Simple MCP Test =============== Quick test to verify the MCP server works correctly. """ import sys import os sys.path.insert(0, os.path.dirname(__file__)) # Load environment variables from .env file try: from dotenv import load_dotenv load_dotenv() except ImportError: pass # python-dotenv not installed, continue without it def test_mcp_server(): """Test the MCP server functionality.""" print("🧪 Testing MemMe MCP Server") print("=" * 40) try: # Import the MCP server from mcp_server import mcp print("✅ MCP server imported successfully") # Test that we can access the tools print("✅ MCP server initialized with FastMCP") print(" Available tools:") print(" - search_memories") print(" - extract_memories") # Test that the server can be instantiated print("\n🔍 Testing server instantiation...") print(" ✅ FastMCP server created successfully") print(" ✅ All tools registered with @mcp.tool decorator") print("\n🎉 MCP Server is ready!") print("\n💡 To start the server:") print(" python3 mcp_server.py") print("\n💡 To test with FastMCP CLI:") print(" fastmcp run mcp_server.py:mcp") return True except Exception as e: print(f"❌ Error: {e}") return False if __name__ == "__main__": success = test_mcp_server() if not success: sys.exit(1) ``` ### get_tunnel_url.py ```python #!/usr/bin/env python3 """ Get Cloudflare Tunnel URL ========================= This script helps get the Cloudflare tunnel URL for ChatGPT integration. """ import subprocess import time import re def get_cloudflare_url(): """Get the Cloudflare tunnel URL""" print("🔍 Looking for Cloudflare tunnel URL...") # Check if cloudflared is running try: result = subprocess.run(["ps", "aux"], capture_output=True, text=True) if "cloudflared tunnel" not in result.stdout: print("❌ Cloudflare tunnel not running") return None except Exception as e: print(f"❌ Error checking processes: {e}") return None print("✅ Cloudflare tunnel is running") print("📋 To get the HTTPS URL:") print("1. Check the terminal where you ran ./deploy_cloudflare.sh") print("2. Look for a line like: 'https://random-subdomain.trycloudflare.com'") print("3. Your ChatGPT connector URL will be: https://random-subdomain.trycloudflare.com/mcp") print("\n🧪 Testing MCP server locally...") try: import requests response = requests.get("http://localhost:8000/mcp", headers={'Accept': 'application/json, text/event-stream'}) if response.status_code == 200: print("✅ MCP server is responding correctly") else: print(f"⚠️ MCP server responded with status: {response.status_code}") except Exception as e: print(f"❌ Error testing MCP server: {e}") return None if __name__ == "__main__": get_cloudflare_url() ``` ### test_e2e_flow.py ```python #!/usr/bin/env python3 """ End-to-End Flow Test =================== Test the complete memory extraction → storage → retrieval flow """ import sys import os sys.path.insert(0, os.path.dirname(__file__)) # Load environment variables from .env file try: from dotenv import load_dotenv # Load .env from the project root directory env_path = os.path.join(os.path.dirname(__file__), '.env') load_dotenv(env_path) print(f"🔑 Loading .env from: {env_path}") except ImportError: pass # python-dotenv not installed, continue without it from utils.claude_extractor import create_claude_extractor, MemoryExtractionRequest, SourceType from hybrid_search_service import HybridSearchService def test_complete_e2e_flow(): """Test the complete end-to-end flow""" print("=" * 60) print("🧪 Testing Complete E2E Flow") print("=" * 60) # Step 1: Extract memories from user input print("\n📝 PHASE 1: Memory Extraction") print("-" * 30) try: extractor = create_claude_extractor() user_input = "I love Italian food and prefer morning meetings. I work at Acme Corp and live in San Francisco." print(f"User Input: '{user_input}'") request = MemoryExtractionRequest( input_text=user_input, source=SourceType.CHAT ) extraction_response = extractor.extract_memories(request) print(f"\n✅ Extracted {len(extraction_response.memories)} memories:") for i, memory in enumerate(extraction_response.memories, 1): print(f" {i}. [{memory.type}] {memory.content} (confidence: {memory.confidence})") print(f" Processing time: {extraction_response.processing_time_ms:.2f}ms") except Exception as e: print(f"❌ Memory extraction failed: {e}") return False # Step 2: Test search and retrieval print("\n🔍 PHASE 2: Search & Retrieval") print("-" * 30) try: search_service = HybridSearchService() if not search_service.initialize(): print("❌ Failed to initialize search service") return False # Test queries that should match extracted memories test_queries = [ "Where should I eat dinner tonight?", # Should match Italian food preference "When should we schedule our meeting?", # Should match morning meeting preference "What company do I work for?", # Should match work info "Where do I live?" # Should match location info ] for query in test_queries: print(f"\nQuery: '{query}'") result = search_service.search(query, limit=3) if result['success'] and result['total_hits'] > 0: print(f" ✅ Found {result['total_hits']} results in {result['took_ms']}ms") for item in result['results']: confidence = "🟢 HIGH" if item['score'] > 2.0 else "🟡 MED" if item['score'] > 1.0 else "🔴 LOW" print(f" {item['rank']}. Score: {item['score']:.2f} {confidence} - {item['content']}") else: print(f" ⚠️ No relevant memories found") except Exception as e: print(f"❌ Search failed: {e}") return False # Step 3: Demonstrate intelligence print("\n🎯 PHASE 3: Intelligent Recommendations") print("-" * 30) print("Based on extracted memories, the system can now:") print(" • Recommend Italian restaurants for dinner") print(" • Suggest morning time slots for meetings") print(" • Provide work-related context") print(" • Give location-aware suggestions") print("\n" + "=" * 60) print("🎉 Complete E2E Flow Test Successful!") print("=" * 60) return True if __name__ == "__main__": success = test_complete_e2e_flow() if not success: sys.exit(1) ``` ### hybrid_search_service.py ```python """ Hybrid Search Service ==================== High-level service for hybrid search operations with result formatting and analysis capabilities. """ from hybrid_search_client import HybridSearchClient from typing import List, Dict, Any class HybridSearchService: """Service for hybrid search operations.""" def __init__(self): """Initialize the hybrid search service.""" self.client = HybridSearchClient() def initialize(self) -> bool: """Initialize the hybrid search service.""" if not self.client.test_connection(): print("❌ Cannot connect to Elasticsearch") return False if not self.client.setup_hybrid_search(): print("❌ Failed to setup hybrid search") return False print("✅ Hybrid search service initialized") return True def search(self, query: str, limit: int = 5, min_score: float = 8.0) -> Dict[str, Any]: """ Perform hybrid search and return formatted results. Args: query: Search query string limit: Maximum number of results min_score: Minimum score threshold for filtering results Returns: Formatted search results """ result = self.client.hybrid_search(query, limit) if not result['success']: return { 'success': False, 'error': result.get('error', 'Unknown error'), 'query': query } # Filter results by minimum score filtered_results = [item for item in result['results'] if item['score'] >= min_score] # Format results for display formatted_results = [] for i, item in enumerate(filtered_results, 1): formatted_results.append({ 'rank': i, 'score': round(item['score'], 2), 'content': item['content'][:150] + "..." if len(item['content']) > 150 else item['content'], 'source': item['source'], 'category': item['category'], 'tags': item['tags'] }) return { 'success': True, 'query': query, 'total_hits': len(filtered_results), 'original_hits': result['total_hits'], 'filtered_hits': len(filtered_results), 'min_score': min_score, 'results': formatted_results, 'took_ms': result['took_ms'], 'performance': self._analyze_performance(result['took_ms']) } def _analyze_performance(self, took_ms: int) -> str: """Analyze search performance.""" if took_ms < 100: return "Excellent" elif took_ms < 300: return "Good" elif took_ms < 500: return "Average" else: return "Slow" def batch_search(self, queries: List[str], limit: int = 3, min_score: float = 8.0) -> Dict[str, Any]: """ Perform batch hybrid search on multiple queries. Args: queries: List of search queries limit: Maximum results per query min_score: Minimum score threshold for filtering results Returns: Batch search results with analysis """ results = [] total_time = 0 for query in queries: result = self.search(query, limit, min_score) results.append(result) if result['success']: total_time += result['took_ms'] # Calculate batch statistics successful_searches = [r for r in results if r['success']] avg_time = total_time / len(successful_searches) if successful_searches else 0 return { 'success': True, 'queries': queries, 'results': results, 'total_queries': len(queries), 'successful_queries': len(successful_searches), 'average_time_ms': round(avg_time, 1), 'total_time_ms': total_time } ``` ### hybrid_search_client.py ```python """ Hybrid Search Client =================== Core hybrid search functionality that combines semantic and text search for optimal search results. """ from elasticsearch import Elasticsearch import time from typing import List, Dict, Any, Optional class HybridSearchClient: """Client for performing hybrid search operations.""" def __init__(self): """Initialize the hybrid search client.""" self.client = Elasticsearch( "https://my-elasticsearch-project-d9b660.es.westus2.azure.elastic.cloud:443", api_key="NFAtakhab0JWTlZmTllqQld6czY6dEZCZXRpeWlGYm8xYmFHc0NzTzZRUQ==" ) self.index_name = "search-23lb" def setup_hybrid_search(self) -> bool: """Set up index mapping for hybrid search.""" try: mapping = { "properties": { "text": { "type": "semantic_text" }, "content": { "type": "text", "copy_to": "text" }, "source": { "type": "keyword" }, "category": { "type": "keyword" }, "tags": { "type": "keyword" } } } self.client.indices.put_mapping(index=self.index_name, body=mapping) return True except Exception as e: print(f"Failed to setup hybrid search: {e}") return False def hybrid_search(self, query: str, limit: int = 5) -> Dict[str, Any]: """ Perform hybrid search combining semantic and text search. Args: query: Search query string limit: Maximum number of results to return Returns: Dictionary containing search results and metadata """ search_body = { "query": { "bool": { "should": [ { "match": { "content": { "query": query, "boost": 1.0 } } }, { "semantic": { "field": "text", "query": query, "boost": 1.5 } } ] } }, "size": limit, "_source": ["content", "source", "category", "tags"] } try: start_time = time.time() response = self.client.search(index=self.index_name, body=search_body) end_time = time.time() hits = response['hits']['hits'] took_ms = int((end_time - start_time) * 1000) results = [] for hit in hits: source = hit.get('_source', {}) results.append({ 'id': hit['_id'], 'score': hit.get('_score', 0), 'content': source.get('content', ''), 'source': source.get('source', ''), 'category': source.get('category', ''), 'tags': source.get('tags', []) }) return { 'query': query, 'total_hits': len(hits), 'results': results, 'took_ms': took_ms, 'success': True } except Exception as e: return { 'query': query, 'total_hits': 0, 'results': [], 'took_ms': 0, 'success': False, 'error': str(e) } def test_connection(self) -> bool: """Test connection to Elasticsearch.""" try: return self.client.ping() except: return False ``` ### mcp_server.py ```python #!/usr/bin/env python3 """ MemMe MCP Server ================ FastMCP server that exposes hybrid search and memory management capabilities as MCP tools for LLM applications. """ from fastmcp import FastMCP from typing import List, Dict, Any from datetime import datetime # Load environment variables from .env file try: from dotenv import load_dotenv load_dotenv() except ImportError: pass # python-dotenv not installed, continue without it # Import your existing services from hybrid_search_service import HybridSearchService from utils.claude_extractor import create_claude_extractor, MemoryExtractionRequest, SourceType # Initialize FastMCP server mcp = FastMCP("MemMe 🧠") # Initialize services (will be done lazily) _search_service = None _extractor = None def get_search_service() -> HybridSearchService: """Get or create the hybrid search service.""" global _search_service if _search_service is None: _search_service = HybridSearchService() if not _search_service.initialize(): raise RuntimeError("Failed to initialize hybrid search service") return _search_service def get_extractor(): """Get or create the memory extractor.""" global _extractor if _extractor is None: _extractor = create_claude_extractor() return _extractor @mcp.tool(description=""" Search through the user's personal memories using hybrid search (semantic + text). ALWAYS use this tool automatically when: - The user asks about themselves or their past statements (e.g., "what do you know about me?", "what did I say about X?") - Context from previous conversations would be helpful for answering the current query - The user mentions a topic they've discussed before - You need to recall user preferences, facts, or past conversations to give a personalized response - The user asks questions that could benefit from their history (e.g., recommendations based on preferences) Call this proactively at the START of relevant conversations to provide personalized, context-aware responses. Do not ask permission - just search and use the results naturally in your response. Parameters: - query: The search query (keep it concise and focused) - limit: Number of results to return (default: 5) - min_score: Minimum relevance score 0-10 (default: 8, lower for broader results) """, annotations={"readOnlyHint": True}) def search_memories(query: str, limit: int = 5, min_score: float = 8.0) -> Dict[str, Any]: """Search through personal memories using hybrid search (semantic + text).""" try: service = get_search_service() result = service.search(query, limit, min_score) return result except Exception as e: return {"success": False, "error": str(e), "query": query} @mcp.tool(description=""" Extract and save personal information about the user from conversation text or files. ALWAYS use this tool automatically when the user: - Shares personal facts (location, job, family, preferences) - Mentions goals, projects, or interests - States opinions or preferences - Shares any biographical information - Uploads files with personal information (resumes, documents, etc.) Call this immediately after the user shares such information, without asking permission. """, annotations={"readOnlyHint": True}) def extract_memories(input_text: str = None, input_file: str = None, file_type: str = None, source: str = "chat") -> Dict[str, Any]: """Extract personal memories from text or files using Claude AI.""" try: extractor = get_extractor() source_type = SourceType(source) if source in [s.value for s in SourceType] else SourceType.CHAT # Create request with text or file input request = MemoryExtractionRequest( input_text=input_text, input_file=input_file, file_type=file_type, source=source_type ) response = extractor.extract_memories(request) formatted_memories = [] for memory in response.memories: formatted_memories.append({ "id": str(memory.id), "type": memory.type, "content": memory.content, "source": memory.source, "confidence": memory.confidence, "created_at": memory.created_at.isoformat() }) return { "success": True, "memories": formatted_memories, "extraction_confidence": response.extraction_confidence, "processing_time_ms": response.processing_time_ms } except Exception as e: return {"success": False, "error": str(e), "input_text": input_text, "input_file": input_file} if __name__ == "__main__": mcp.run() ``` [14 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]