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

feat: add database mode for LLM message storage and analytics

Browse Source 
Implements a database mode that stores LLM prompts and responses in PostgreSQL
instead of making API calls. This enables:
- Testing without consuming API credits
- Collecting analytics on usage patterns
- Development and debugging workflows

Changes:
- Added DatabaseClient class for PostgreSQL operations
- Modified LLMAnalyzer to support database/API mode toggle
- Added USE_DATABASE config flag to switch between modes
- Included Docker Compose setup for PostgreSQL
- Added utility scripts for database init and analytics viewing
- Comprehensive documentation in DATABASE_MODE.md

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

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
0xWheatyz
2026-03-10 21:13:13 -04:00
parent 11a4aba46f
commit 44456cb073
11 changed files with 952 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
SPARC/llm.py
+89 -4
View File
@@ -2,22 +2,33 @@
from openai import OpenAI
from SPARC import config
from SPARC.database import DatabaseClient
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):
def __init__(self, api_key: str | None = None, test_mode: bool = False, use_database: 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
"""
self.test_mode = test_mode
self.use_database = use_database if use_database is not None else config.use_database
self.db_client = None
if (api_key or config.openrouter_api_key) and not test_mode:
# Initialize database client if in database mode
if self.use_database:
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:
self.client = OpenAI(
api_key=api_key or config.openrouter_api_key,
base_url="https://openrouter.ai/api/v1"
@@ -57,13 +68,47 @@ 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(
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)}
)
return response_text
# API mode: send to OpenRouter
if self.client:
response = self.client.chat.completions.create(
model=self.model,
max_tokens=1024,
messages=[{"role": "user", "content": prompt}],
)
return 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:
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
def analyze_patent_portfolio(
self, patents_data: list[Dict[str, str]], company_name: str
@@ -105,6 +150,25 @@ 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."
self.db_client.store_message(
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]
}
)
return response_text
# API mode: send to OpenRouter
try:
response = self.client.chat.completions.create(
model=self.model,
@@ -112,7 +176,28 @@ Provide a comprehensive analysis (4-5 paragraphs) with a final verdict on the co
messages=[{"role": "user", "content": prompt}],
)
return 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:
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]
},
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
except AttributeError:
return prompt