AI 中台架构设计——一家中型企业的落地方案
AI 中台架构设计——一家中型企业的落地方案
适读人群：技术架构师、技术负责人、AI 工程师 | 阅读时长：约 15 分钟 | 核心价值：一个真实可落地的中型企业 AI 中台设计方案
上个季度我参与了一家制造业企业的 AI 中台选型。这家公司大概 300 人的技术团队，分散在十几个业务系统里。CEO 开了个会说要做 AI 中台，然后问我有什么建议。
会议上有个架构师拿出来一份 PPT，里面全是大厂 AI 中台的架构图：向量数据库集群、流式计算引擎、模型训练平台、A/B 测试框架……密密麻麻，看起来很厉害。
我当时问了一个问题：你们现在 AI 相关的日活用户有多少？
对方沉默了一会儿，说大概……五千左右？
我说那个方案可能不适合你们。
这篇文章就是聊那次经历。中型企业做 AI 中台，不是要复制大厂，是要找到适合自己规模和阶段的方案。
为什么中型企业需要 AI 中台
在聊架构之前，先说清楚为什么需要中台。这家企业在我接触之前已经有了 AI 的散装实践：
- 客服团队搭了一个基于 GPT-4 的问答机器人
- 研发团队用 GitLab CI 接了一个代码审查的 AI 脚本
- 数据团队有一个用 Python 写的报告生成脚本，也调 LLM
- 销售部门刚采购了一个第三方 AI CRM 工具
四个地方，四套 API Key，四套配置，四套计费，四套故障处理方式。有一次 OpenAI 涨价，他们得改四个地方。有一次某个 Key 超配额了，排查了两个小时才找到是哪个系统。
AI 中台要解决的核心问题就这么朴素：统一管理，减少重复建设。
中型企业 AI 中台的设计原则
在开始画架构图之前，我定了几个原则，这些原则决定了后面所有的技术选型：
原则一：够用就好，不要过度设计。 300 人的技术团队，不可能有专职的 MLOps 团队来维护复杂的基础设施。
原则二：可观测性优先。 不知道钱花在哪里、不知道哪个服务出问题，AI 中台就是个黑盒，用不起来。
原则三：渐进式落地。 不要试图一次把所有东西都做完。先做 API 统一管理，再做缓存，再做 RAG，按优先级来。
原则四：保留逃生通道。 万一中台挂了，各业务系统要能快速切回直接调 API 的方式，不能因为中台故障导致所有业务都挂。
整体架构设计
外部业务系统
(客服/研发/数据/销售)
│
│ HTTP/gRPC
▼
+──────────────────────────────────────────+
│ AI 网关层 │
│ - 统一鉴权（JWT/API Key） │
│ - 路由分发（按业务/模型类型） │
│ - 限流熔断（Sentinel） │
│ - 请求日志 │
+──────────────────────────────────────────+
│
├──────────────┬────────────────┐
▼ ▼ ▼
+──────────+ +──────────────+ +──────────────+
│ LLM 代理 │ │ 语义缓存 │ │ RAG 服务 │
│ 服务 │ │ (Redis) │ │ (向量检索) │
+──────────+ +──────────────+ +──────────────+
│
├────────────────────────────┐
▼ ▼
+──────────────────+ +──────────────────────+
│ 模型路由层 │ │ 可观测性平台 │
│ - OpenAI │ │ - 费用统计 │
│ - Azure OpenAI │ │ - 延迟监控 │
│ - 国产模型 │ │ - 错误率统计 │
│ - 本地模型 │ │ - 按业务分摊 │
+──────────────────+ +──────────────────────+这个架构有四个核心模块，下面逐一说。
模块一：AI 网关层
网关是整个中台的入口，所有 AI 请求都要经过这里。
我们用 Spring Cloud Gateway 来实现，主要处理这几件事：
@Configuration
public class AiGatewayConfig {
@Bean
public RouteLocator aiGatewayRoutes(RouteLocatorBuilder builder) {
return builder.routes()
// 客服系统的请求路由
.route("customer-service-ai", r -> r
.path("/api/ai/customer/**")
.filters(f -> f
.filter(new AuthenticationFilter()) // 鉴权
.filter(new RateLimitFilter("cs", 100)) // 限流：每分钟100次
.filter(new CostTrackingFilter("customer-service")) // 费用追踪
.rewritePath("/api/ai/customer/(?<segment>.*)", "/${segment}")
)
.uri("lb://ai-proxy-service")
)
// 研发工具的请求路由
.route("dev-tools-ai", r -> r
.path("/api/ai/devtools/**")
.filters(f -> f
.filter(new AuthenticationFilter())
.filter(new RateLimitFilter("dev", 50))
.filter(new CostTrackingFilter("dev-tools"))
)
.uri("lb://ai-proxy-service")
)
.build();
}
}鉴权这里，我们没有用 API Key 透传，而是让各业务系统用内部的 JWT，网关负责验证 JWT，然后用统一的主 API Key 去调上游模型。这样 API Key 只存在一个地方，轮换的时候改一处就够了。
模块二：LLM 代理服务
代理服务是核心，负责把请求转发给实际的 LLM，同时做语义缓存、失败重试、模型降级。
@Service
@RequiredArgsConstructor
public class LlmProxyService {
private final SemanticCacheService cacheService;
private final ModelRouter modelRouter;
private final CostTracker costTracker;
public LlmResponse proxy(LlmRequest request) {
// 1. 尝试语义缓存
Optional<LlmResponse> cached = cacheService.get(request);
if (cached.isPresent()) {
costTracker.recordCacheHit(request.getBizUnit());
return cached.get();
}
// 2. 路由到合适的模型
LlmProvider provider = modelRouter.route(request);
// 3. 调用模型（带重试）
LlmResponse response = callWithRetry(provider, request);
// 4. 缓存结果
if (response.isCacheable()) {
cacheService.put(request, response);
}
// 5. 记录费用
costTracker.record(request.getBizUnit(), response.getUsage());
return response;
}
private LlmResponse callWithRetry(LlmProvider provider, LlmRequest request) {
int maxRetry = 3;
for (int i = 0; i < maxRetry; i++) {
try {
return provider.call(request);
} catch (RateLimitException e) {
if (i < maxRetry - 1) {
// 限流时自动切换到备用模型
provider = modelRouter.fallback(provider);
log.warn("Rate limited, switching to fallback: {}", provider.getName());
}
} catch (Exception e) {
if (i == maxRetry - 1) throw e;
log.warn("LLM call failed, retry {}/{}", i + 1, maxRetry, e);
}
}
throw new RuntimeException("All retries exhausted");
}
}模型路由策略
不同的任务用不同的模型，这是降成本的关键：
@Component
public class ModelRouter {
public LlmProvider route(LlmRequest request) {
// 简单的分类任务，用便宜的小模型
if (request.getTaskType() == CLASSIFICATION || request.getTaskType() == EXTRACTION) {
return providers.get("gpt-3.5-turbo");
}
// 代码生成、复杂推理，用强模型
if (request.getTaskType() == CODE_GEN || request.getTaskType() == COMPLEX_REASONING) {
return providers.get("gpt-4o");
}
// 敏感业务（合同分析），走私有化部署的模型
if (request.isSensitive()) {
return providers.get("private-llm");
}
// 默认
return providers.get("gpt-4o-mini");
}
}模块三：语义缓存
这是降成本效果最明显的功能。
普通缓存是完全匹配，"今天天气怎么样" 和 "今天的天气如何" 是两条不同的缓存。语义缓存通过向量相似度匹配，语义相近的问题命中同一条缓存。
@Service
@RequiredArgsConstructor
public class SemanticCacheService {
private final EmbeddingService embeddingService; // 生成向量
private final VectorStore vectorStore; // 存储向量（我们用 Redis Stack）
private static final double SIMILARITY_THRESHOLD = 0.92; // 相似度阈值
public Optional<LlmResponse> get(LlmRequest request) {
// 1. 把请求 prompt 转成向量
float[] queryVector = embeddingService.embed(request.getPrompt());
// 2. 在向量库里找相似的历史请求
List<SimilarResult> similar = vectorStore.findSimilar(
queryVector,
SIMILARITY_THRESHOLD,
request.getModel() // 只在同模型的缓存里找
);
if (similar.isEmpty()) {
return Optional.empty();
}
// 3. 返回最相似的缓存结果
SimilarResult best = similar.get(0);
log.debug("Cache hit with similarity: {}", best.getSimilarity());
return Optional.of(best.getResponse());
}
public void put(LlmRequest request, LlmResponse response) {
float[] vector = embeddingService.embed(request.getPrompt());
vectorStore.store(request.getPrompt(), vector, response, request.getModel());
}
}语义缓存的缓存命中率在我们的场景里大概有 35%，也就是说节省了三分之一的 API 调用费用。对于重复性查询比较多的业务（比如客服 FAQ），效果更好。
模块四：可观测性平台
这个是最容易被忽视但最重要的模块。
我们需要知道：每个业务系统花了多少钱？哪个模型的效果更好？哪个 Prompt 导致了高延迟？
@Component
@RequiredArgsConstructor
public class AiMetricsCollector {
private final MeterRegistry meterRegistry;
private final InfluxDBClient influxDB;
public void recordRequest(AiRequestMetrics metrics) {
// Prometheus metrics（用于告警）
meterRegistry.counter("ai.requests.total",
"biz_unit", metrics.getBizUnit(),
"model", metrics.getModel(),
"status", metrics.getStatus()
).increment();
meterRegistry.timer("ai.request.latency",
"model", metrics.getModel()
).record(metrics.getLatency());
// InfluxDB（用于费用统计和趋势分析）
Point point = Point.measurement("ai_usage")
.addTag("biz_unit", metrics.getBizUnit())
.addTag("model", metrics.getModel())
.addTag("task_type", metrics.getTaskType())
.addField("prompt_tokens", metrics.getPromptTokens())
.addField("completion_tokens", metrics.getCompletionTokens())
.addField("cost_usd", metrics.getCostUsd())
.addField("latency_ms", metrics.getLatencyMs())
.addField("cache_hit", metrics.isCacheHit() ? 1 : 0)
.time(Instant.now(), WritePrecision.MS);
influxDB.getWriteApiBlocking().writePoint(point);
}
}这套数据接上 Grafana，就可以出费用看板、延迟看板、缓存命中率看板。每周我会给各业务团队发一份"AI 使用账单"，这个东西推出来之后大家对 Prompt 的质量开始认真对待了——因为他们知道自己在花多少钱。
渐进式落地计划
这个架构不是一次性建完的。我给这家企业设计了一个三个月的落地计划：
第一个月：API 网关 + 统一管理
- 搭 AI 网关，接入所有已有的 AI 业务
- 统一 API Key 管理
- 基础监控（请求量、费用）
- 预期效果：API Key 统一管理，成本可见
第二个月：语义缓存 + 模型路由
- 上语义缓存，降低 API 调用成本
- 实现简单的模型路由（便宜任务用小模型）
- 完善费用分摊报表
- 预期效果：API 成本降低 30-40%
第三个月：RAG 基础能力 + 规范化
- 搭 RAG 服务，让各业务系统可以接入私有知识库
- 制定 AI 使用规范（哪些场景用什么模型，Prompt 如何管理）
- 建立 AI 能力目录（已有哪些 AI 能力可以复用）
- 预期效果：新业务接入 AI 能力的时间从 2 周降到 2 天
最常见的问题
Q：要不要自己部署模型？
除非有极强的数据隐私要求，中型企业不要自己部署大模型。运维成本和技术门槛都很高。先把调用托管模型的工程能力做好，等规模上来了再考虑。
Q：用 LangChain 还是自己写？
我们最终选择了自己写核心逻辑，LangChain 用于快速原型验证。LangChain 对于线上系统的可观测性、可定制性都不够好，而且版本迭代很快，维护成本高。
Q：向量数据库选什么？
初期用 Redis Stack（带向量搜索的 Redis），团队熟悉，运维简单。等向量数据量上了百万再考虑 Milvus 或者 Weaviate。
AI 中台不是技术问题，是工程问题。真正的挑战不是架构设计，而是怎么让各业务团队愿意用、能用起来。所以我给的架构建议，一定比你在大厂 PPT 上看到的简单很多——因为复杂的架构，在中型企业里没人能维护住。