企业级大模型K8s部署方案(开源千问)
·
基于 K8s 虚拟化的水平扩容架构
目录
一、架构概述
1.1 整体架构图
┌─────────────────────────────────────────────────────────────────────────────┐
│ 企业应用层 │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 智能客服系统 │ │ 企业知识库 │ │ AI机器人 │ │ 数据分析平台 │ │
│ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │
│ │ │ │ │ │
│ ┌──────┴─────────────────┴─────────────────┴─────────────────┴───────┐ │
│ │ API 网关层 (Kong / APISIX) │ │
│ │ 认证 · 限流 · 路由 · 负载均衡 · 审计 │ │
│ └──────────────────────────────┬──────────────────────────────────────┘ │
│ │ │
│ ┌──────────────────────────────┴──────────────────────────────────────┐ │
│ │ 服务编排层 (RAG / Agent) │ │
│ │ ┌────────────┐ ┌────────────┐ ┌────────────┐ ┌────────────┐ │ │
│ │ │ RAG 服务 │ │ Agent 服务 │ │ 工具调用 │ │ 会话管理 │ │ │
│ │ └────────────┘ └────────────┘ └────────────┘ └────────────┘ │ │
│ └──────────────────────────────┬──────────────────────────────────────┘ │
│ │ │
│ ┌──────────────────────────────┴──────────────────────────────────────┐ │
│ │ 模型推理层 (vLLM / TGI) │ │
│ │ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ │ │
│ │ │ Pod-1 │ │ Pod-2 │ │ Pod-3 │ │ Pod-4 │ │ Pod-N │ HPA │ │
│ │ │ 7B模型 │ │ 7B模型 │ │ 14B模型│ │ 72B模型│ │ ... │ 自动扩缩 │ │
│ │ └────────┘ └────────┘ └────────┘ └────────┘ └────────┘ │ │
│ └──────────────────────────────┬──────────────────────────────────────┘ │
│ │ │
│ ┌──────────────────────────────┴──────────────────────────────────────┐ │
│ │ 向量数据库层 (Milvus / pgvector) │ │
│ │ 知识库检索 · 语义搜索 · 相似度匹配 │ │
│ └──────────────────────────────┬──────────────────────────────────────┘ │
│ │ │
│ ┌──────────────────────────────┴──────────────────────────────────────┐ │
│ │ 数据存储层 │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ Redis │ │ PostgreSQL│ │ MinIO │ │ Elasticsearch│ │ │
│ │ │ 缓存/会话 │ │ 业务数据 │ │ 文件/模型 │ │ 日志/检索 │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │ │
│ └──────────────────────────────┬──────────────────────────────────────┘ │
│ │ │
│ ┌──────────────────────────────┴──────────────────────────────────────┐ │
│ │ Kubernetes 集群 (GPU Node Pool) │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ GPU Node │ │ GPU Node │ │ GPU Node │ │ CPU Node │ │ │
│ │ │ A100x4 │ │ A100x4 │ │ A100x4 │ │ 32C128G │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────┘
1.2 设计原则
| 原则 | 说明 |
|---|---|
| 高可用 | 多副本部署，跨节点/跨可用区容灾 |
| 弹性伸缩 | HPA/VPA 基于 GPU 利用率/QPS 自动扩缩容 |
| 资源隔离 | Namespace 隔离不同业务，LimitRange 控制资源 |
| 离线安全 | 完全内网部署，无外网依赖，私有镜像/模型仓库 |
| 统一接口 | OpenAI 兼容 API，便于应用层对接 |
二、技术栈选型
2.1 核心技术组件
| 组件 | 选型 | 版本 | 说明 |
|---|---|---|---|
| 容器编排 | Kubernetes | 1.28+ | 生产级容器编排 |
| GPU 调度 | NVIDIA Device Plugin | 0.14+ | GPU 资源暴露 |
| GPU 运行时 | NVIDIA Container Toolkit | 1.13+ | 容器 GPU 支持 |
| 模型推理 | vLLM | 0.5+ | 高吞吐推理引擎 |
| 向量数据库 | Milvus | 2.4+ | 分布式向量检索 |
| API 网关 | APISIX | 3.8+ | 高性能 API 网关 |
| 缓存 | Redis | 7.2+ | 会话/缓存 |
| 数据库 | PostgreSQL | 16+ | 业务数据存储 |
| 对象存储 | MinIO | 17+ | 模型/文件存储 |
| 服务网格 | Istio | 1.20+ | 流量管理(可选) |
| 监控 | Prometheus + Grafana | - | 指标监控 |
| 日志 | ELK / Loki | - | 日志收集 |
2.2 模型选型
| 场景 | 推荐模型 | 参数 | 量化 | 单卡显存 |
|---|---|---|---|---|
| 客服对话 | Qwen2.5-7B-Instruct | 7B | INT4 | ~6GB |
| 知识库问答 | Qwen2.5-14B-Instruct | 14B | INT4 | ~10GB |
| 复杂推理 | Qwen2.5-32B-Instruct | 32B | INT4 | ~20GB |
| 企业级 | Qwen2.5-72B-Instruct | 72B | INT4 (4卡) | ~40GB/卡 |
| 嵌入向量 | BGE-M3 | 567M | FP16 | ~1GB |
三、集群规划
3.1 节点规划
┌─────────────────────────────────────────────────────────────┐
│ Kubernetes 集群 │
├──────────────┬──────────────────────────────────────────────┤
│ 节点类型 │ 配置规格 │
├──────────────┼──────────────────────────────────────────────┤
│ Master x3 │ 16C / 64G / 500SSD (高可用控制平面) │
│ CPU Worker │ 32C / 128G / 2TB (CPU 推理/服务) │
│ GPU Worker │ 64C / 512G / 4xA100-80G (GPU 推理) │
│ │ 或 64C / 512G / 8xA100-80G (大模型推理) │
│ Storage │ 分布式存储 Ceph / Longhorn (模型/数据) │
└──────────────┴──────────────────────────────────────────────┘
3.2 资源配额
# namespace-quota.yaml
apiVersion: v1
kind: ResourceQuota
metadata:
name: llm-quota
namespace: llm-system
spec:
hard:
requests.cpu: "256"
requests.memory: 1Ti
limits.cpu: "512"
limits.memory: 2Ti
# GPU 资源通过 NVIDIA Device Plugin 管理
requests.nvidia.com/gpu: "32"
pods: "100"
persistentvolumeclaims: "20"
四、核心组件部署
4.1 NVIDIA GPU 插件部署
# nvidia-device-plugin.yaml
apiVersion: apps/v1
kind: DaemonSet
metadata:
name: nvidia-device-plugin-daemonset
namespace: kube-system
spec:
selector:
matchLabels:
name: nvidia-device-plugin-ds
template:
metadata:
labels:
name: nvidia-device-plugin-ds
spec:
tolerations:
- key: nvidia.com/gpu
operator: Exists
effect: NoSchedule
containers:
- image: registry.k8s.io/nvidia-device-plugin:0.14.1
name: nvidia-device-plugin-ctr
securityContext:
allowPrivilegeEscalation: false
capabilities:
drop: ["ALL"]
volumeMounts:
- name: device-plugin
mountPath: /var/lib/kubelet/device-plugins
volumes:
- name: device-plugin
hostPath:
path: /var/lib/kubelet/device-plugins
4.2 模型推理服务 (vLLM)
# vllm-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: qwen25-7b-deployment
namespace: llm-system
labels:
app: qwen25-7b
model-size: 7b
spec:
replicas: 3
selector:
matchLabels:
app: qwen25-7b
strategy:
type: RollingUpdate
rollingUpdate:
maxSurge: 1
maxUnavailable: 0
template:
metadata:
labels:
app: qwen25-7b
spec:
tolerations:
- key: nvidia.com/gpu
operator: Exists
effect: NoSchedule
affinity:
nodeAffinity:
requiredDuringSchedulingIgnoredDuringExecution:
nodeSelectorTerms:
- matchExpressions:
- key: nvidia.com/gpu.product
operator: In
values: ["NVIDIA-A100-SXM4-80GB"]
podAntiAffinity:
preferredDuringSchedulingIgnoredDuringExecution:
- weight: 100
podAffinityTerm:
labelSelector:
matchExpressions:
- key: app
operator: In
values: ["qwen25-7b"]
topologyKey: kubernetes.io/hostname
containers:
- name: vllm
image: registry.internal/vllm/qwen25-7b:latest
imagePullPolicy: IfNotPresent
ports:
- containerPort: 8000
name: http
- containerPort: 8001
name: metrics
env:
- name: HUGGING_FACE_HUB_TOKEN
valueFrom:
secretKeyRef:
name: hf-secret
key: token
- name: VLLM_MODEL
value: /models/Qwen2.5-7B-Instruct
- name: VLLM_TENSOR_PARALLEL_SIZE
value: "1"
- name: VLLM_MAX_MODEL_LEN
value: "8192"
- name: VLLM_GPU_MEMORY_UTILIZATION
value: "0.9"
- name: VLLM_MAX_NUM_BATCHED_TOKENS
value: "8192"
- name: VLLM_MAX_NUM_SEQS
value: "256"
resources:
requests:
cpu: "8"
memory: "32Gi"
nvidia.com/gpu: "1"
limits:
cpu: "16"
memory: "64Gi"
nvidia.com/gpu: "1"
volumeMounts:
- name: model-storage
mountPath: /models
readOnly: true
- name: shm
mountPath: /dev/shm
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 120
periodSeconds: 30
timeoutSeconds: 10
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 60
periodSeconds: 10
volumes:
- name: model-storage
persistentVolumeClaim:
claimName: model-pvc
- name: shm
emptyDir:
medium: Memory
sizeLimit: "16Gi"
---
apiVersion: v1
kind: Service
metadata:
name: qwen25-7b-service
namespace: llm-system
spec:
selector:
app: qwen25-7b
ports:
- port: 8000
targetPort: 8000
name: http
type: ClusterIP
---
# HPA 水平自动扩缩容
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: qwen25-7b-hpa
namespace: llm-system
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: qwen25-7b-deployment
minReplicas: 2
maxReplicas: 20
metrics:
# 基于 CPU 利用率
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
# 基于 GPU 利用率 (需要 custom metrics)
- type: Pods
pods:
metric:
name: gpu_utilization
target:
type: AverageValue
averageValue: "75"
behavior:
scaleUp:
stabilizationWindowSeconds: 60
policies:
- type: Pods
value: 2
periodSeconds: 60
scaleDown:
stabilizationWindowSeconds: 300
policies:
- type: Pods
value: 1
periodSeconds: 120
4.3 多模型路由服务
# model-router-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: model-router
namespace: llm-system
spec:
replicas: 2
selector:
matchLabels:
app: model-router
template:
metadata:
labels:
app: model-router
spec:
containers:
- name: router
image: registry.internal/model-router:latest
ports:
- containerPort: 8080
env:
- name: MODELS_CONFIG
value: |
{
"qwen25-7b": {
"service": "qwen25-7b-service.llm-system.svc.cluster.local:8000",
"max_tokens": 8192,
"priority": 1
},
"qwen25-14b": {
"service": "qwen25-14b-service.llm-system.svc.cluster.local:8000",
"max_tokens": 8192,
"priority": 2
},
"qwen25-72b": {
"service": "qwen25-72b-service.llm-system.svc.cluster.local:8000",
"max_tokens": 4096,
"priority": 3
}
}
resources:
requests:
cpu: "2"
memory: "4Gi"
limits:
cpu: "4"
memory: "8Gi"
---
apiVersion: v1
kind: Service
metadata:
name: model-router-service
namespace: llm-system
spec:
selector:
app: model-router
ports:
- port: 8080
targetPort: 8080
type: ClusterIP
4.4 向量数据库 (Milvus)
# milvus-values.yaml (Helm 部署)
# helm install milvus milvus/milvus -f milvus-values.yaml -n vector-db
replicaManager:
enabled: true
replicas: 2
etcd:
replicaCount: 3
persistence:
size: 100Gi
minio:
mode: distributed
replicas: 4
persistence:
size: 500Gi
pulsar:
bookkeeper:
volumeSize: 200Gi
bookies:
replicas: 3
standalone:
enabled: false
cluster:
enabled: true
dataCoord:
replicas: 2
dataNode:
replicas: 2
indexCoord:
replicas: 2
indexNode:
replicas: 2
proxy:
replicas: 2
rootCoord:
replicas: 2
queryCoord:
replicas: 2
queryNode:
replicas: 3
resources:
requests:
cpu: "4"
memory: "16Gi"
limits:
cpu: "8"
memory: "32Gi"
4.5 Redis 缓存
# redis-statefulset.yaml
apiVersion: apps/v1
kind: StatefulSet
metadata:
name: redis
namespace: llm-system
spec:
serviceName: redis
replicas: 3
selector:
matchLabels:
app: redis
template:
metadata:
labels:
app: redis
spec:
containers:
- name: redis
image: registry.internal/redis:7.2
ports:
- containerPort: 6379
command: ["redis-server"]
args:
- "--requirepass"
- "$(REDIS_PASSWORD)"
- "--appendonly"
- "yes"
- "--maxmemory"
- "4gb"
- "--maxmemory-policy"
- "allkeys-lru"
env:
- name: REDIS_PASSWORD
valueFrom:
secretKeyRef:
name: redis-secret
key: password
resources:
requests:
cpu: "2"
memory: "4Gi"
limits:
cpu: "4"
memory: "8Gi"
volumeMounts:
- name: redis-data
mountPath: /data
volumeClaimTemplates:
- metadata:
name: redis-data
spec:
accessModes: ["ReadWriteOnce"]
resources:
requests:
storage: 50Gi
---
apiVersion: v1
kind: Service
metadata:
name: redis-service
namespace: llm-system
spec:
selector:
app: redis
ports:
- port: 6379
targetPort: 6379
clusterIP: None
五、应用场景实现
5.1 智能客服系统
┌─────────────────────────────────────────────────────────┐
│ 智能客服系统架构 │
├─────────────────────────────────────────────────────────┤
│ │
│ 用户 ──► Web/APP/微信 ──► API网关 ──► 客服服务 │
│ │ │
│ ▼ │
│ ┌───────────────────────────┐ │
│ │ 意图识别服务 │ │
│ │ (小模型快速分类) │ │
│ └───────┬───────────────────┘ │
│ │ │
│ ┌─────────────┼─────────────┐ │
│ ▼ ▼ ▼ │
│ ┌────────┐ ┌────────┐ ┌────────┐ │
│ │ FAQ问答 │ │ 业务办理│ │ 人工转接│ │
│ │ RAG+LLM│ │ Agent │ │ 排队系统│ │
│ └────┬───┘ └────┬───┘ └────────┘ │
│ │ │ │
│ ▼ ▼ │
│ 向量检索 工具调用 │
│ 知识库 API集成 │
│ │
└─────────────────────────────────────────────────────────┘
客服服务核心代码
# customer_service.py
import asyncio
import json
from typing import Dict, List, Optional
from dataclasses import dataclass, field
import redis
import httpx
from milvus import MilvusClient
@dataclass
class ConversationContext:
"""对话上下文"""
session_id: str
user_id: str
messages: List[Dict] = field(default_factory=list)
intent: Optional[str] = None
slots: Dict = field(default_factory=dict)
turn_count: int = 0
class CustomerServiceEngine:
"""智能客服引擎"""
def __init__(self, config: Dict):
self.config = config
self.redis_client = redis.Redis(
host=config['redis_host'],
port=config['redis_port'],
password=config['redis_password'],
decode_responses=True
)
self.milvus_client = MilvusClient(
uri=config['milvus_uri']
)
self.llm_client = httpx.AsyncClient(
base_url=config['llm_api_url'],
timeout=60.0
)
self.intent_model_url = config['intent_model_url']
async def handle_message(self, user_id: str, message: str) -> Dict:
"""处理用户消息"""
# 1. 获取或创建会话
session_id = await self._get_session_id(user_id)
context = await self._load_context(session_id)
# 2. 意图识别
intent = await self._classify_intent(message, context)
context.intent = intent
# 3. 根据意图路由
if intent == "faq":
response = await self._handle_faq(message, context)
elif intent == "business":
response = await self._handle_business(message, context)
elif intent == "complaint":
response = await self._handle_complaint(message, context)
elif intent == "transfer":
response = await self._transfer_to_human(user_id, context)
else:
response = await self._general_chat(message, context)
# 4. 更新上下文
context.messages.append({"role": "user", "content": message})
context.messages.append({"role": "assistant", "content": response})
context.turn_count += 1
# 保持最近20轮对话
if len(context.messages) > 40:
context.messages = context.messages[-40:]
await self._save_context(session_id, context)
return {
"session_id": session_id,
"response": response,
"intent": intent,
"suggestions": self._generate_suggestions(intent, response)
}
async def _handle_faq(self, message: str, context: ConversationContext) -> str:
"""FAQ 问答 - RAG 流程"""
# 1. 向量化查询
embedding = await self._get_embedding(message)
# 2. 向量检索
results = self.milvus_client.search(
collection_name="faq_knowledge",
data=[embedding],
limit=5,
output_fields=["content", "title", "source"]
)
# 3. 构建 Prompt
knowledge_context = self._format_knowledge(results)
prompt = self._build_faq_prompt(message, knowledge_context, context)
# 4. 调用 LLM
response = await self._call_llm(prompt, model="qwen25-7b")
return response
async def _handle_business(self, message: str, context: ConversationContext) -> str:
"""业务办理 - Agent 流程"""
# 1. 提取槽位信息
slots = await self._extract_slots(message, context)
context.slots.update(slots)
# 2. 检查槽位是否完整
missing_slots = self._check_slots(context.intent, context.slots)
if missing_slots:
return f"请您提供{missing_slots}，以便我为您办理"
# 3. 调用业务 API
result = await self._call_business_api(context.intent, context.slots)
# 4. 生成回复
prompt = self._build_business_response_prompt(
context.intent, context.slots, result
)
return await self._call_llm(prompt, model="qwen25-7b")
async def _classify_intent(self, message: str, context: ConversationContext) -> str:
"""意图分类"""
prompt = f"""你是一个意图分类器，请判断用户意图。
可选意图：faq, business, complaint, transfer, chitchat
对话历史：
{self._format_history(context.messages[-10:])}
用户消息：{message}
只返回意图名称："""
result = await self._call_llm(prompt, model="qwen25-7b", max_tokens=10)
return result.strip().lower()
async def _get_embedding(self, text: str) -> List[float]:
"""获取文本向量"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.config['embedding_api_url']}/embed",
json={"texts": [text]}
)
return response.json()["embeddings"][0]
async def _call_llm(self, prompt: str, model: str = "qwen25-7b", **kwargs) -> str:
"""调用大模型"""
response = await self.llm_client.post(
"/v1/chat/completions",
json={
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": kwargs.get("temperature", 0.3),
"max_tokens": kwargs.get("max_tokens", 1024),
"stream": False
}
)
return response.json()["choices"][0]["message"]["content"]
async def _get_session_id(self, user_id: str) -> str:
"""获取会话ID"""
session_id = self.redis_client.get(f"session:{user_id}")
if not session_id:
import uuid
session_id = str(uuid.uuid4())
# 2小时过期
self.redis_client.setex(f"session:{user_id}", 7200, session_id)
return session_id
async def _load_context(self, session_id: str) -> ConversationContext:
"""加载对话上下文"""
data = self.redis_client.get(f"context:{session_id}")
if data:
d = json.loads(data)
ctx = ConversationContext(**d)
return ctx
return ConversationContext(session_id=session_id, user_id="")
async def _save_context(self, session_id: str, context: ConversationContext):
"""保存对话上下文"""
data = json.dumps({
"session_id": context.session_id,
"user_id": context.user_id,
"messages": context.messages,
"intent": context.intent,
"slots": context.slots,
"turn_count": context.turn_count
})
self.redis_client.setex(f"context:{session_id}", 7200, data)
def _generate_suggestions(self, intent: str, response: str) -> List[str]:
"""生成建议问题"""
suggestions_map = {
"faq": ["还有其他问题吗？", "查看常见问题", "转人工服务"],
"business": ["办理进度查询", "修改信息", "取消办理"],
"complaint": ["查看处理进度", "转人工投诉", "满意度评价"],
"transfer": [],
"chitchat": ["查看服务列表", "常见问题", "转人工服务"]
}
return suggestions_map.get(intent, [])
5.2 企业知识库系统
┌─────────────────────────────────────────────────────────┐
│ 知识库系统架构 │
├─────────────────────────────────────────────────────────┤
│ │
│ 文档上传 ──► 文档解析 ──► 文本分块 ──► 向量化 │
│ (PDF/Word/ (滑动窗口/ (BGE-M3) │
│ Markdown) 语义分块) │
│ │ │
│ ▼ │
│ ┌───────────────────────────┐ │
│ │ Milvus 向量数据库 │ │
│ │ - 文档向量索引 │ │
│ │ - 元数据过滤 │ │
│ │ - 混合搜索 │ │
│ └───────────────────────────┘ │
│ │
│ 用户查询 ──► 查询理解 ──► 混合检索 ──► 重排序 │
│ │ │
│ ├─ 向量检索 (Top 20) │
│ ├─ 关键词检索 (BM25) │
│ └─ 元数据过滤 │
│ │ │
│ ▼ │
│ ┌───────────────────────────┐ │
│ │ RAG 生成 (LLM) │ │
│ │ - 上下文组装 │ │
│ │ - 引用标注 │ │
│ │ - 答案生成 │ │
│ └───────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────┘
知识库服务
# knowledge_base.py
import asyncio
from typing import Dict, List, Optional
from dataclasses import dataclass
from langchain.text_splitter import RecursiveCharacterTextSplitter
from milvus import MilvusClient
import httpx
@dataclass
class DocumentChunk:
"""文档分块"""
id: str
content: str
metadata: Dict
embedding: Optional[List[float]] = None
class KnowledgeBaseService:
"""企业知识库服务"""
def __init__(self, config: Dict):
self.config = config
self.milvus_client = MilvusClient(uri=config['milvus_uri'])
self.embedding_url = config['embedding_api_url']
self.llm_url = config['llm_api_url']
# 文本分块器
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=512,
chunk_overlap=50,
separators=["\n\n", "\n", "。", "！", "？", "；", "；", " ", ""]
)
# 创建集合
self._create_collection()
def _create_collection(self):
"""创建向量集合"""
collection_schema = {
"name": "enterprise_knowledge",
"description": "企业知识库向量集合",
"fields": [
{"name": "id", "type": "VARCHAR", "is_primary": True, "max_length": 64},
{"name": "embedding", "type": "FLOAT_VECTOR", "dim": 1024},
{"name": "content", "type": "VARCHAR", "max_length": 4096},
{"name": "title", "type": "VARCHAR", "max_length": 256},
{"name": "source", "type": "VARCHAR", "max_length": 256},
{"name": "department", "type": "VARCHAR", "max_length": 64},
{"name": "created_at", "type": "VARCHAR", "max_length": 32},
],
"index_config": {
"index_type": "HNSW",
"metric_type": "COSINE",
"params": {"M": 16, "efConstruction": 256}
}
}
try:
self.milvus_client.create_collection(collection_schema)
except Exception:
pass # 集合已存在
async def ingest_document(self, content: str, metadata: Dict) -> int:
"""入库文档"""
# 1. 文本分块
chunks = self.text_splitter.split_text(content)
# 2. 向量化
embeddings = await self._batch_embed(chunks)
# 3. 构建数据
import uuid
from datetime import datetime
data = []
for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
data.append({
"id": f"{metadata.get('doc_id', uuid.uuid4())}_{i}",
"embedding": embedding,
"content": chunk,
"title": metadata.get("title", ""),
"source": metadata.get("source", ""),
"department": metadata.get("department", ""),
"created_at": datetime.now().isoformat()
})
# 4. 插入向量数据库
self.milvus_client.insert(
collection_name="enterprise_knowledge",
data=data
)
return len(data)
async def search(
self,
query: str,
department: Optional[str] = None,
top_k: int = 10
) -> List[Dict]:
"""知识检索"""
# 1. 查询向量化
query_embedding = await self._embed(query)
# 2. 构建过滤表达式
expr = ""
if department:
expr = f'department == "{department}"'
# 3. 向量检索
results = self.milvus_client.search(
collection_name="enterprise_knowledge",
data=[query_embedding],
limit=top_k * 2, # 多取一些用于重排序
expression=expr,
output_fields=["content", "title", "source", "department"]
)
# 4. 重排序 (基于相关性分数)
ranked_results = []
for hit in results[0][:top_k]:
ranked_results.append({
"content": hit["entity"]["content"],
"title": hit["entity"]["title"],
"source": hit["entity"]["source"],
"score": hit["distance"]
})
return ranked_results
async def query_with_rag(
self,
question: str,
department: Optional[str] = None,
conversation_history: List[Dict] = None
) -> Dict:
"""RAG 问答"""
# 1. 检索相关知识
knowledge_results = await self.search(question, department, top_k=8)
# 2. 组装上下文
context_text = self._format_knowledge_context(knowledge_results)
# 3. 构建 Prompt
prompt = self._build_rag_prompt(
question, context_text, conversation_history
)
# 4. 调用 LLM 生成答案
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
f"{self.llm_url}/v1/chat/completions",
json={
"model": "qwen25-14b",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.1,
"max_tokens": 2048
}
)
answer = response.json()["choices"][0]["message"]["content"]
return {
"answer": answer,
"references": [
{"title": r["title"], "source": r["source"]}
for r in knowledge_results[:3]
],
"confidence": knowledge_results[0]["score"] if knowledge_results else 0
}
def _build_rag_prompt(
self,
question: str,
context: str,
history: List[Dict] = None
) -> str:
"""构建 RAG Prompt"""
history_text = ""
if history:
history_text = "\n".join([
f"{'用户' if m['role'] == 'user' else '助手'}: {m['content']}"
for m in history[-6:]
])
prompt = f"""你是一个专业的企业知识问答助手。请根据以下参考资料回答用户问题。
## 参考资料
{context}
## 对话历史
{history_text}
## 用户问题
{question}
## 回答要求
1. 基于参考资料回答问题，不要编造信息
2. 如果参考资料中没有相关信息，请明确说明"根据现有资料无法回答此问题"
3. 回答要简洁准确，引用具体来源
4. 使用中文回答
请回答："""
return prompt
def _format_knowledge_context(self, results: List[Dict]) -> str:
"""格式化知识上下文"""
context_parts = []
for i, result in enumerate(results, 1):
context_parts.append(
f"[{i}] {result['title']} (来源: {result['source']})\n{result['content']}"
)
return "\n\n".join(context_parts)
async def _embed(self, text: str) -> List[float]:
"""单文本向量化"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.embedding_url}/embed",
json={"texts": [text]}
)
return response.json()["embeddings"][0]
async def _batch_embed(self, texts: List[str]) -> List[List[float]]:
"""批量向量化"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.embedding_url}/embed",
json={"texts": texts}
)
return response.json()["embeddings"]
5.3 AI 机器人系统
┌─────────────────────────────────────────────────────────┐
│ AI 机器人系统架构 │
├─────────────────────────────────────────────────────────┤
│ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ 微信机器人│ │钉钉机器人 │ │Web机器人 │ ... │
│ └────┬────┘ └────┬────┘ └────┬────┘ │
│ │ │ │ │
│ └──────────────┼──────────────┘ │
│ │ │
│ ┌───────┴───────┐ │
│ │ 消息统一接入层 │ │
│ │ (协议适配) │ │
│ └───────┬───────┘ │
│ │ │
│ ┌───────┴───────┐ │
│ │ Agent 引擎 │ │
│ │ │ │
│ │ - 任务规划 │ │
│ │ - 工具调用 │ │
│ │ - 记忆管理 │ │
│ │ - 安全过滤 │ │
│ └───────┬───────┘ │
│ │ │
│ ┌────────────────┼────────────────┐ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌────────┐ ┌────────┐ ┌────────┐ │
│ │ 工具库 │ │ 知识库 │ │ 数据源 │ │
│ │ │ │ (RAG) │ │ │ │
│ │ - API │ │ │ │ - DB │ │
│ │ - 搜索 │ │ │ │ - 文件 │ │
│ │ - 计算 │ │ │ │ - 外部 │ │
│ └────────┘ └────────┘ └────────┘ │
│ │
└─────────────────────────────────────────────────────────┘
Agent 引擎
# agent_engine.py
import json
import asyncio
from typing import Dict, List, Callable, Optional
from dataclasses import dataclass, field
from enum import Enum
class ToolType(Enum):
API_CALL = "api_call"
SEARCH = "search"
CALCULATE = "calculate"
DATABASE = "database"
KNOWLEDGE = "knowledge"
@dataclass
class Tool:
"""工具定义"""
name: str
description: str
parameters: Dict
handler: Callable
tool_type: ToolType
@dataclass
class AgentMemory:
"""Agent 记忆"""
short_term: List[Dict] = field(default_factory=list)
long_term_summary: str = ""
tools_used: List[str] = field(default_factory=list)
class AgentEngine:
"""AI Agent 引擎"""
def __init__(self, config: Dict):
self.config = config
self.tools: Dict[str, Tool] = {}
self.max_iterations = 10
# 注册内置工具
self._register_builtin_tools()
def register_tool(self, tool: Tool):
"""注册工具"""
self.tools[tool.name] = tool
def _register_builtin_tools(self):
"""注册内置工具"""
# 知识库搜索工具
self.register_tool(Tool(
name="search_knowledge",
description="在企业知识库中搜索相关信息",
parameters={
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"},
"department": {"type": "string", "description": "部门过滤"}
},
"required": ["query"]
},
handler=self._search_knowledge_handler,
tool_type=ToolType.KNOWLEDGE
))
# 数据库查询工具
self.register_tool(Tool(
name="query_database",
description="查询业务数据库获取实时数据",
parameters={
"type": "object",
"properties": {
"table": {"type": "string", "description": "表名"},
"conditions": {"type": "string", "description": "查询条件"}
},
"required": ["table"]
},
handler=self._query_database_handler,
tool_type=ToolType.DATABASE
))
# 计算器工具
self.register_tool(Tool(
name="calculate",
description="执行数学计算",
parameters={
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式"}
},
"required": ["expression"]
},
handler=self._calculate_handler,
tool_type=ToolType.CALCULATE
))
async def run(self, task: str, memory: Optional[AgentMemory] = None) -> Dict:
"""执行 Agent 任务"""
if not memory:
memory = AgentMemory()
# 构建工具描述
tools_description = self._build_tools_description()
# 初始 Prompt
prompt = f"""你是一个智能Agent，可以调用工具来完成任务。
## 可用工具
{tools_description}
## 任务
{task}
## 历史操作
{self._format_memory(memory)}
请分析任务，决定是否需要调用工具。如果需要，请返回JSON格式的工具调用：
{{"action": "tool_name", "action_input": {{"param": "value"}}}}
如果不需要工具，直接返回：
{{"action": "finish", "action_input": {{"answer": "最终答案"}}}}
你的响应："""
iteration = 0
while iteration < self.max_iterations:
iteration += 1
# 调用 LLM 获取决策
response = await self._call_llm(prompt)
try:
decision = json.loads(response)
except json.JSONDecodeError:
decision = {"action": "finish", "action_input": {"answer": response}}
action = decision.get("action")
action_input = decision.get("action_input", {})
if action == "finish":
memory.short_term.append({
"type": "response",
"content": action_input.get("answer", "")
})
return {
"answer": action_input.get("answer", ""),
"iterations": iteration,
"tools_used": memory.tools_used
}
# 执行工具
if action in self.tools:
tool = self.tools[action]
try:
result = await tool.handler(**action_input)
memory.short_term.append({
"type": "tool_call",
"tool": action,
"input": action_input,
"output": str(result)
})
memory.tools_used.append(action)
# 继续迭代
prompt = f"""工具执行结果：
工具: {action}
输入: {action_input}
输出: {result}
请根据结果继续决策："""
except Exception as e:
prompt = f"""工具执行出错：
工具: {action}
错误: {str(e)}
请根据错误信息继续决策："""
else:
prompt = f"""未知工具: {action}
请选择可用的工具或返回最终答案："""
return {
"answer": "Agent 执行达到最大迭代次数",
"iterations": iteration,
"tools_used": memory.tools_used
}
def _build_tools_description(self) -> str:
"""构建工具描述"""
descriptions = []
for name, tool in self.tools.items():
descriptions.append(
f"- {name}: {tool.description}\n 参数: {json.dumps(tool.parameters, ensure_ascii=False)}"
)
return "\n".join(descriptions)
def _format_memory(self, memory: AgentMemory) -> str:
"""格式化记忆"""
if not memory.short_term:
return "无历史操作"
parts = []
for item in memory.short_term[-6:]:
if item["type"] == "tool_call":
parts.append(f"调用工具 {item['tool']}: {item['output'][:200]}")
elif item["type"] == "response":
parts.append(f"回答: {item['content'][:200]}")
return "\n".join(parts)
async def _call_llm(self, prompt: str) -> str:
"""调用 LLM"""
import httpx
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
f"{self.config['llm_api_url']}/v1/chat/completions",
json={
"model": "qwen25-14b",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.1,
"max_tokens": 512
}
)
return response.json()["choices"][0]["message"]["content"]
# 工具处理器
async def _search_knowledge_handler(self, query: str, department: str = None) -> Dict:
"""知识库搜索"""
# 调用知识库服务
return {"query": query, "results": "搜索结果..."}
async def _query_database_handler(self, table: str, conditions: str = None) -> Dict:
"""数据库查询"""
# 安全过滤后执行查询
return {"table": table, "data": "查询结果..."}
async def _calculate_handler(self, expression: str) -> Dict:
"""计算"""
try:
# 安全计算
result = eval(expression, {"__builtins__": {}}, {})
return {"result": result}
except Exception as e:
return {"error": str(e)}
六、水平扩容策略
6.1 HPA 自动扩缩容
# hpa-gpu.yaml - 基于 GPU 指标的扩缩容
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: llm-gpu-hpa
namespace: llm-system
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: qwen25-7b-deployment
minReplicas: 2
maxReplicas: 50
metrics:
# GPU 显存利用率
- type: Pods
pods:
metric:
name: gpu_memory_utilization
target:
type: AverageValue
averageValue: "80"
# GPU 计算利用率
- type: Pods
pods:
metric:
name: gpu_utilization
target:
type: AverageValue
averageValue: "75"
# 请求队列长度
- type: Pods
pods:
metric:
name: request_queue_length
target:
type: AverageValue
averageValue: "10"
# QPS
- type: Pods
pods:
metric:
name: requests_per_second
target:
type: AverageValue
averageValue: "100"
behavior:
scaleUp:
stabilizationWindowSeconds: 30
policies:
- type: Pods
value: 5
periodSeconds: 60
- type: Percent
value: 50
periodSeconds: 60
selectPolicy: Max
scaleDown:
stabilizationWindowSeconds: 300
policies:
- type: Pods
value: 1
periodSeconds: 120
selectPolicy: Min
6.2 KEDA 事件驱动扩缩容
# keda-scaler.yaml - 基于消息队列的扩缩容
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
name: llm-queue-scaler
namespace: llm-system
spec:
scaleTargetRef:
name: qwen25-7b-deployment
minReplicaCount: 2
maxReplicaCount: 30
cooldownPeriod: 300
triggers:
# Redis 队列长度
- type: redis
metadata:
address: redis-service.llm-system:6379
passwordSecretName: redis-secret
passwordKey: password
command: llen request_queue
threshold: "50"
# RabbitMQ 队列
- type: rabbitmq
metadata:
host: rabbitmq-service.llm-system:5672
queueName: llm_requests
mode: queue_length
value: "100"
6.3 扩缩容策略矩阵
| 场景 | 扩缩容触发条件 | 最小副本 | 最大副本 | 扩容速度 |
|---|---|---|---|---|
| 客服系统 | QPS > 100/Pod | 3 | 20 | 快速 (30s) |
| 知识库 | GPU显存 > 80% | 2 | 10 | 中等 (60s) |
| 机器人 | 队列长度 > 50 | 2 | 15 | 快速 (30s) |
| 离线批处理 | 任务队列非空 | 0 | 30 | 慢 (5min) |
6.4 多模型资源共享
# gpu-sharing.yaml - GPU 时间片共享
apiVersion: v1
kind: ConfigMap
metadata:
name: nvidia-device-plugin-config
namespace: kube-system
data:
config.json: |
{
"version": "v1",
"flags": {
"migStrategy": "none"
},
"sharing": {
"timeSlicing": {
"resources": [
{"name": "nvidia.com/gpu", "timeSlicing": {"reusePorts": false}}
],
"groups": [
{
"resources": ["nvidia.com/gpu"],
"replicas": 4,
"failurePolicy": "Recycle"
}
]
}
}
}
七、监控与运维
7.1 Prometheus 监控指标
# prometheus-rules.yaml
apiVersion: monitoring.coreos.com/v1
kind: PrometheusRule
metadata:
name: llm-alerts
namespace: monitoring
spec:
groups:
- name: llm.rules
rules:
# GPU 利用率告警
- alert: HighGPUUtilization
expr: gpu_utilization > 90
for: 5m
labels:
severity: warning
annotations:
summary: "GPU 利用率过高 {{ $value }}%"
# 推理延迟告警
- alert: HighInferenceLatency
expr: histogram_quantile(0.95, rate(llm_inference_latency_seconds_bucket[5m])) > 10
for: 5m
labels:
severity: warning
annotations:
summary: "P95 推理延迟超过 10s"
# 错误率告警
- alert: HighErrorRate
expr: rate(llm_request_errors_total[5m]) / rate(llm_requests_total[5m]) > 0.05
for: 5m
labels:
severity: critical
annotations:
summary: "错误率超过 5%"
# OOM 告警
- alert: GPUOOM
expr: increase(vllm_gpu_cache_usage_perc[5m]) > 0.95
for: 2m
labels:
severity: critical
annotations:
summary: "GPU 显存即将耗尽"
# Pod 重启告警
- alert: PodFrequentRestart
expr: increase(kube_pod_container_status_restarts_total[1h]) > 3
for: 10m
labels:
severity: warning
annotations:
summary: "Pod 频繁重启"
7.2 Grafana 仪表板
{
"dashboard": {
"title": "LLM 推理服务监控",
"panels": [
{
"title": "QPS",
"targets": [
{"expr": "rate(llm_requests_total[1m])", "legendFormat": "{{model}}"}
]
},
{
"title": "推理延迟 (P50/P95/P99)",
"targets": [
{"expr": "histogram_quantile(0.50, rate(llm_inference_latency_seconds_bucket[5m]))"},
{"expr": "histogram_quantile(0.95, rate(llm_inference_latency_seconds_bucket[5m]))"},
{"expr": "histogram_quantile(0.99, rate(llm_inference_latency_seconds_bucket[5m]))"}
]
},
{
"title": "GPU 利用率",
"targets": [
{"expr": "gpu_utilization", "legendFormat": "{{pod}}"}
]
},
{
"title": "GPU 显存使用",
"targets": [
{"expr": "gpu_memory_used_bytes / gpu_memory_total_bytes * 100", "legendFormat": "{{pod}}"}
]
},
{
"title": "请求队列长度",
"targets": [
{"expr": "llm_request_queue_length"}
]
},
{
"title": "错误率",
"targets": [
{"expr": "rate(llm_request_errors_total[5m]) / rate(llm_requests_total[5m]) * 100"}
]
},
{
"title": "HPA 副本数",
"targets": [
{"expr": "kube_horizontalpodautoscaler_status_current_replicas", "legendFormat": "{{hpa}}"}
]
}
]
}
}
7.3 日志收集
# fluentbit-config.yaml
apiVersion: v1
kind: ConfigMap
metadata:
name: fluentbit-config
namespace: logging
data:
fluent-bit.conf: |
[SERVICE]
Flush 1
Log_Level info
[INPUT]
Name tail
Path /var/log/containers/*llm-system*.log
Parser docker
Tag llm.*
[FILTER]
Name kubernetes
Match llm.*
K8S-Logging.Parser On
[OUTPUT]
Name elasticsearch
Match *
Host elasticsearch.logging.svc.cluster.local
Port 9200
Logstash_Format On
Logstash_Prefix llm-logs
Replace_Dots On
八、安全与合规
8.1 网络安全策略
# network-policy.yaml
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
name: llm-network-policy
namespace: llm-system
spec:
podSelector:
matchLabels:
app: qwen25-7b
policyTypes:
- Ingress
- Egress
ingress:
# 只允许来自 API 网关的流量
- from:
- namespaceSelector:
matchLabels:
name: gateway
- podSelector:
matchLabels:
app: model-router
ports:
- protocol: TCP
port: 8000
egress:
# 不允许外网访问
- to:
- namespaceSelector:
matchLabels:
name: llm-system
ports:
- protocol: TCP
port: 6379 # Redis
- protocol: TCP
port: 19530 # Milvus
8.2 密钥管理
# secrets.yaml
apiVersion: v1
kind: Secret
metadata:
name: llm-secrets
namespace: llm-system
type: Opaque
stringData:
redis-password: "${REDIS_PASSWORD}"
milvus-token: "${MILVUS_TOKEN}"
api-key: "${API_KEY}"
---
# 生产环境建议使用 Vault
# helm install vault hashicorp/vault -n vault
8.3 内容安全过滤
# content_safety.py
import re
from typing import Tuple
class ContentSafetyFilter:
"""内容安全过滤器"""
def __init__(self):
# 敏感词库 (从安全服务器加载)
self.blocked_patterns = []
self.sensitive_keywords = set()
def check_input(self, text: str) -> Tuple[bool, str]:
"""检查输入内容"""
# 1. 长度限制
if len(text) > 10000:
return False, "输入内容过长"
# 2. 敏感词检测
for keyword in self.sensitive_keywords:
if keyword in text.lower():
return False, "包含敏感内容"
# 3. 模式匹配
for pattern in self.blocked_patterns:
if re.search(pattern, text):
return False, "包含违规内容"
return True, ""
def check_output(self, text: str) -> Tuple[bool, str]:
"""检查输出内容"""
# 类似的输出过滤逻辑
return True, ""
def sanitize(self, text: str) -> str:
"""内容清洗"""
# 移除潜在的危险内容
text = re.sub(r'<script[^>]*>.*?</script>', '', text, flags=re.IGNORECASE)
text = re.sub(r'javascript:', '', text, flags=re.IGNORECASE)
return text.strip()
九、部署脚本
9.1 一键部署脚本
#!/bin/bash
# deploy.sh - 一键部署脚本
set -e
NAMESPACE="llm-system"
CHART_VERSION="1.0.0"
echo "=== 企业级大模型 K8s 部署 ==="
# 1. 创建命名空间
echo "[1/8] 创建命名空间..."
kubectl create namespace ${NAMESPACE} --dry-run=client -o yaml | kubectl apply -f -
kubectl label ns ${NAMESPACE} istio-injection=enabled --overwrite
# 2. 部署 GPU 插件
echo "[2/8] 部署 NVIDIA Device Plugin..."
kubectl apply -f https://raw.githubusercontent.com/NVIDIA/k8s-device-plugin/v0.14.1/nvidia-device-plugin.yml
# 3. 部署存储
echo "[3/8] 部署存储组件..."
helm install minio minio/minio -n ${NAMESPACE} \
--set replicas=4 \
--set persistence.size=1Ti
# 4. 部署向量数据库
echo "[4/8] 部署 Milvus..."
helm install milvus milvus/milvus -n ${NAMESPACE} \
-f milvus-values.yaml
# 5. 部署 Redis
echo "[5/8] 部署 Redis..."
helm install redis bitnami/redis -n ${NAMESPACE} \
--set architecture=replication \
--set auth.password=${REDIS_PASSWORD}
# 6. 部署模型推理服务
echo "[6/8] 部署模型推理服务..."
kubectl apply -f vllm-deployment.yaml
kubectl apply -f model-router-deployment.yaml
# 7. 部署应用服务
echo "[7/8] 部署应用服务..."
kubectl apply -f customer-service-deployment.yaml
kubectl apply -f knowledge-base-deployment.yaml
kubectl apply -f agent-engine-deployment.yaml
# 8. 部署 API 网关
echo "[8/8] 部署 API 网关..."
helm install apisix apache/apisix -n ${NAMESPACE} \
--set gateway.type=LoadBalancer
echo "=== 部署完成 ==="
echo "API 网关地址: $(kubectl get svc apisix-gateway -n ${NAMESPACE} -o jsonpath='{.status.loadBalancer.ingress[0].ip}')"
9.2 Helm Chart 结构
llm-platform/
├── Chart.yaml
├── values.yaml
└── templates/
├── namespace.yaml
├── configmap.yaml
├── secrets.yaml
├── model-inference/
│ ├── deployment.yaml
│ ├── service.yaml
│ └── hpa.yaml
├── vector-db/
│ └── milvus-values.yaml
├── cache/
│ └── redis-values.yaml
├── applications/
│ ├── customer-service.yaml
│ ├── knowledge-base.yaml
│ └── agent-engine.yaml
├── gateway/
│ └── apisix-values.yaml
└── monitoring/
├── prometheus-rules.yaml
└── grafana-dashboard.yaml
9.3 values.yaml 配置
# values.yaml
global:
namespace: llm-system
registry: registry.internal
model:
name: qwen25-7b-instruct
version: latest
replicas: 3
minReplicas: 2
maxReplicas: 20
gpu:
requests: 1
limits: 1
type: nvidia.com/gpu
resources:
cpu: "8"
memory: "32Gi"
vllm:
maxModelLen: 8192
gpuMemoryUtilization: 0.9
tensorParallelSize: 1
vectorDb:
enabled: true
type: milvus
milvus:
replicas: 3
storageSize: 500Gi
cache:
enabled: true
type: redis
redis:
replicas: 3
memory: "4Gi"
gateway:
enabled: true
type: apisix
rateLimit:
enabled: true
requestsPerMinute: 100
monitoring:
enabled: true
prometheus:
retention: 30d
grafana:
adminPassword: changeme
applications:
customerService:
enabled: true
replicas: 3
knowledgeBase:
enabled: true
replicas: 2
agentEngine:
enabled: true
replicas: 2
十、总结
10.1 方案优势
| 优势 | 说明 |
|---|---|
| 弹性伸缩 | HPA/KEDA 支持自动扩缩容，应对流量高峰 |
| 高可用 | 多副本 + 跨节点部署，单点故障不影响服务 |
| 资源高效 | GPU 时间片共享，提高硬件利用率 |
| 统一接口 | OpenAI 兼容 API，应用层无缝对接 |
| 离线安全 | 完全内网部署，数据不出域 |
| 多场景支持 | 客服、知识库、机器人统一平台 |
10.2 成本估算
| 配置 | GPU 数量 | 月成本估算 | 支持并发 |
|---|---|---|---|
| 小型 (7B) | 4x A100 | ¥50,000-80,000 | 100-200 QPS |
| 中型 (14B) | 8x A100 | ¥100,000-150,000 | 200-500 QPS |
| 大型 (72B) | 16x A100 | ¥200,000-300,000 | 500-1000 QPS |
10.3 后续优化方向
- 模型蒸馏：大模型能力蒸馏到小模型，降低推理成本
- Speculative Decoding：推测解码加速推理
- Continuous Batching：持续批处理提高吞吐
- MIG 切分：GPU 实例切分，更细粒度资源利用
- 多模态支持：扩展图像、语音等多模态能力
附录
A. 常见问题
Q: 如何选择合适的模型大小？
A: 根据场景复杂度选择。简单问答用 7B，复杂推理用 14B-32B，企业级用 72B。
Q: INT4 量化会影响效果吗？
A: 现代量化技术 (AWQ/GPTQ) 在 INT4 下性能损失 < 2%，性价比极高。
Q: 如何保证数据安全？
A: 网络隔离 + 加密传输 + 访问控制 + 审计日志 + 内容过滤。
Q: 扩容需要多长时间？
A: 冷启动约 2-5 分钟 (模型加载)，可用 GPU 预加载优化到秒级。
文档版本: v1.0
更新日期: 2026-07-31
免费领 150 小时云算力，进群参与显卡、AI PC 幸运抽奖
更多推荐
4
0- 0
已为社区贡献3条内容
所有评论(0)