81 lines
3.0 KiB
Python
81 lines
3.0 KiB
Python
# CLAUDE AI CLIENT - Same pattern as working tech-stack-selector
|
|
|
|
import os
|
|
import json
|
|
import re
|
|
from typing import Dict, Any
|
|
from loguru import logger
|
|
|
|
try:
|
|
import anthropic
|
|
except ImportError:
|
|
anthropic = None
|
|
|
|
class ClaudeClient:
|
|
"""Claude API client for AI-powered architecture generation"""
|
|
|
|
def __init__(self):
|
|
self.api_key = os.getenv("ANTHROPIC_API_KEY")
|
|
if not self.api_key:
|
|
logger.warning("ANTHROPIC_API_KEY not found - Claude AI will not work")
|
|
self.client = None
|
|
elif not anthropic:
|
|
logger.error("Anthropic library not installed")
|
|
self.client = None
|
|
else:
|
|
try:
|
|
# Use the same initialization pattern as tech-stack-selector
|
|
self.client = anthropic.Client(api_key=self.api_key)
|
|
logger.info("🤖 Claude AI client initialized successfully")
|
|
except Exception as e:
|
|
logger.error(f"Failed to initialize Claude client: {e}")
|
|
self.client = None
|
|
|
|
async def generate_architecture(self, prompt: str) -> Dict[str, Any]:
|
|
"""Generate architecture using Claude AI"""
|
|
try:
|
|
if not self.client:
|
|
logger.warning("Claude AI not available - using fallback response")
|
|
return {"success": False, "error": "Claude AI not configured"}
|
|
|
|
logger.info("🤖 Sending prompt to Claude AI...")
|
|
|
|
# Use the same API call pattern as tech-stack-selector
|
|
response = self.client.messages.create(
|
|
model="claude-3-5-sonnet-20241022",
|
|
max_tokens=4000,
|
|
temperature=0.1,
|
|
messages=[{"role": "user", "content": prompt}]
|
|
)
|
|
|
|
response_text = response.content[0].text
|
|
architecture_data = self._extract_json_from_response(response_text)
|
|
|
|
if architecture_data:
|
|
logger.info("✅ Claude AI generated architecture successfully")
|
|
return {"success": True, "data": architecture_data}
|
|
else:
|
|
return {"success": False, "error": "Invalid JSON response"}
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ Claude AI call failed: {e}")
|
|
return {"success": False, "error": str(e)}
|
|
|
|
def _extract_json_from_response(self, response_text: str) -> Dict[str, Any]:
|
|
"""Extract JSON from Claude response"""
|
|
try:
|
|
# Try JSON block first
|
|
json_match = re.search(r'```json\s*(.*?)\s*```', response_text, re.DOTALL)
|
|
if json_match:
|
|
return json.loads(json_match.group(1))
|
|
|
|
# Try direct JSON
|
|
json_match = re.search(r'\{.*\}', response_text, re.DOTALL)
|
|
if json_match:
|
|
return json.loads(json_match.group(0))
|
|
|
|
return json.loads(response_text)
|
|
|
|
except json.JSONDecodeError:
|
|
return None
|