Clean and organize project structure
Major reorganization: - Created scripts/ directory for all utility scripts - Created config/ directory for configuration files - Moved all test files to tests/ directory - Updated all script paths to work with new structure - Updated README.md with new project structure diagram New structure: ├── src/ # Source code (API + MCP) ├── scripts/ # Utility scripts (start-*.sh, docs_server.py, etc.) ├── tests/ # All test files and debug utilities ├── config/ # Configuration files (JSON, Caddy config) ├── docs/ # Documentation website └── logs/ # Log files All scripts updated to use relative paths from project root. Documentation updated with new folder structure. 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Docker Config Backup
174
tests/simple_test.py
Normal file
174
tests/simple_test.py
Normal file
@@ -0,0 +1,174 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Simple test of LangMem API core functionality
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import asyncpg
|
||||
import httpx
|
||||
import json
|
||||
from datetime import datetime
|
||||
from uuid import uuid4
|
||||
|
||||
# Configuration
|
||||
OLLAMA_URL = "http://localhost:11434"
|
||||
SUPABASE_DB_URL = "postgresql://postgres:CzkaYmRvc26Y@localhost:5435/postgres"
|
||||
API_KEY = "langmem_api_key_2025"
|
||||
|
||||
async def get_embedding(text: str):
|
||||
"""Generate embedding using Ollama"""
|
||||
try:
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.post(
|
||||
f"{OLLAMA_URL}/api/embeddings",
|
||||
json={
|
||||
"model": "nomic-embed-text",
|
||||
"prompt": text
|
||||
},
|
||||
timeout=30.0
|
||||
)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
return data["embedding"]
|
||||
except Exception as e:
|
||||
print(f"❌ Error generating embedding: {e}")
|
||||
return None
|
||||
|
||||
async def test_database_connection():
|
||||
"""Test database connection"""
|
||||
try:
|
||||
conn = await asyncpg.connect(SUPABASE_DB_URL)
|
||||
result = await conn.fetchval("SELECT 1")
|
||||
await conn.close()
|
||||
print(f"✅ Database connection successful: {result}")
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"❌ Database connection failed: {e}")
|
||||
return False
|
||||
|
||||
async def test_ollama_connection():
|
||||
"""Test Ollama connection"""
|
||||
try:
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.get(f"{OLLAMA_URL}/api/tags")
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
models = [model["name"] for model in data.get("models", [])]
|
||||
print(f"✅ Ollama connection successful, models: {models}")
|
||||
return True
|
||||
else:
|
||||
print(f"❌ Ollama connection failed: {response.status_code}")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"❌ Ollama connection failed: {e}")
|
||||
return False
|
||||
|
||||
async def test_embedding_generation():
|
||||
"""Test embedding generation"""
|
||||
test_text = "This is a test memory for the LangMem system"
|
||||
print(f"🧪 Testing embedding generation for: '{test_text}'")
|
||||
|
||||
embedding = await get_embedding(test_text)
|
||||
if embedding:
|
||||
print(f"✅ Embedding generated successfully, dimension: {len(embedding)}")
|
||||
return embedding
|
||||
else:
|
||||
print("❌ Failed to generate embedding")
|
||||
return None
|
||||
|
||||
async def test_vector_storage():
|
||||
"""Test vector storage in Supabase"""
|
||||
try:
|
||||
conn = await asyncpg.connect(SUPABASE_DB_URL)
|
||||
|
||||
# Create test table
|
||||
await conn.execute("""
|
||||
CREATE TABLE IF NOT EXISTS test_langmem_documents (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
content TEXT NOT NULL,
|
||||
embedding vector(768),
|
||||
user_id TEXT NOT NULL,
|
||||
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
|
||||
);
|
||||
""")
|
||||
|
||||
# Generate test embedding
|
||||
test_content = "FastAPI is a modern web framework for building APIs with Python"
|
||||
embedding = await get_embedding(test_content)
|
||||
|
||||
if not embedding:
|
||||
print("❌ Cannot test vector storage without embedding")
|
||||
return False
|
||||
|
||||
# Store test document
|
||||
doc_id = uuid4()
|
||||
await conn.execute("""
|
||||
INSERT INTO test_langmem_documents (id, content, embedding, user_id)
|
||||
VALUES ($1, $2, $3, $4)
|
||||
""", doc_id, test_content, str(embedding), "test_user")
|
||||
|
||||
# Test similarity search
|
||||
query_embedding = await get_embedding("Python web framework")
|
||||
if query_embedding:
|
||||
results = await conn.fetch("""
|
||||
SELECT id, content, 1 - (embedding <=> $1) as similarity
|
||||
FROM test_langmem_documents
|
||||
WHERE user_id = 'test_user'
|
||||
ORDER BY embedding <=> $1
|
||||
LIMIT 5
|
||||
""", str(query_embedding))
|
||||
|
||||
print(f"✅ Vector storage and similarity search successful")
|
||||
for row in results:
|
||||
print(f" - Content: {row['content'][:50]}...")
|
||||
print(f" - Similarity: {row['similarity']:.3f}")
|
||||
|
||||
# Cleanup
|
||||
await conn.execute("DROP TABLE IF EXISTS test_langmem_documents")
|
||||
await conn.close()
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Vector storage test failed: {e}")
|
||||
return False
|
||||
|
||||
async def main():
|
||||
"""Run all tests"""
|
||||
print("🚀 Starting LangMem API Tests")
|
||||
print("=" * 50)
|
||||
|
||||
# Test 1: Database connection
|
||||
print("\n1. Testing Database Connection...")
|
||||
db_ok = await test_database_connection()
|
||||
|
||||
# Test 2: Ollama connection
|
||||
print("\n2. Testing Ollama Connection...")
|
||||
ollama_ok = await test_ollama_connection()
|
||||
|
||||
# Test 3: Embedding generation
|
||||
print("\n3. Testing Embedding Generation...")
|
||||
embedding_ok = await test_embedding_generation()
|
||||
|
||||
# Test 4: Vector storage
|
||||
print("\n4. Testing Vector Storage...")
|
||||
vector_ok = await test_vector_storage()
|
||||
|
||||
# Summary
|
||||
print("\n" + "=" * 50)
|
||||
print("📊 Test Results Summary:")
|
||||
print(f" Database Connection: {'✅ PASS' if db_ok else '❌ FAIL'}")
|
||||
print(f" Ollama Connection: {'✅ PASS' if ollama_ok else '❌ FAIL'}")
|
||||
print(f" Embedding Generation: {'✅ PASS' if embedding_ok else '❌ FAIL'}")
|
||||
print(f" Vector Storage: {'✅ PASS' if vector_ok else '❌ FAIL'}")
|
||||
|
||||
all_passed = all([db_ok, ollama_ok, embedding_ok, vector_ok])
|
||||
print(f"\n🎯 Overall Status: {'✅ ALL TESTS PASSED' if all_passed else '❌ SOME TESTS FAILED'}")
|
||||
|
||||
if all_passed:
|
||||
print("\n🎉 LangMem API core functionality is working!")
|
||||
print(" Ready to proceed with full API deployment.")
|
||||
else:
|
||||
print("\n⚠️ Some tests failed. Please check the configuration.")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user