多模态大模型在组件化设计系统中的可观测性建设与故障闭环

multimodal-design-observability/
├── config/
│   └── settings.py
├── core/
│   ├── __init__.py
│   ├── design_components.py
│   ├── observability.py
│   └── issue_manager.py
├── multimodal/
│   ├── __init__.py
│   └── analyzer.py
├── cli/
│   └── simulator.py
├── app.py
├── requirements.txt
└── run.sh

"""
项目配置文件
"""
import os

class Config:
    # 模拟设置
    DESIGN_SYSTEM_VERSION = "1.5.0"
    
    # 可观测性
    TRACE_HEADER = "X-Trace-ID"
    
    # 多模态AI服务 (模拟或真实)
    # 使用模拟模式避免真实API调用
    MULTIMODAL_AI_MODE = "simulated"  # 可选: "simulated", "openai", "azure"
    # 如果使用OpenAI
    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "")
    OPENAI_MODEL_VISION = "gpt-4-vision-preview"
    
    # 故障管理
    ISSUE_STORAGE_FILE = "data/issues.json"
    
    # 服务器
    HOST = "0.0.0.0"
    PORT = 5000
    DEBUG = True

config = Config()

"""
模拟的设计系统组件库及其可能出现的故障模式。
"""
from dataclasses import dataclass
from enum import Enum
import random
from typing import Optional, Dict, Any

class ComponentType(Enum):
    BUTTON = "Button"
    INPUT = "Input"
    MODAL = "Modal"
    DATA_TABLE = "DataTable"
    NAV_BAR = "NavBar"

class ComponentDefect(Enum):
    """组件缺陷枚举"""
    VISUAL_MISALIGNMENT = "visual_misalignment"
    STYLE_LOST = "style_lost"
    INTERACTION_BROKEN = "interaction_broken"
    RESPONSIVE_FAILURE = "responsive_failure"
    CONTENT_OVERFLOW = "content_overflow"

@dataclass
class DesignComponent:
    """设计系统组件实例"""
    id: str
    type: ComponentType
    name: str
    version: str
    props: Dict[str, Any]
    defect: Optional[ComponentDefect] = None
    defect_description: Optional[str] = None
    
    def render(self, trace_id: str) -> Dict[str, Any]:
        """模拟组件渲染，可能引入缺陷"""
        result = {
            "component_id": self.id,
            "type": self.type.value,
            "name": self.name,
            "version": self.version,
            "props": self.props,
            "trace_id": trace_id,
            "ok": True
        }
        
        # 模拟缺陷：10%的几率渲染出有问题的组件
        if random.random() < 0.1:
            self.defect = random.choice(list(ComponentDefect))
            defect_map = {
                ComponentDefect.VISUAL_MISALIGNMENT: "Component appears misaligned by 5px on certain viewports.",
                ComponentDefect.STYLE_LOST: "Primary color and border-radius styles are not applied.",
                ComponentDefect.INTERACTION_BROKEN: "Click handler does not fire on mobile Safari.",
                ComponentDefect.RESPONSIVE_FAILURE: "Layout breaks on screen widths between 768px and 992px.",
                ComponentDefect.CONTENT_OVERFLOW: "Long text content overflows container without ellipsis."
            }
            self.defect_description = defect_map[self.defect]
            result.update({
                "ok": False,
                "defect": self.defect.value,
                "defect_description": self.defect_description,
                "recommended_action": "Capture screenshot and analyze with multimodal AI."
            })
        return result

# 组件注册表
COMPONENT_REGISTRY: Dict[str, DesignComponent] = {}

def init_component_registry():
    """初始化模拟组件库"""
    global COMPONENT_REGISTRY
    components = [
        DesignComponent("btn-primary-1", ComponentType.BUTTON, "PrimaryButton", "1.2.0", {"variant": "primary", "size": "medium"}),
        DesignComponent("input-search-1", ComponentType.INPUT, "SearchInput", "1.1.5", {"placeholder": "Search...", "disabled": False}),
        DesignComponent("modal-confirm-1", ComponentType.MODAL, "ConfirmModal", "1.3.2", {"title": "Confirm Action", "showFooter": True}),
        DesignComponent("table-user-1", ComponentType.DATA_TABLE, "UserTable", "1.4.0", {"pageSize": 10, "striped": True}),
        DesignComponent("nav-main-1", ComponentType.NAV_BAR, "MainNavigation", "1.0.8", {"items": ["Home", "Dashboard", "Settings"]}),
    ]
    COMPONENT_REGISTRY = {comp.id: comp for comp in components}

"""
可观测性核心：追踪(Trace)、日志(Log)、指标(Metric)的生成与关联。
"""
import uuid
import time
from datetime import datetime
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
import json

from config.settings import config

@dataclass
class Span:
    """分布式追踪中的一个跨度"""
    span_id: str
    trace_id: str
    operation: str
    component_id: Optional[str] = None
    start_time: float = field(default_factory=time.time)
    end_time: Optional[float] = None
    tags: Dict[str, Any] = field(default_factory=dict)
    logs: List[Dict[str, Any]] = field(default_factory=list)
    
    def finish(self):
        self.end_time = time.time()
    
    def add_log(self, message: str, level: str = "INFO", **kwargs):
        self.logs.append({
            "timestamp": datetime.utcnow().isoformat(),
            "level": level,
            "message": message,
            **kwargs
        })
    
    def to_dict(self):
        return {
            "span_id": self.span_id,
            "trace_id": self.trace_id,
            "operation": self.operation,
            "component_id": self.component_id,
            "duration_ms": round((self.end_time or time.time()) - self.start_time, 3) * 1000,
            "tags": self.tags,
            "logs": self.logs
        }

class TraceContext:
    """追踪上下文管理器"""
    
    def __init__(self, trace_id: Optional[str] = None, parent_span_id: Optional[str] = None):
        self.trace_id = trace_id or str(uuid.uuid4())
        self.parent_span_id = parent_span_id
        self._spans: List[Span] = []
    
    def start_span(self, operation: str, component_id: Optional[str] = None) -> Span:
        span = Span(
            span_id=str(uuid.uuid4()),
            trace_id=self.trace_id,
            operation=operation,
            component_id=component_id
        )
        self._spans.append(span)
        return span
    
    def get_all_spans(self) -> List[Dict[str, Any]]:
        """获取所有已完成的span数据"""
        return [span.to_dict() for span in self._spans if span.end_time is not None]
    
    def to_trace_header(self) -> Dict[str, str]:
        """生成用于HTTP传播的追踪头"""
        return {config.TRACE_HEADER: self.trace_id}

# 全局追踪收集器（简化版，生产环境应使用Jaeger, Tempo等）
TRACE_COLLECTOR: List[Dict[str, Any]] = []

def record_trace(trace_data: Dict[str, Any]):
    """记录追踪数据到收集器"""
    TRACE_COLLECTOR.append({
        "timestamp": datetime.utcnow().isoformat(),
        **trace_data
    })

"""
故障工单管理，实现状态机与持久化。
"""
import json
import uuid
from enum import Enum
from dataclasses import dataclass, field, asdict
from datetime import datetime
from typing import List, Optional, Dict, Any
from pathlib import Path

from config.settings import config

class IssueStatus(Enum):
    OPEN = "open"
    ANALYZING = "analyzing"
    FIX_PROPOSED = "fix_proposed"
    RESOLVED = "resolved"
    IGNORED = "ignored"

@dataclass
class Issue:
    """故障工单"""
    id: str
    title: str
    description: str
    component_id: str
    trace_id: str
    status: IssueStatus = IssueStatus.OPEN
    screenshot_ref: Optional[str] = None  # 截图存储路径或URL
    ai_analysis: Optional[Dict[str, Any]] = None
    proposed_fix: Optional[Dict[str, Any]] = None  # 修复建议，如代码diff
    created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())
    updated_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())
    
    def update_status(self, new_status: IssueStatus, metadata: Optional[Dict] = None):
        self.status = new_status
        self.updated_at = datetime.utcnow().isoformat()
        if metadata:
            if new_status == IssueStatus.ANALYZING:
                self.ai_analysis = metadata
            elif new_status == IssueStatus.FIX_PROPOSED:
                self.proposed_fix = metadata

class IssueStore:
    """简单的文件存储（生产环境应使用数据库）"""
    
    def __init__(self, storage_path: str):
        self.storage_path = Path(storage_path)
        self.storage_path.parent.mkdir(parents=True, exist_ok=True)
        self._issues: Dict[str, Issue] = self._load()
    
    def _load(self) -> Dict[str, Issue]:
        if not self.storage_path.exists():
            return {}
        with open(self.storage_path, 'r') as f:
            data = json.load(f)
        issues = {}
        for issue_id, issue_data in data.items():
            # 反序列化时转换status字符串为枚举
            issue_data['status'] = IssueStatus(issue_data['status'])
            issues[issue_id] = Issue(**issue_data)
        return issues
    
    def _save(self):
        data = {issue_id: asdict(issue) for issue_id, issue in self._issues.items()}
        # 序列化时枚举转为字符串
        for issue in data.values():
            issue['status'] = issue['status'].value
        with open(self.storage_path, 'w') as f:
            json.dump(data, f, indent=2)
    
    def create(self, title: str, description: str, component_id: str, trace_id: str, screenshot_ref: Optional[str] = None) -> Issue:
        issue_id = f"ISSUE-{uuid.uuid4().hex[:8].upper()}"
        issue = Issue(
            id=issue_id,
            title=title,
            description=description,
            component_id=component_id,
            trace_id=trace_id,
            screenshot_ref=screenshot_ref
        )
        self._issues[issue_id] = issue
        self._save()
        return issue
    
    def get(self, issue_id: str) -> Optional[Issue]:
        return self._issues.get(issue_id)
    
    def update(self, issue: Issue):
        self._issues[issue.id] = issue
        self._save()
    
    def list_by_status(self, status: Optional[IssueStatus] = None) -> List[Issue]:
        if status:
            return [issue for issue in self._issues.values() if issue.status == status]
        return list(self._issues.values())

# 全局存储实例
issue_store = IssueStore(config.ISSUE_STORAGE_FILE)

"""
多模态大模型分析服务。
在模拟模式下，根据预设规则返回分析结果；真实模式下调用AI API。
"""
import base64
import logging
from typing import Dict, Any, Optional
import random

from config.settings import config

logger = logging.getLogger(__name__)

def encode_image_to_base64(image_path: str) -> str:
    """将本地图片文件编码为base64字符串（模拟用）"""
    try:
        with open(image_path, "rb") as image_file:
            return base64.b64encode(image_file.read()).decode('utf-8')
    except FileNotFoundError:
        # 返回一个模拟的占位符
        return "simulated_base64_image_data"

def analyze_with_multimodal_ai(
    screenshot_data: str,  # 可以是base64，也可以是URL
    description: str,
    context: Dict[str, Any]  # 包含trace, 组件信息等
) -> Dict[str, Any]:
    """
    核心多模态分析函数。
    输入：截图、问题描述、技术上下文。
    输出：根因分析、置信度、修复建议。
    """
    
    if config.MULTIMODAL_AI_MODE == "simulated":
        # 模拟AI分析逻辑
        logger.info("Running simulated multimodal analysis...")
        component_type = context.get("component_type", "Unknown")
        defect = context.get("defect", "unknown")
        
        # 模拟一些"智能"分析规则
        analysis_rules = {
            ("Button", "visual_misalignment"): {
                "root_cause": "The parent container uses `display: flex` with `align-items: center`, but the button has `margin-top: 2px` overriding alignment. Likely a CSS specificity conflict with global button styles.",
                "confidence": 0.85,
                "suggested_fix": {
                    "file": "src/components/Button/Button.css",
                    "diff": "// Remove conflicting margin\n- .button-primary {\n-   margin-top: 2px;\n- }\n+ /* Alignment handled by flex container */",
                    "component_version": "Upgrade to Button v1.2.1 which fixes this conflict."
                }
            },
            ("Input", "style_lost"): {
                "root_cause": "The new theme provider is not passing down CSS variables to shadow DOM encapsulated Input component. The `--primary-color` variable is undefined.",
                "confidence": 0.92,
                "suggested_fix": {
                    "file": "src/theme/ThemeProvider.jsx",
                    "diff": "// Ensure CSS variables propagate\n<ThemeProvider>\n+ <style>\n+   :root, :host {\n+     --primary-color: #007bff;\n+   }\n+ </style>\n  {children}\n</ThemeProvider>"
                }
            },
            ("DataTable", "responsive_failure"): {
                "root_cause": "Table column width calculations do not account for `box-sizing: border-box` when container has padding. On mid-breakpoints, total width exceeds 100% causing overflow.",
                "confidence": 0.78,
                "suggested_fix": {
                    "file": "src/components/DataTable/useColumnWidths.js",
                    "diff": "const totalWidth = columns.reduce((sum, col) => sum + col.width, 0);\n- if (totalWidth > containerWidth) {\n+ if (totalWidth > (containerWidth - horizontalPadding)) {\n    // trigger horizontal scroll\n}"
                }
            }
        }
        
        # 查找匹配规则，否则生成通用分析
        key = (component_type, defect)
        if key in analysis_rules:
            result = analysis_rules[key]
        else:
            result = {
                "root_cause": f"Based on the screenshot and description, the {component_type} component is experiencing a `{defect}` issue. This is often related to recent updates in the design system or conflicting styles from a parent component.",
                "confidence": round(random.uniform(0.6, 0.9), 2),
                "suggested_fix": {
                    "file": f"src/components/{component_type}/{component_type}.jsx",
                    "diff": "// Review recent changes and check for style conflicts.\n// Consider rolling back to previous stable version.",
                    "component_version": "Check for patches in the design system changelog."
                }
            }
        
        # 模拟处理时间
        import time
        time.sleep(1)
        
        return {
            "analysis_id": f"AI-ANALYSIS-{random.randint(1000,9999)}",
            "model_used": "simulated-gpt-4-vision",
            **result
        }
    
    elif config.MULTIMODAL_AI_MODE == "openai":
        # 真实调用OpenAI GPT-4V API (需要配置API KEY)
        # 此处为接口示例，实际运行需要有效的API Key
        logger.info("Calling OpenAI GPT-4V API...")
        # 注意：以下代码需要 openai >= 1.0.0
        from openai import OpenAI
        client = OpenAI(api_key=config.OPENAI_API_KEY)
        
        # 构建消息，假设screenshot_data是base64编码
        response = client.chat.completions.create(
            model=config.OPENAI_MODEL_VISION,
            messages=[
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": f"""
                        Analyze this UI defect for a design system component.
                        
                        Problem Description: {description}
                        
                        Technical Context:

                        - Component Type: {context.get('component_type')}
                        - Component Version: {context.get('component_version')}
                        - Recent Deployments: {context.get('recent_deployments', 'N/A')}
                        - User Actions (from trace): {context.get('user_actions', [])}
                        
                        Provide a concise root cause analysis, confidence score (0-1), and a specific code fix suggestion.
                        """},
                        {
                            "type": "image_url",
                            "image_url": {
                                "url": f"data:image/jpeg;base64,{screenshot_data}" if screenshot_data.startswith('/') else screenshot_data
                            }
                        }
                    ]
                }
            ],
            max_tokens=500
        )
        
        # 解析响应（此处简化，真实情况需更复杂的解析）
        ai_text = response.choices[0].message.content
        return {
            "analysis_id": response.id,
            "model_used": config.OPENAI_MODEL_VISION,
            "root_cause": ai_text,  # 实际应使用结构化输出或解析
            "confidence": 0.88,  # 示例
            "suggested_fix": {
                "note": "Fix extracted from AI response. Manual review required."
            }
        }
    
    else:
        raise ValueError(f"Unsupported MULTIMODAL_AI_MODE: {config.MULTIMODAL_AI_MODE}")

"""
主Flask应用，提供REST API端点。
"""
import json
import logging
from flask import Flask, request, jsonify
from flask_cors import CORS

from config.settings import config
from core.design_components import init_component_registry, COMPONENT_REGISTRY, DesignComponent
from core.observability import TraceContext, record_trace, TRACE_COLLECTOR
from core.issue_manager import issue_store, IssueStatus
from multimodal.analyzer import analyze_with_multimodal_ai

app = Flask(__name__)
CORS(app)
logging.basicConfig(level=logging.INFO)

# 初始化
init_component_registry()

@app.before_request
def start_trace():
    """为每个请求创建追踪上下文"""
    trace_id = request.headers.get(config.TRACE_HEADER) or str(request.url_rule) if request.url_rule else "unknown"
    request.trace_ctx = TraceContext(trace_id=trace_id)

@app.after_request
def finish_trace(response):
    """请求结束后记录追踪"""
    if hasattr(request, 'trace_ctx'):
        # 可以在这里记录请求级别的span
        pass
    return response

@app.route('/api/v1/component/render', methods=['POST'])
def render_component():
    """模拟渲染组件端点"""
    data = request.get_json()
    component_id = data.get('component_id')
    
    if component_id not in COMPONENT_REGISTRY:
        return jsonify({"error": "Component not found"}), 404
    
    component = COMPONENT_REGISTRY[component_id]
    span = request.trace_ctx.start_span("render_component", component_id)
    
    # 模拟渲染过程
    result = component.render(request.trace_ctx.trace_id)
    
    span.add_log(f"Rendered component {component_id}", 
                 defect=result.get('defect'),
                 ok=result['ok'])
    span.finish()
    
    # 记录本次渲染的追踪
    trace_data = {
        "trace_id": request.trace_ctx.trace_id,
        "spans": request.trace_ctx.get_all_spans(),
        "component_operation": "render"
    }
    record_trace(trace_data)
    
    # 如果渲染出错，自动创建故障工单
    if not result['ok']:
        # 模拟截图路径 (实际中应从监控系统获取)
        screenshot_ref = f"screenshots/{request.trace_ctx.trace_id}.png"
        issue = issue_store.create(
            title=f"Defect in {component.name}: {result.get('defect', 'unknown')}",
            description=result.get('defect_description', 'No description'),
            component_id=component_id,
            trace_id=request.trace_ctx.trace_id,
            screenshot_ref=screenshot_ref
        )
        result["issue_id"] = issue.id
        result["issue_created"] = True
    
    return jsonify(result)

@app.route('/api/v1/issue/<issue_id>/analyze', methods=['POST'])
def analyze_issue(issue_id):
    """触发对某个故障的多模态AI分析"""
    issue = issue_store.get(issue_id)
    if not issue:
        return jsonify({"error": "Issue not found"}), 404
    
    # 获取相关追踪数据
    relevant_traces = [t for t in TRACE_COLLECTOR if t.get('trace_id') == issue.trace_id]
    component = COMPONENT_REGISTRY.get(issue.component_id)
    
    # 准备分析上下文
    context = {
        "trace_id": issue.trace_id,
        "component_type": component.type.value if component else "Unknown",
        "component_version": component.version if component else "Unknown",
        "defect": issue.description.split(':')[0].lower() if ':' in issue.description else "unknown",
        "user_actions": relevant_traces[:3]  # 取前3个相关追踪
    }
    
    # 更新状态为分析中
    issue.update_status(IssueStatus.ANALYZING)
    
    # 调用多模态分析器 (使用模拟截图)
    screenshot_data = issue.screenshot_ref or "simulated_screenshot.png"
    try:
        ai_result = analyze_with_multimodal_ai(
            screenshot_data=screenshot_data,
            description=issue.description,
            context=context
        )
    except Exception as e:
        logging.error(f"AI analysis failed: {e}")
        issue.update_status(IssueStatus.OPEN, {"error": str(e)})
        issue_store.update(issue)
        return jsonify({"error": "Analysis failed", "details": str(e)}), 500
    
    # 更新工单状态和分析结果
    issue.update_status(IssueStatus.FIX_PROPOSED, ai_result)
    issue_store.update(issue)
    
    return jsonify({
        "issue_id": issue.id,
        "status": issue.status.value,
        "ai_analysis": ai_result
    })

@app.route('/api/v1/issue/<issue_id>/resolve', methods=['POST'])
def resolve_issue(issue_id):
    """标记故障为已解决（模拟修复部署后调用）"""
    issue = issue_store.get(issue_id)
    if not issue:
        return jsonify({"error": "Issue not found"}), 404
    
    data = request.get_json()
    fix_verification = data.get('verification', {})
    
    issue.update_status(IssueStatus.RESOLVED, fix_verification)
    issue_store.update(issue)
    
    return jsonify({
        "issue_id": issue.id,
        "status": issue.status.value,
        "resolved_at": issue.updated_at
    })

@app.route('/api/v1/issues', methods=['GET'])
def list_issues():
    """列出所有故障工单，可按状态过滤"""
    status_param = request.args.get('status')
    status = IssueStatus(status_param) if status_param else None
    
    issues = issue_store.list_by_status(status)
    return jsonify({
        "count": len(issues),
        "issues": [{
            "id": issue.id,
            "title": issue.title,
            "status": issue.status.value,
            "component_id": issue.component_id,
            "created_at": issue.created_at
        } for issue in issues]
    })

@app.route('/api/v1/traces', methods=['GET'])
def get_traces():
    """获取追踪数据（仅用于演示）"""
    return jsonify({
        "count": len(TRACE_COLLECTOR),
        "traces": TRACE_COLLECTOR[-10:]  # 返回最近10条
    })

@app.route('/api/v1/health', methods=['GET'])
def health():
    return jsonify({"status": "healthy", "service": "multimodal-design-observability"})

if __name__ == '__main__':
    app.run(host=config.HOST, port=config.PORT, debug=config.DEBUG)

"""
命令行模拟器，用于生成负载和测试故障闭环流程。
"""
import sys
import time
import random
import requests
import logging
from typing import Optional

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
BASE_URL = "http://localhost:5000/api/v1"

def simulate_user_session(session_id: str):
    """模拟一个用户会话：渲染多个组件，可能触发故障"""
    trace_id = f"TRACE-{session_id}"
    headers = {"X-Trace-ID": trace_id}
    
    components_to_render = ["btn-primary-1", "input-search-1", "modal-confirm-1", "table-user-1"]
    logging.info(f"[Session {session_id}] Starting with trace_id: {trace_id}")
    
    issues_created = []
    
    for comp_id in components_to_render:
        time.sleep(0.5)  # 模拟用户操作间隔
        payload = {"component_id": comp_id}
        try:
            resp = requests.post(f"{BASE_URL}/component/render", json=payload, headers=headers, timeout=5)
            data = resp.json()
            if resp.status_code == 200:
                if not data.get('ok'):
                    issue_id = data.get('issue_id')
                    if issue_id:
                        issues_created.append(issue_id)
                        logging.warning(f"  [!] Defect detected in {comp_id}. Issue created: {issue_id}")
                else:
                    logging.info(f"  [+] Rendered {comp_id} successfully.")
            else:
                logging.error(f"  [x] Failed to render {comp_id}: {data.get('error')}")
        except requests.exceptions.ConnectionError:
            logging.error("  [x] Cannot connect to server. Is it running?")
            sys.exit(1)
    
    return issues_created

def trigger_ai_analysis(issue_id: str):
    """触发对指定故障的AI分析"""
    logging.info(f"  [AI] Triggering analysis for issue {issue_id}...")
    try:
        resp = requests.post(f"{BASE_URL}/issue/{issue_id}/analyze", timeout=30)  # 分析可能较慢
        if resp.status_code == 200:
            data = resp.json()
            logging.info(f"  [AI] Analysis complete. Status: {data['status']}")
            if 'ai_analysis' in data:
                analysis = data['ai_analysis']
                logging.info(f"      Root Cause: {analysis.get('root_cause', 'N/A')[:100]}...")
                logging.info(f"      Confidence: {analysis.get('confidence')}")
            return data
        else:
            logging.error(f"  [AI] Analysis failed: {resp.json()}")
    except requests.exceptions.Timeout:
        logging.warning("  [AI] Analysis timed out (simulated slow AI).")

def close_issue(issue_id: str):
    """模拟修复后关闭工单"""
    logging.info(f"  [Fix] Marking issue {issue_id} as resolved...")
    payload = {"verification": {"verified_by": "auto-test", "result": "passed"}}
    resp = requests.post(f"{BASE_URL}/issue/{issue_id}/resolve", json=payload)
    if resp.status_code == 200:
        logging.info(f"  [Fix] Issue {issue_id} resolved successfully.")
    else:
        logging.error(f"  [Fix] Failed to resolve issue: {resp.json()}")

def run_demo_loop(num_sessions: int = 3):
    """运行完整的演示循环"""
    all_issues = []
    
    # 第1阶段：模拟用户会话，生成潜在故障
    logging.info("="*50)
    logging.info("PHASE 1: Simulating User Sessions & Defect Detection")
    logging.info("="*50)
    for i in range(num_sessions):
        issues = simulate_user_session(f"USER-{i+1}")
        all_issues.extend(issues)
        time.sleep(1)
    
    if not all_issues:
        logging.info("No defects were generated in this run. Try again!")
        # 为了演示，我们可以手动创建一个模拟问题
        logging.info("Creating a manual test issue for demo purposes...")
        # 这里可以添加手动创建工单的逻辑，但为了简洁，我们直接退出。
        return
    
    # 第2阶段：对每个故障进行AI分析
    logging.info("\n" + "="*50)
    logging.info("PHASE 2: Multimodal AI Root Cause Analysis")
    logging.info("="*50)
    for issue_id in all_issues:
        trigger_ai_analysis(issue_id)
        time.sleep(1)
    
    # 第3阶段：模拟修复与闭环
    logging.info("\n" + "="*50)
    logging.info("PHASE 3: Fault Resolution & Closure")
    logging.info("="*50)
    for issue_id in all_issues:
        close_issue(issue_id)
        time.sleep(0.5)
    
    # 最终状态查看
    logging.info("\n" + "="*50)
    logging.info("FINAL STATE: Listing all issues")
    logging.info("="*50)
    resp = requests.get(f"{BASE_URL}/issues")
    if resp.status_code == 200:
        data = resp.json()
        for issue in data['issues']:
            logging.info(f"  {issue['id']}: {issue['title']} -> {issue['status']}")

if __name__ == '__main__':
    run_demo_loop(2)

Flask==2.3.3
Flask-CORS==4.0.0
requests==2.31.0
# 如果使用真实的OpenAI API，取消下面一行的注释并配置API Key
# openai>=1.0.0

#!/bin/bash
# 启动脚本

echo "Installing dependencies..."
pip install -r requirements.txt

echo "Starting the Multimodal Design Observability Server..."
python app.py &
SERVER_PID=$!

# 等待服务器启动
sleep 3

echo "Running the simulation CLI..."
python cli/simulator.py

# 关闭服务器
echo "Stopping server (PID: $SERVER_PID)..."
kill $SERVER_PID

mkdir multimodal-design-observability
    cd multimodal-design-observability
    # 将上述所有代码文件放入对应目录

pip install -r requirements.txt

# 在Linux/macOS上
    chmod +x run.sh
    ./run.sh

    # 或在Windows上手动执行
    # 第一个终端：
    python app.py
    # 第二个终端：
    python cli/simulator.py

{"component_id": "btn-primary-1"}

import sys
sys.path.insert(0, '.')

from core.design_components import DesignComponent, ComponentType, ComponentDefect, init_component_registry, COMPONENT_REGISTRY
from core.issue_manager import Issue, IssueStatus, IssueStore
from multimodal.analyzer import analyze_with_multimodal_ai
import tempfile
import json
import os

def test_component_defect_detection():
    """测试组件缺陷检测逻辑"""
    init_component_registry()
    component = COMPONENT_REGISTRY["btn-primary-1"]
    # 由于缺陷是随机引入的，我们多次渲染以增加触发几率或直接测试状态
    print("Testing component registry... OK")
    assert component.id == "btn-primary-1"
    assert component.type == ComponentType.BUTTON
    print("Component instantiation... OK")

def test_issue_lifecycle():
    """测试故障工单状态流转"""
    with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
        temp_path = f.name
    
    try:
        store = IssueStore(temp_path)
        issue = store.create(
            title="Test Button Misalignment",
            description="Button appears 5px off",
            component_id="btn-test-1",
            trace_id="trace-123"
        )
        assert issue.status == IssueStatus.OPEN
        assert store.get(issue.id) is not None
        
        issue.update_status(IssueStatus.ANALYZING, {"analysis_started": True})
        store.update(issue)
        retrieved = store.get(issue.id)
        assert retrieved.status == IssueStatus.ANALYZING
        
        open_issues = store.list_by_status(IssueStatus.OPEN)
        assert len(open_issues) == 0  # 我们的测试工单状态已不是OPEN
        
        print("Issue lifecycle... OK")
    finally:
        os.unlink(temp_path)

def test_simulated_ai_analysis():
    """测试模拟的多模态分析"""
    # 使用模拟模式进行分析
    result = analyze_with_multimodal_ai(
        screenshot_data="simulated_path.png",
        description="Button misaligned",
        context={"component_type": "Button", "defect": "visual_misalignment"}
    )
    assert "root_cause" in result
    assert "confidence" in result
    assert 0 <= result["confidence"] <= 1
    assert "suggested_fix" in result
    print("Simulated AI analysis... OK")

if __name__ == "__main__":
    test_component_defect_detection()
    test_issue_lifecycle()
    test_simulated_ai_analysis()
    print("\nAll basic tests passed!")

python test_basic.py