deletion of unnecessary file

This commit is contained in:
Madhu 2025-02-02 01:17:12 +05:30
parent 6497f1b515
commit 5bd5227288
5 changed files with 0 additions and 522 deletions

View file

@ -1,35 +0,0 @@
## Example Test Prompts
### 1. Real-time Event Processing System
"We're building a real-time event processing system for a smart city infrastructure that needs to handle 100,000 IoT sensors, process environmental data, manage traffic flows, and provide emergency response capabilities. The system needs to be highly available, handle peak loads during emergencies, and maintain data integrity. Budget constraints exist but reliability is critical."
### 2. Healthcare Data Platform
"Design a HIPAA-compliant healthcare data platform that needs to integrate with legacy systems, handle real-time patient monitoring, support ML-based diagnostics, and manage secure data sharing between different healthcare providers. The system should scale to handle data from 50 hospitals and support both real-time analytics and batch processing."
### 3. Financial Trading Platform
"We need to build a high-frequency trading platform that processes market data streams, executes trades with sub-millisecond latency, maintains audit trails, and handles complex risk calculations. The system needs to be globally distributed, handle 100,000 transactions per second, and have robust disaster recovery capabilities."
### 4. Multi-tenant SaaS Platform
"Design a multi-tenant SaaS platform for enterprise resource planning that needs to support customization per tenant, handle different data residency requirements, support offline capabilities, and maintain performance isolation between tenants. The system should scale to 10,000 concurrent users and support custom integrations."
### 5. Digital Content Delivery Network
"We're building a global content delivery platform for streaming high-definition video content, supporting live streaming, VOD, and interactive content. The system needs to handle dynamic transcoding, support DRM, manage user-generated content, and optimize delivery based on network conditions and device capabilities."
### 6. Supply Chain Management System
"Design a blockchain-based supply chain management system that needs to track products from source to retail, integrate with IoT sensors for condition monitoring, support smart contracts for automated settlements, and provide real-time visibility across the supply chain. The system should handle 1000 partners and support regulatory compliance reporting."
Each of these prompts presents complex architectural challenges that require careful consideration of:
- Scalability patterns
- Data consistency requirements
- Security and compliance needs
- Integration complexities
- Performance optimization
- Cost-benefit trade-offs
- Technical debt implications
- Team expertise requirements
The DeepSeek model will analyze these requirements and provide structured recommendations using the ProjectAnalysis schema, which Claude can then use to provide detailed implementation guidance.
Startups making critical technical decisions
Enterprise architecture modernization projects

View file

@ -1,295 +0,0 @@
from typing import Optional, List, Dict, Any, Union
import os
import time
import streamlit as st
from openai import OpenAI
import anthropic
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from enum import Enum
import json
from phi.agent import Agent, RunResponse
from phi.model.anthropic import Claude
# Model Constants
DEEPSEEK_MODEL: str = "deepseek-reasoner"
CLAUDE_MODEL: str = "claude-3-5-sonnet-20241022"
# Load environment variables
load_dotenv()
system_prompt = """You are a Senior Software Expert and Technical Documentation Assistant. Your role is to analyze the structured JSON response from DeepSeek, which contains architectural and technical recommendations across various domains, along with the original user query describing the software system they want to build.
The input consists of:
- The user's original query describing their software requirements
- A structured JSON response containing recommendations for architecture, security, infrastructure, compliance and other technical domains
For each key-value pair in the JSON:
1. Present the key and its corresponding value in a readable report format
2. Format the information in a clear, organized way
3. Do not add your own opinions or suggestions
4. Do not modify or reinterpret the provided information
Keep your responses factual and directly based on the JSON content provided."""
class ArchitecturePattern(str, Enum):
"""Architectural patterns for system design."""
MICROSERVICES = "microservices" # Decomposed into small, independent services
MONOLITHIC = "monolithic" # Single, unified codebase
SERVERLESS = "serverless" # Function-as-a-Service architecture
EVENT_DRIVEN = "event_driven" # Asynchronous event-based communication
class DatabaseType(str, Enum):
"""Types of database systems."""
SQL = "sql" # Relational databases with ACID properties
NOSQL = "nosql" # Non-relational databases for flexible schemas
HYBRID = "hybrid" # Combined SQL and NoSQL approach
class ComplianceStandard(str, Enum):
"""Regulatory compliance standards."""
HIPAA = "hipaa" # Healthcare data protection
GDPR = "gdpr" # EU data privacy regulation
SOC2 = "soc2" # Service organization security controls
ISO27001 = "iso27001" # Information security management
class ArchitectureDecision(BaseModel):
"""Represents architectural decisions and their justifications."""
pattern: ArchitecturePattern
rationale: str = Field(..., min_length=50) # Detailed explanation for the choice
trade_offs: Dict[str, List[str]] = Field(..., alias="trade_offs") # Pros and cons
estimated_cost: Dict[str, float] # Cost breakdown
class SecurityMeasure(BaseModel):
"""Security controls and implementation details."""
measure_type: str # Type of security measure
implementation_priority: int = Field(..., ge=1, le=5) # Priority level 1-5
compliance_standards: List[ComplianceStandard] # Applicable standards
data_classification: str # Data sensitivity level
class InfrastructureResource(BaseModel):
"""Infrastructure components and specifications."""
resource_type: str # Type of infrastructure resource
specifications: Dict[str, str] # Technical specifications
scaling_policy: Dict[str, str] # Scaling rules and thresholds
estimated_cost: float # Estimated cost per resource
class TechnicalAnalysis(BaseModel):
"""Complete technical analysis of the system architecture."""
architecture_decision: ArchitectureDecision # Core architecture choices
infrastructure_resources: List[InfrastructureResource] # Required resources
security_measures: List[SecurityMeasure] # Security controls
database_choice: DatabaseType # Database architecture
compliance_requirements: List[ComplianceStandard] = [] # Required standards
performance_requirements: List[Dict[str, Union[str, float]]] = [] # Performance metrics
risk_assessment: Dict[str, str] = {} # Identified risks and mitigations
class ModelChain:
def __init__(self, deepseek_api_key: str, anthropic_api_key: str) -> None:
self.client = OpenAI(
api_key=deepseek_api_key,
base_url="https://api.deepseek.com"
)
self.claude_client = anthropic.Anthropic(api_key=anthropic_api_key)
self.agent = Agent(
model=Claude(id="claude-3-5-sonnet-20241022", api_key=anthropic_api_key),
system_prompt=system_prompt,
markdown=True
)
self.deepseek_messages: List[Dict[str, str]] = []
self.claude_messages: List[Dict[str, Any]] = []
self.current_model: str = CLAUDE_MODEL
def get_deepseek_reasoning(self, user_input: str) -> tuple[str, str]:
start_time = time.time()
system_prompt = """You are an expert software architect and technical advisor. Analyze the user's project requirements
and provide structured reasoning about architecture, tools, and implementation strategies.
IMPORTANT: Reason why you are choosing a particular architecture pattern, database type, etc. for user understanding in your reasoning.
IMPORTANT: Your response must be a valid JSON object (not a string or any other format) that matches the schema provided below.
Do not include any explanatory text, markdown formatting, or code blocks - only return the JSON object.
Schema:
{
"architecture_decision": {
"pattern": "one of: microservices|monolithic|serverless|event_driven|layered",
"rationale": "string",
"trade_offs": {"advantage": ["list of strings"], "disadvantage": ["list of strings"]},
"estimated_cost": {"implementation": float, "maintenance": float}
},
"infrastructure_resources": [{
"resource_type": "string",
"specifications": {"key": "value"},
"scaling_policy": {"key": "value"},
"estimated_cost": float
}],
"security_measures": [{
"measure_type": "string",
"implementation_priority": "integer 1-5",
"compliance_standards": ["hipaa", "gdpr", "soc2", "hitech", "iso27001", "pci_dss"],
"estimated_setup_time_days": "integer",
"data_classification": "one of: protected_health_information|personally_identifiable_information|confidential|public",
"encryption_requirements": {"key": "value"},
"access_control_policy": {"role": ["permissions"]},
"audit_requirements": ["list of strings"]
}],
"database_choice": "one of: sql|nosql|graph|time_series|hybrid",
"ml_capabilities": [{
"model_type": "string",
"training_frequency": "string",
"input_data_types": ["list of strings"],
"performance_requirements": {"metric": float},
"hardware_requirements": {"resource": "specification"},
"regulatory_constraints": ["list of strings"]
}],
"data_integrations": [{
"integration_type": "one of: hl7|fhir|dicom|rest|soap|custom",
"data_format": "string",
"frequency": "string",
"volume": "string",
"security_requirements": {"key": "value"}
}],
"performance_requirements": [{
"metric_name": "string",
"target_value": float,
"measurement_unit": "string",
"priority": "integer 1-5"
}],
"audit_config": {
"log_retention_period": "integer",
"audit_events": ["list of strings"],
"compliance_mapping": {"standard": ["requirements"]}
},
"api_config": {
"version": "string",
"auth_method": "string",
"rate_limits": {"role": "requests_per_minute"},
"documentation_url": "string"
},
"error_handling": {
"retry_policy": {"key": "value"},
"fallback_strategies": ["list of strings"],
"notification_channels": ["list of strings"]
},
"estimated_team_size": "integer",
"critical_path_components": ["list of strings"],
"risk_assessment": {"risk": "mitigation"},
"maintenance_considerations": ["list of strings"],
"compliance_requirements": ["list of compliance standards"],
"data_retention_policy": {"data_type": "retention_period"},
"disaster_recovery": {"key": "value"},
"interoperability_standards": ["list of strings"]
}
Consider scalability, security, maintenance, and technical debt in your analysis.
Focus on practical, modern solutions while being mindful of trade-offs."""
try:
deepseek_response = self.client.chat.completions.create(
model="deepseek-reasoner",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_input}
],
max_tokens=3000,
stream=False
)
reasoning_content = deepseek_response.choices[0].message.reasoning_content
normal_content = deepseek_response.choices[0].message.content
# Display the reasoning separately
with st.expander("DeepSeek Reasoning", expanded=True):
st.markdown(reasoning_content)
with st.expander("💭 Technical Analysis", expanded=True):
st.markdown(normal_content)
elapsed_time = time.time() - start_time
time_str = f"{elapsed_time/60:.1f} minutes" if elapsed_time >= 60 else f"{elapsed_time:.1f} seconds"
st.caption(f"⏱️ Analysis completed in {time_str}")
# Return both reasoning and normal content
return reasoning_content, normal_content
except Exception as e:
st.error(f"Error in DeepSeek analysis: {str(e)}")
return "Error occurred while analyzing", ""
def get_claude_response(self, user_input: str, deepseek_output: tuple[str, str]) -> str:
try:
reasoning_content, normal_content = deepseek_output
# Create expander for Claude's response
with st.expander("🤖 Claude's Response", expanded=True):
response_placeholder = st.empty()
# Prepare the message with user input, reasoning and normal output
message = f"""User Query: {user_input}
DeepSeek Reasoning: {reasoning_content}
DeepSeek Technical Analysis: {normal_content}"""
# Use Phi Agent to get response
response: RunResponse = self.agent.run(
message=message
)
return response.content
except Exception as e:
st.error(f"Error in Claude response: {str(e)}")
return "Error occurred while getting response"
def main() -> None:
"""Main function to run the Streamlit app."""
st.title("🤖 AI Project with Deepseek + R1")
# Sidebar for API keys
with st.sidebar:
st.header("⚙️ Configuration")
deepseek_api_key = st.text_input("DeepSeek API Key", type="password")
anthropic_api_key = st.text_input("Anthropic API Key", type="password")
if st.button("🗑️ Clear Chat History"):
st.session_state.messages = []
st.rerun()
# Initialize session state for messages
if "messages" not in st.session_state:
st.session_state.messages = []
# Display chat messages
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.markdown(message["content"])
# Chat input
if prompt := st.chat_input("What would you like to know?"):
if not deepseek_api_key or not anthropic_api_key:
st.error("⚠️ Please enter both API keys in the sidebar.")
return
# Initialize ModelChain
chain = ModelChain(deepseek_api_key, anthropic_api_key)
# Add user message to chat
st.session_state.messages.append({"role": "user", "content": prompt})
with st.chat_message("user"):
st.markdown(prompt)
# Get AI response
with st.chat_message("assistant"):
with st.spinner("🤔 Thinking..."):
deepseek_output = chain.get_deepseek_reasoning(prompt)
with st.spinner("✍️ Responding..."):
response = chain.get_claude_response(prompt, deepseek_output)
st.session_state.messages.append({"role": "assistant", "content": response})
if __name__ == "__main__":
main()

View file

@ -1,4 +0,0 @@
streamlit
openai
anthropic
python-dotenv

View file

@ -1,184 +0,0 @@
from enum import Enum
from typing import List, Dict, Union
from pydantic import BaseModel, Field, ValidationError
import streamlit as st
from openai import OpenAI
import anthropic
import json
import re
import os
from dotenv import load_dotenv
from phi.agent import Agent, RunResponse
from phi.model.anthropic import Claude
load_dotenv()
# --------------------------
# Enums & Data Models
# --------------------------
class ArchitecturePattern(str, Enum):
MICROSERVICES = "microservices"
MONOLITHIC = "monolithic"
SERVERLESS = "serverless"
EVENT_DRIVEN = "event_driven"
class DatabaseType(str, Enum):
SQL = "sql"
NOSQL = "nosql"
HYBRID = "hybrid"
class ComplianceStandard(str, Enum):
HIPAA = "hipaa"
GDPR = "gdpr"
SOC2 = "soc2"
ISO27001 = "iso27001"
class ArchitectureDecision(BaseModel):
pattern: ArchitecturePattern
rationale: str = Field(..., min_length=50)
trade_offs: Dict[str, List[str]] = Field(..., alias="trade_offs")
estimated_cost: Dict[str, float]
class SecurityMeasure(BaseModel):
measure_type: str
implementation_priority: int = Field(..., ge=1, le=5)
compliance_standards: List[ComplianceStandard]
data_classification: str
class InfrastructureResource(BaseModel):
resource_type: str
specifications: Dict[str, str]
scaling_policy: Dict[str, str]
estimated_cost: float
class TechnicalAnalysis(BaseModel):
architecture_decision: ArchitectureDecision
infrastructure_resources: List[InfrastructureResource]
security_measures: List[SecurityMeasure]
database_choice: DatabaseType
compliance_requirements: List[ComplianceStandard] = []
performance_requirements: List[Dict[str, Union[str, float]]] = []
risk_assessment: Dict[str, str] = {}
# --------------------------
# Core Implementation
# --------------------------
class ArchitectureAnalyzer:
def __init__(self, deepseek_api_key: str, anthropic_api_key: str):
self.deepseek_client = OpenAI(
api_key=deepseek_api_key,
base_url="https://api.deepseek.com"
)
self.claude_agent = Agent(
model=Claude(
id="claude-3-5-sonnet-20241022",
api_key=anthropic_api_key
),
markdown=True,
)
self.reasoning_content = ""
self.deepseek_prompt = f"""Analyze software requirements and return JSON with:
{{
"architecture_decision": {{
"pattern": "{'|'.join([e.value for e in ArchitecturePattern])}",
"rationale": "technical justification",
"trade_offs": {{"pros": [], "cons": []}},
"estimated_cost": {{"development": float, "maintenance": float}}
}},
"infrastructure_resources": [{{"resource_type": "...", "specifications": {{}}, ...}}],
"security_measures": [{{"measure_type": "...", "priority": 1-5, ...}}],
"database_choice": "{'|'.join([e.value for e in DatabaseType])}",
"compliance_requirements": ["..."],
"performance_requirements": [{{"metric": "...", "target": float}}]
}}"""
def _extract_json(self, text: str) -> dict:
try:
json_str = re.search(r'\{.*\}', text, re.DOTALL).group()
return json.loads(json_str)
except (AttributeError, json.JSONDecodeError) as e:
st.error(f"JSON extraction failed: {str(e)}")
st.text("Raw response:\n" + text)
raise
def analyze_requirements(self, user_input: str) -> TechnicalAnalysis:
try:
response1 = self.deepseek_client.chat.completions.create(
model="deepseek-reasoner",
messages=[
{"role": "system", "content": self.deepseek_prompt},
{"role": "user", "content": user_input}
],
temperature=0.2,
max_tokens=2000
)
self.reasoning_content = response1.choices[0].message.reasoning_content
json_data = self._extract_json(response1.choices[0].message.content)
return TechnicalAnalysis(**json_data)
except ValidationError as e:
st.error(f"Validation error: {e.errors()}")
st.json(json_data)
raise
def generate_report(self, analysis: TechnicalAnalysis) -> str:
report_prompt = f"""Convert this technical analysis into a executive report:
{analysis.model_dump_json(indent=2)}
Use markdown with:
# Title
## Sections
- Bullet points
**Bold important items**
Tables for cost/performance"""
response = self.claude_agent.run(report_prompt)
return response.content
# --------------------------
# Streamlit UI
# --------------------------
def main():
st.title("🏗️ AI Architecture Advisor")
with st.sidebar:
st.header("🔑 Setup")
deepseek_api_key = st.text_input("DeepSeek Key", type="password")
anthropic_api_key = st.text_input("Claude Key", type="password")
if "analysis" not in st.session_state:
st.session_state.analysis = None
if prompt := st.chat_input("Describe your system requirements:"):
if not all([deepseek_api_key, anthropic_api_key]):
st.error("Missing API keys")
return
analyzer = ArchitectureAnalyzer(deepseek_api_key, anthropic_api_key)
with st.status("🔨 Processing...", expanded=True):
try:
# Analysis Phase
st.write("🧠 Analyzing requirements...")
analysis = analyzer.analyze_requirements(prompt)
st.session_state.analysis = analysis
with st.expander("reasoning"):
st.markdown(analyzer.reasoning_content)
# Reporting Phase
st.write("📊 Generating report...")
report = analyzer.generate_report(analysis)
# Display Results
st.success("Analysis complete!")
st.markdown(report)
with st.expander("📁 Raw Analysis Data"):
st.json(analysis.model_dump_json())
except Exception as e:
st.error(f"Processing failed: {str(e)}")
if __name__ == "__main__":
main()

View file

@ -15,10 +15,6 @@ from phi.model.anthropic import Claude
DEEPSEEK_MODEL: str = "deepseek-reasoner"
CLAUDE_MODEL: str = "claude-3-5-sonnet-20241022"
# Load environment variables
load_dotenv()
class ArchitecturePattern(str, Enum):
"""Architectural patterns for system design."""
MICROSERVICES = "microservices" # Decomposed into small, independent services