企业级 AI 中台架构教程

SEO 信息

• 名称：企业级AI中台架构教程
• 描述：零基础企业级AI中台架构教程，涵盖AI中台架构设计、模型服务网关、数据标注平台、实验管理、特征工程、应用编排、监控运维等核心技能，适合AI架构师和技术管理者系统学习。
• 关键词：AI中台, 模型服务平台, AI架构, 企业AI, MLOps
• 长尾关键词：企业级AI中台架构设计教程, AI模型服务平台搭建教程, MLOps平台开发实战, 企业AI基础设施建设教程

一、AI 中台概念与价值

1.1 什么是 AI 中台

AI 中台（AI Middle Platform）是企业级的人工智能基础设施层，它将 AI 能力从业务系统中抽象出来，以平台化、服务化的方式统一管理和供给。如果说数据中台解决的是"数据如何治理"的问题，那么 AI 中台解决的就是"AI 能力如何复用"的问题。

在传统模式下，每个业务团队独立训练模型、独立部署服务、独立管理数据，导致大量重复建设和资源浪费。AI 中台的核心理念是：一次建设，多方复用。

1.2 AI 中台的核心价值

1.3 适用场景

AI 中台并不是所有企业都需要的。以下场景建议建设 AI 中台：

• 企业有 3 个以上 业务线同时使用 AI 能力
• 存在明显的 重复建设 问题（多个团队训练相似模型）
• GPU 资源紧张，需要 统一调度
• 对模型的 合规性、可追溯性 有严格要求
• 希望 AI 能力 快速赋能 新业务线

二、AI 中台架构设计

2.1 整体架构

一个完整的企业级 AI 中台通常包含以下六层：

┌─────────────────────────────────────────────┐
│            AI 应用层（Application Layer）      │
│   智能客服 │ 智能推荐 │ 文档审核 │ 代码助手     │
├─────────────────────────────────────────────┤
│          AI 编排层（Orchestration Layer）      │
│   Agent 编排 │ 工作流引擎 │ Prompt 管理       │
├─────────────────────────────────────────────┤
│          模型服务层（Model Serving Layer）      │
│   统一推理网关 │ 模型仓库 │ A/B 测试          │
├─────────────────────────────────────────────┤
│          模型训练层（Training Layer）           │
│   实验管理 │ 分布式训练 │ 自动调参             │
├─────────────────────────────────────────────┤
│          数据与特征层（Data & Feature Layer）   │
│   数据标注 │ 特征工程 │ 数据版本管理          │
├─────────────────────────────────────────────┤
│          基础设施层（Infrastructure Layer）     │
│   GPU 调度 │ 资源编排 │ 网络 │ 存储           │
└─────────────────────────────────────────────┘

2.2 设计原则

1. 松耦合：各层之间通过标准 API 交互，可独立演进
2. 可扩展：支持新模型、新框架、新硬件的快速接入
3. 多租户：不同业务线、不同团队之间资源隔离
4. 可观测：从请求到推理，全链路可追踪
5. 安全合规：模型和数据的访问控制、审计日志

2.3 技术选型参考

三、模型服务层：统一推理网关

3.1 为什么需要统一推理网关

当企业内部有多个模型、多个业务线时，直接让业务方调用各个模型服务会导致：

• 地址管理混乱，业务方需要知道每个模型的部署位置
• 认证鉴权分散，安全风险高
• 无法统一限流、熔断、灰度
• 缺乏统一的用量统计和计费依据

统一推理网关（Inference Gateway）就是解决这些问题的。

3.2 网关核心能力

# 推理网关的路由逻辑示意
class InferenceGateway:
    def __init__(self):
        self.registry = ModelRegistry()      # 模型注册中心
        self.rate_limiter = RateLimiter()     # 限流器
        self.auth = AuthManager()             # 认证管理
        self.monitor = MetricsCollector()     # 指标采集

    async def infer(self, request: InferenceRequest) -> InferenceResponse:
        # 1. 认证鉴权
        tenant = self.auth.verify(request.api_key)

        # 2. 限流检查
        if not self.rate_limiter.allow(tenant.id, request.model):
            raise RateLimitError()

        # 3. 路由到具体模型实例
        model = self.registry.resolve(
            model_name=request.model,
            version=request.version,  # 可选：指定版本
            strategy="canary"         # 灰度策略
        )

        # 4. 执行推理
        start_time = time.time()
        result = await model.predict(request.input)
        latency = time.time() - start_time

        # 5. 记录指标
        self.monitor.record(
            tenant=tenant.id,
            model=request.model,
            latency=latency,
            tokens=result.usage
        )

        return result

3.3 支持 OpenAI 兼容接口

为了让业务方低成本接入，推理网关通常会暴露 OpenAI 兼容的 API 格式：

# 业务方调用方式与调用 OpenAI 完全一致
curl -X POST https://ai-gateway.internal.com/v1/chat/completions \
  -H "Authorization: Bearer <company-api-key>" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "company-llm-7b",
    "messages": [
      {"role": "user", "content": "帮我总结一下今天的会议纪要"}
    ],
    "temperature": 0.7
  }'

这种方式让业务方可以无缝切换底层模型，只需修改 model 字段即可。

3.4 灰度发布与 A/B 测试

推理网关应支持将流量按比例分配到不同模型版本：

# 模型路由配置示例
routes:
  - model: "customer-service-llm"
    targets:
      - version: "v2.1"
        weight: 90        # 90% 流量
      - version: "v3.0-rc"
        weight: 10        # 10% 灰度流量
    fallback: "v2.1"      # 失败时回退

四、数据管理与标注平台

4.1 数据管理的挑战

AI 中台的数据管理远比传统数据仓库复杂，因为需要同时管理：

• 原始数据：文本、图片、音频、视频
• 标注数据：人工标注的标签、边界框、文本对齐
• 模型数据：模型权重、训练配置、评估结果
• 版本数据：同一数据集的不同版本

4.2 数据版本管理

推荐使用 DVC（Data Version Control）实现数据的版本管理：

# 初始化 DVC
dvc init

# 添加远程存储（如 S3、MinIO）
dvc remote add -d storage s3://ai-platform-data/datasets

# 追踪数据集
dvc add datasets/customer-service/v1/

# 提交版本
git add datasets/customer-service/v1.dvc .gitignore
git commit -m "customer-service dataset v1: 10k samples"

# 推送到远程存储
dvc push

# 切换到历史版本
git checkout v0.9 -- datasets/customer-service/v1.dvc
dvc checkout

4.3 数据标注平台集成

以 Label Studio 为例，展示如何通过 API 集成标注平台：

from label_studio_sdk import Client

# 连接标注平台
ls = Client(url='http://label-studio.internal.com', api_key='your-key')

# 创建项目
project = ls.start_project(
    title='客服对话意图标注',
    label_config='''
    <View>
      <Labels name="intent" toName="text">
        <Label value="咨询" />
        <Label value="投诉" />
        <Label value="退换货" />
        <Label value="其他" />
      </Labels>
      <Text name="text" value="$text" />
    </View>
    '''
)

# 导入待标注数据
tasks = [
    {"data": {"text": "我想退货，商品有质量问题"}},
    {"data": {"text": "请问你们的发货时间是多久"}},
    {"data": {"text": "这个月的账单不对"}},
]
project.import_tasks(tasks)

# 获取标注结果
annotations = project.get_labeled_tasks()

4.4 数据质量保障

建立数据质量检查流水线：

class DataQualityChecker:
    """数据质量检查器"""

    def check(self, dataset: list[dict]) -> QualityReport:
        checks = [
            self._check_completeness,     # 完整性检查
            self._check_duplicates,        # 重复检查
            self._check_label_distribution,# 标签分布检查
            self._check_text_length,       # 文本长度检查
            self._check_encoding,          # 编码检查
        ]
        report = QualityReport()
        for check_fn in checks:
            result = check_fn(dataset)
            report.add(result)
        return report

    def _check_completeness(self, dataset):
        """检查缺失字段"""
        required_fields = ['text', 'label']
        missing = []
        for i, item in enumerate(dataset):
            for field in required_fields:
                if field not in item or not item[field]:
                    missing.append((i, field))
        return CheckResult(
            name="completeness",
            passed=len(missing) == 0,
            details=f"缺失 {len(missing)} 个字段",
            samples=missing[:5]
        )

    def _check_duplicates(self, dataset):
        """检查重复数据"""
        seen = set()
        duplicates = []
        for i, item in enumerate(dataset):
            key = item.get('text', '')
            if key in seen:
                duplicates.append(i)
            seen.add(key)
        return CheckResult(
            name="duplicates",
            passed=len(duplicates) == 0,
            details=f"发现 {len(duplicates)} 条重复数据"
        )

五、模型训练与实验管理

5.1 实验管理的必要性

在企业级 AI 场景中，一个模型可能经历上百次实验迭代。没有实验管理，团队会面临：

• "上周那个效果最好的参数是什么来着？"
• "这个模型是用哪个版本的数据训练的？"
• "张三和李四的实验结果能不能对比？"

5.2 基于 MLflow 的实验管理

import mlflow
import mlflow.pytorch
from transformers import AutoModelForCausalLM, TrainingArguments, Trainer

# 设置实验
mlflow.set_tracking_uri("http://mlflow.internal.com")
mlflow.set_experiment("customer-service-llm-finetune")

# 开始一次实验运行
with mlflow.start_run(run_name="lora-r16-lr2e-4"):
    # 记录超参数
    params = {
        "base_model": "Qwen2.5-7B",
        "method": "LoRA",
        "lora_r": 16,
        "lora_alpha": 32,
        "learning_rate": 2e-4,
        "batch_size": 8,
        "epochs": 3,
        "max_seq_length": 2048,
    }
    mlflow.log_params(params)

    # 训练（简化示例）
    training_args = TrainingArguments(
        output_dir="./output",
        num_train_epochs=params["epochs"],
        per_device_train_batch_size=params["batch_size"],
        learning_rate=params["learning_rate"],
        logging_steps=10,
    )

    # 每个 epoch 记录指标
    for epoch in range(params["epochs"]):
        # ... 训练逻辑 ...
        metrics = {
            "train_loss": 0.85 - epoch * 0.1,
            "eval_accuracy": 0.78 + epoch * 0.05,
            "eval_f1": 0.75 + epoch * 0.06,
        }
        mlflow.log_metrics(metrics, step=epoch)

    # 保存模型
    mlflow.pytorch.log_model(model, "model")

    # 记录数据集版本
    mlflow.log_artifact("datasets/customer-service/v1/meta.json")

5.3 分布式训练调度

在 Kubernetes 集群中使用 PyTorchJob 进行分布式训练：

# pytorch-job.yaml
apiVersion: kubeflow.org/v1
kind: PyTorchJob
metadata:
  name: llm-finetune-lora
  namespace: ai-training
spec:
  pytorchReplicaSpecs:
    Master:
      replicas: 1
      template:
        spec:
          containers:
            - name: pytorch
              image: ai-platform/training:latest
              resources:
                limits:
                  nvidia.com/gpu: 2
                  memory: "32Gi"
              env:
                - name: MODEL_NAME
                  value: "Qwen2.5-7B"
                - name: DATASET_PATH
                  value: "/data/customer-service/v1"
              volumeMounts:
                - name: data-volume
                  mountPath: /data
                - name: model-cache
                  mountPath: /cache
          volumes:
            - name: data-volume
              persistentVolumeClaim:
                claimName: ai-datasets-pvc
            - name: model-cache
              persistentVolumeClaim:
                claimName: model-cache-pvc
    Worker:
      replicas: 2
      # ... 配置类似 Master

六、特征工程平台

6.1 特征工程的核心问题

在实时推理场景中（如推荐系统、风控系统），模型需要的不仅仅是原始数据，还需要经过计算的特征。特征工程平台解决的问题：

• 特征复用：不同模型共享特征，避免重复计算
• 线上线下一致性：训练和推理使用相同的特征计算逻辑
• 实时性：支持实时特征和离线特征的统一管理

6.2 基于 Feast 的特征平台

from feast import FeatureStore, Entity, FeatureView, Field
from feast.types import Float32, Int64
from feast.infra.offline_stores.file_source import FileSource

# 定义实体
user = Entity(
    name="user_id",
    join_keys=["user_id"],
    description="用户ID"
)

# 定义离线特征源
user_behavior_source = FileSource(
    path="s3://features/user_behavior.parquet",
    timestamp_field="event_timestamp"
)

# 定义特征视图
user_behavior_fv = FeatureView(
    name="user_behavior_features",
    entities=[user],
    schema=[
        Field(name="total_orders_30d", dtype=Int64),
        Field(name="avg_order_amount_30d", dtype=Float32),
        Field(name="days_since_last_order", dtype=Int64),
        Field(name="preferred_category", dtype=Int64),
    ],
    source=user_behavior_source,
    ttl=timedelta(days=1),
)

# 注册到特征仓库
store = FeatureStore(repo_path=".")
store.apply([user, user_behavior_fv])

# 在线推理时获取特征
features = store.get_online_features(
    features=[
        "user_behavior_features:total_orders_30d",
        "user_behavior_features:avg_order_amount_30d",
    ],
    entity_rows=[{"user_id": "U12345"}]
).to_dict()

6.3 实时特征计算

对于需要秒级更新的特征，使用流式计算：

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment

env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(env)

# 从 Kafka 读取用户行为事件
t_env.execute_sql("""
    CREATE TABLE user_events (
        user_id STRING,
        action STRING,
        amount DOUBLE,
        event_time TIMESTAMP(3),
        WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'user-events',
        'properties.bootstrap.servers' = 'kafka:9092',
        'format' = 'json'
    )
""")

# 计算滑动窗口特征
t_env.execute_sql("""
    CREATE TABLE user_features AS
    SELECT
        user_id,
        COUNT(*) as order_count_1h,
        SUM(amount) as total_amount_1h,
        AVG(amount) as avg_amount_1h
    FROM user_events
    WHERE action = 'purchase'
    GROUP BY
        user_id,
        TUMBLE(event_time, INTERVAL '1' HOUR)
""")

七、AI 应用编排引擎

7.1 为什么需要编排引擎

现代 AI 应用很少只调用单个模型。一个典型的智能客服应用可能涉及：

1. 意图识别模型 → 判断用户意图
2. RAG 检索 → 从知识库中检索相关信息
3. LLM 生成 → 基于检索结果生成回答
4. 安全审核模型 → 检查输出是否合规
5. 情感分析模型 → 判断用户情绪，决定是否转人工

编排引擎就是将这些步骤串联起来的"指挥家"。

7.2 工作流定义

使用 YAML 定义 AI 应用工作流：

# ai-workflow.yaml
apiVersion: ai-platform/v1
kind: AIWorkflow
metadata:
  name: smart-customer-service
  version: "1.2"
spec:
  entry: intent_router

  nodes:
    intent_router:
      type: model
      model: intent-classifier-v3
      input: "{{user_message}}"
      outputs:
        intent: "{{result.label}}"
        confidence: "{{result.score}}"
      next:
        - condition: "confidence > 0.8"
          goto: "{{intent}}"
        - condition: "confidence <= 0.8"
          goto: fallback_llm

    complaint:
      type: chain
      steps:
        - id: rag_retrieval
          type: retrieval
          knowledge_base: complaint-policies
          query: "{{user_message}}"
          top_k: 3
        - id: llm_response
          type: model
          model: company-llm-14b
          prompt: |
            基于以下政策文档，回答用户的投诉问题。
            政策文档：{{rag_retrieval.results}}
            用户问题：{{user_message}}
        - id: safety_check
          type: model
          model: content-safety-v2
          input: "{{llm_response.text}}"
          next:
            - condition: "result.safe == true"
              goto: respond
            - condition: "result.safe == false"
              goto: human_handoff

    respond:
      type: output
      template: "{{llm_response.text}}"

    human_handoff:
      type: action
      action: transfer_to_agent
      params:
        reason: "safety_concern"

7.3 编排引擎实现

class AIWorkflowEngine:
    """AI 工作流执行引擎"""

    def __init__(self, model_gateway, retrieval_service):
        self.model_gateway = model_gateway
        self.retrieval = retrieval_service
        self.node_handlers = {
            "model": self._handle_model,
            "retrieval": self._handle_retrieval,
            "chain": self._handle_chain,
            "output": self._handle_output,
            "action": self._handle_action,
        }

    async def execute(self, workflow: dict, context: dict) -> dict:
        current_node = workflow["spec"]["entry"]
        trace = ExecutionTrace()

        while current_node and current_node != "__end__":
            node_def = workflow["spec"]["nodes"][current_node]
            node_type = node_def["type"]

            handler = self.node_handlers[node_type]
            result = await handler(node_def, context)

            trace.record(current_node, result)
            context.update(result)

            # 路由到下一个节点
            current_node = self._resolve_next(node_def, result, context)

        return {
            "response": context.get("final_response"),
            "trace": trace.to_dict()
        }

    async def _handle_model(self, node_def, context):
        prompt = self._render_template(node_def.get("prompt", ""), context)
        response = await self.model_gateway.infer(
            model=node_def["model"],
            input=prompt or context.get("user_message")
        )
        return {"result": response}

    async def _handle_retrieval(self, node_def, context):
        query = self._render_template(node_def["query"], context)
        results = await self.retrieval.search(
            knowledge_base=node_def["knowledge_base"],
            query=query,
            top_k=node_def.get("top_k", 3)
        )
        return {"results": results}

八、监控与可观测性

8.1 监控维度

AI 中台的监控需要覆盖三个维度：

基础设施层监控：

• GPU 利用率、显存占用
• CPU/内存使用率
• 网络带宽、存储 I/O

模型服务层监控：

• 推理延迟（P50/P95/P99）
• 吞吐量（QPS/TPS）
• 错误率、超时率
• 模型加载时间

业务效果层监控：

• 模型准确率/召回率的变化
• 用户满意度
• 业务指标（转化率、留存率等）

8.2 基于 Prometheus 的指标采集

from prometheus_client import Histogram, Counter, Gauge, start_http_server

# 定义指标
INFERENCE_LATENCY = Histogram(
    'ai_inference_latency_seconds',
    '模型推理延迟',
    ['model', 'version', 'tenant'],
    buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0]
)

INFERENCE_REQUESTS = Counter(
    'ai_inference_requests_total',
    '推理请求总数',
    ['model', 'version', 'tenant', 'status']
)

MODEL_TOKENS = Counter(
    'ai_inference_tokens_total',
    'Token 使用量',
    ['model', 'tenant', 'type']  # type: input/output
)

GPU_UTILIZATION = Gauge(
    'ai_gpu_utilization_percent',
    'GPU 利用率',
    ['gpu_id', 'node']
)

# 在推理网关中使用
class MonitoredInferenceGateway:
    @INFERENCE_LATENCY.labels(model=model, version=ver, tenant=tid).time()
    async def infer(self, request):
        try:
            result = await self._do_inference(request)
            INFERENCE_REQUESTS.labels(
                model=request.model, version=request.version,
                tenant=request.tenant, status="success"
            ).inc()
            MODEL_TOKENS.labels(
                model=request.model, tenant=request.tenant, type="input"
            ).inc(result.usage.prompt_tokens)
            MODEL_TOKENS.labels(
                model=request.model, tenant=request.tenant, type="output"
            ).inc(result.usage.completion_tokens)
            return result
        except Exception as e:
            INFERENCE_REQUESTS.labels(
                model=request.model, version=request.version,
                tenant=request.tenant, status="error"
            ).inc()
            raise

8.3 Grafana 仪表板配置

{
  "dashboard": {
    "title": "AI 中台监控面板",
    "panels": [
      {
        "title": "推理延迟 P95",
        "type": "graph",
        "targets": [
          {
            "expr": "histogram_quantile(0.95, rate(ai_inference_latency_seconds_bucket[5m]))",
            "legendFormat": "{{model}} - {{version}}"
          }
        ]
      },
      {
        "title": "各租户 Token 消耗",
        "type": "piechart",
        "targets": [
          {
            "expr": "sum by (tenant) (rate(ai_inference_tokens_total[1h]))",
            "legendFormat": "{{tenant}}"
          }
        ]
      }
    ]
  }
}

九、成本治理与资源调度

9.1 AI 成本的构成

企业 AI 成本通常包括：

• 算力成本：GPU/TPU 租赁或采购（通常占 60%-70%）
• 存储成本：数据集、模型权重、日志存储
• 人力成本：平台开发、运维、标注人员
• API 调用成本：第三方模型 API 费用

9.2 成本分摊模型

class CostAllocator:
    """AI 资源成本分摊器"""

    def __init__(self, gpu_price_per_hour: float):
        self.gpu_price = gpu_price_per_hour

    def calculate_monthly_cost(self, tenant_usage: dict) -> dict:
        """
        tenant_usage 格式:
        {
            "team_a": {
                "gpu_hours": 720,      # GPU 使用小时
                "storage_gb": 500,     # 存储 GB
                "api_calls": 100000,   # API 调用次数
                "tokens": 5000000      # Token 消耗
            }
        }
        """
        costs = {}
        for team, usage in tenant_usage.items():
            gpu_cost = usage["gpu_hours"] * self.gpu_price
            storage_cost = usage["storage_gb"] * 0.02  # $0.02/GB/月
            api_cost = usage["tokens"] / 1000 * 0.002  # 按 token 计费

            costs[team] = {
                "gpu_cost": round(gpu_cost, 2),
                "storage_cost": round(storage_cost, 2),
                "api_cost": round(api_cost, 2),
                "total": round(gpu_cost + storage_cost + api_cost, 2)
            }

        return costs

9.3 智能资源调度

class GPUResourceScheduler:
    """GPU 资源智能调度器"""

    def schedule(self, training_jobs: list, available_gpus: int):
        """
        基于优先级和资源需求的调度策略
        """
        # 按优先级排序
        jobs = sorted(training_jobs, key=lambda j: j.priority, reverse=True)

        allocated = []
        remaining_gpus = available_gpus

        for job in jobs:
            if job.gpu_count <= remaining_gpus:
                allocated.append({
                    "job_id": job.id,
                    "gpus": job.gpu_count,
                    "node": self._find_best_node(job)
                })
                remaining_gpus -= job.gpu_count
            else:
                # 尝试抢占低优先级任务
                preempted = self._try_preempt(job, allocated, remaining_gpus)
                if preempted:
                    allocated = preempted
                    remaining_gpus = available_gpus - sum(
                        a["gpus"] for a in allocated
                    )

        return allocated

    def _find_best_node(self, job):
        """选择最优节点（考虑数据本地性、GPU 型号等）"""
        # 简化逻辑：优先选择数据所在节点
        if job.data_locality:
            return job.data_locality
        return self._least_loaded_node()

十、实战：从零搭建企业 AI 中台

10.1 项目规划

假设我们是一家有 500 人的中型企业，有 3 个业务线需要使用 AI 能力。我们分三个阶段搭建 AI 中台：

第一阶段（1-2 个月）：基础能力

• 搭建 Kubernetes 集群 + GPU 节点
• 部署统一推理网关
• 基础监控

第二阶段（2-3 个月）：训练与数据

• 实验管理平台
• 数据标注平台
• 特征工程平台

第三阶段（3-4 个月）：高级能力

• 工作流编排引擎
• 成本治理
• 自动化运维

10.2 第一阶段：搭建推理网关

使用开源方案快速搭建：

# 1. 准备 Kubernetes 集群
# 假设已有 K8s 集群，且有 GPU 节点

# 2. 部署 NVIDIA 设备插件
kubectl apply -f https://raw.githubusercontent.com/NVIDIA/k8s-device-plugin/v0.14.0/nvidia-device-plugin.yml

# 3. 部署 vLLM 作为推理引擎
cat <<EOF | kubectl apply -f -
apiVersion: apps/v1
kind: Deployment
metadata:
  name: qwen-7b-service
  namespace: ai-serving
spec:
  replicas: 2
  selector:
    matchLabels:
      app: qwen-7b
  template:
    metadata:
      labels:
        app: qwen-7b
    spec:
      containers:
        - name: vllm
          image: vllm/vllm-openai:latest
          args:
            - "--model"
            - "Qwen/Qwen2.5-7B-Instruct"
            - "--served-model-name"
            - "company-llm-7b"
            - "--tensor-parallel-size"
            - "1"
            - "--max-model-len"
            - "4096"
          ports:
            - containerPort: 8000
          resources:
            limits:
              nvidia.com/gpu: 1
              memory: "16Gi"
          readinessProbe:
            httpGet:
              path: /health
              port: 8000
            initialDelaySeconds: 60
            periodSeconds: 10
---
apiVersion: v1
kind: Service
metadata:
  name: qwen-7b-svc
  namespace: ai-serving
spec:
  selector:
    app: qwen-7b
  ports:
    - port: 8000
      targetPort: 8000
EOF

# 4. 部署 API 网关（以 APISIX 为例）
helm repo add apisix https://charts.apiseven.com
helm install ai-gateway apisix/apisix \
  --namespace ai-gateway --create-namespace \
  --set gateway.type=LoadBalancer

10.3 第二阶段：部署 MLflow 实验管理

# 使用 Helm 部署 MLflow
helm repo add community-charts https://community-charts.github.io/helm-charts
helm install mlflow community-charts/mlflow \
  --namespace ai-platform --create-namespace \
  --set "backendStore.postgres.host=mlflow-db" \
  --set "backendStore.postgres.database=mlflow" \
  --set "artifactRoot.s3.bucket=mlflow-artifacts"

10.4 第三阶段：编排引擎集成

# 使用 Python SDK 编排 AI 应用
from ai_platform import WorkflowBuilder, ModelNode, RetrievalNode

workflow = (
    WorkflowBuilder("smart-customer-service")
    .add_node(ModelNode(
        name="intent_classifier",
        model="intent-classifier-v3",
        input="{{user_message}}"
    ))
    .add_node(RetrievalNode(
        name="knowledge_retrieval",
        knowledge_base="complaint-policies",
        query="{{user_message}}",
        condition="intent_classifier.result.label == 'complaint'"
    ))
    .add_node(ModelNode(
        name="response_generator",
        model="company-llm-14b",
        prompt="基于以下信息回答：{{knowledge_retrieval.results}}\n用户问题：{{user_message}}"
    ))
    .add_node(ModelNode(
        name="safety_checker",
        model="content-safety-v2",
        input="{{response_generator.result.text}}"
    ))
    .connect("intent_classifier", "knowledge_retrieval")
    .connect("knowledge_retrieval", "response_generator")
    .connect("response_generator", "safety_checker")
    .build()
)

# 部署并测试
workflow.deploy()
result = workflow.run({"user_message": "我要投诉你们的物流太慢了"})
print(result.response)

总结

企业级 AI 中台的建设是一个系统工程，核心要点：

1. 分层设计：基础设施层、数据层、模型层、服务层、应用层各司其职
2. 统一网关：所有模型服务通过统一入口暴露，便于管理和治理
3. 数据闭环：从标注、训练到推理，数据全生命周期管理
4. 可观测性：从基础设施到业务效果，全链路监控
5. 成本治理：建立清晰的成本分摊模型，避免资源浪费

建设 AI 中台不是一蹴而就的，建议按照"先解决最痛的点"的原则，逐步迭代。通常推理网关是第一个要建设的组件，因为它能最快地解决多团队协作和资源管理的问题。