The user wants me to generate an AI system architecture review bot. This is a comprehensive request that involves designing a system capable of reviewing software/system architecture documents, diagrams, or descriptions. Let me think about what this entails: 1. **Purpose**: An AI bot that reviews system architecture - looking at design decisions, patterns, scalability, security, reliability, etc. 2. **Components needed**: - Input handling (architecture documents, diagrams, descriptions) - Analysis engine - Knowledge base of best practices/patterns - Scoring/rating system - Output/recommendations generator 3. **Architecture of the bot itself**: - Frontend (UI for uploads/interaction) - Backend (processing pipeline) - AI/ML layer (LLM, NLP, potentially computer vision for diagrams) - Knowledge base - Reporting engine Let me design a complete architecture review bot with: - Core features - Technical architecture - Code example (Python with FastAPI + LLM integration) - Workflow - Evaluation criteria - Sample interaction I'll provide a thorough, production-ready design that includes: 1. System overview 2. Architecture diagram description3. Component breakdown 4. Implementation code 5. Review criteria/framework 6. Sample usage Let me build a comprehensive response. # AI System Architecture Review Bot A comprehensive, production-ready design for an AI-powered architecture review system. --- ## 1. System Overview The **Architecture Review Bot** analyzes system architecture descriptions, documents (PDF/Markdown), and diagrams (images) to provide expert-level feedback on: - **Design Quality** (modularity, coupling, cohesion) - **Scalability** (horizontal/vertical scaling, bottlenecks) - **Reliability** (fault tolerance, redundancy, SPOFs) - **Security** (threat surface, auth/authz, data protection) - **Performance** (latency, throughput, caching) - **Cost & Maintainability** (cloud spend, technical debt) - **Best Practice Adherence** (12-factor, microservices patterns, etc.) --- ## 2. High-Level Architecture ``` ┌─────────────────────────────────────────────────────────────────┐ │ USER INTERFACE (Web UI / CLI / API) │ └────────────────────────────┬────────────────────────────────────┘ │ ┌────────────────────────────▼────────────────────────────────────┐ │ INGESTION LAYER │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ Markdown │ │ PDF │ │ Diagram │ │ │ │ Parser │ │ Extractor │ │ (Vision) │ │ │ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │ │ └─────────────────┼─────────────────┘ │ │ ▼ │ │ ┌──────────────────────┐ │ │ │ Normalized AST/Graph │ │ │ └──────────┬───────────┘ │ └─────────────────────────┼───────────────────────────────────────┘ │ ┌─────────────────────────▼───────────────────────────────────────┐ │ ANALYSIS ENGINE │ │ ┌─────────────────────────────────────────────────────────┐ │ │ │ Multi-Modal LLM (GPT-4V / Claude 3.5 / Gemini) │ │ │ │ + Embedding Model (sentence-transformers) │ │ │ │ + Pattern Matching Engine │ │ │ └────────────────────────┬────────────────────────────────┘ │ │ ▼ │ │ ┌─────────────────────────────────────────────────────────┐ │ │ │ REVIEW FRAMEWORK (7 Axes) │ │ │ │ • Scalability • Reliability • Security │ │ │ │ • Performance • Cost • Maintainability │ │ │ │ • Best Practices │ │ │ └────────────────────────┬────────────────────────────────┘ │ └───────────────────────────┼─────────────────────────────────────┘ │ ┌───────────────────────────▼─────────────────────────────────────┐ │ KNOWLEDGE BASE │ │ ┌────────────┐ ┌────────────┐ ┌──────────────────────────┐ │ │ │ Patterns │ │ Anti- │ │ Industry Benchmarks │ │ │ │ Catalog │ │ Patterns │ │ (RAG: Vector DB) │ │ │ └────────────┘ └────────────┘ └──────────────────────────┘ │ └───────────────────────────┬─────────────────────────────────────┘ │ ┌───────────────────────────▼─────────────────────────────────────┐ │ REPORT GENERATOR │ │ • Executive Summary • Detailed Findings │ │ • Risk Heatmap • Actionable Recommendations │ │ • Scorecard (0-100) • Comparison with Best Practices │ └─────────────────────────────────────────────────────────────────┘ ``` --- ## 3. Component Breakdown | Component | Technology | Responsibility | |-----------|-----------|----------------| | **UI** | React / Streamlit / CLI | User interaction | | **Ingestion** | PyMuPDF, python-markdown, Pytesseract | Multi-format parsing | | **Vision AI** | GPT-4o / Claude 3.5 Sonnet | Diagram interpretation | | **Analysis LLM** | Claude 3.5 / GPT-4o | Reasoning & critique | | **Embeddings** | `text-embedding-3-large` | Semantic search | | **Vector DB** | Pinecone / Chroma / Qdrant | Pattern lookup (RAG) | | **Orchestration** | LangChain / LlamaIndex | Pipeline management | | **Storage** | PostgreSQL + S3 | Persistence & history | | **Backend** | FastAPI | API layer | --- ## 4. Implementation (Python) ### 4.1 Project Structure ``` arch-review-bot/ ├── core/ │ ├── __init__.py │ ├── ingestor.py # Document parsing │ ├── analyzer.py # LLM-based analysis │ ├── reviewer.py # 7-axis review framework │ ├── rag.py # Knowledge retrieval │ └── reporter.py # Report generation ├── knowledge/ │ ├── patterns.json # Architecture patterns │ ├── anti_patterns.json # Common mistakes │ └── embeddings/ # Cached vectors ├── api/ │ └── server.py # FastAPI server ├── ui/ │ └── app.py # Streamlit UI ├── main.py # Entry point └── requirements.txt ``` ### 4.2 Core Modules **`core/ingestor.py`** — Multi-format input handler ```python from pathlib import Path import fitz # PyMuPDF import markdown from PIL import Image import base64 class ArchitectureIngestor: """Parses architecture inputs into structured representations.""" def ingest(self, source: str | Path) -> dict: path = Path(source) suffix = path.suffix.lower() if suffix in {".md", ".markdown", ".txt"}: return self._from_markdown(path) elif suffix == ".pdf": return self._from_pdf(path) elif suffix in {".png", ".jpg", ".jpeg"}: return self._from_diagram(path) else: raise ValueError(f"Unsupported format: {suffix}") def _from_markdown(self, path: Path) -> dict: text = path.read_text() return { "type": "text", "content": text, "format": "markdown", "metadata": {"source": str(path), "length": len(text)}, } def _from_pdf(self, path: Path) -> dict: doc = fitz.open(path) text = "\n".join(page.get_text() for page in doc) images = [] for page in doc: for img in page.get_images(full=True): # Extract embedded diagrams xref = img[0] base_img = doc.extract_image(xref) images.append(base_img["image"]) return { "type": "mixed", "content": text, "images": images, "format": "pdf", "metadata": {"pages": len(doc)}, } def _from_diagram(self, path: Path) -> dict: # Encode for vision model with open(path, "rb") as f: img_b64 = base64.b64encode(f.read()).decode() return { "type": "image", "content": img_b64, "format": path.suffix.lstrip("."), "metadata": {"source": str(path)}, } ``` **`knowledge/patterns.py`** — Best practice catalog ```python PATTERNS = { "scalability": { "good": [ "Stateless application servers behind load balancer", "Database read replicas for read-heavy workloads", "Async messaging (Kafka/RabbitMQ) for decoupling", "CDN for static content delivery", "Horizontal pod autoscaling (HPA) with CPU/memory metrics", ], "anti_patterns": [ "Sticky sessions preventing horizontal scaling", "Synchronous chains of microservices (cascading failures)", "Single shared database without sharding strategy", "Caching only at origin, not at edge", ], }, "reliability": { "good": [ "Circuit breakers (Hystrix/resilience4j)", "Bulkheads isolating critical resources", "Multi-AZ deployment with automated failover", "Idempotent operations with retry + dead-letter queues", "Health checks and graceful shutdown", ], "anti_patterns": [ "Single point of failure (single DB instance, single LB)", "No retry/backoff causing cascading failures", "Missing chaos testing for failure scenarios", ], }, "security": { "good": [ "Zero-trust networking with mTLS between services", "Secrets in vault (HashiCorp Vault / AWS Secrets Manager)", "OAuth2/OIDC with short-lived JWT tokens", "Encryption at rest AND in transit", "Principle of least privilege for IAM roles", "WAF + DDoS protection at edge", ], "anti_patterns": [ "Hardcoded credentials in code/config", "Public S3 buckets or open databases", "No input validation at API gateway", ], }, "performance": { "good": [ "Multi-layer caching (L1 in-memory, L2 Redis, L3 CDN)", "Database connection pooling (PgBouncer, HikariCP)", "Async I/O and non-blocking APIs", "Index optimization and query plan analysis", "Pagination for large result sets", ], "anti_patterns": [ "N+1 query patterns", "Large payloads without compression", "Cold starts from serverless overuse", ], }, "cost": { "good": [ "Right-sized instances with autoscaling", "Spot/preemptible instances for fault-tolerant workloads", "S3 lifecycle policies for cold data", "Reserved capacity for predictable workloads", ], "anti_patterns": [ "Always-on large instances for low-traffic services", "Cross-AZ data transfer without justification", "Over-provisioned databases", ], }, "maintainability": { "good": [ "Domain-driven design with bounded contexts", "Infrastructure as Code (Terraform/Pulumi)", "Comprehensive observability (logs, metrics, traces)", "Documented ADRs (Architecture Decision Records)", "Automated CI/CD with progressive delivery", ], "anti_patterns": [ "Distributed monolith (microservices without autonomy)", "Snowflake servers with manual configuration", "Missing or stale documentation", ], }, } ``` **`core/analyzer.py`** — LLM-powered analysis ```python from anthropic import Anthropic from openai import OpenAI import json class ArchitectureAnalyzer: """Orchestrates multi-model analysis of architecture.""" REVIEW_AXES = [ "scalability", "reliability", "security", "performance", "cost", "maintainability", "best_practices", ] def __init__(self, model: str = "claude-3-5-sonnet"): self.client = Anthropic() if "claude" in model else OpenAI() self.model = model def analyze(self, ingestion: dict, rag_context: str) -> dict: """Run full multi-axis analysis.""" if ingestion["type"] == "image": return self._analyze_diagram(ingestion, rag_context) elif ingestion["type"] == "mixed": return self._analyze_mixed(ingestion, rag_context) else: return self._analyze_text(ingestion, rag_context) def _build_prompt(self, content: str, rag_context: str) -> str: return f"""You are a Principal Software Architect with 20+ years of experience reviewing system architectures at companies like Google, Netflix, and Amazon. ## Reference Best Practices (from knowledge base): {rag_context} ## Architecture Under Review: {content} ## Your Task: Analyze this architecture across these 7 axes: 1. **Scalability** — Can it handle 10x, 100x growth? 2. **Reliability** — What happens when components fail? 3. **Security** — Threat surface, auth, data protection? 4. **Performance** — Latency, throughput, bottlenecks? 5. **Cost** — Cloud spend efficiency, waste? 6. **Maintainability** — Tech debt, operability, documentation? 7. **Best Practices** — Industry-standard patterns? ## Output Format (strict JSON): {{ "executive_summary": "2-3 sentence high-level assessment", "overall_score": 0-100, "axes": {{ "scalability": {{ "score": 0-100, "strengths": ["..."], "weaknesses": ["..."], "recommendations": ["actionable fix 1", "actionable fix 2"] }}, "reliability": {{ ... }}, ... (all 7 axes) }}, "critical_issues": [ {{"severity": "high|medium|low", "issue": "...", "impact": "...", "fix": "..."}} ], "architecture_style": "monolith | microservices | serverless | event-driven | ...", "key_components_identified": ["..."], "missing_components": ["..."], "swot": {{ "strengths": ["..."], "weaknesses": ["..."], "opportunities": ["..."], "threats": ["..."] }} }} """ def _analyze_text(self, ingestion: dict, rag_context: str) -> dict: prompt = self._build_prompt(ingestion["content"], rag_context) msg = self.client.messages.create( model=self.model, max_tokens=4096, messages=[{"role": "user", "content": prompt}], ) return self._parse_json(msg.content[0].text) def _analyze_diagram(self, ingestion: dict, rag_context: str) -> dict: # Vision-capable model required prompt = self._build_prompt("[See architecture diagram]", rag_context) msg = self.client.messages.create( model="claude-3-5-sonnet", max_tokens=4096, messages=[{ "role": "user", "content": [ {"type": "image", "source": {"type": "base64", "media_type": f"image/{ingestion['format']}", "data": ingestion["content"]}}, {"type": "text", "text": prompt}, ], }], ) return self._parse_json(msg.content[0].text) def _parse_json(self, text: str) -> dict: # Robust JSON extraction if "```json" in text: text = text.split("```json")[1].split("```")[0] return json.loads(text) ``` **`core/rag.py`** — Retrieval-Augmented Generation ```python import chromadb from chromadb.utils import embedding_functions class PatternRetriever: """RAG over architecture knowledge base.""" def __init__(self): self.client = chromadb.PersistentClient(path="./knowledge/embeddings") self.ef = embedding_functions.OpenAIEmbeddingFunction( api_key="...", model_name="text-embedding-3-small" ) self.collection = self.client.get_or_create_collection( name="architecture_patterns", embedding_function=self.ef ) def retrieve(self, query: str, n: int = 8) -> str: results = self.collection.query(query_texts=[query], n_results=n) docs = results["documents"][0] return "\n\n".join(f"- {doc}" for doc in docs) def index_patterns(self, patterns: dict): """One-time indexing of knowledge base.""" docs = [] for axis, content in patterns.items(): for pattern in content["good"]: docs.append(f"[{axis.upper()} - GOOD] {pattern}") for anti in content["anti_patterns"]: docs.append(f"[{axis.upper()} - ANTI-PATTERN] {anti}") self.collection.add( documents=docs, ids=[f"doc_{i}" for i in range(len(docs))], ) ``` **`core/reporter.py`** — Rich report generation ```python from datetime import datetime import json class ReportGenerator: """Generates human-readable reports from analysis.""" def generate(self, analysis: dict, format: str = "markdown") -> str: if format == "markdown": return self._markdown(analysis) elif format == "html": return self._html(analysis) raise ValueError(f"Unknown format: {format}") def _markdown(self, a: dict) -> str: score_emoji = { range(90, 101): "🌟 Excellent", range(75, 90): "✅ Strong", range(60, 75): "⚠️ Adequate", range(40, 60): "🟠 Needs Improvement", range(0, 40): "🔴 Critical Issues", } score = a["overall_score"] rating = next(v for k, v in score_emoji.items() if score in k) md = f"""# 🏛️ Architecture Review Report **Date:** {datetime.now().isoformat()} **Overall Score:** **{score}/100** — {rating} **Style:** `{a.get('architecture_style', 'unknown')}` ## 📋 Executive Summary {a['executive_summary']} ## 🚨 Critical Issues """ for issue in a.get("critical_issues", []): emoji = {"high": "🔴", "medium": "🟡", "low": "🟢"}[issue["severity"]] md += f"- {emoji} **[{issue['severity'].upper()}]** {issue['issue']}\n" md += f" - *Impact:* {issue['impact']}\n" md += f" - *Fix:* {issue['fix']}\n\n" md += "\n## 📊 Axis-by-Axis Analysis\n" for axis, data in a["axes"].items(): md += f"\n### {axis.title()} — {data['score']}/100\n" if data.get("strengths"): md += "**Strengths:**\n" + "\n".join(f"- ✅ {s}" for s in data["strengths"]) + "\n" if data.get("weaknesses"): md += "\n**Weaknesses:**\n" + "\n".join(f"- ⚠️ {w}" for w in data["weaknesses"]) + "\n" if data.get("recommendations"): md += "\n**Recommendations:**\n" + "\n".join(f"- 💡 {r}" for r in data["recommendations"]) + "\n" md += "\n## 🎯 SWOT Analysis\n" swot = a.get("swot", {}) md += f"**Strengths:** {', '.join(swot.get('strengths', []))}\n\n" md += f"**Weaknesses:** {', '.join(swot.get('weaknesses', []))}\n\n" md += f"**Opportunities:** {', '.join(swot.get('opportunities', []))}\n\n" md += f"**Threats:** {', '.join(swot.get('threats', []))}\n" return md def _html(self, analysis: dict) -> str: # Render markdown to HTML with CSS styling import markdown as md css = """ """ return f"{css}{md.markdown(self._markdown(analysis))}" ``` **`core/reviewer.py`** — Orchestrator ```python class ArchitectureReviewer: """End-to-end orchestration.""" def __init__(self): self.ingestor = ArchitectureIngestor() self.retriever = PatternRetriever() self.analyzer = ArchitectureAnalyzer() self.reporter = ReportGenerator() def review(self, source: str, format: str = "markdown") -> str: # 1. Ingest ingestion = self.ingestor.ingest(source) # 2. Retrieve relevant patterns query_text = ingestion.get("content", "")[:2000] or "system architecture" rag_context = self.retriever.retrieve(query_text) # 3. Analyze analysis = self.analyzer.analyze(ingestion, rag_context) # 4. Generate report return self.reporter.generate(analysis, format=format) ``` ### 4.3 FastAPI Server **`api/server.py`** ```python from fastapi import FastAPI, UploadFile, File, HTTPException from fastapi.responses import HTMLResponse, JSONResponse from core.reviewer import ArchitectureReviewer import tempfile from pathlib import Path app = FastAPI(title="AI Architecture Review Bot", version="1.0") reviewer = ArchitectureReviewer() @app.post("/review/upload") async def review_upload(file: UploadFile = File(...), format: str = "markdown"): """Upload architecture document/diagram for review.""" suffix = Path(file.filename).suffix with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp: content = await file.read() tmp.write(content) tmp_path = tmp.name try: report = reviewer.review(tmp_path, format=format) if format == "html": return HTMLResponse(report) return JSONResponse({"report": report, "format": format}) except Exception as e: raise HTTPException(500, f"Review failed: {str(e)}") finally: Path(tmp_path).unlink(missing_ok=True) @app.post("/review/text") async def review_text(architecture: str): """Review architecture from text description.""" with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".md") as tmp: tmp.write(architecture) tmp_path = tmp.name try: return {"report": reviewer.review(tmp_path)} finally: Path(tmp_path).unlink(missing_ok=True) @app.get("/health") async def health(): return {"status": "ok", "service": "arch-review-bot"} ``` ### 4.4 Streamlit UI **`ui/app.py`** ```python import streamlit as st from core.reviewer import ArchitectureReviewer import tempfile st.set_page_config(page_title="🏛️ AI Architecture Review", layout="wide") st.title("🏛️ AI Architecture Review Bot") st.write("Upload your architecture document or paste a description for expert review.") reviewer = ArchitectureReviewer() tab1, tab2 = st.tabs(["📤 Upload File", "✍️ Paste Text"]) with tab1: file = st.file_uploader("Upload architecture", type=["md", "pdf", "png", "jpg", "txt"]) if file and st.button("Review Architecture", type="primary"): with st.spinner("🔍 Analyzing architecture..."): with tempfile.NamedTemporaryFile(delete=False, suffix=f".{file.name.split('.')[-1]}") as tmp: tmp.write(file.getvalue()) tmp_path = tmp.name report = reviewer.review(tmp_path) st.markdown(report) st.download_button("📥 Download Report", report, "architecture_review.md") with tab2: text = st.text_area("Describe your architecture:", height=300, placeholder="E.g., 'A 3-tier web app using React, Node.js, and PostgreSQL...'") if text and st.button("Review", type="primary"): with st.spinner("🔍 Analyzing..."): with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".md") as tmp: tmp.write(text) report = reviewer.review(tmp.name) st.markdown(report) ``` --- ## 5. Workflow ``` ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ Input │───▶│ Ingestion│───▶│ RAG │───▶│ Analyzer │───▶│ Report │ │ (file/ │ │ Parse │ │ Retrieve │ │ LLM │ │ Format │ │ text) │ │ │ │ Patterns │ │ Reason │ │ │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │ │ ▼ ▼ ┌──────────┐ ┌──────────┐ │ Knowledge│ │ Critique │ │ Base │ │ + Score │ └──────────┘ └──────────┘ ``` --- ## 6. Sample Interaction **Input:** ```markdown # My Architecture - React frontend - Node.js API on single EC2 instance - MongoDB on same EC2 - No backups, no monitoring ``` **Output:** ```markdown # 🏛️ Architecture Review Report **Overall Score:** 23/100 — 🔴 Critical Issues ## 📋 Executive Summary This prototype-grade architecture lacks fundamental scalability, reliability, and security controls. Not production-ready. ## 🚨 Critical Issues - 🔴 **[HIGH]** Single EC2 instance = single point of failure - *Fix:* Migrate to ECS/EKS with multi-AZ deployment - 🔴 **[HIGH]** No backups or disaster recovery - *Fix:* Enable automated backups + cross-region replication - 🟡 **[MEDIUM]** Co-located DB and app cause resource contention - *Fix:* Use managed MongoDB Atlas ## 📊 Axis-by-Axis ### Reliability — 15/100 **Weaknesses:** SPOF, no failover, no backups **Recommendations:** - 💡 Implement managed database with HA - 💡 Add multi-AZ redundancy - 💡 Set up automated backups with point-in-time recovery ``` --- ## 7. Running It ```bash # Install pip install fastapi uvicorn anthropic openai chromadb pymupdf streamlit pillow # Index knowledge base (one-time) python -c "from core.rag import PatternRetriever; from knowledge.patterns import PATTERNS; PatternRetriever().index_patterns(PATTERNS)" # Run API uvicorn api.server:app --reload # Run UI streamlit run ui/app.py ``` --- ## 8. Extension Points | Enhancement | Implementation | |------------|----------------| | **Multi-repo analysis** | Clone repos, scan for patterns (e.g., DB migrations, IaC) | | **Cost estimation** | Integrate Infracost / AWS pricing API | | **Compliance checks** | Add SOC2/HIPAA/PCI rule packs | | **Diff between versions** | Compare architecture versions over time | | **Team collaboration** | Comments, voting, approval workflows | | **Auto-fix suggestions** | Generate Terraform/Pulumi code patches | | **Slack/Teams bot** | `/review-arch` command in chat | | **CI/CD gate** | Block PRs that introduce anti-patterns | --- This bot can serve as a **first-pass reviewer** before involving senior architects, dramatically reducing review cycles while enforcing consistency. It complements (not replaces) human judgment by surfacing issues systematically and citing relevant best practices.

Test