推理服务平台技术债治理：替代方案评估与决策框架

inference_platform/
├── configs
│   └── models_config.yaml      # 模型定义与部署配置
├── src
│   ├── __init__.py
│   ├── core
│   │   ├── __init__.py
│   │   ├── config.py           # 配置加载与管理
│   │   └── model_registry.py   # 模型注册表（中心化管理）
│   ├── routing
│   │   ├── __init__.py
│   │   ├── base_router.py      # 路由器抽象基类
│   │   ├── load_aware_router.py # 基于负载的路由策略
│   │   └── router_factory.py   # 路由器工厂
│   ├── processing
│   │   ├── __init__.py
│   │   └── batch_processor.py  # 请求批处理核心逻辑
│   ├── service
│   │   ├── __init__.py
│   │   └── inference_service.py # 主服务逻辑，集成路由与处理
│   └── decision
│       ├── __init__.py
│       └── framework.py        # 技术债替代方案评估框架
├── tests
│   ├── __init__.py
│   ├── test_router.py
│   └── test_batch_processor.py
├── requirements.txt            # 项目依赖
├── run_server.py              # 服务启动入口
└── evaluate_decision.py       # 执行方案评估的脚本

model_registry:
  model-a:
    model_class: "TextClassifier"
    model_path: "/opt/models/text_classifier_v1.onnx"
    max_batch_size: 16
    required_memory_gb: 2
    backend: "onnxruntime"
    endpoints:

      - name: "gpu-node-1"
        url: "localhost:8001"
        capacity: 10  # 并发容量

      - name: "gpu-node-2"
        url: "localhost:8002"
        capacity: 15
  model-b:
    model_class: "ImageEmbedder"
    model_path: "/opt/models/image_embedder_v2.trt"
    max_batch_size: 8
    required_memory_gb: 4
    backend: "tensorrt"
    endpoints:

      - name: "trt-node-1"
        url: "localhost:9001"
        capacity: 6

import yaml
import os
from typing import Dict, Any

class ConfigManager:
    _instance = None
    _config = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(ConfigManager, cls).__new__(cls)
            cls._instance._load_config()
        return cls._instance

    def _load_config(self):
        config_path = os.path.join(
            os.path.dirname(__file__), '..', '..', 'configs', 'models_config.yaml'
        )
        with open(config_path, 'r') as f:
            self._config = yaml.safe_load(f)

    def get_model_config(self, model_id: str) -> Dict[str, Any]:
        """获取指定模型的配置"""
        return self._config['model_registry'].get(model_id)

    def get_all_models(self) -> Dict[str, Any]:
        """获取所有模型配置"""
        return self._config['model_registry']

    def update_config(self, new_config: Dict[str, Any]):
        """动态更新配置（模拟热更新）"""
        # 注意：生产环境需要加锁和持久化
        self._config['model_registry'].update(new_config)
        print("Config updated dynamically.")

from .config import ConfigManager
from typing import Dict, Any, Optional

class ModelInstance:
    """模型实例，包含配置与运行时状态"""
    def __init__(self, model_id: str):
        self.model_id = model_id
        self.config = ConfigManager().get_model_config(model_id)
        if not self.config:
            raise ValueError(f"Model {model_id} not found in config.")
        self.current_load = 0  # 当前负载（如并发请求数）
        self.is_healthy = True

    def increment_load(self):
        self.current_load += 1

    def decrement_load(self):
        if self.current_load > 0:
            self.current_load -= 1

    def get_available_capacity(self, endpoint_name: str) -> int:
        """获取指定端点的剩余容量"""
        for ep in self.config['endpoints']:
            if ep['name'] == endpoint_name:
                return ep['capacity'] - self.current_load
        return 0

class ModelRegistry:
    """全局模型注册表"""
    _models: Dict[str, ModelInstance] = {}

    @classmethod
    def get_model(cls, model_id: str) -> Optional[ModelInstance]:
        if model_id not in cls._models:
            try:
                cls._models[model_id] = ModelInstance(model_id)
            except ValueError:
                return None
        return cls._models[model_id]

    @classmethod
    def get_all_models(cls) -> Dict[str, ModelInstance]:
        return cls._models

    @classmethod
    def update_load(cls, model_id: str, endpoint: str, delta: int):
        """更新模型负载，delta为+1或-1"""
        model = cls.get_model(model_id)
        if model:
            if delta > 0:
                model.increment_load()
            else:
                model.decrement_load()

from abc import ABC, abstractmethod
from typing import Dict, Any, List
from src.core.model_registry import ModelRegistry, ModelInstance

class BaseRouter(ABC):
    """路由器抽象基类，所有具体路由策略需继承此类"""

    def __init__(self):
        self.registry = ModelRegistry()

    @abstractmethod
    def route(self, model_id: str, request_data: Any) -> Dict[str, Any]:
        """
        路由请求到具体的模型端点。
        返回: {'endpoint': url, 'model_id': model_id, 'strategy': name}
        """
        pass

    def _get_available_endpoints(self, model: ModelInstance) -> List[Dict[str, Any]]:
        """辅助方法：获取健康且有容量的端点列表"""
        available = []
        for ep in model.config['endpoints']:
            if model.is_healthy and model.get_available_capacity(ep['name']) > 0:
                available.append(ep)
        return available

from .base_router import BaseRouter
import random
from typing import Dict, Any

class LoadAwareRouter(BaseRouter):
    """最小负载路由策略：选择当前负载最小的可用端点"""

    def route(self, model_id: str, request_data: Any) -> Dict[str, Any]:
        model = self.registry.get_model(model_id)
        if not model:
            raise ValueError(f"Model {model_id} not registered.")

        available_endpoints = self._get_available_endpoints(model)
        if not available_endpoints:
            # 降级策略：返回一个健康的端点，即使超载
            for ep in model.config['endpoints']:
                if model.is_healthy:
                    return {
                        'endpoint': ep['url'],
                        'model_id': model_id,
                        'strategy': 'load_aware_fallback'
                    }
            raise RuntimeError(f"No healthy endpoint for model {model_id}.")

        # 选择负载最小的端点 (容量 - 当前负载 = 剩余容量，取最大值)
        selected_endpoint = max(
            available_endpoints,
            key=lambda ep: model.get_available_capacity(ep['name'])
        )
        return {
            'endpoint': selected_endpoint['url'],
            'model_id': model_id,
            'strategy': 'load_aware'
        }

import threading
import time
import asyncio
from queue import Queue, Empty
from typing import List, Any, Callable, Tuple
from dataclasses import dataclass
import uuid

@dataclass
class BatchRequest:
    """批处理请求单元"""
    request_id: str
    data: Any
    model_id: str
    future: asyncio.Future  # 用于异步返回结果

class BatchProcessor:
    """
    批处理器：收集一段时间内或达到一定数量的请求，合并后发送给后端。
    采用生产者-消费者模式。
    """
    def __init__(self, model_id: str, batch_timeout: float = 0.1, max_batch_size: int = 16):
        self.model_id = model_id
        self.batch_timeout = batch_timeout
        self.max_batch_size = max_batch_size
        self.queue = Queue()
        self.processing_lock = threading.Lock()
        self.stop_event = threading.Event()
        # 启动后台批处理线程
        self.worker_thread = threading.Thread(target=self._batch_worker, daemon=True)
        self.worker_thread.start()

    def submit(self, data: Any) -> asyncio.Future:
        """提交一个请求，返回一个Future对象"""
        loop = asyncio.get_event_loop()
        future = loop.create_future()
        request = BatchRequest(
            request_id=str(uuid.uuid4()),
            data=data,
            model_id=self.model_id,
            future=future
        )
        self.queue.put(request)
        return future

    def _batch_worker(self):
        """后台工作线程，负责组批和执行"""
        while not self.stop_event.is_set():
            batch: List[BatchRequest] = []
            start_time = time.time()

            # 收集批：超时或达到最大尺寸
            try:
                while len(batch) < self.max_batch_size:
                    time_left = start_time + self.batch_timeout - time.time()
                    if time_left <= 0:
                        break
                    try:
                        req = self.queue.get(timeout=time_left)
                        batch.append(req)
                    except Empty:
                        break  # 超时
            except Exception as e:
                print(f"Error collecting batch: {e}")
                continue

            if not batch:
                continue

            # 执行批量推理（模拟）
            self._execute_batch(batch)

    def _execute_batch(self, batch: List[BatchRequest]):
        """执行批量推理，并设置每个请求的结果"""
        try:
            # 模拟批量数据处理：合并请求数据
            batch_data = [req.data for req in batch]
            # 此处应调用真实的批量推理后端，例如：
            # batch_results = inference_backend.predict(self.model_id, batch_data)
            # 为示例，我们模拟一个简单的处理：返回带批注的结果
            simulated_results = [
                f"Processed {req.request_id} for {req.model_id} in batch of size {len(batch)}"
                for req in batch
            ]

            # 将结果设置到各自的Future中
            for req, result in zip(batch, simulated_results):
                if not req.future.done():
                    req.future.set_result(result)

        except Exception as e:
            # 如果批处理失败，将错误设置到每个请求的Future
            for req in batch:
                if not req.future.done():
                    req.future.set_exception(e)

    def shutdown(self):
        """关闭处理器"""
        self.stop_event.set()
        self.worker_thread.join(timeout=5)

import asyncio
from src.core.config import ConfigManager
from src.routing.router_factory import RouterFactory
from src.processing.batch_processor import BatchProcessor
from typing import Dict, Any

class InferenceService:
    """
    推理服务主类。
    治理了以下技术债：

    1. 通过ConfigManager消除硬编码。
    2. 通过RouterFactory支持可插拔路由策略。
    3. 通过BatchProcessor（可选）优化吞吐量。
    """
    def __init__(self, use_batching: bool = False):
        self.config_mgr = ConfigManager()
        self.router = RouterFactory.create_router('load_aware')  # 策略可配置
        self.use_batching = use_batching
        self.batch_processors: Dict[str, BatchProcessor] = {}
        if use_batching:
            self._init_batch_processors()

    def _init_batch_processors(self):
        """为每个支持批处理的模型初始化批处理器"""
        all_models = self.config_mgr.get_all_models()
        for model_id, config in all_models.items():
            max_batch_size = config.get('max_batch_size', 1)
            if max_batch_size > 1:
                self.batch_processors[model_id] = BatchProcessor(
                    model_id=model_id,
                    max_batch_size=max_batch_size
                )

    async def infer(self, model_id: str, input_data: Any) -> Dict[str, Any]:
        """
        处理推理请求的入口。

        1. 路由决策。
        2. 负载记录（模拟）。
        3. 选择处理路径（直接或批处理）。
        """
        # 1. 路由决策
        routing_decision = self.router.route(model_id, input_data)
        endpoint = routing_decision['endpoint']

        # 2. 模拟更新负载（在实际中，应在请求开始和结束时调用）
        from src.core.model_registry import ModelRegistry
        ModelRegistry.update_load(model_id, endpoint, +1)

        # 3. 处理请求
        if self.use_batching and model_id in self.batch_processors:
            # 批处理路径
            future = self.batch_processors[model_id].submit(input_data)
            try:
                processed_result = await asyncio.wait_for(future, timeout=5.0)
                result = {
                    'result': processed_result,
                    'endpoint': endpoint,
                    'processing_mode': 'batched'
                }
            except asyncio.TimeoutError:
                result = {'error': 'Batch processing timeout'}
        else:
            # 实时处理路径（模拟）
            await asyncio.sleep(0.05)  # 模拟推理延迟
            result = {
                'result': f'Processed directly at {endpoint}',
                'endpoint': endpoint,
                'processing_mode': 'real-time'
            }

        # 4. 模拟请求结束，减少负载
        ModelRegistry.update_load(model_id, endpoint, -1)
        return {**result, 'model_id': model_id, 'strategy': routing_decision['strategy']}

    def update_routing_strategy(self, strategy_name: str):
        """动态切换路由策略，展示治理的灵活性"""
        self.router = RouterFactory.create_router(strategy_name)
        print(f"Routing strategy switched to: {strategy_name}")

from typing import List, Dict, Any
import numpy as np

class Alternative:
    """代表一个待评估的治理方案"""
    def __init__(self, name: str, description: str):
        self.name = name
        self.description = description
        self.scores: Dict[str, float] = {}  # 维度 -> 得分

class DecisionFramework:
    """
    决策框架：使用多维度加权评分模型。
    维度包括：开发成本、运维成本、性能收益、可维护性收益、实施风险。
    """
    def __init__(self, alternatives: List[Alternative], weights: Dict[str, float]):
        """
        :param alternatives: 候选方案列表
        :param weights: 各评估维度的权重，e.g., {'cost': 0.3, 'benefit': 0.7}
        """
        self.alternatives = alternatives
        self.weights = weights
        # 预定义评估维度
        self.dimensions = ['dev_cost', 'ops_cost', 'perf_gain', 'maintainability_gain', 'risk']

    def evaluate(self, assessment_matrix: Dict[str, Dict[str, float]]):
        """
        执行评估。
        :param assessment_matrix: 形如 {‘方案A': {‘dev_cost': 5, ‘perf_gain': 8...}, ...}
                               得分范围建议1-10，成本类得分越低越好，收益类得分越高越好。
        """
        results = []
        for alt in self.alternatives:
            if alt.name not in assessment_matrix:
                raise ValueError(f"Missing assessment for alternative: {alt.name}")
            scores = assessment_matrix[alt.name]

            # 归一化处理（将成本类得分反转，使得所有维度都是越高越好）
            normalized_scores = {}
            for dim in self.dimensions:
                raw_score = scores.get(dim, 5.0)  # 默认5分
                # 如果是成本维度，得分反转（10 - score），因为成本越低越好
                if dim in ['dev_cost', 'ops_cost']:
                    normalized_scores[dim] = 10.0 - raw_score
                else:
                    normalized_scores[dim] = raw_score
                alt.scores[dim] = normalized_scores[dim]

            # 计算加权总分
            total_score = 0.0
            for dim, weight in self.weights.items():
                # 这里简化处理，将维度映射到权重类别。更精细的做法是每个维度独立赋权。
                if dim == 'cost':
                    sub_score = (normalized_scores['dev_cost'] + normalized_scores['ops_cost']) / 2
                    total_score += weight * sub_score
                elif dim == 'benefit':
                    sub_score = (normalized_scores['perf_gain'] + normalized_scores['maintainability_gain']) / 2
                    total_score += weight * sub_score
                elif dim == 'risk':
                    # 风险是负向指标，得分越高风险越低，所以直接加权
                    total_score += weight * normalized_scores['risk']
                else:
                    total_score += weight * normalized_scores.get(dim, 5.0)

            results.append({
                'alternative': alt.name,
                'total_score': round(total_score, 2),
                'dimension_scores': normalized_scores
            })

        # 按总分排序
        results.sort(key=lambda x: x['total_score'], reverse=True)
        return results

    @staticmethod
    def print_recommendation(evaluation_results: List[Dict[str, Any]]):
        """格式化输出评估结果与建议"""
        print("\n" + "="*60)
        print("技术债治理替代方案评估报告")
        print("="*60)
        for i, res in enumerate(evaluation_results):
            print(f"\n{i+1}. {res['alternative']}")
            print(f"   综合得分: {res['total_score']}")
            print(f"   维度详情: {res['dimension_scores']}")
        print("\n" + "="*60)
        top_choice = evaluation_results[0]
        print(f"推荐方案: 【{top_choice['alternative']}】 (综合得分: {top_choice['total_score']})")
        print("="*60)

# requirements.txt
fastapi==0.104.1
uvicorn[standard]==0.24.0
PyYAML==6.0.1
numpy==1.24.3

# 安装命令
pip install -r requirements.txt

# run_server.py
import uvicorn
import asyncio
from src.service.inference_service import InferenceService
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI(title="Inference Platform with Tech Debt Governance")

# 初始化服务，可选择是否启用批处理
service = InferenceService(use_batching=True)

class InferenceRequest(BaseModel):
    model_id: str
    input_data: str

@app.post("/v1/infer")
async def infer_endpoint(request: InferenceRequest):
    try:
        result = await service.infer(request.model_id, request.input_data)
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/admin/switch_router/{strategy}")
async def switch_router(strategy: str):
    """动态切换路由策略的管理端点"""
    service.update_routing_strategy(strategy)
    return {"message": f"Router strategy switched to {strategy}"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8080)

python run_server.py

# 测试请求
curl -X POST "http://localhost:8080/v1/infer" \
  -H "Content-Type: application/json" \
  -d '{"model_id": "model-a", "input_data": "Sample text for classification"}'

# 预期返回（示例）:
# {"result":"Processed 89f2e... in batch of size 3","endpoint":"localhost:8001","processing_mode":"batched","model_id":"model-a","strategy":"load_aware"}

# 动态切换路由策略（如果需要）
curl -X GET "http://localhost:8080/admin/switch_router/load_aware"

cd /path/to/project
python -m pytest tests/ -v

#!/usr/bin/env python3
from src.decision.framework import Alternative, DecisionFramework

def main():
    # 1. 定义待评估的替代方案
    alternatives = [
        Alternative("现状维持", "保持现有硬编码和实时处理模式。"),
        Alternative("重构路由", "引入配置化与智能路由策略，但不改批处理。"),
        Alternative("全面优化", "同时实施配置化、智能路由和请求批处理。"),
    ]

    # 2. 定义决策者偏好的权重 (成本30%，收益50%，风险20%)
    weights = {'cost': 0.3, 'benefit': 0.5, 'risk': 0.2}

    # 3. 专家/团队评估各方案在各维度的得分 (1-10分)
    # dev_cost: 开发成本（分越高成本越高）
    # ops_cost: 运维复杂度（分越高越复杂）
    # perf_gain: 性能提升预期（分越高提升越大）
    # maintainability_gain: 可维护性提升（分越高越好）
    # risk: 实施风险（分越高风险越大）
    assessment_matrix = {
        "现状维持": {
            "dev_cost": 1,   # 无新开发
            "ops_cost": 8,   # 运维复杂，难以应对变化
            "perf_gain": 2,  # 无提升
            "maintainability_gain": 2, # 代码僵硬，难以维护
            "risk": 1        # 无变更风险
        },
        "重构路由": {
            "dev_cost": 6,
            "ops_cost": 4,   # 运维简化
            "perf_gain": 5,  # 负载均衡带来一定性能提升
            "maintainability_gain": 8,
            "risk": 4        # 中等风险，涉及核心路由逻辑变更
        },
        "全面优化": {
            "dev_cost": 9,   # 开发成本最高
            "ops_cost": 3,   # 运维更简单，但批处理需要监控
            "perf_gain": 9,  # 性能提升显著
            "maintainability_gain": 9,
            "risk": 7        # 风险最高，涉及异步、批处理等复杂机制
        }
    }

    # 4. 执行评估
    framework = DecisionFramework(alternatives, weights)
    results = framework.evaluate(assessment_matrix)

    # 5. 输出结果与建议
    DecisionFramework.print_recommendation(results)

if __name__ == "__main__":
    main()

python evaluate_decision.py