leeworks-agents/SPARC
5
0
Fork 0
forked from 0xWheatyz/SPARC
Code Issues 10 Pull Requests 9 Activity

feat(backend): add response caching and user management

Browse Source 
Replace USE_DATABASE toggle with USE_CACHE for smarter LLM response handling:
- Add prompt hashing for efficient cache lookups
- Cache API responses in database to reduce token usage
- Always store responses for analytics (cache or fresh)

Add user authentication infrastructure:
- User table with bcrypt password hashing
- CRUD operations for user management
- Role-based access control (admin/user)

Dependencies: add bcrypt and PyJWT for auth

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
0xWheatyz
2026-03-14 13:40:34 -04:00
parent 0107691c90
commit af52107ed8
4 changed files with 442 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
SPARC/config.py
+9 -4
View File
@@ -13,10 +13,15 @@ api_key = os.getenv("API_KEY")
# OpenRouter API key for LLM analysis
openrouter_api_key = os.getenv("OPENROUTER_API_KEY")
# Database configuration
# Database configuration - all messages are stored in the database
# The database serves as both a persistent store and a cache layer
database_url = os.getenv("DATABASE_URL", "postgresql://postgres:postgres@localhost:5432/sparc")
# Toggle between database mode and API mode
# When True: stores all messages in database instead of sending to OpenRouter
# When False: sends messages to OpenRouter API as normal
# Cache configuration
# When enabled (default), the system checks the database for cached responses
# before making API calls, saving tokens and reducing latency
use_cache = os.getenv("USE_CACHE", "true").lower() in ("true", "1", "yes")
# Legacy compatibility - USE_DATABASE is deprecated, database is always used
# This variable is kept for backwards compatibility but has no effect
use_database = os.getenv("USE_DATABASE", "false").lower() in ("true", "1", "yes")
SPARC/database.py
+324 -4
View File
@@ -1,10 +1,12 @@
"""Database client for storing and retrieving LLM messages."""
"""Database client for storing and retrieving LLM messages and user authentication."""
import psycopg2
from psycopg2.extras import RealDictCursor
from typing import Dict, List, Optional
from datetime import datetime
import json
import hashlib
import bcrypt
class DatabaseClient:
@@ -43,10 +45,12 @@ class DatabaseClient:
analysis_type VARCHAR(50),
model VARCHAR(100),
prompt TEXT NOT NULL,
prompt_hash VARCHAR(64),
response TEXT,
metadata JSONB,
token_usage JSONB,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
is_cached BOOLEAN DEFAULT FALSE
)
""")
@@ -62,8 +66,109 @@ class DatabaseClient:
ON llm_messages(company_name)
""")
# Add prompt_hash and is_cached columns if they don't exist (for existing tables)
# This must run BEFORE creating the index on prompt_hash
cursor.execute("""
DO $$
BEGIN
IF NOT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'llm_messages' AND column_name = 'prompt_hash'
) THEN
ALTER TABLE llm_messages ADD COLUMN prompt_hash VARCHAR(64);
END IF;
IF NOT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'llm_messages' AND column_name = 'is_cached'
) THEN
ALTER TABLE llm_messages ADD COLUMN is_cached BOOLEAN DEFAULT FALSE;
END IF;
END $$;
""")
# Create index on prompt_hash for cache lookups
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_messages_prompt_hash
ON llm_messages(prompt_hash)
""")
# Create users table for authentication
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (
id SERIAL PRIMARY KEY,
email VARCHAR(255) UNIQUE NOT NULL,
password_hash VARCHAR(255) NOT NULL,
role VARCHAR(20) DEFAULT 'user' CHECK (role IN ('admin', 'user')),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# Create index on email for fast lookups
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_users_email
ON users(email)
""")
self.conn.commit()
@staticmethod
def hash_prompt(prompt: str) -> str:
"""Generate a hash of the prompt for cache lookups.
Args:
prompt: The prompt text to hash
Returns:
SHA-256 hash of the prompt
"""
return hashlib.sha256(prompt.encode()).hexdigest()
def get_cached_response(
self,
prompt: str,
company_name: Optional[str] = None,
analysis_type: Optional[str] = None,
) -> Optional[Dict]:
"""Look up a cached response for a given prompt.
Args:
prompt: The prompt to look up
company_name: Optional company name filter
analysis_type: Optional analysis type filter
Returns:
Cached message dict if found, None otherwise
"""
self.connect()
prompt_hash = self.hash_prompt(prompt)
query = """
SELECT * FROM llm_messages
WHERE prompt_hash = %s
AND response IS NOT NULL
AND response NOT LIKE '[DATABASE MODE]%%'
AND response NOT LIKE '[TEST MODE]%%'
AND response NOT LIKE '[NO API]%%'
"""
params = [prompt_hash]
if company_name:
query += " AND company_name = %s"
params.append(company_name)
if analysis_type:
query += " AND analysis_type = %s"
params.append(analysis_type)
query += " ORDER BY timestamp DESC LIMIT 1"
with self.conn.cursor(cursor_factory=RealDictCursor) as cursor:
cursor.execute(query, params)
result = cursor.fetchone()
return dict(result) if result else None
def store_message(
self,
prompt: str,
@@ -73,6 +178,7 @@ class DatabaseClient:
model: Optional[str] = None,
metadata: Optional[Dict] = None,
token_usage: Optional[Dict] = None,
is_cached: bool = False,
) -> int:
"""Store an LLM message exchange in the database.
@@ -84,28 +190,33 @@ class DatabaseClient:
model: Model identifier used
metadata: Additional metadata as dict
token_usage: Token usage information
is_cached: Whether this response was served from cache
Returns:
The ID of the inserted record
"""
self.connect()
prompt_hash = self.hash_prompt(prompt)
with self.conn.cursor() as cursor:
cursor.execute(
"""
INSERT INTO llm_messages
(prompt, response, company_name, analysis_type, model, metadata, token_usage)
VALUES (%s, %s, %s, %s, %s, %s, %s)
(prompt, prompt_hash, response, company_name, analysis_type, model, metadata, token_usage, is_cached)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
RETURNING id
""",
(
prompt,
prompt_hash,
response,
company_name,
analysis_type,
model,
json.dumps(metadata) if metadata else None,
json.dumps(token_usage) if token_usage else None,
is_cached,
),
)
@@ -208,3 +319,212 @@ class DatabaseClient:
"by_type": [dict(row) for row in by_type],
"period_days": days,
}
# User Authentication Methods
@staticmethod
def hash_password(password: str) -> str:
"""Hash a password using bcrypt.
Args:
password: Plain text password
Returns:
Hashed password string
"""
return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
@staticmethod
def verify_password(password: str, password_hash: str) -> bool:
"""Verify a password against its hash.
Args:
password: Plain text password
password_hash: Stored hash
Returns:
True if password matches
"""
return bcrypt.checkpw(password.encode(), password_hash.encode())
def create_user(
self,
email: str,
password: str,
role: str = "user",
) -> Optional[Dict]:
"""Create a new user.
Args:
email: User email
password: Plain text password
role: User role ('admin' or 'user')
Returns:
Created user dict or None if email exists
"""
self.connect()
password_hash = self.hash_password(password)
try:
with self.conn.cursor(cursor_factory=RealDictCursor) as cursor:
cursor.execute(
"""
INSERT INTO users (email, password_hash, role)
VALUES (%s, %s, %s)
RETURNING id, email, role, created_at
""",
(email, password_hash, role),
)
user = cursor.fetchone()
self.conn.commit()
return dict(user) if user else None
except psycopg2.errors.UniqueViolation:
self.conn.rollback()
return None
def authenticate_user(self, email: str, password: str) -> Optional[Dict]:
"""Authenticate a user by email and password.
Args:
email: User email
password: Plain text password
Returns:
User dict if authenticated, None otherwise
"""
self.connect()
with self.conn.cursor(cursor_factory=RealDictCursor) as cursor:
cursor.execute(
"SELECT * FROM users WHERE email = %s",
(email,),
)
user = cursor.fetchone()
if user and self.verify_password(password, user["password_hash"]):
return {
"id": user["id"],
"email": user["email"],
"role": user["role"],
"created_at": user["created_at"],
}
return None
def get_user_by_id(self, user_id: int) -> Optional[Dict]:
"""Get a user by ID.
Args:
user_id: User ID
Returns:
User dict or None
"""
self.connect()
with self.conn.cursor(cursor_factory=RealDictCursor) as cursor:
cursor.execute(
"SELECT id, email, role, created_at FROM users WHERE id = %s",
(user_id,),
)
user = cursor.fetchone()
return dict(user) if user else None
def get_user_by_email(self, email: str) -> Optional[Dict]:
"""Get a user by email.
Args:
email: User email
Returns:
User dict or None
"""
self.connect()
with self.conn.cursor(cursor_factory=RealDictCursor) as cursor:
cursor.execute(
"SELECT id, email, role, created_at FROM users WHERE email = %s",
(email,),
)
user = cursor.fetchone()
return dict(user) if user else None
def get_all_users(self, limit: int = 100, offset: int = 0) -> List[Dict]:
"""Get all users (admin only).
Args:
limit: Maximum number of users
offset: Offset for pagination
Returns:
List of user dicts
"""
self.connect()
with self.conn.cursor(cursor_factory=RealDictCursor) as cursor:
cursor.execute(
"""
SELECT id, email, role, created_at
FROM users
ORDER BY created_at DESC
LIMIT %s OFFSET %s
""",
(limit, offset),
)
return [dict(row) for row in cursor.fetchall()]
def update_user_role(self, user_id: int, role: str) -> Optional[Dict]:
"""Update a user's role (admin only).
Args:
user_id: User ID
role: New role ('admin' or 'user')
Returns:
Updated user dict or None
"""
self.connect()
with self.conn.cursor(cursor_factory=RealDictCursor) as cursor:
cursor.execute(
"""
UPDATE users
SET role = %s, updated_at = CURRENT_TIMESTAMP
WHERE id = %s
RETURNING id, email, role, created_at
""",
(role, user_id),
)
user = cursor.fetchone()
self.conn.commit()
return dict(user) if user else None
def delete_user(self, user_id: int) -> bool:
"""Delete a user (admin only).
Args:
user_id: User ID
Returns:
True if deleted
"""
self.connect()
with self.conn.cursor() as cursor:
cursor.execute("DELETE FROM users WHERE id = %s", (user_id,))
deleted = cursor.rowcount > 0
self.conn.commit()
return deleted
def get_user_count(self) -> int:
"""Get total user count.
Returns:
Number of users
"""
self.connect()
with self.conn.cursor() as cursor:
cursor.execute("SELECT COUNT(*) FROM users")
return cursor.fetchone()[0]
SPARC/llm.py
+107 -69
View File
@@ -9,31 +9,29 @@ from typing import Dict
class LLMAnalyzer:
"""Handles LLM-based analysis of patent content."""
def __init__(self, api_key: str | None = None, test_mode: bool = False, use_database: bool | None = None):
def __init__(self, api_key: str | None = None, test_mode: bool = False, use_cache: bool | None = None):
"""Initialize the LLM analyzer.
Args:
api_key: OpenRouter API key. If None, will attempt to load from config.
test_mode: If True, print prompts instead of making API calls
use_database: If True, store messages in database instead of calling API.
If None, will use config.use_database
use_cache: If True, check database cache before making API calls.
If None, uses config.use_cache (default: True)
"""
self.test_mode = test_mode
self.use_database = use_database if use_database is not None else config.use_database
self.db_client = None
self.use_cache = use_cache if use_cache is not None else config.use_cache
self.model = "anthropic/claude-3.5-sonnet"
# Initialize database client if in database mode
if self.use_database:
self.db_client = DatabaseClient(config.database_url)
self.db_client.initialize_schema()
# Always initialize database client for storage and caching
self.db_client = DatabaseClient(config.database_url)
self.db_client.initialize_schema()
# Initialize OpenRouter client if not in database mode
if (api_key or config.openrouter_api_key) and not test_mode and not self.use_database:
# Initialize OpenRouter client if API key is available
if (api_key or config.openrouter_api_key) and not test_mode:
self.client = OpenAI(
api_key=api_key or config.openrouter_api_key,
base_url="https://openrouter.ai/api/v1"
)
self.model = "anthropic/claude-3.5-sonnet"
else:
self.client = None
@@ -68,22 +66,31 @@ Provide a concise analysis (2-3 paragraphs) focusing on what this patent reveals
print("=" * 80)
return "[TEST MODE - No API call made]"
# Database mode: store the prompt and return a placeholder response
if self.use_database:
response_text = "[DATABASE MODE] Message stored for testing/analytics. Enable API mode to get actual analysis."
self.db_client.store_message(
# Check cache first
if self.use_cache:
cached = self.db_client.get_cached_response(
prompt=prompt,
response=response_text,
company_name=company_name,
analysis_type="single_patent",
model=self.model if hasattr(self, 'model') else None,
metadata={"patent_content_length": len(patent_content)}
analysis_type="single_patent"
)
if cached:
# Log the cache hit
self.db_client.store_message(
prompt=prompt,
response=cached["response"],
company_name=company_name,
analysis_type="single_patent",
model=self.model,
metadata={
"patent_content_length": len(patent_content),
"cache_hit": True,
"original_message_id": cached["id"]
},
is_cached=True
)
return cached["response"]
return response_text
# API mode: send to OpenRouter
# Call API if no cache hit and client is available
if self.client:
response = self.client.chat.completions.create(
model=self.model,
@@ -92,23 +99,34 @@ Provide a concise analysis (2-3 paragraphs) focusing on what this patent reveals
)
response_text = response.choices[0].message.content
# Store in database if db_client is available (for logging even in API mode)
if self.db_client:
self.db_client.store_message(
prompt=prompt,
response=response_text,
company_name=company_name,
analysis_type="single_patent",
model=self.model,
metadata={"patent_content_length": len(patent_content)},
token_usage={
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens,
"total_tokens": response.usage.total_tokens
} if hasattr(response, 'usage') else None
)
# Store in database for future cache lookups
self.db_client.store_message(
prompt=prompt,
response=response_text,
company_name=company_name,
analysis_type="single_patent",
model=self.model,
metadata={"patent_content_length": len(patent_content)},
token_usage={
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens,
"total_tokens": response.usage.total_tokens
} if hasattr(response, 'usage') else None
)
return response_text
# No API client available - store prompt for later processing
placeholder = "[NO API] Prompt stored in database. Configure OPENROUTER_API_KEY to enable analysis."
self.db_client.store_message(
prompt=prompt,
response=placeholder,
company_name=company_name,
analysis_type="single_patent",
model=self.model,
metadata={"patent_content_length": len(patent_content), "pending": True}
)
return placeholder
def analyze_patent_portfolio(
self, patents_data: list[Dict[str, str]], company_name: str
@@ -150,46 +168,54 @@ Provide a comprehensive analysis (4-5 paragraphs) with a final verdict on the co
print(prompt)
return "[TEST MODE]"
# Database mode: store the prompt and return a placeholder response
if self.use_database:
response_text = "[DATABASE MODE] Message stored for testing/analytics. Enable API mode to get actual analysis."
metadata = {
"patent_count": len(patents_data),
"patent_ids": [p['patent_id'] for p in patents_data]
}
self.db_client.store_message(
# Check cache first
if self.use_cache:
cached = self.db_client.get_cached_response(
prompt=prompt,
response=response_text,
company_name=company_name,
analysis_type="portfolio",
model=self.model if hasattr(self, 'model') else None,
metadata={
"patent_count": len(patents_data),
"patent_ids": [p['patent_id'] for p in patents_data]
}
analysis_type="portfolio"
)
if cached:
# Log the cache hit
self.db_client.store_message(
prompt=prompt,
response=cached["response"],
company_name=company_name,
analysis_type="portfolio",
model=self.model,
metadata={
**metadata,
"cache_hit": True,
"original_message_id": cached["id"]
},
is_cached=True
)
return cached["response"]
return response_text
# Call API if no cache hit and client is available
if self.client:
try:
response = self.client.chat.completions.create(
model=self.model,
max_tokens=2048,
messages=[{"role": "user", "content": prompt}],
)
# API mode: send to OpenRouter
try:
response = self.client.chat.completions.create(
model=self.model,
max_tokens=2048,
messages=[{"role": "user", "content": prompt}],
)
response_text = response.choices[0].message.content
response_text = response.choices[0].message.content
# Store in database if db_client is available (for logging even in API mode)
if self.db_client:
# Store in database for future cache lookups
self.db_client.store_message(
prompt=prompt,
response=response_text,
company_name=company_name,
analysis_type="portfolio",
model=self.model,
metadata={
"patent_count": len(patents_data),
"patent_ids": [p['patent_id'] for p in patents_data]
},
metadata=metadata,
token_usage={
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens,
@@ -197,7 +223,19 @@ Provide a comprehensive analysis (4-5 paragraphs) with a final verdict on the co
} if hasattr(response, 'usage') else None
)
return response_text
except AttributeError:
return prompt
return response_text
except AttributeError:
return prompt
# No API client available - store prompt for later processing
placeholder = "[NO API] Prompt stored in database. Configure OPENROUTER_API_KEY to enable analysis."
self.db_client.store_message(
prompt=prompt,
response=placeholder,
company_name=company_name,
analysis_type="portfolio",
model=self.model,
metadata={**metadata, "pending": True}
)
return placeholder
requirements.txt
+2 -2
View File
@@ -9,7 +9,7 @@ psycopg2-binary
fastapi
uvicorn[standard]
httpx
streamlit
plotly
numpy
pandas
bcrypt
PyJWT