云端AI IDE架构：Client、Gateway、Agent、Executor、Storage

// 代码示例2-1：Monaco Editor初始化配置
import * as monaco from 'monaco-editor';

// 配置加载器
self.MonacoEnvironment = {
    async getWorker(_, label) {
        const getWorkerModule = (moduleUrl, localUrl) => {
            return new Worker(
                self.MonacoEnvironment.getWorkerUrl(moduleUrl, localUrl),
                { type: 'module' }
            );
        };

        // 根据语言加载对应的语言服务Worker
        switch (label) {
            case 'json':
                return getWorkerModule('/monaco-editor/esm/vs/language/json/json.worker?worker', 'json');
            case 'css':
            case 'scss':
            case 'less':
                return getWorkerModule('/monaco-editor/esm/vs/language/css/css.worker?worker', 'css');
            case 'html':
            case 'handlebars':
            case 'razor':
                return getWorkerModule('/monaco-editor/esm/vs/language/html/html.worker?worker', 'html');
            case 'typescript':
            case 'javascript':
                return getWorkerModule('/monaco-editor/esm/vs/language/typescript/ts.worker?worker', 'ts');
            default:
                return getWorkerModule('/monaco-editor/esm/vs/editor/editor.worker?worker', 'editor');
        }
    }
};

// 编辑器实例创建
const editor = monaco.editor.create(document.getElementById('container'), {
    value: '',
    language: 'typescript',
    theme: 'vs-dark',
    fontSize: 14,
    fontFamily: "'JetBrains Mono', 'Fira Code', Consolas, monospace",
    fontLigatures: true,
    lineNumbers: 'on',
    minimap: { enabled: true },
    scrollBeyondLastLine: false,
    automaticLayout: true,
    tabSize: 4,
    insertSpaces: true,
    wordWrap: 'on',
    cursorBlinking: 'smooth',
    cursorSmoothCaretAnimation: 'on',
    smoothScrolling: true,
    renderLineHighlight: 'all',
    renderWhitespace: 'selection',
    bracketPairColorization: { enabled: true },
    guides: {
        bracketPairs: true,
        indentation: true
    },
    // AI辅助功能集成
    quickSuggestions: {
        other: true,
        comments: false,
        strings: true
    },
    parameterHints: { enabled: true },
    suggestOnTriggerCharacters: true,
    acceptSuggestionOnEnter: 'smart',
    tabCompletion: 'on'
});

// 监听内容变更，触发AI上下文同步
editor.onDidChangeModelContent(() => {
    const model = editor.getModel();
    const content = model.getValue();
    const cursorPosition = editor.getPosition();
    
    // 向后端发送增量更新
    contextSyncService.sendIncrementalUpdate({
        content,
        cursor: { line: cursorPosition.lineNumber, column: cursorPosition.column },
        language: model.getLanguageId(),
        version: model.getVersionId()
    });
});

// 代码示例2-2：WebSocket通信管理器实现
class GatewayClient {
    constructor(config) {
        this.config = config;
        this.ws = null;
        this.reconnectAttempts = 0;
        this.maxReconnectAttempts = 5;
        this.reconnectDelay = 1000;
        this.heartbeatInterval = null;
        this.pendingRequests = new Map();
        this.messageQueue = [];
        this.requestId = 0;
    }

    async connect() {
        return new Promise((resolve, reject) => {
            const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:';
            const wsUrl = `${protocol}//${this.config.gatewayHost}:${this.config.gatewayPort}/ai-ide`;

            this.ws = new WebSocket(wsUrl);
            this.ws.binaryType = 'arraybuffer';

            this.ws.onopen = () => {
                console.log('[GatewayClient] WebSocket connected');
                this.reconnectAttempts = 0;
                this.startHeartbeat();
                this.flushMessageQueue();
                this.performAuth();
                resolve();
            };

            this.ws.onmessage = (event) => {
                this.handleMessage(event);
            };

            this.ws.onerror = (error) => {
                console.error('[GatewayClient] WebSocket error:', error);
            };

            this.ws.onclose = (event) => {
                console.log('[GatewayClient] WebSocket closed:', event.code, event.reason);
                this.stopHeartbeat();
                this.handleReconnect();
            };
        });
    }

    performAuth() {
        const authMessage = {
            type: 'auth',
            payload: {
                token: this.config.authToken,
                workspaceId: this.config.workspaceId,
                clientVersion: this.config.clientVersion
            }
        };
        this.send(authMessage);
    }

    send(message) {
        const msgStr = JSON.stringify(message);
        if (this.ws && this.ws.readyState === WebSocket.OPEN) {
            this.ws.send(msgStr);
        } else {
            this.messageQueue.push(message);
        }
    }

    async sendRequest(type, payload, timeout = 30000) {
        const requestId = `req_${++this.requestId}_${Date.now()}`;
        
        return new Promise((resolve, reject) => {
            const timer = setTimeout(() => {
                this.pendingRequests.delete(requestId);
                reject(new Error(`Request ${type} timeout after ${timeout}ms`));
            }, timeout);

            this.pendingRequests.set(requestId, { resolve, reject, timer });

            this.send({
                type: 'request',
                id: requestId,
                payload: { type, ...payload }
            });
        });
    }

    async requestAICompletion(context) {
        return this.sendRequest('ai.completion', {
            prompt: context.prompt,
            language: context.language,
            cursorContext: context.cursorContext,
            maxTokens: context.maxTokens || 500,
            temperature: context.temperature || 0.7,
            stream: context.stream !== false
        });
    }

    async requestCodeExecution(code, language, args = {}) {
        return this.sendRequest('executor.run', {
            code,
            language,
            stdin: args.stdin || '',
            timeout: args.timeout || 30000,
            resourceLimits: args.resourceLimits || { memory: '256MB', cpu: '1' }
        });
    }

    subscribeFileChanges(callback) {
        this.on('file.change', callback);
    }

    subscribeTerminalOutput(callback) {
        this.on('terminal.output', callback);
    }

    on(eventType, callback) {
        if (!this.eventHandlers) {
            this.eventHandlers = new Map();
        }
        if (!this.eventHandlers.has(eventType)) {
            this.eventHandlers.set(eventType, []);
        }
        this.eventHandlers.get(eventType).push(callback);
    }

    handleMessage(event) {
        let message;
        if (event.data instanceof ArrayBuffer) {
            const decoder = new TextDecoder();
            message = JSON.parse(decoder.decode(event.data));
        } else {
            message = JSON.parse(event.data);
        }

        const { type, id, payload } = message;

        if (type === 'response' && id) {
            const pending = this.pendingRequests.get(id);
            if (pending) {
                clearTimeout(pending.timer);
                this.pendingRequests.delete(id);
                if (payload.error) {
                    pending.reject(new Error(payload.error));
                } else {
                    pending.resolve(payload);
                }
            }
        } else if (type === 'event') {
            const handlers = this.eventHandlers?.get(payload.eventType) || [];
            handlers.forEach(handler => handler(payload.data));
        } else if (type === 'stream') {
            const handlers = this.eventHandlers?.get('ai.stream') || [];
            handlers.forEach(handler => handler(payload));
        }
    }

    startHeartbeat() {
        this.heartbeatInterval = setInterval(() => {
            if (this.ws?.readyState === WebSocket.OPEN) {
                this.send({ type: 'ping' });
            }
        }, 30000);
    }

    stopHeartbeat() {
        if (this.heartbeatInterval) {
            clearInterval(this.heartbeatInterval);
            this.heartbeatInterval = null;
        }
    }

    flushMessageQueue() {
        while (this.messageQueue.length > 0) {
            const msg = this.messageQueue.shift();
            this.send(msg);
        }
    }

    handleReconnect() {
        if (this.reconnectAttempts < this.maxReconnectAttempts) {
            this.reconnectAttempts++;
            const delay = this.reconnectDelay * Math.pow(2, this.reconnectAttempts - 1);
            console.log(`[GatewayClient] Reconnecting in ${delay}ms (attempt ${this.reconnectAttempts})`);
            setTimeout(() => this.connect(), delay);
        } else {
            console.error('[GatewayClient] Max reconnection attempts reached');
            this.emit('disconnected');
        }
    }

    disconnect() {
        this.stopHeartbeat();
        if (this.ws) {
            this.ws.close(1000, 'Client disconnect');
            this.ws = null;
        }
    }
}

export default GatewayClient;

// 代码示例2-3：虚拟滚动实现原理
class VirtualScrollManager {
    constructor(options) {
        this.container = options.container;
        this.itemHeight = options.itemHeight || 20;
        this.bufferSize = options.bufferSize || 10;
        this.items = [];
        this.renderedItems = new Map();
        
        this.scrollTop = 0;
        this.containerHeight = 0;
        this.totalHeight = 0;
        
        this.init();
    }

    init() {
        this.container.style.overflow = 'auto';
        this.container.style.position = 'relative';
        
        this.contentElement = document.createElement('div');
        this.contentElement.style.position = 'absolute';
        this.contentElement.style.left = '0';
        this.contentElement.style.right = '0';
        this.contentElement.style.top = '0';
        this.contentElement.style.pointerEvents = 'none';
        this.container.appendChild(this.contentElement);
        
        this.container.addEventListener('scroll', this.onScroll.bind(this));
        
        this.updateContainerHeight();
        window.addEventListener('resize', this.updateContainerHeight.bind(this));
    }

    updateContainerHeight() {
        this.containerHeight = this.container.clientHeight;
    }

    setItems(items) {
        this.items = items;
        this.totalHeight = items.length * this.itemHeight;
        this.contentElement.style.height = `${this.totalHeight}px`;
        this.render();
    }

    onScroll() {
        this.scrollTop = this.container.scrollTop;
        requestAnimationFrame(() => this.render());
    }

    render() {
        const visibleStart = Math.floor(this.scrollTop / this.itemHeight);
        const visibleEnd = Math.ceil((this.scrollTop + this.containerHeight) / this.itemHeight);
        
        const renderStart = Math.max(0, visibleStart - this.bufferSize);
        const renderEnd = Math.min(this.items.length, visibleEnd + this.bufferSize);
        
        for (const [index, element] of this.renderedItems) {
            if (index < renderStart || index >= renderEnd) {
                element.remove();
                this.renderedItems.delete(index);
            }
        }
        
        for (let i = renderStart; i < renderEnd; i++) {
            if (!this.renderedItems.has(i)) {
                const element = this.createItemElement(i, this.items[i]);
                element.style.position = 'absolute';
                element.style.top = `${i * this.itemHeight}px`;
                element.style.width = '100%';
                this.contentElement.appendChild(element);
                this.renderedItems.set(i, element);
            }
        }
    }

    createItemElement(index, item) {
        const element = document.createElement('div');
        element.className = 'virtual-scroll-item';
        element.innerHTML = this.itemRenderer(item, index);
        return element;
    }

    itemRenderer(item, index) {
        return `<span>${item.content}</span>`;
    }

    dispose() {
        this.container.removeEventListener('scroll', this.onScroll);
        window.removeEventListener('resize', this.updateContainerHeight.bind(this));
        this.contentElement.remove();
    }
}

# 代码示例3-1：Gateway路由配置示例
routes:
  - path: /api/ai/*
    upstream: ai-service-pool
    loadBalancer:
      algorithm: weighted_least_connections
      weights:
        gpu-node-1: 3
        gpu-node-2: 3
        gpu-node-3: 2
    retry:
      attempts: 2
      perTryTimeout: 10s
    timeout: 60s

  - path: /api/executor/*
    upstream: executor-service-pool
    loadBalancer:
      algorithm: round_robin
    retry:
      attempts: 3
      perTryTimeout: 30s
    timeout: 300s

  - path: /api/storage/*
    upstream: storage-service-pool
    loadBalancer:
      algorithm: consistent_hash
      hashKeys:
        - workspace_id
    retry:
      attempts: 3
      perTryTimeout: 5s
    timeout: 10s

  - path: /ws/*
    upstream: websocket-pool
    loadBalancer:
      algorithm: sticky_session
      cookieName: AIIDE_SESSION
    timeout: 3600s

healthChecks:
  - name: ai-service-health
    interval: 10s
    timeout: 5s
    healthyThreshold: 2
    unhealthyThreshold: 3
    httpGet:
      path: /health
      port: 8080

# 代码示例3-2：WebSocket连接管理器实现
import asyncio
import uuid
import time
from typing import Dict, Optional, Set
from dataclasses import dataclass, field
from enum import Enum

class ConnectionState(Enum):
    CONNECTING = "connecting"
    AUTHENTICATING = "authenticating"
    AUTHENTICATED = "authenticated"
    READY = "ready"
    RECONNECTING = "reconnecting"
    DISCONNECTED = "disconnected"
    CLOSED = "closed"

@dataclass
class WebSocketConnection:
    connection_id: str
    websocket: object
    user_id: str
    workspace_id: str
    state: ConnectionState = ConnectionState.CONNECTING
    created_at: float = field(default_factory=time.time)
    last_active: float = field(default_factory=time.time)
    message_count: int = 0
    bytes_received: int = 0
    bytes_sent: int = 0
    tags: Dict[str, str] = field(default_factory=dict)

class WebSocketConnectionManager:
    def __init__(self, config):
        self.config = config
        self.connections: Dict[str, WebSocketConnection] = {}
        self.user_connections: Dict[str, Set[str]] = {}
        self.workspace_connections: Dict[str, Set[str]] = {}
        self.backend_pools: Dict[str, asyncio.Queue] = {}
        self.heartbeat_task: Optional[asyncio.Task] = None
        self.stats = {
            "total_connections": 0,
            "active_connections": 0,
            "messages_processed": 0,
            "bytes_transferred": 0,
            "reconnections": 0,
            "disconnections": 0
        }
    
    async def start(self):
        self.heartbeat_task = asyncio.create_task(self._heartbeat_loop())
        asyncio.create_task(self._cleanup_loop())
    
    async def stop(self):
        if self.heartbeat_task:
            self.heartbeat_task.cancel()
            await self.heartbeat_task
        
        for conn_id in list(self.connections.keys()):
            await self.disconnect(conn_id, code=1001, reason="Server shutdown")
    
    async def connect(self, websocket, user_id: str, workspace_id: str, metadata: Dict = None) -> WebSocketConnection:
        conn_id = str(uuid.uuid4())
        
        connection = WebSocketConnection(
            connection_id=conn_id,
            websocket=websocket,
            user_id=user_id,
            workspace_id=workspace_id,
            state=ConnectionState.AUTHENTICATING,
            tags=metadata or {}
        )
        
        self.connections[conn_id] = connection
        self.stats["total_connections"] += 1
        self.stats["active_connections"] += 1
        
        if user_id not in self.user_connections:
            self.user_connections[user_id] = set()
        self.user_connections[user_id].add(conn_id)
        
        if workspace_id not in self.workspace_connections:
            self.workspace_connections[workspace_id] = set()
        self.workspace_connections[workspace_id].add(conn_id)
        
        await self.send_to_connection(conn_id, {
            "type": "connection.established",
            "connectionId": conn_id,
            "serverTime": time.time()
        })
        
        return connection
    
    async def authenticate(self, conn_id: str, token: str) -> bool:
        connection = self.connections.get(conn_id)
        if not connection:
            return False
        
        is_valid = await self._verify_token(token)
        if is_valid:
            connection.state = ConnectionState.AUTHENTICATED
            await self.send_to_connection(conn_id, {
                "type": "auth.success",
                "connectionId": conn_id
            })
        else:
            await self.disconnect(conn_id, code=4001, reason="Authentication failed")
        
        return is_valid
    
    async def disconnect(self, conn_id: str, code: int = 1000, reason: str = "Normal closure"):
        connection = self.connections.pop(conn_id, None)
        if not connection:
            return
        
        self.user_connections.get(connection.user_id, set()).discard(conn_id)
        self.workspace_connections.get(connection.workspace_id, set()).discard(conn_id)
        
        try:
            await connection.websocket.close(code, reason)
        except Exception:
            pass
        
        self.stats["active_connections"] -= 1
        self.stats["disconnections"] += 1
        
        await self._emit_event("connection.closed", {
            "connectionId": conn_id,
            "userId": connection.user_id,
            "workspaceId": connection.workspace_id,
            "reason": reason,
            "duration": time.time() - connection.created_at,
            "messageCount": connection.message_count
        })
    
    async def send_to_connection(self, conn_id: str, message: dict):
        connection = self.connections.get(conn_id)
        if not connection:
            return False
        
        try:
            await connection.websocket.send_json(message)
            connection.last_active = time.time()
            connection.bytes_sent += len(str(message))
            return True
        except Exception as e:
            print(f"Failed to send to connection {conn_id}: {e}")
            await self.disconnect(conn_id, code=1011, reason="Send error")
            return False
    
    async def broadcast_to_workspace(self, workspace_id: str, message: dict, exclude: Set[str] = None):
        conn_ids = self.workspace_connections.get(workspace_id, set())
        exclude = exclude or set()
        
        tasks = [
            self.send_to_connection(conn_id, message)
            for conn_id in conn_ids
            if conn_id not in exclude
        ]
        
        if tasks:
            await asyncio.gather(*tasks, return_exceptions=True)
    
    async def _heartbeat_loop(self):
        while True:
            try:
                await asyncio.sleep(self.config.get("heartbeat_interval", 30))
                await self._check_idle_connections()
            except asyncio.CancelledError:
                break
            except Exception as e:
                print(f"Heartbeat error: {e}")
    
    async def _check_idle_connections(self):
        now = time.time()
        idle_threshold = self.config.get("idle_timeout", 300)
        
        for conn_id, conn in list(self.connections.items()):
            if conn.state in (ConnectionState.CONNECTING, ConnectionState.AUTHENTICATING):
                if now - conn.created_at > 30:
                    await self.disconnect(conn_id, code=4002, reason="Authentication timeout")
            elif now - conn.last_active > idle_threshold:
                try:
                    await conn.websocket.ping()
                    conn.last_active = now
                except Exception:
                    await self.disconnect(conn_id, code=1000, reason="Connection timeout")
    
    async def _verify_token(self, token: str) -> bool:
        return True
    
    async def _emit_event(self, event_type: str, data: dict):
        pass
    
    def get_stats(self) -> dict:
        return {
            **self.stats,
            "connection_states": {
                state.value: sum(1 for c in self.connections.values() if c.state == state)
                for state in ConnectionState
            },
            "active_workspaces": len(self.workspace_connections),
            "active_users": len(self.user_connections)
        }

# 代码示例3-3：JWT认证中间件实现
from fastapi import Request, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, JWTError, ExpiredSignatureError
from datetime import datetime, timedelta
import hashlib

security = HTTPBearer()

class JWTAuthenticator:
    def __init__(self, config):
        self.config = config
        self.algorithm = "RS256"
        self.access_token_expire = 15
        self.refresh_token_expire = 7
    
    def create_access_token(self, user_id: str, workspace_id: str, permissions: list) -> str:
        now = datetime.utcnow()
        expire = now + timedelta(minutes=self.access_token_expire)
        
        payload = {
            "sub": user_id,
            "workspace": workspace_id,
            "permissions": permissions,
            "iat": now.timestamp(),
            "exp": expire.timestamp(),
            "jti": self._generate_jti(),
            "type": "access"
        }
        
        return jwt.encode(payload, self.config.private_key, algorithm=self.algorithm)
    
    def create_refresh_token(self, user_id: str) -> str:
        now = datetime.utcnow()
        expire = now + timedelta(days=self.refresh_token_expire)
        
        payload = {
            "sub": user_id,
            "iat": now.timestamp(),
            "exp": expire.timestamp(),
            "jti": self._generate_jti(),
            "type": "refresh"
        }
        
        return jwt.encode(payload, self.config.private_key, algorithm=self.algorithm)
    
    def verify_token(self, token: str) -> dict:
        try:
            payload = jwt.decode(
                token,
                self.config.public_key,
                algorithms=[self.algorithm]
            )
            
            if payload.get("type") not in ("access", "refresh"):
                raise JWTError("Invalid token type")
            
            return payload
            
        except ExpiredSignatureError:
            raise HTTPException(status_code=401, detail="Token expired")
        except JWTError as e:
            raise HTTPException(status_code=401, detail=f"Invalid token: {str(e)}")
    
    def _generate_jti(self) -> str:
        import secrets
        random_bytes = secrets.token_bytes(16)
        return hashlib.sha256(random_bytes).hexdigest()

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    authenticator: JWTAuthenticator = Depends(get_authenticator)
) -> dict:
    token = credentials.credentials
    payload = authenticator.verify_token(token)
    
    if payload["type"] != "access":
        raise HTTPException(status_code=401, detail="Invalid token type")
    
    return {
        "user_id": payload["sub"],
        "workspace_id": payload["workspace"],
        "permissions": payload["permissions"]
    }

def require_permission(permission: str):
    async def check_permission(user: dict = Depends(get_current_user)):
        if permission not in user.get("permissions", []):
            raise HTTPException(
                status_code=403,
                detail=f"Permission denied: {permission}"
            )
        return user
    return check_permission

# 代码示例3-4：分布式限流实现
from fastapi import Request, HTTPException, Depends
from typing import Callable
import time

class DistributedRateLimiter:
    def __init__(self, redis_client, config):
        self.redis = redis_client
        self.config = config
    
    async def is_allowed(self, key: str, limit: int, window: int) -> bool:
        now = time.time()
        window_start = now - window
        
        pipe = self.redis.pipeline()
        pipe.zremrangebyscore(key, 0, window_start)
        pipe.zcard(key)
        results = await pipe.execute()
        current_count = results[1]
        
        if current_count >= limit:
            return False
        
        await self.redis.zadd(key, {str(now): now})
        await self.redis.expire(key, window)
        
        return True

class RateLimitMiddleware:
    def __init__(self, rate_limiter: DistributedRateLimiter):
        self.rate_limiter = rate_limiter
        self.limits = {
            "/api/ai/completion": {"limit": 60, "window": 60},
            "/api/ai/chat": {"limit": 30, "window": 60},
            "/api/executor/run": {"limit": 20, "window": 60},
            "/api/storage/*": {"limit": 100, "window": 60},
        }
    
    async def __call__(self, request: Request, call_next):
        client_id = await self._get_client_id(request)
        path = request.url.path
        limit_config = self._match_path(path)
        
        if limit_config:
            key = f"rate_limit:{client_id}:{path}"
            is_allowed = await self.rate_limiter.is_allowed(
                key,
                limit_config["limit"],
                limit_config["window"]
            )
            
            if not is_allowed:
                raise HTTPException(
                    status_code=429,
                    detail="Rate limit exceeded",
                    headers={"Retry-After": str(limit_config["window"])}
                )
        
        response = await call_next(request)
        return response
    
    async def _get_client_id(self, request: Request) -> str:
        if hasattr(request.state, "user_id"):
            return f"user:{request.state.user_id}"
        
        client_ip = request.client.host
        forwarded = request.headers.get("X-Forwarded-For")
        if forwarded:
            client_ip = forwarded.split(",")[0].strip()
        
        return f"ip:{client_ip}"
    
    def _match_path(self, path: str) -> dict:
        for pattern, config in self.limits.items():
            if pattern.endswith("*"):
                if path.startswith(pattern[:-1]):
                    return config
            elif path == pattern:
                return config
        return None

# 代码示例4-1：模型路由器实现
from typing import List, Dict, Optional, Callable
from dataclasses import dataclass
from enum import Enum
import asyncio
import time
import hashlib

class ModelProvider(Enum):
    OPENAI = "openai"
    ANTHROPIC = "anthropic"
    GOOGLE = "google"
    LOCAL = "local"
    CUSTOM = "custom"

@dataclass
class ModelConfig:
    provider: ModelProvider
    model_name: str
    display_name: str
    max_tokens: int
    context_window: int
    capabilities: List[str]
    latency_target_ms: int
    cost_per_1k_tokens: float
    routing_weight: float

class ModelRouter:
    def __init__(self, config):
        self.config = config
        self.models: Dict[str, ModelConfig] = {}
        self.model_clients: Dict[str, "ModelClient"] = {}
        self.routing_cache: Dict[str, str] = {}
        self.strategy = config.get("strategy", "latency_cost_balanced")
    
    def register_model(self, model_id: str, model_config: ModelConfig, client: "ModelClient"):
        self.models[model_id] = model_config
        self.model_clients[model_id] = client
    
    async def route(self, task: "Task", context: dict) -> str:
        cache_key = self._compute_cache_key(task, context)
        
        if cache_key in self.routing_cache:
            return self.routing_cache[cache_key]
        
        candidates = self._filter_candidates(task)
        
        if not candidates:
            raise ValueError(f"No suitable model for task: {task.type}")
        
        if self.strategy == "latency_optimal":
            selected = await self._select_by_latency(candidates, task)
        elif self.strategy == "cost_optimal":
            selected = self._select_by_cost(candidates, task)
        elif self.strategy == "quality_optimal":
            selected = await self._select_by_quality(candidates, task)
        else:
            selected = await self._select_balanced(candidates, task)
        
        self.routing_cache[cache_key] = selected
        
        return selected
    
    def _filter_candidates(self, task: "Task") -> List[str]:
        required_capabilities = self._get_required_capabilities(task)
        
        candidates = []
        for model_id, config in self.models.items():
            if all(cap in config.capabilities for cap in required_capabilities):
                if config.context_window >= task.context_length:
                    candidates.append(model_id)
        
        return candidates
    
    async def _select_by_latency(self, candidates: List[str], task: "Task") -> str:
        scores = {}
        
        for model_id in candidates:
            config = self.models[model_id]
            client = self.model_clients[model_id]
            
            current_latency = await client.get_current_latency()
            adjusted_latency = current_latency * (1 + task.complexity * 0.1)
            
            scores[model_id] = adjusted_latency
        
        return min(scores, key=scores.get)
    
    def _select_by_cost(self, candidates: List[str], task: "Task") -> str:
        estimated_tokens = self._estimate_tokens(task)
        
        costs = {}
        for model_id in candidates:
            config = self.models[model_id]
            cost = config.cost_per_1k_tokens * (estimated_tokens / 1000)
            costs[model_id] = cost
        
        return min(costs, key=costs.get)
    
    async def _select_by_quality(self, candidates: List[str], task: "Task") -> str:
        quality_scores = {}
        
        for model_id in candidates:
            config = self.models[model_id]
            
            quality = 0
            if "reasoning" in config.capabilities:
                quality += 3
            if "code_generation" in config.capabilities:
                quality += 3
            if config.context_window > 100000:
                quality += 2
            
            historical_quality = await self._get_historical_quality(model_id, task.type)
            quality += historical_quality
            
            quality_scores[model_id] = quality
        
        return max(quality_scores, key=quality_scores.get)
    
    async def _select_balanced(self, candidates: List[str], task: "Task") -> str:
        estimated_tokens = self._estimate_tokens(task)
        
        scores = {}
        for model_id in candidates:
            config = self.models[model_id]
            client = self.model_clients[model_id]
            
            current_latency = await client.get_current_latency()
            estimated_cost = config.cost_per_1k_tokens * (estimated_tokens / 1000)
            
            latency_score = current_latency / 1000
            cost_score = estimated_cost / 0.1
            
            quality_weight = 0.4
            latency_weight = 0.3
            cost_weight = 0.3
            
            quality_score = self._estimate_quality(model_id, task)
            scores[model_id] = (
                quality_weight * (1 - quality_score) +
                latency_weight * latency_score +
                cost_weight * cost_score
            )
        
        return min(scores, key=scores.get)
    
    def _compute_cache_key(self, task: "Task", context: dict) -> str:
        content = f"{task.type}:{task.prompt[:100]}:{context.get('language', 'unknown')}"
        return hashlib.md5(content.encode()).hexdigest()
    
    def _get_required_capabilities(self, task: "Task") -> List[str]:
        capability_map = {
            "code_completion": ["code_generation"],
            "code_generation": ["code_generation"],
            "code_review": ["code_generation", "reasoning"],
            "debugging": ["reasoning"],
            "architecture_design": ["reasoning", "code_generation"],
            "multimodal": ["multimodal"]
        }
        return capability_map.get(task.type, ["code_generation"])
    
    def _estimate_tokens(self, task: "Task") -> int:
        prompt_tokens = len(task.prompt.split()) * 1.3
        return int(prompt_tokens * 1.5)
    
    async def _get_historical_quality(self, model_id: str, task_type: str) -> float:
        return 0.8
    
    def _estimate_quality(self, model_id: str, task: "Task") -> float:
        config = self.models[model_id]
        quality = 0.5
        
        if "reasoning" in config.capabilities:
            quality += 0.2
        if "code_generation" in config.capabilities:
            quality += 0.2
        if config.context_window > 50000:
            quality += 0.1
        
        return min(quality, 1.0)

# 代码示例4-2：提示词引擎实现
from typing import List, Dict, Optional, Any
from dataclasses import dataclass, field
import json
import re

@dataclass
class PromptTemplate:
    name: str
    system_template: str
    user_template: str
    examples: List[Dict[str, str]] = field(default_factory=list)
    variables: List[str] = field(default_factory=list)

@dataclass
class ContextEntry:
    content: str
    source: str
    relevance_score: float = 1.0
    token_count: int = 0
    metadata: Dict[str, Any] = field(default_factory=dict)

class PromptEngine:
    def __init__(self, config):
        self.config = config
        self.templates: Dict[str, PromptTemplate] = {}
        self.context_window = config.get("context_window", 128000)
        self.reserved_tokens = config.get("reserved_tokens", 4000)
        
        self._register_default_templates()
    
    def _register_default_templates(self):
        self.templates["code_generation"] = PromptTemplate(
            name="code_generation",
            system_template="""You are an expert {language} programmer. Generate clean, efficient, and well-documented code based on the user's request.

Code should follow these principles:
- Type hints for all function signatures
- Docstrings for complex logic
- Error handling for expected failure cases
- Performance considerations for hot paths

Available context from the project:
{files_context}""",
            user_template="""## Task
{instruction}

## Current File
The code should be added to: `{file_path}`

## Existing Code (first {max_lines} lines)
```{language}
{existing_code}

    self.templates["code_review"] = PromptTemplate(
        name="code_review",
        system_template="""You are a senior code reviewer with expertise in {language}, software architecture, and security. Review the provided code and give constructive feedback.

{code}

def build_prompt(
    self,
    template_name: str,
    context: List[ContextEntry],
    variables: Dict[str, Any],
    max_context_tokens: Optional[int] = None
) -> Dict[str, str]:
    if template_name not in self.templates:
        raise ValueError(f"Unknown template: {template_name}")
    
    template = self.templates[template_name]
    
    filled_system = self._fill_template(template.system_template, variables)
    filled_user = self._fill_template(template.user_template, variables)
    
    available_tokens = (max_context_tokens or self.context_window) - self.reserved_tokens
    available_tokens -= self._estimate_tokens(filled_system)
    available_tokens -= self._estimate_tokens(filled_user)
    available_tokens -= self._estimate_examples_tokens(template.examples)
    
    compressed_context = self._compress_context(context, available_tokens)
    
    files_context = self._format_files_context(compressed_context)
    filled_system = filled_system.format(files_context=files_context)
    
    return {
        "system": filled_system,
        "user": filled_user,
        "examples": template.examples,
        "context_entries": compressed_context
    }

def _fill_template(self, template: str, variables: Dict[str, Any]) -> str:
    result = template
    
    for key, value in variables.items():
        if isinstance(value, str):
            result = result.replace(f"{{{key}}}", value)
        elif value is None:
            result = result.replace(f"{{{key}}}", "(not provided)")
    
    return result

def _compress_context(
    self,
    context: List[ContextEntry],
    max_tokens: int
) -> List[ContextEntry]:
    if not context:
        return []
    
    current_tokens = sum(entry.token_count for entry in context)
    
    if current_tokens <= max_tokens:
        return context
    
    sorted_context = sorted(context, key=lambda x: x.relevance_score, reverse=True)
    
    compressed = []
    used_tokens = 0
    
    for entry in sorted_context:
        if used_tokens + entry.token_count > max_tokens:
            remaining_tokens = max_tokens - used_tokens
            compressed_entry = await self._summarize_entry(entry, remaining_tokens)
            compressed.append(compressed_entry)
            used_tokens += compressed_entry.token_count
            break
        
        compressed.append(entry)
        used_tokens += entry.token_count
    
    return compressed

async def _summarize_entry(self, entry: ContextEntry, max_tokens: int) -> ContextEntry:
    summarization_prompt = f"""Summarize the following code in no more than {max_tokens} tokens,

    summary = await self._call_summary_model(summarization_prompt)
    
    return ContextEntry(
        content=summary,
        source=f"{entry.source} (summarized)",
        relevance_score=entry.relevance_score * 0.9,
        token_count=max_tokens,
        metadata={**entry.metadata, "original_length": entry.token_count}
    )

def _format_files_context(self, context: List[ContextEntry]) -> str:
    file_entries = [e for e in context if e.source == "file"]
    
    if not file_entries:
        return "No project files available in context."
    
    formatted = []
    for entry in file_entries:
        metadata = entry.metadata
        formatted.append(f"""

{entry.content}

    return "\n".join(formatted)

def _estimate_tokens(self, text: str) -> int:
    chinese_chars = len(re.findall(r'[\u4e00-\u9fff]', text))
    english_words = len(re.findall(r'[a-zA-Z]+', text))
    other_chars = len(text) - chinese_chars
    
    return int(chinese_chars * 1.5 + english_words * 1.3 + other_chars)

def _estimate_examples_tokens(self, examples: List[Dict[str, str]]) -> int:
    total = 0
    for ex in examples:
        total += self._estimate_tokens(ex.get("input", ""))
        total += self._estimate_tokens(ex.get("output", ""))
    return total

async def _call_summary_model(self, prompt: str) -> str:
    return prompt[:500] + "..."

### 4.4 工具调用与Agent编排

现代AI Agent需要能够调用各种工具来完成任务。OpenAI的Function Calling[^19]和Anthropic的Tool Use[^20]规范为工具调用提供了标准化接口。云端AI IDE的Agent层需要实现灵活的工具调用编排机制：

```python
# 代码示例4-3：工具调用编排器实现
from typing import List, Dict, Optional, Callable, Any
from dataclasses import dataclass, field
from enum import Enum
import asyncio
import json
import re
from datetime import datetime

class ToolStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"

@dataclass
class ToolDefinition:
    name: str
    description: str
    parameters: Dict[str, Any]
    handler: Callable
    requires_confirmation: bool = False
    resource_requirements: Dict[str, Any] = field(default_factory=dict)

@dataclass
class ToolCall:
    call_id: str
    tool_name: str
    arguments: Dict[str, Any]
    status: ToolStatus = ToolStatus.PENDING
    result: Any = None
    error: str = None
    started_at: datetime = None
    completed_at: datetime = None
    execution_time_ms: int = 0

@dataclass
class AgentResponse:
    content: str
    tool_calls: List[ToolCall] = field(default_factory=list)
    suggestions: List[str] = field(default_factory=list)
    context_updates: List[Dict] = field(default_factory=list)

class ToolOrchestrator:
    def __init__(self, config):
        self.config = config
        self.tools: Dict[str, ToolDefinition] = {}
        self.pending_calls: Dict[str, ToolCall] = {}
        self.completed_calls: List[ToolCall] = []
        
        self._register_builtin_tools()
    
    def _register_builtin_tools(self):
        self.register_tool(ToolDefinition(
            name="read_file",
            description="Read the contents of a file",
            parameters={
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "File path to read"},
                    "start_line": {"type": "integer", "description": "Start line number"},
                    "end_line": {"type": "integer", "description": "End line number"},
                    "max_lines": {"type": "integer", "description": "Maximum lines to read"}
                },
                "required": ["path"]
            },
            handler=self._read_file_handler,
            resource_requirements={"type": "io", "priority": "high"}
        ))
        
        self.register_tool(ToolDefinition(
            name="write_file",
            description="Write content to a file",
            parameters={
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "File path to write"},
                    "content": {"type": "string", "description": "Content to write"},
                    "create_directory": {"type": "boolean", "description": "Create directory if not exists"}
                },
                "required": ["path", "content"]
            },
            handler=self._write_file_handler,
            requires_confirmation=True,
            resource_requirements={"type": "io", "priority": "high"}
        ))
        
        self.register_tool(ToolDefinition(
            name="execute_code",
            description="Execute code in a sandboxed environment",
            parameters={
                "type": "object",
                "properties": {
                    "language": {"type": "string", "enum": ["python", "javascript", "bash", "go", "rust"]},
                    "code": {"type": "string", "description": "Code to execute"},
                    "timeout_ms": {"type": "integer", "description": "Execution timeout"},
                    "stdin": {"type": "string", "description": "Standard input"}
                },
                "required": ["language", "code"]
            },
            handler=self._execute_code_handler,
            requires_confirmation=True,
            resource_requirements={"type": "compute", "priority": "medium"}
        ))
        
        self.register_tool(ToolDefinition(
            name="search_codebase",
            description="Search for code patterns in the project",
            parameters={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search query"},
                    "search_type": {"type": "string", "enum": ["semantic", "regex", "keyword"]},
                    "file_pattern": {"type": "string", "description": "File glob pattern"},
                    "max_results": {"type": "integer", "description": "Maximum results"}
                },
                "required": ["query"]
            },
            handler=self._search_codebase_handler,
            resource_requirements={"type": "search", "priority": "low"}
        ))
    
    def register_tool(self, tool_def: ToolDefinition):
        self.tools[tool_def.name] = tool_def
    
    async def execute_tool_calls(
        self,
        calls: List[Dict[str, Any]],
        user_context: Dict[str, Any],
        confirm_callback: Optional[Callable] = None
    ) -> List[ToolCall]:
        results = []
        
        for call_spec in calls:
            tool_name = call_spec.get("name")
            arguments = call_spec.get("arguments", {})
            
            if tool_name not in self.tools:
                results.append(ToolCall(
                    call_id=call_spec.get("id", f"call_{len(results)}"),
                    tool_name=tool_name,
                    arguments=arguments,
                    status=ToolStatus.FAILED,
                    error=f"Unknown tool: {tool_name}"
                ))
                continue
            
            tool_def = self.tools[tool_name]
            call = ToolCall(
                call_id=call_spec.get("id", f"call_{len(results)}"),
                tool_name=tool_name,
                arguments=arguments
            )
            
            if tool_def.requires_confirmation and confirm_callback:
                confirmed = await confirm_callback(tool_name, arguments)
                if not confirmed:
                    call.status = ToolStatus.CANCELLED
                    results.append(call)
                    continue
            
            try:
                call.status = ToolStatus.RUNNING
                call.started_at = datetime.now()
                
                result = await self._execute_tool(tool_def, arguments, user_context)
                
                call.status = ToolStatus.COMPLETED
                call.result = result
                call.completed_at = datetime.now()
                call.execution_time_ms = int(
                    (call.completed_at - call.started_at).total_seconds() * 1000
                )
                
            except Exception as e:
                call.status = ToolStatus.FAILED
                call.error = str(e)
                call.completed_at = datetime.now()
                call.execution_time_ms = int(
                    (call.completed_at - call.started_at).total_seconds() * 1000
                )
            
            results.append(call)
        
        return results
    
    async def _execute_tool(
        self,
        tool_def: ToolDefinition,
        arguments: Dict[str, Any],
        user_context: Dict[str, Any]
    ) -> Any:
        self._validate_arguments(tool_def, arguments)
        
        handler = tool_def.handler
        if asyncio.iscoroutinefunction(handler):
            return await handler(arguments, user_context)
        else:
            return handler(arguments, user_context)
    
    def _validate_arguments(self, tool_def: ToolDefinition, arguments: Dict[str, Any]):
        params = tool_def.parameters.get("properties", {})
        required = tool_def.parameters.get("required", [])
        
        for req_param in required:
            if req_param not in arguments:
                raise ValueError(f"Missing required parameter: {req_param}")
        
        for param_name, param_value in arguments.items():
            if param_name not in params:
                raise ValueError(f"Unknown parameter: {param_name}")
            
            expected_type = params[param_name].get("type")
            if not self._check_type(param_value, expected_type):
                raise ValueError(
                    f"Invalid type for {param_name}: expected {expected_type}, "
                    f"got {type(param_value).__name__}"
                )
    
    def _check_type(self, value: Any, expected_type: str) -> bool:
        type_map = {
            "string": str,
            "integer": int,
            "number": (int, float),
            "boolean": bool,
            "array": list,
            "object": dict
        }
        
        if expected_type not in type_map:
            return True
        
        return isinstance(value, type_map[expected_type])
    
    async def _read_file_handler(self, args: Dict, context: Dict) -> Dict:
        return {"path": args["path"], "content": "...", "lines": 100}
    
    async def _write_file_handler(self, args: Dict, context: Dict) -> Dict:
        return {"path": args["path"], "bytes_written": len(args["content"])}
    
    async def _execute_code_handler(self, args: Dict, context: Dict) -> Dict:
        return {
            "language": args["language"],
            "stdout": "...",
            "stderr": "",
            "exit_code": 0,
            "execution_time_ms": 100
        }
    
    async def _search_codebase_handler(self, args: Dict, context: Dict) -> Dict:
        return {"query": args["query"], "results": [], "total_matches": 0}
    
    def get_tool_definitions(self) -> List[Dict]:
        return [
            {
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.parameters
            }
            for tool in self.tools.values()
        ]

# 代码示例5-1：Kubernetes执行环境配置
apiVersion: v1
kind: Pod
metadata:
  name: executor-pod-{session_id}
  labels:
    app: ai-ide-executor
    session: {session_id}
spec:
  restartPolicy: Never
  terminationGracePeriodSeconds: 30
  
  securityContext:
    runAsUser: 1000
    runAsGroup: 1000
    fsGroup: 1000
    runAsNonRoot: true
    seccompProfile:
      type: RuntimeDefault
  
  containers:
  - name: executor
    image: {executor_image}
    imagePullPolicy: Always
    
    env:
    - name: SESSION_ID
      value: "{session_id}"
    - name: USER_ID
      value: "{user_id}"
    - name: WORKSPACE_ID
      value: "{workspace_id}"
    - name: RESOURCE_LIMIT_CPU
      value: "{resource_limit_cpu}"
    - name: RESOURCE_LIMIT_MEMORY
      value: "{resource_limit_memory}"
    
    resources:
      requests:
        memory: "{requested_memory}"
        cpu: "{requested_cpu}"
      limits:
        memory: "{limit_memory}"
        cpu: "{limit_cpu}"
        ephemeral-storage: "{limit_ephemeral}"
    
    livenessProbe:
      httpGet:
        path: /health
        port: 8080
      initialDelaySeconds: 5
      periodSeconds: 10
    
    volumeMounts:
    - name: workspace
      mountPath: /workspace
      readOnly: true
    - name: temp-storage
      mountPath: /tmp
    
    securityContext:
      allowPrivilegeEscalation: false
      capabilities:
        drop:
        - ALL
    
    ports:
    - containerPort: 8080
      protocol: TCP
    
    command: ["/opt/executor/bin/start.sh"]
    args:
    - "--session-id=$(SESSION_ID)"
    - "--user-id=$(USER_ID)"
  
  volumes:
  - name: workspace
    persistentVolumeClaim:
      claimName: workspace-{workspace_id}
  - name: temp-storage
    emptyDir:
      medium: Memory
      sizeLimit: 512Mi
  
  nodeSelector:
    executor-pool: "true"
  
  tolerations:
  - key: "executor"
    operator: "Exists"
    effect: "NoSchedule"
  
  readinessProbe:
    httpGet:
      path: /ready
      port: 8080
    initialDelaySeconds: 3
    periodSeconds: 5

# 代码示例5-2：代码执行服务实现
import asyncio
import subprocess
import tempfile
import os
import shutil
import resource
import signal
import time
from typing import Dict, Optional, Any
from dataclasses import dataclass
from enum import Enum
from abc import ABC, abstractmethod

class ExecutionStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    TIMEOUT = "timeout"
    OOM_KILLED = "oom_killed"
    SIGKILL = "sigkill"
    ERROR = "error"

@dataclass
class ExecutionResult:
    status: ExecutionStatus
    stdout: str
    stderr: str
    exit_code: int
    execution_time_ms: int
    peak_memory_bytes: int
    error_message: Optional[str] = None

@dataclass
class ExecutionRequest:
    session_id: str
    language: str
    code: str
    stdin: str = ""
    timeout_ms: int = 30000
    memory_limit_bytes: int = 512 * 1024 * 1024
    cpu_limit: float = 1.0

class ILanguageRuntime(ABC):
    @abstractmethod
    async def get_executor_command(self, code_path: str, stdin_path: str) -> list:
        pass
    
    @abstractmethod
    def get_image_tag(self) -> str:
        pass

class PythonRuntime(ILanguageRuntime):
    def get_executor_command(self, code_path: str, stdin_path: str) -> list:
        return ["python3", code_path]
    
    def get_image_tag(self) -> str:
        return "python:3.11-slim"

class JavaScriptRuntime(ILanguageRuntime):
    def get_executor_command(self, code_path: str, stdin_path: str) -> list:
        return ["node", code_path]
    
    def get_image_tag(self) -> str:
        return "node:20-slim"

class GoRuntime(ILanguageRuntime):
    def get_executor_command(self, code_path: str, stdin_path: str) -> list:
        return ["go", "run", code_path]
    
    def get_image_tag(self) -> str:
        return "golang:1.21-alpine"

class RustRuntime(ILanguageRuntime):
    def get_executor_command(self, code_path: str, stdin_path: str) -> list:
        return ["sh", "-c", f"rustc {code_path} -o /tmp/a.out && /tmp/a.out"]
    
    def get_image_tag(self) -> str:
        return "rust:1.75-slim"

class CodeExecutor:
    def __init__(self, config: Dict):
        self.config = config
        self.runtimes: Dict[str, ILanguageRuntime] = {
            "python": PythonRuntime(),
            "javascript": JavaScriptRuntime(),
            "go": GoRuntime(),
            "rust": RustRuntime(),
        }
        self.temp_dir = tempfile.mkdtemp(prefix="aiide_executor_")
        self.active_processes: Dict[str, asyncio.subprocess.Process] = {}
    
    async def execute(self, request: ExecutionRequest) -> ExecutionResult:
        start_time = time.time()
        
        code_path, stdin_path = await self._prepare_files(request)
        
        try:
            runtime = self.runtimes.get(request.language)
            if not runtime:
                return ExecutionResult(
                    status=ExecutionStatus.ERROR,
                    stdout="",
                    stderr=f"Unsupported language: {request.language}",
                    exit_code=-1,
                    execution_time_ms=0,
                    peak_memory_bytes=0,
                    error_message="Language not supported"
                )
            
            cmd = await runtime.get_executor_command(code_path, stdin_path)
            
            process = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                stdin=asyncio.subprocess.PIPE,
                cwd=self.temp_dir,
                limit=10 * 1024 * 1024
            )
            
            self.active_processes[request.session_id] = process
            
            try:
                stdout, stderr = await asyncio.wait_for(
                    process.communicate(input=request.stdin.encode() if request.stdin else None),
                    timeout=request.timeout_ms / 1000
                )
                
                status = ExecutionStatus.COMPLETED if process.returncode == 0 else ExecutionStatus.ERROR
                
            except asyncio.TimeoutError:
                process.kill()
                await process.wait()
                status = ExecutionStatus.TIMEOUT
            
            finally:
                self.active_processes.pop(request.session_id, None)
            
            execution_time = int((time.time() - start_time) * 1000)
            
            return ExecutionResult(
                status=status,
                stdout=stdout.decode('utf-8', errors='replace') if stdout else "",
                stderr=stderr.decode('utf-8', errors='replace') if stderr else "",
                exit_code=process.returncode or 0,
                execution_time_ms=execution_time,
                peak_memory_bytes=0,
                error_message=None if status == ExecutionStatus.COMPLETED else f"Execution {status.value}"
            )
        
        except Exception as e:
            return ExecutionResult(
                status=ExecutionStatus.ERROR,
                stdout="",
                stderr=str(e),
                exit_code=-1,
                execution_time_ms=int((time.time() - start_time) * 1000),
                peak_memory_bytes=0,
                error_message=str(e)
            )
        
        finally:
            await self._cleanup_files(code_path, stdin_path)
    
    async def _prepare_files(self, request: ExecutionRequest) -> tuple:
        suffix_map = {
            "python": ".py",
            "javascript": ".js",
            "go": ".go",
            "rust": ".rs"
        }
        suffix = suffix_map.get(request.language, ".txt")
        
        code_path = os.path.join(self.temp_dir, f"code_{request.session_id}{suffix}")
        stdin_path = os.path.join(self.temp_dir, f"stdin_{request.session_id}.txt")
        
        with open(code_path, 'w', encoding='utf-8') as f:
            f.write(request.code)
        
        with open(stdin_path, 'w', encoding='utf-8') as f:
            f.write(request.stdin)
        
        return code_path, stdin_path
    
    async def _cleanup_files(self, code_path: str, stdin_path: str):
        for path in [code_path, stdin_path]:
            try:
                if os.path.exists(path):
                    os.remove(path)
            except Exception:
                pass
    
    async def terminate(self, session_id: str):
        process = self.active_processes.get(session_id)
        if process:
            try:
                process.kill()
                await process.wait()
            except ProcessLookupError:
                pass
    
    def dispose(self):
        for session_id in list(self.active_processes.keys()):
            asyncio.create_task(self.terminate(session_id))
        
        try:
            shutil.rmtree(self.temp_dir)
        except Exception:
            pass

# 代码示例5-3：网络隔离策略配置
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: executor-network-policy
  namespace: ai-ide-production
spec:
  podSelector:
    matchLabels:
      app: ai-ide-executor
  
  policyTypes:
  - Ingress
  - Egress
  
  ingress:
  - from:
    - namespaceSelector:
        matchLabels:
          name: ai-ide-gateway
    ports:
    - protocol: TCP
      port: 8080
  
  egress:
  - to:
    - namespaceSelector:
        matchLabels:
          name: ai-ide-storage
    ports:
    - protocol: TCP
      port: 2049
  
  - to:
    - namespaceSelector:
        matchLabels:
          kubernetes.io/name: kube-dns
    ports:
    - protocol: UDP
      port: 53

# 代码示例6-1：向量数据库服务实现
from typing import List, Dict, Optional, Any, Tuple
from dataclasses import dataclass
import numpy as np
import hashlib
from datetime import datetime

@dataclass
class CodeChunk:
    chunk_id: str
    content: str
    language: str
    file_path: str
    start_line: int
    end_line: int
    vector: Optional[np.ndarray] = None
    metadata: Dict[str, Any] = None

@dataclass
class SearchResult:
    chunk_id: str
    content: str
    file_path: str
    start_line: int
    end_line: int
    similarity: float
    metadata: Dict[str, Any]

class VectorStore:
    def __init__(self, config: dict):
        self.config = config
        self.embedding_dim = config.get("embedding_dim", 1536)
        self.index_type = config.get("index_type", "HNSW")
    
    def _initialize_collection(self):
        schema = {
            "name": "code_chunks",
            "fields": [
                {"name": "chunk_id", "type": "string", "is_primary": True},
                {"name": "content", "type": "string"},
                {"name": "embedding", "type": "float", "dim": self.embedding_dim},
                {"name": "file_path", "type": "string"},
                {"name": "language", "type": "string"},
                {"name": "start_line", "type": "int"},
                {"name": "end_line", "type": "int"},
                {"name": "workspace_id", "type": "string"},
                {"name": "updated_at", "type": "datetime"}
            ],
            "indexes": [
                {"field": "embedding", "index_type": self.index_type, "params": {"M": 16, "efConstruction": 200}},
                {"field": "file_path", "index_type": "inverted"},
                {"field": "workspace_id", "index_type": "hashed"}
            ]
        }
    
    async def add_chunks(self, chunks: List[CodeChunk], workspace_id: str):
        entities = []
        
        for chunk in chunks:
            chunk_id = self._generate_chunk_id(chunk, workspace_id)
            
            entity = {
                "chunk_id": chunk_id,
                "content": chunk.content,
                "embedding": chunk.vector.tolist() if chunk.vector is not None else None,
                "file_path": chunk.file_path,
                "language": chunk.language,
                "start_line": chunk.start_line,
                "end_line": chunk.end_line,
                "workspace_id": workspace_id,
                "updated_at": datetime.now()
            }
            entities.append(entity)
        
        return len(entities)
    
    async def search(
        self,
        query_vector: np.ndarray,
        workspace_id: str,
        language: Optional[str] = None,
        file_path: Optional[str] = None,
        limit: int = 10,
        similarity_threshold: float = 0.7
    ) -> List[SearchResult]:
        filter_expr = f'workspace_id == "{workspace_id}"'
        if language:
            filter_expr += f' and language == "{language}"'
        if file_path:
            filter_expr += f' and file_path like "{file_path}%"'
        
        search_params = {
            "collection": "code_chunks",
            "data": [query_vector.tolist()],
            "filter": filter_expr,
            "limit": limit,
            "param": {"M": 16, "ef": 128}
        }
        
        results = []
        
        for result in results:
            if result.get("score", 0) >= similarity_threshold:
                results.append(SearchResult(
                    chunk_id=result["entity"]["chunk_id"],
                    content=result["entity"]["content"],
                    file_path=result["entity"]["file_path"],
                    start_line=result["entity"]["start_line"],
                    end_line=result["entity"]["end_line"],
                    similarity=result["score"],
                    metadata={}
                ))
        
        return results
    
    def _generate_chunk_id(self, chunk: CodeChunk, workspace_id: str) -> str:
        content = f"{workspace_id}:{chunk.file_path}:{chunk.start_line}:{chunk.end_line}"
        return hashlib.sha256(content.encode()).hexdigest()[:16]

# 代码示例7-1：协议路由实现
from enum import Enum
from typing import Dict, Callable, Any
import asyncio

class Protocol(Enum):
    WEBSOCKET = "websocket"
    HTTP2 = "http2"
    GRPC = "grpc"

class Request:
    def __init__(self, path: str, method: str, headers: Dict, body: Any):
        self.path = path
        self.method = method
        self.headers = headers
        self.body = body

class ProtocolRouter:
    def __init__(self, config: dict):
        self.config = config
        
        self.route_rules: Dict[str, Protocol] = {
            "/ws/terminal": Protocol.WEBSOCKET,
            "/ws/ai-stream": Protocol.WEBSOCKET,
            "/ws/collab": Protocol.WEBSOCKET,
            "/ws/events": Protocol.WEBSOCKET,
            "/api/files/*": Protocol.HTTP2,
            "/api/projects/*": Protocol.HTTP2,
            "/api/user/*": Protocol.HTTP2,
            "/api/ai/completion": Protocol.HTTP2,
            "/api/search": Protocol.HTTP2,
            "/grpc.executor.v1.ExecutorService/*": Protocol.GRPC,
            "/grpc.storage.v1.StorageService/*": Protocol.GRPC,
            "/grpc.ai.v1.AgentService/*": Protocol.GRPC,
        }
        
        self.handlers: Dict[Protocol, Any] = {}
    
    def register_handler(self, protocol: Protocol, handler: Any):
        self.handlers[protocol] = handler
    
    async def route(self, request: Request) -> Any:
        protocol = self._match_route(request.path)
        
        if protocol is None:
            raise ValueError(f"No route for path: {request.path}")
        
        handler = self.handlers.get(protocol)
        if not handler:
            raise ValueError(f"No handler for protocol: {protocol}")
        
        return await self._dispatch(handler, protocol, request)
    
    def _match_route(self, path: str) -> Protocol:
        if path in self.route_rules:
            return self.route_rules[path]
        
        for pattern, protocol in self.route_rules.items():
            if pattern.endswith("*"):
                prefix = pattern[:-1]
                if path.startswith(prefix):
                    return protocol
        
        return None
    
    async def _dispatch(self, handler: Any, protocol: Protocol, request: Request) -> Any:
        if protocol == Protocol.WEBSOCKET:
            return await handler.handle_websocket(request)
        elif protocol == Protocol.HTTP2:
            return await handler.handle_http2(request)
        elif protocol == Protocol.GRPC:
            return await handler.handle_grpc(request)

# 代码示例8-1：延迟优化策略实现
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field
import asyncio
import time
import hashlib
from collections import OrderedDict
import heapq

@dataclass
class CacheEntry:
    key: str
    value: Any
    created_at: float
    last_accessed: float
    access_count: int
    size_bytes: int
    ttl: Optional[float] = None
    
    def is_expired(self) -> bool:
        if self.ttl is None:
            return False
        return time.time() - self.created_at > self.ttl

class OptimisticCache:
    def __init__(self, config: dict):
        self.config = config
        self.max_size_mb = config.get("cache_size_mb", 100)
        self.default_ttl = config.get("default_ttl", 3600)
        
        self._cache: OrderedDict[str, CacheEntry] = OrderedDict()
        self._current_size = 0
        self._hits = 0
        self._misses = 0
    
    def get(self, key: str) -> Optional[Any]:
        entry = self._cache.get(key)
        
        if entry is None:
            self._misses += 1
            return None
        
        if entry.is_expired():
            self._remove(key)
            self._misses += 1
            return None
        
        entry.last_accessed = time.time()
        entry.access_count += 1
        
        self._cache.move_to_end(key)
        
        self._hits += 1
        return entry.value
    
    def set(self, key: str, value: Any, ttl: Optional[float] = None, size_bytes: int = 0):
        if key in self._cache:
            self._remove(key)
        
        size = size_bytes or self._estimate_size(value)
        max_size = self.max_size_mb * 1024 * 1024
        
        while self._current_size + size > max_size and self._cache:
            self._evict_lru()
        
        entry = CacheEntry(
            key=key,
            value=value,
            created_at=time.time(),
            last_accessed=time.time(),
            access_count=1,
            size_bytes=size,
            ttl=ttl or self.default_ttl
        )
        
        self._cache[key] = entry
        self._current_size += size
    
    def _evict_lru(self):
        if not self._cache:
            return
        
        key, entry = self._cache.popitem(last=False)
        self._current_size -= entry.size_bytes
    
    def _remove(self, key: str):
        entry = self._cache.pop(key, None)
        if entry:
            self._current_size -= entry.size_bytes
    
    def _estimate_size(self, value: Any) -> int:
        import sys
        return sys.getsizeof(value)
    
    def get_stats(self) -> dict:
        total = self._hits + self._misses
        hit_rate = self._hits / total if total > 0 else 0
        return {
            "hits": self._hits,
            "misses": self._misses,
            "hit_rate": hit_rate,
            "size_mb": self._current_size / (1024 * 1024),
            "entries": len(self._cache)
        }

class PrefetchEngine:
    def __init__(self, config: dict):
        self.config = config
        self.prefetch_queue: List[tuple] = []
        self.active_tasks: Dict[str, asyncio.Task] = {}
    
    async def predict_and_prefetch(self, user_context: Dict[str, Any]):
        current_file = user_context.get("current_file")
        cursor_position = user_context.get("cursor_position")
        open_files = user_context.get("open_files", [])
        
        predictions = await self._predict_resources(current_file, cursor_position, open_files)
        
        for priority, resource in predictions:
            asyncio.create_task(self._prefetch_resource(resource))
    
    async def _predict_resources(self, current_file: str, cursor: tuple, open_files: List[str]) -> List[tuple]:
        predictions = []
        
        if current_file:
            imports = await self._get_import_dependencies(current_file)
            for imp in imports:
                predictions.append((0.9, {"type": "file", "path": imp}))
        
        file_dir = current_file.rsplit("/", 1)[0] if "/" in current_file else ""
        adjacent_files = await self._list_directory(file_dir)
        for f in adjacent_files[:5]:
            predictions.append((0.7, {"type": "file", "path": f}))
        
        config_files = await self._get_related_configs(current_file)
        for cf in config_files:
            predictions.append((0.8, {"type": "config", "path": cf}))
        
        predictions.append((0.6, {"type": "ai_context", "file": current_file}))
        
        predictions.sort(key=lambda x: x[0], reverse=True)
        
        return predictions[:10]
    
    async def _prefetch_resource(self, resource: Dict[str, Any]):
        resource_type = resource.get("type")
        resource_id = f"{resource_type}:{resource.get('path', '')}"
        
        if resource_id in self.active_tasks:
            return
        
        task = asyncio.create_task(self._do_prefetch(resource))
        self.active_tasks[resource_id] = task
        
        try:
            await task
        finally:
            self.active_tasks.pop(resource_id, None)
    
    async def _do_prefetch(self, resource: Dict[str, Any]) -> Any:
        resource_type = resource.get("type")
        
        if resource_type == "file":
            path = resource.get("path")
            return await self._read_file_async(path)
        
        elif resource_type == "ai_context":
            file_path = resource.get("file")
            return await self._precompute_ai_context(file_path)
        
        return None
    
    async def _get_import_dependencies(self, file_path: str) -> List[str]:
        return []
    
    async def _list_directory(self, path: str) -> List[str]:
        return []
    
    async def _get_related_configs(self, file_path: str) -> List[str]:
        return []
    
    async def _read_file_async(self, path: str) -> str:
        import aiofiles
        async with aiofiles.open(path, 'r') as f:
            return await f.read()
    
    async def _precompute_ai_context(self, file_path: str) -> Dict:
        return {}

# 代码示例10-1：资源计量服务实现
from typing import Dict, List, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
import asyncio

@dataclass
class ResourceUsage:
    user_id: str
    workspace_id: str
    resource_type: str
    quantity: float
    unit: str
    timestamp: datetime
    metadata: Dict

class BillingMeter:
    def __init__(self, config: dict):
        self.config = config
        self.usage_records: List[ResourceUsage] = []
        self.pricing_config = self._load_pricing_config()
    
    def _load_pricing_config(self) -> dict:
        return {
            "compute": {
                "unit": "core_hour",
                "tiers": [
                    {"limit": 100, "price": 0.05},
                    {"limit": 500, "price": 0.04},
                    {"limit": float('inf'), "price": 0.03}
                ]
            },
            "memory": {
                "unit": "gb_hour",
                "tiers": [
                    {"limit": 50, "price": 0.01},
                    {"limit": float('inf'), "price": 0.008}
                ]
            },
            "storage": {
                "unit": "gb_month",
                "tiers": [
                    {"limit": 10, "price": 0.10},
                    {"limit": 100, "price": 0.08},
                    {"limit": float('inf'), "price": 0.05}
                ]
            },
            "ai_requests": {
                "unit": "per_1k_tokens",
                "tiers": [
                    {"limit": 100, "price": 0.50},
                    {"limit": 1000, "price": 0.40},
                    {"limit": float('inf'), "price": 0.30}
                ]
            },
            "network": {
                "unit": "gb",
                "tiers": [
                    {"limit": 10, "price": 0.12},
                    {"limit": 100, "price": 0.10},
                    {"limit": float('inf'), "price": 0.08}
                ]
            }
        }
    
    async def record_usage(self, usage: ResourceUsage):
        self.usage_records.append(usage)
        
        await self._persist_usage_record(usage)
    
    async def calculate_cost(self, user_id: str, billing_period: tuple) -> Dict:
        start_date, end_date = billing_period
        
        user_usage = [
            u for u in self.usage_records
            if u.user_id == user_id and start_date <= u.timestamp <= end_date
        ]
        
        cost_by_type = {}
        
        for resource_type in self.pricing_config.keys():
            type_usage = [u for u in user_usage if u.resource_type == resource_type]
            total_quantity = sum(u.quantity for u in type_usage)
            
            if total_quantity > 0:
                unit_price = self._calculate_tiered_price(resource_type, total_quantity)
                cost_by_type[resource_type] = {
                    "quantity": total_quantity,
                    "unit": self.pricing_config[resource_type]["unit"],
                    "unit_price": unit_price,
                    "total": total_quantity * unit_price
                }
        
        return {
            "user_id": user_id,
            "billing_period": {"start": start_date, "end": end_date},
            "cost_by_type": cost_by_type,
            "total_cost": sum(c["total"] for c in cost_by_type.values())
        }
    
    def _calculate_tiered_price(self, resource_type: str, quantity: float) -> float:
        tiers = self.pricing_config[resource_type]["tiers"]
        
        remaining = quantity
        total_cost = 0
        
        for tier in tiers:
            tier_limit = tier["limit"]
            tier_price = tier["price"]
            
            if remaining <= 0:
                break
            
            tier_quantity = min(remaining, tier_limit)
            total_cost += tier_quantity * tier_price
            remaining -= tier_quantity
        
        return total_cost / quantity
    
    async def _persist_usage_record(self, usage: ResourceUsage):
        pass

Client Layer (Port 3000-4000)
    │
    ▼
Gateway Layer (Port 8000-9000)
    │
    ├──▶ Agent Layer (Internal)
    │       │
    │       ├──▶ Model Router
    │       ├──▶ Prompt Engine
    │       └──▶ Tool Orchestrator
    │
    ├──▶ Executor Layer (Internal)
    │       │
    │       ├──▶ Container Pool
    │       └──▶ Code Runner
    │
    └──▶ Storage Layer (Internal)
            │
            ├──▶ File Service
            ├──▶ Vector Store
            └──▶ Object Storage