部署配置：环境配置·YAML·环境变量·热重载·多环境管理
Agent部署与运维
内容：完整Docker Compose编排、Kubernetes部署、环境配置管理、配置热重载、多环境管理
部署配置体系，含Docker Compose、K8s部署、环境配置管理、配置热重载、多环境管理、Secret管理、健康检查、优雅关闭、CI/CD流水线
| 字段 | 值 |
|---|---|
| 文档版本 | v1.0 |
| 创建日期 | 2026-08-19 |
| 适用范围 | AI Agent生产部署配置 |
| 技术栈 | Docker / Kubernetes / Python / YAML |
一、Docker Compose 编排
生产环境Agent服务通常包含：Agent应用、PostgreSQL（状态）、Redis（会话缓存）、Chroma（向量库）四个容器。
services:
agent:
build: .
ports: ["8000:8000"]
environment:
- DATABASE_URL=postgresql://agent:agent@db:5432/agent
- REDIS_URL=redis://redis:6379
- CHROMA_HOST=chroma
depends_on: [db, redis, chroma]
deploy:
replicas: 3
resources: {limits: {memory: 2G, cpus: '2'}}
restart_policy: {condition: on-failure, max_attempts: 3}
db:
image: postgres:16
environment: {POSTGRES_USER: agent, POSTGRES_PASSWORD: agent, POSTGRES_DB: agent}
volumes: [pgdata:/var/lib/postgresql/data]
redis:
image: redis:7-alpine
command: redis-server --maxmemory 512mb --maxmemory-policy allkeys-lru
chroma:
image: chromadb/chroma:latest
ports: ["8001:8000"]
volumes: [chromadata:/chroma/chroma]
volumes: {pgdata: , chromadata: }
要点：依赖编排（depends_on）、资源限制（2G/2CPU）、失败重启策略、数据卷持久化（pgdata/chromadata）。
1.1 完整生产级 docker-compose.yml
# docker-compose.prod.yml - 生产环境完整编排
version: "3.9"
services:
# ============ Agent 应用层 ============
agent:
build:
context: .
dockerfile: Dockerfile.prod
args:
- APP_VERSION=${APP_VERSION:-latest}
image: ${REGISTRY:-registry.cn}/agent:${APP_VERSION:-latest}
container_name: agent-app
ports:
- "${AGENT_PORT:-8000}:8000"
environment:
- ENVIRONMENT=production
- LOG_LEVEL=INFO
- DATABASE_URL=postgresql://agent:${DB_PASSWORD}@db:5432/agent
- REDIS_URL=redis://redis:6379/0
- CHROMA_HOST=chroma
- CHROMA_PORT=8000
- OPENAI_API_KEY=${OPENAI_API_KEY}
- OPENAI_MODEL=gpt-4o
- MAX_CONCURRENT_REQUESTS=100
- REQUEST_TIMEOUT=300
- ENABLE_METRICS=true
- METRICS_PORT=9090
depends_on:
db:
condition: service_healthy
redis:
condition: service_started
chroma:
condition: service_started
deploy:
replicas: 3
resources:
limits:
memory: 2G
cpus: "2"
reservations:
memory: 512M
cpus: "0.5"
restart_policy:
condition: on-failure
max_attempts: 5
delay: 10s
window: 60s
update_config:
parallelism: 1
delay: 30s
order: start-first
failure_action: rollback
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 10s
retries: 3
start_period: 40s
logging:
driver: "json-file"
options:
max-size: "50m"
max-file: "5"
networks:
- agent-network
# ============ 数据库层 ============
db:
image: postgres:16
container_name: agent-db
environment:
POSTGRES_USER: agent
POSTGRES_PASSWORD: ${DB_PASSWORD}
POSTGRES_DB: agent
PGDATA: /var/lib/postgresql/data/pgdata
volumes:
- pgdata:/var/lib/postgresql/data
- ./db/init:/docker-entrypoint-initdb.d
ports:
- "${DB_PORT:-5432}:5432"
healthcheck:
test: ["CMD-SHELL", "pg_isready -U agent -d agent"]
interval: 10s
timeout: 5s
retries: 5
deploy:
resources:
limits:
memory: 1G
cpus: "1"
placement:
constraints:
- node.role != manager
backup:
- schedule: "0 2 * * *"
command: "pg_dump -U agent agent | gzip > /backup/db_$(date +%Y%m%d).sql.gz"
networks:
- agent-network
# ============ 缓存层 ============
redis:
image: redis:7-alpine
container_name: agent-redis
command: >
redis-server
--maxmemory 512mb
--maxmemory-policy allkeys-lru
--appendonly yes
--appendfsync everysec
--requirepass ${REDIS_PASSWORD}
--save 900 1
--save 300 10
--save 60 10000
volumes:
- redisdata:/data
ports:
- "${REDIS_PORT:-6379}:6379"
healthcheck:
test: ["CMD", "redis-cli", "-a", "${REDIS_PASSWORD}", "ping"]
interval: 10s
timeout: 5s
retries: 3
deploy:
resources:
limits:
memory: 768M
cpus: "0.5"
networks:
- agent-network
# ============ 向量数据库 ============
chroma:
image: chromadb/chroma:latest
container_name: agent-chroma
environment:
- CHROMA_SERVER_HOST=0.0.0.0
- CHROMA_SERVER_HTTP_PORT=8000
- CHROMA_PERSISTENCE_PATH=/chroma/chroma
volumes:
- chromadata:/chroma/chroma
ports:
- "${CHROMA_PORT:-8001}:8000"
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/api/v1/heartbeat"]
interval: 30s
timeout: 10s
retries: 3
deploy:
resources:
limits:
memory: 1G
cpus: "1"
networks:
- agent-network
# ============ Nginx 反向代理 ============
nginx:
image: nginx:alpine
container_name: agent-nginx
ports:
- "80:80"
- "443:443"
volumes:
- ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
- ./nginx/certs:/etc/nginx/certs:ro
- nginx_logs:/var/log/nginx
depends_on:
- agent
deploy:
resources:
limits:
memory: 256M
cpus: "0.25"
networks:
- agent-network
# ============ Prometheus 监控 ============
prometheus:
image: prom/prometheus:latest
container_name: agent-prometheus
volumes:
- ./prometheus/prometheus.yml:/etc/prometheus/prometheus.yml:ro
- prometheus_data:/prometheus
ports:
- "${PROMETHEUS_PORT:-9091}:9090"
command:
- "--config.file=/etc/prometheus/prometheus.yml"
- "--storage.tsdb.path=/prometheus"
- "--storage.tsdb.retention.time=30d"
- "--web.enable-lifecycle"
deploy:
resources:
limits:
memory: 512M
cpus: "0.5"
networks:
- agent-network
# ============ Grafana 仪表盘 ============
grafana:
image: grafana/grafana:latest
container_name: agent-grafana
environment:
- GF_SECURITY_ADMIN_USER=${GRAFANA_USER:-admin}
- GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_PASSWORD}
- GF_INSTALL_PLUGINS=grafana-piechart-panel
volumes:
- grafana_data:/var/lib/grafana
- ./grafana/provisioning:/etc/grafana/provisioning:ro
ports:
- "${GRAFANA_PORT:-3000}:3000"
depends_on:
- prometheus
deploy:
resources:
limits:
memory: 256M
cpus: "0.25"
networks:
- agent-network
# ============ Jaeger 链路追踪 ============
jaeger:
image: jaegertracing/all-in-one:1
container_name: agent-jaeger
environment:
- COLLECTOR_OTLP_ENABLED=true
- JAEGER_AGENT_HOST=jaeger
ports:
- "${JAEGER_UI_PORT:-16686}:16686"
- "${JAEGER_OTLP_PORT:-4317}:4317"
deploy:
resources:
limits:
memory: 512M
cpus: "0.5"
networks:
- agent-network
volumes:
pgdata:
driver: local
redisdata:
driver: local
chromadata:
driver: local
nginx_logs:
driver: local
prometheus_data:
driver: local
grafana_data:
driver: local
networks:
agent-network:
driver: bridge
ipam:
config:
- subnet: 172.20.0.0/16
1.2 开发环境 docker-compose.dev.yml
# docker-compose.dev.yml - 开发环境
version: "3.9"
services:
agent:
build:
context: .
dockerfile: Dockerfile.dev
container_name: agent-dev
ports:
- "8000:8000"
- "5678:5678" # debugpy
environment:
- ENVIRONMENT=development
- LOG_LEVEL=DEBUG
- DATABASE_URL=postgresql://agent:agent@db:5432/agent
- REDIS_URL=redis://redis:6379/0
- CHROMA_HOST=chroma
- OPENAI_API_KEY=${OPENAI_API_KEY}
- ENABLE_HOT_RELOAD=true
- PYTHONUNBUFFERED=1
volumes:
- ./src:/app/src
- ./tests:/app/tests
- ./config:/app/config
command: >
python -m debugpy --listen 0.0.0.0:5678
--wait-for-client -m uvicorn main:app --reload --host 0.0.0.0 --port 8000
depends_on:
- db
- redis
- chroma
deploy:
resources:
limits:
memory: 1G
cpus: "1"
db:
image: postgres:16
environment:
POSTGRES_USER: agent
POSTGRES_PASSWORD: agent
POSTGRES_DB: agent
ports:
- "5432:5432"
volumes:
- pgdata_dev:/var/lib/postgresql/data
redis:
image: redis:7-alpine
ports:
- "6379:6379"
chroma:
image: chromadb/chroma:latest
ports:
- "8001:8000"
volumes:
- chromadata_dev:/chroma/chroma
volumes:
pgdata_dev:
chromadata_dev:
1.3 使用 docker-compose override 合并配置
# docker-compose.override.yml - 自动覆盖（开发时自动加载）
version: "3.9"
services:
agent:
environment:
- LOG_LEVEL=DEBUG
- ENABLE_SWAGGER=true
volumes:
- ./src:/app/src
command: python -m uvicorn main:app --reload --host 0.0.0.0 --port 8000
db:
ports:
- "5432:5432" # 开发时暴露数据库端口
# 启动开发环境（自动合并 override）
docker compose up -d
# 启动生产环境（指定文件）
docker compose -f docker-compose.prod.yml up -d
# 查看服务状态
docker compose ps
# 查看日志
docker compose logs -f agent
# 重新构建
docker compose build --no-cache agent
# 扩缩容
docker compose up -d --scale agent=5
二、Kubernetes 部署
2.1 基础 Deployment
apiVersion: apps/v1
kind: Deployment
metadata: {name: agent-service}
spec:
replicas: 3
template:
spec:
containers:
- name: agent
image: my-agent:latest
ports: [{containerPort: 8000}]
env:
- name: OPENAI_API_KEY
valueFrom: {secretKeyRef: {name: agent-secrets, key: openai-api-key}}
resources:
requests: {memory: "512Mi", cpu: "500m"}
limits: {memory: "2Gi", cpu: "2000m"}
livenessProbe: {httpGet: {path: /health, port: 8000}}
readinessProbe: {httpGet: {path: /ready, port: 8000}}
K8s要点：Secret管理API Key、资源requests/limits、liveness探针（存活检查）、readiness探针（就绪检查）。
2.2 完整生产级 Deployment
# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: agent-service
namespace: agent-prod
labels:
app: agent
version: v1
tier: backend
spec:
replicas: 3
strategy:
type: RollingUpdate
rollingUpdate:
maxSurge: 1
maxUnavailable: 0
selector:
matchLabels:
app: agent
template:
metadata:
labels:
app: agent
version: v1
tier: backend
annotations:
prometheus.io/scrape: "true"
prometheus.io/port: "9090"
prometheus.io/path: "/metrics"
spec:
serviceAccountName: agent-sa
securityContext:
runAsNonRoot: true
runAsUser: 1000
fsGroup: 1000
containers:
- name: agent
image: registry.cn/agent:v1.0.0
imagePullPolicy: IfNotPresent
ports:
- containerPort: 8000
name: http
protocol: TCP
- containerPort: 9090
name: metrics
protocol: TCP
envFrom:
- configMapRef:
name: agent-config
env:
- name: OPENAI_API_KEY
valueFrom:
secretKeyRef:
name: agent-secrets
key: openai-api-key
- name: DATABASE_URL
valueFrom:
secretKeyRef:
name: agent-secrets
key: database-url
- name: REDIS_URL
valueFrom:
secretKeyRef:
name: agent-secrets
key: redis-url
- name: POD_NAME
valueFrom:
fieldRef:
fieldPath: metadata.name
- name: POD_NAMESPACE
valueFrom:
fieldRef:
fieldPath: metadata.namespace
resources:
requests:
memory: "512Mi"
cpu: "500m"
limits:
memory: "2Gi"
cpu: "2000m"
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 40
periodSeconds: 30
timeoutSeconds: 5
failureThreshold: 3
successThreshold: 1
readinessProbe:
httpGet:
path: /ready
port: 8000
initialDelaySeconds: 10
periodSeconds: 10
timeoutSeconds: 5
failureThreshold: 3
startupProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 0
periodSeconds: 10
failureThreshold: 30
lifecycle:
preStop:
exec:
command: ["/bin/sh", "-c", "sleep 15 && curl -X POST http://localhost:8000/shutdown"]
volumeMounts:
- name: config-volume
mountPath: /app/config
readOnly: true
- name: tmp-volume
mountPath: /tmp
- name: cache-volume
mountPath: /app/.cache
volumes:
- name: config-volume
configMap:
name: agent-config-files
- name: tmp-volume
emptyDir: {}
- name: cache-volume
emptyDir:
sizeLimit: 1Gi
terminationGracePeriodSeconds: 60
affinity:
podAntiAffinity:
preferredDuringSchedulingIgnoredDuringExecution:
- weight: 100
podAffinityTerm:
labelSelector:
matchLabels:
app: agent
topologyKey: kubernetes.io/hostname
tolerations:
- key: "dedicated"
operator: "Equal"
value: "agent"
effect: "NoSchedule"
nodeSelector:
node-role: worker
2.3 Service 定义
# k8s/service.yaml
apiVersion: v1
kind: Service
metadata:
name: agent-service
namespace: agent-prod
labels:
app: agent
spec:
type: ClusterIP
selector:
app: agent
ports:
- name: http
port: 80
targetPort: 8000
protocol: TCP
- name: metrics
port: 9090
targetPort: 9090
protocol: TCP
---
# Headless Service（用于 StatefulSet）
apiVersion: v1
kind: Service
metadata:
name: agent-headless
namespace: agent-prod
spec:
clusterIP: None
selector:
app: agent
ports:
- port: 8000
name: http
2.4 Ingress 配置
# k8s/ingress.yaml
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: agent-ingress
namespace: agent-prod
annotations:
nginx.ingress.kubernetes.io/ssl-redirect: "true"
nginx.ingress.kubernetes.io/proxy-body-size: "50m"
nginx.ingress.kubernetes.io/proxy-read-timeout: "300"
nginx.ingress.kubernetes.io/proxy-send-timeout: "300"
nginx.ingress.kubernetes.io/proxy-buffering: "off"
nginx.ingress.kubernetes.io/configuration-snippet: |
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Request-ID $request_id;
cert-manager.io/cluster-issuer: "letsencrypt-prod"
spec:
tls:
- hosts:
- agent.example.com
secretName: agent-tls
rules:
- host: agent.example.com
http:
paths:
- path: /
pathType: Prefix
backend:
service:
name: agent-service
port:
number: 80
- path: /ws
pathType: Prefix
backend:
service:
name: agent-service
port:
number: 80
2.5 HPA 自动扩缩容
# k8s/hpa.yaml
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: agent-hpa
namespace: agent-prod
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: agent-service
minReplicas: 3
maxReplicas: 20
metrics:
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
- type: Resource
resource:
name: memory
target:
type: Utilization
averageUtilization: 80
- type: Pods
pods:
metric:
name: http_requests_per_second
target:
type: AverageValue
averageValue: "100"
behavior:
scaleUp:
stabilizationWindowSeconds: 30
policies:
- type: Percent
value: 100
periodSeconds: 30
- type: Pods
value: 4
periodSeconds: 30
selectPolicy: Max
scaleDown:
stabilizationWindowSeconds: 300
policies:
- type: Percent
value: 25
periodSeconds: 60
selectPolicy: Min
2.6 ConfigMap 与 Secret
# k8s/configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:
name: agent-config
namespace: agent-prod
data:
ENVIRONMENT: "production"
LOG_LEVEL: "INFO"
OPENAI_MODEL: "gpt-4o"
MAX_CONCURRENT_REQUESTS: "100"
REQUEST_TIMEOUT: "300"
ENABLE_METRICS: "true"
METRICS_PORT: "9090"
CHROMA_HOST: "chroma-service"
CHROMA_PORT: "8000"
JAEGER_ENDPOINT: "http://jaeger-collector:14268/api/traces"
REDIS_HOST: "redis-service"
REDIS_PORT: "6379"
DB_HOST: "postgres-service"
DB_PORT: "5432"
---
# k8s/config-files.yaml
apiVersion: v1
kind: ConfigMap
metadata:
name: agent-config-files
namespace: agent-prod
data:
agent.yaml: |
model:
name: gpt-4o
temperature: 0.7
max_tokens: 4096
tools:
max_concurrent: 5
timeout: 60
memory:
type: persistent
max_entries: 100
logging:
level: INFO
format: json
prompts.yaml: |
system: |
You are a helpful AI assistant.
greeting: "Hello! How can I help you?"
---
# k8s/secret.yaml
apiVersion: v1
kind: Secret
metadata:
name: agent-secrets
namespace: agent-prod
type: Opaque
stringData:
openai-api-key: "sk-xxxxxxxxxxxxxxxxxxxx"
database-url: "postgresql://agent:password@postgres-service:5432/agent"
redis-url: "redis://:password@redis-service:6379/0"
jwt-secret: "your-jwt-secret-key"
2.7 PodDisruptionBudget
# k8s/pdb.yaml
apiVersion: policy/v1
kind: PodDisruptionBudget
metadata:
name: agent-pdb
namespace: agent-prod
spec:
minAvailable: 2
selector:
matchLabels:
app: agent
2.8 StatefulSet（有状态服务）
# k8s/statefulset.yaml - 用于有状态的Agent服务
apiVersion: apps/v1
kind: StatefulSet
metadata:
name: agent-stateful
namespace: agent-prod
spec:
serviceName: agent-headless
replicas: 3
selector:
matchLabels:
app: agent-stateful
template:
metadata:
labels:
app: agent-stateful
spec:
containers:
- name: agent
image: registry.cn/agent:v1.0.0
ports:
- containerPort: 8000
env:
- name: POD_NAME
valueFrom:
fieldRef:
fieldPath: metadata.name
- name: POD_INDEX
valueFrom:
fieldRef:
fieldPath: metadata.annotations['apps.kubernetes.io/pod-index']
volumeMounts:
- name: agent-data
mountPath: /app/data
volumeClaimTemplates:
- metadata:
name: agent-data
spec:
accessModes: ["ReadWriteOnce"]
storageClassName: "fast-ssd"
resources:
requests:
storage: 10Gi
三、环境配置管理
3.1 环境变量层级
3.2 Python 配置管理器
import os
import yaml
from pathlib import Path
from typing import Any, Optional
from dataclasses import dataclass, field
from enum import Enum
class Environment(str, Enum):
"""环境枚举"""
DEVELOPMENT = "development"
STAGING = "staging"
PRODUCTION = "production"
TEST = "test"
@dataclass
class DatabaseConfig:
"""数据库配置"""
url: str = "postgresql://localhost:5432/agent"
pool_size: int = 10
max_overflow: int = 20
pool_timeout: int = 30
pool_recycle: int = 3600
echo: bool = False
@dataclass
class RedisConfig:
"""Redis配置"""
url: str = "redis://localhost:6379/0"
max_connections: int = 20
socket_timeout: int = 5
socket_connect_timeout: int = 5
retry_on_timeout: bool = True
@dataclass
class LLMConfig:
"""LLM配置"""
provider: str = "openai"
model: str = "gpt-4o"
api_key: str = ""
temperature: float = 0.7
max_tokens: int = 4096
timeout: int = 120
max_retries: int = 3
streaming: bool = True
@dataclass
class AgentConfig:
"""Agent配置"""
max_concurrent_requests: int = 100
request_timeout: int = 300
max_steps: int = 10
max_context_tokens: int = 100000
enable_streaming: bool = True
enable_tools: bool = True
enable_memory: bool = True
enable_tracing: bool = True
@dataclass
class ObservabilityConfig:
"""可观测性配置"""
enable_metrics: bool = True
metrics_port: int = 9090
enable_tracing: bool = True
jaeger_endpoint: str = "http://localhost:14268/api/traces"
enable_logging: bool = True
log_level: str = "INFO"
log_format: str = "json"
@dataclass
class SecurityConfig:
"""安全配置"""
secret_key: str = "change-me-in-production"
jwt_secret: str = "change-me-in-production"
jwt_algorithm: str = "HS256"
jwt_expire_minutes: int = 60
cors_origins: list = field(default_factory=lambda: ["*"])
rate_limit_per_minute: int = 60
@dataclass
class AppConfig:
"""应用总配置"""
environment: Environment = Environment.DEVELOPMENT
app_name: str = "agent-service"
app_version: str = "1.0.0"
host: str = "0.0.0.0"
port: int = 8000
debug: bool = False
database: DatabaseConfig = field(default_factory=DatabaseConfig)
redis: RedisConfig = field(default_factory=RedisConfig)
llm: LLMConfig = field(default_factory=LLMConfig)
agent: AgentConfig = field(default_factory=AgentConfig)
observability: ObservabilityConfig = field(default_factory=ObservabilityConfig)
security: SecurityConfig = field(default_factory=SecurityConfig)
# 自定义配置
custom: dict = field(default_factory=dict)
class ConfigManager:
"""配置管理器"""
def __init__(self, config_dir: str = "config"):
self.config_dir = Path(config_dir)
self._config: Optional[AppConfig] = None
def load(self, environment: Optional[str] = None) -> AppConfig:
"""加载配置"""
# 1. 确定环境
env = environment or os.getenv("ENVIRONMENT", "development")
# 2. 加载默认配置
config = self._load_yaml("default.yaml")
# 3. 加载环境特定配置（覆盖默认）
env_config = self._load_yaml(f"{env}.yaml")
config = self._deep_merge(config, env_config)
# 4. 加载.env文件
config = self._load_env_file(config, env)
# 5. 加载环境变量（覆盖所有）
config = self._load_env_vars(config)
# 6. 转换为dataclass
self._config = self._dict_to_config(config)
return self._config
def _load_yaml(self, filename: str) -> dict:
"""加载YAML文件"""
filepath = self.config_dir / filename
if not filepath.exists():
return {}
with open(filepath, "r", encoding="utf-8") as f:
return yaml.safe_load(f) or {}
def _load_env_file(self, config: dict, env: str) -> dict:
"""加载.env文件"""
env_file = self.config_dir / f".env.{env}"
if not env_file.exists():
env_file = self.config_dir / ".env"
if not env_file.exists():
return config
with open(env_file, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line or line.startswith("#"):
continue
if "=" in line:
key, value = line.split("=", 1)
key = key.strip()
value = value.strip().strip('"').strip("'")
config[key] = value
return config
def _load_env_vars(self, config: dict) -> dict:
"""加载环境变量"""
for key, value in os.environ.items():
if key.startswith("AGENT_"):
config_key = key[6:].lower() # 去掉AGENT_前缀
config[config_key] = value
return config
def _deep_merge(self, base: dict, override: dict) -> dict:
"""深度合并字典"""
result = base.copy()
for key, value in override.items():
if key in result and isinstance(result[key], dict) and isinstance(value, dict):
result[key] = self._deep_merge(result[key], value)
else:
result[key] = value
return result
def _dict_to_config(self, config: dict) -> AppConfig:
"""字典转dataclass"""
app_config = AppConfig()
# 基础配置
app_config.environment = Environment(config.get("environment", "development"))
app_config.app_name = config.get("app_name", app_config.app_name)
app_config.app_version = config.get("app_version", app_config.app_version)
app_config.host = config.get("host", app_config.host)
app_config.port = int(config.get("port", app_config.port))
app_config.debug = config.get("debug", app_config.debug)
# 数据库配置
db_config = config.get("database", {})
app_config.database = DatabaseConfig(
url=db_config.get("url", os.getenv("DATABASE_URL", app_config.database.url)),
pool_size=int(db_config.get("pool_size", app_config.database.pool_size)),
max_overflow=int(db_config.get("max_overflow", app_config.database.max_overflow)),
pool_timeout=int(db_config.get("pool_timeout", app_config.database.pool_timeout)),
pool_recycle=int(db_config.get("pool_recycle", app_config.database.pool_recycle)),
echo=db_config.get("echo", app_config.database.echo),
)
# Redis配置
redis_config = config.get("redis", {})
app_config.redis = RedisConfig(
url=redis_config.get("url", os.getenv("REDIS_URL", app_config.redis.url)),
max_connections=int(redis_config.get("max_connections", app_config.redis.max_connections)),
)
# LLM配置
llm_config = config.get("llm", {})
app_config.llm = LLMConfig(
provider=llm_config.get("provider", app_config.llm.provider),
model=llm_config.get("model", os.getenv("OPENAI_MODEL", app_config.llm.model)),
api_key=llm_config.get("api_key", os.getenv("OPENAI_API_KEY", app_config.llm.api_key)),
temperature=float(llm_config.get("temperature", app_config.llm.temperature)),
max_tokens=int(llm_config.get("max_tokens", app_config.llm.max_tokens)),
timeout=int(llm_config.get("timeout", app_config.llm.timeout)),
max_retries=int(llm_config.get("max_retries", app_config.llm.max_retries)),
streaming=llm_config.get("streaming", app_config.llm.streaming),
)
# Agent配置
agent_config = config.get("agent", {})
app_config.agent = AgentConfig(
max_concurrent_requests=int(agent_config.get("max_concurrent_requests", app_config.agent.max_concurrent_requests)),
request_timeout=int(agent_config.get("request_timeout", app_config.agent.request_timeout)),
max_steps=int(agent_config.get("max_steps", app_config.agent.max_steps)),
max_context_tokens=int(agent_config.get("max_context_tokens", app_config.agent.max_context_tokens)),
enable_streaming=agent_config.get("enable_streaming", app_config.agent.enable_streaming),
enable_tools=agent_config.get("enable_tools", app_config.agent.enable_tools),
enable_memory=agent_config.get("enable_memory", app_config.agent.enable_memory),
enable_tracing=agent_config.get("enable_tracing", app_config.agent.enable_tracing),
)
# 可观测性配置
obs_config = config.get("observability", {})
app_config.observability = ObservabilityConfig(
enable_metrics=obs_config.get("enable_metrics", app_config.observability.enable_metrics),
metrics_port=int(obs_config.get("metrics_port", app_config.observability.metrics_port)),
enable_tracing=obs_config.get("enable_tracing", app_config.observability.enable_tracing),
jaeger_endpoint=obs_config.get("jaeger_endpoint", os.getenv("JAEGER_ENDPOINT", app_config.observability.jaeger_endpoint)),
enable_logging=obs_config.get("enable_logging", app_config.observability.enable_logging),
log_level=obs_config.get("log_level", os.getenv("LOG_LEVEL", app_config.observability.log_level)),
log_format=obs_config.get("log_format", app_config.observability.log_format),
)
# 安全配置
sec_config = config.get("security", {})
app_config.security = SecurityConfig(
secret_key=sec_config.get("secret_key", os.getenv("SECRET_KEY", app_config.security.secret_key)),
jwt_secret=sec_config.get("jwt_secret", os.getenv("JWT_SECRET", app_config.security.jwt_secret)),
jwt_algorithm=sec_config.get("jwt_algorithm", app_config.security.jwt_algorithm),
jwt_expire_minutes=int(sec_config.get("jwt_expire_minutes", app_config.security.jwt_expire_minutes)),
cors_origins=sec_config.get("cors_origins", app_config.security.cors_origins),
rate_limit_per_minute=int(sec_config.get("rate_limit_per_minute", app_config.security.rate_limit_per_minute)),
)
# 自定义配置
app_config.custom = config.get("custom", {})
return app_config
def get(self) -> AppConfig:
"""获取当前配置"""
if self._config is None:
return self.load()
return self._config
def reload(self) -> AppConfig:
"""重新加载配置"""
return self.load()
3.3 YAML 配置文件
# config/default.yaml - 默认配置
environment: development
app_name: agent-service
app_version: 1.0.0
host: 0.0.0.0
port: 8000
debug: false
database:
url: postgresql://localhost:5432/agent
pool_size: 10
max_overflow: 20
pool_timeout: 30
pool_recycle: 3600
echo: false
redis:
url: redis://localhost:6379/0
max_connections: 20
llm:
provider: openai
model: gpt-4o
temperature: 0.7
max_tokens: 4096
timeout: 120
max_retries: 3
streaming: true
agent:
max_concurrent_requests: 100
request_timeout: 300
max_steps: 10
max_context_tokens: 100000
enable_streaming: true
enable_tools: true
enable_memory: true
enable_tracing: true
observability:
enable_metrics: true
metrics_port: 9090
enable_tracing: true
jaeger_endpoint: http://localhost:14268/api/traces
enable_logging: true
log_level: INFO
log_format: json
security:
jwt_algorithm: HS256
jwt_expire_minutes: 60
cors_origins:
- "*"
rate_limit_per_minute: 60
custom:
features:
enable_caching: true
enable_cost_monitor: true
enable_model_router: true
# config/development.yaml - 开发环境覆盖
environment: development
debug: true
database:
echo: true
pool_size: 5
llm:
model: gpt-4o-mini # 开发用小模型
temperature: 0.9
observability:
log_level: DEBUG
log_format: text
security:
cors_origins:
- "http://localhost:3000"
- "http://localhost:8080"
# config/production.yaml - 生产环境覆盖
environment: production
debug: false
database:
pool_size: 20
max_overflow: 50
pool_recycle: 1800
llm:
model: gpt-4o
temperature: 0.7
timeout: 60
max_retries: 5
agent:
max_concurrent_requests: 200
max_steps: 8 # 生产环境限制步数
observability:
log_level: INFO
log_format: json
security:
cors_origins:
- "https://agent.example.com"
rate_limit_per_minute: 100
# config/staging.yaml - 预发布环境
environment: staging
debug: false
database:
pool_size: 15
llm:
model: gpt-4o
temperature: 0.7
observability:
log_level: INFO
四、配置热重载
4.1 文件监听热重载
import asyncio
from watchfiles import awatch
from typing import Callable, Optional
class HotReloader:
"""配置热重载器"""
def __init__(self, config_manager: ConfigManager,
config_dir: str = "config"):
self.config_manager = config_manager
self.config_dir = config_dir
self._watchers: list[asyncio.Task] = []
self._callbacks: list[Callable] = []
self._running = False
def on_reload(self, callback: Callable):
"""注册重载回调"""
self._callbacks.append(callback)
async def start(self):
"""启动文件监听"""
self._running = True
task = asyncio.create_task(self._watch_files())
self._watchers.append(task)
print(f"[HotReloader] 监听目录: {self.config_dir}")
async def stop(self):
"""停止监听"""
self._running = False
for task in self._watchers:
task.cancel()
self._watchers.clear()
async def _watch_files(self):
"""监听配置文件变化"""
async for changes in awatch(self.config_dir):
if not self._running:
break
for change_type, path in changes:
if path.endswith(('.yaml', '.yml', '.env')):
print(f"[HotReloader] 检测到配置变化: {change_type.name} {path}")
await self._reload()
break
async def _reload(self):
"""重新加载配置"""
try:
old_config = self.config_manager.get()
new_config = self.config_manager.reload()
# 对比配置差异
diff = self._diff_configs(old_config, new_config)
if diff:
print(f"[HotReloader] 配置变更: {diff}")
# 通知回调
for callback in self._callbacks:
try:
if asyncio.iscoroutinefunction(callback):
await callback(new_config, diff)
else:
callback(new_config, diff)
except Exception as e:
print(f"[HotReloader] 回调执行失败: {e}")
except Exception as e:
print(f"[HotReloader] 重载失败: {e}")
def _diff_configs(self, old, new) -> dict:
"""对比配置差异"""
diff = {}
old_dict = old.__dict__ if hasattr(old, '__dict__') else old
new_dict = new.__dict__ if hasattr(new, '__dict__') else new
for key in set(list(old_dict.keys()) + list(new_dict.keys())):
old_val = old_dict.get(key)
new_val = new_dict.get(key)
if old_val != new_val:
diff[key] = {"old": str(old_val), "new": str(new_val)}
return diff
4.2 热重载使用示例
# main.py
import asyncio
from config_manager import ConfigManager, HotReloader
async def main():
# 加载配置
config_manager = ConfigManager(config_dir="config")
config = config_manager.load(environment="production")
# 创建热重载器
reloader = HotReloader(config_manager, config_dir="config")
# 注册回调
def on_config_change(new_config, diff):
print(f"配置已更新: {diff}")
# 可以在这里更新全局配置引用
# 更新数据库连接池
# 更新LLM客户端
# 更新日志级别
reloader.on_reload(on_config_change)
# 启动热重载
await reloader.start()
# 启动应用
app = AgentApp(config)
await app.start()
if __name__ == "__main__":
asyncio.run(main())
4.3 配置热重载流程
五、多环境管理
5.1 多环境配置目录结构
config/
├── default.yaml # 默认配置（所有环境共享）
├── development.yaml # 开发环境覆盖
├── staging.yaml # 预发布环境覆盖
├── production.yaml # 生产环境覆盖
├── test.yaml # 测试环境覆盖
├── .env.development # 开发环境变量
├── .env.staging # 预发布环境变量
├── .env.production # 生产环境变量
├── .env.test # 测试环境变量
├── secrets/ # 敏感配置（不入Git）
│ ├── development/
│ │ └── api_keys.yaml
│ ├── staging/
│ │ └── api_keys.yaml
│ └── production/
│ └── api_keys.yaml
└── prompts/ # Prompt模板
├── system.yaml
├── tools.yaml
└── agents.yaml
5.2 多环境管理器
class EnvironmentManager:
"""多环境管理器"""
ENVIRONMENTS = ["development", "staging", "production", "test"]
def __init__(self, config_dir: str = "config"):
self.config_dir = Path(config_dir)
self.current_env: Optional[str] = None
self.config_manager = ConfigManager(config_dir)
def detect_environment(self) -> str:
"""自动检测当前环境"""
# 1. 环境变量
env = os.getenv("ENVIRONMENT")
if env and env in self.ENVIRONMENTS:
return env
# 2. 根据主机名判断
hostname = os.uname().nodename if hasattr(os, 'uname') else os.getenv("COMPUTERNAME", "")
if "prod" in hostname.lower():
return "production"
elif "staging" in hostname.lower():
return "staging"
elif "test" in hostname.lower():
return "test"
# 3. 默认开发环境
return "development"
def load_config(self, env: Optional[str] = None) -> AppConfig:
"""加载指定环境的配置"""
env = env or self.detect_environment()
self.current_env = env
print(f"[EnvironmentManager] 当前环境: {env}")
config = self.config_manager.load(environment=env)
# 验证配置
self._validate_config(config, env)
return config
def _validate_config(self, config: AppConfig, env: str):
"""验证配置"""
errors = []
# 生产环境必须配置API Key
if env == "production":
if not config.llm.api_key or config.llm.api_key == "":
errors.append("生产环境必须配置 OPENAI_API_KEY")
if config.security.secret_key == "change-me-in-production":
errors.append("生产环境必须修改默认 SECRET_KEY")
if config.debug:
errors.append("生产环境不应开启 debug 模式")
if "*" in config.security.cors_origins:
errors.append("生产环境不应使用通配符 CORS")
if errors:
raise ValueError(f"配置验证失败:\n" + "\n".join(f" - {e}" for e in errors))
def get_env_specific_overrides(self, env: str) -> dict:
"""获取环境特定覆盖"""
overrides = {
"development": {
"debug": True,
"llm.model": "gpt-4o-mini",
"observability.log_level": "DEBUG",
"observability.log_format": "text",
},
"staging": {
"debug": False,
"llm.model": "gpt-4o",
"observability.log_level": "INFO",
},
"production": {
"debug": False,
"llm.model": "gpt-4o",
"observability.log_level": "INFO",
"observability.log_format": "json",
"agent.max_steps": 8,
},
"test": {
"debug": True,
"llm.model": "gpt-4o-mini",
"database.url": "sqlite:///test.db",
},
}
return overrides.get(env, {})
5.3 .env 文件管理
# .env.development
ENVIRONMENT=development
LOG_LEVEL=DEBUG
OPENAI_API_KEY=sk-dev-xxxxxxxxxxxxxxxx
OPENAI_MODEL=gpt-4o-mini
DATABASE_URL=postgresql://agent:agent@localhost:5432/agent_dev
REDIS_URL=redis://localhost:6379/0
SECRET_KEY=dev-secret-key-not-for-production
# .env.production
ENVIRONMENT=production
LOG_LEVEL=INFO
OPENAI_API_KEY=sk-prod-xxxxxxxxxxxxxxxx
OPENAI_MODEL=gpt-4o
DATABASE_URL=postgresql://agent:${DB_PASSWORD}@db:5432/agent
REDIS_URL=redis://:${REDIS_PASSWORD}@redis:6379/0
SECRET_KEY=${SECRET_KEY}
JAEGER_ENDPOINT=http://jaeger:14268/api/traces
5.4 .gitignore 配置
# .gitignore
# 敏感配置文件
.env
.env.*
!.env.example
config/secrets/
# 但保留示例
!.env.development.example
六、Secret管理
6.1 Kubernetes Secret管理
# k8s/sealed-secret.yaml - 使用 Sealed Secrets 加密
apiVersion: bitnami.com/v1alpha1
kind: SealedSecret
metadata:
name: agent-secrets
namespace: agent-prod
spec:
encryptedData:
openai-api-key: AgBh...加密内容...
database-url: AgBh...加密内容...
redis-url: AgBh...加密内容...
template:
metadata:
name: agent-secrets
type: Opaque
6.2 HashiCorp Vault 集成
import hvac
class VaultSecretManager:
"""Vault密钥管理"""
def __init__(self, vault_url: str, vault_token: str):
self.client = hvac.Client(url=vault_url, token=vault_token)
def get_secret(self, path: str, key: str) -> str:
"""获取密钥"""
read_response = self.client.secrets.kv.read_secret_version(
path=path,
mount_point="secret"
)
return read_response["data"]["data"][key]
def get_all_secrets(self, path: str) -> dict:
"""获取路径下所有密钥"""
read_response = self.client.secrets.kv.read_secret_version(
path=path,
mount_point="secret"
)
return read_response["data"]["data"]
def put_secret(self, path: str, **secrets):
"""写入密钥"""
self.client.secrets.kv.v2.create_or_update_secret(
path=path,
secret=secrets,
mount_point="secret"
)
class SecretProvider:
"""密钥提供者（支持多种后端）"""
@staticmethod
def create(provider: str = "env", **kwargs):
"""创建密钥提供者"""
if provider == "env":
return EnvSecretProvider()
elif provider == "vault":
return VaultSecretManager(
vault_url=kwargs["url"],
vault_token=kwargs["token"]
)
elif provider == "k8s":
return K8sSecretProvider(
namespace=kwargs.get("namespace", "default")
)
else:
raise ValueError(f"不支持的密钥提供者: {provider}")
class EnvSecretProvider:
"""环境变量密钥提供者"""
def get_secret(self, key: str) -> str:
return os.getenv(key, "")
def get_all_secrets(self, prefix: str = "") -> dict:
return {
k: v for k, v in os.environ.items()
if k.startswith(prefix)
}
七、健康检查与就绪探针
7.1 健康检查端点
from fastapi import FastAPI
from fastapi.responses import JSONResponse
import asyncio
import time
class HealthChecker:
"""健康检查器"""
def __init__(self, config: AppConfig):
self.config = config
self.start_time = time.time()
self._db_healthy = True
self._redis_healthy = True
self._llm_healthy = True
self._last_check = 0
async def check_liveness(self) -> dict:
"""存活检查（进程是否活着）"""
return {
"status": "alive",
"uptime_seconds": time.time() - self.start_time,
"timestamp": datetime.now().isoformat(),
}
async def check_readiness(self) -> dict:
"""就绪检查（是否能处理请求）"""
checks = {}
all_healthy = True
# 检查数据库
try:
db_ok = await self._check_database()
checks["database"] = "ok" if db_ok else "error"
if not db_ok:
all_healthy = False
except Exception as e:
checks["database"] = f"error: {e}"
all_healthy = False
# 检查Redis
try:
redis_ok = await self._check_redis()
checks["redis"] = "ok" if redis_ok else "error"
if not redis_ok:
all_healthy = False
except Exception as e:
checks["redis"] = f"error: {e}"
all_healthy = False
# 检查LLM API（轻量检查）
try:
llm_ok = await self._check_llm()
checks["llm"] = "ok" if llm_ok else "degraded"
except Exception as e:
checks["llm"] = f"error: {e}"
return {
"status": "ready" if all_healthy else "not_ready",
"checks": checks,
"timestamp": datetime.now().isoformat(),
}
async def check_startup(self) -> dict:
"""启动检查"""
return await self.check_readiness()
async def _check_database(self) -> bool:
"""检查数据库连接"""
try:
# 实际实现中执行 SELECT 1
return True
except:
return False
async def _check_redis(self) -> bool:
"""检查Redis连接"""
try:
# 实际实现中执行 PING
return True
except:
return False
async def _check_llm(self) -> bool:
"""检查LLM API"""
try:
# 轻量检查，不实际调用LLM
return True
except:
return False
def setup_health_endpoints(app: FastAPI, health_checker: HealthChecker):
"""设置健康检查端点"""
@app.get("/health")
async def health():
"""存活探针"""
result = await health_checker.check_liveness()
return JSONResponse(content=result)
@app.get("/ready")
async def ready():
"""就绪探针"""
result = await health_checker.check_readiness()
status_code = 200 if result["status"] == "ready" else 503
return JSONResponse(content=result, status_code=status_code)
@app.get("/startup")
async def startup():
"""启动探针"""
result = await health_checker.check_startup()
status_code = 200 if result["status"] == "ready" else 503
return JSONResponse(content=result, status_code=status_code)
@app.get("/metrics")
async def metrics():
"""Prometheus指标"""
return JSONResponse(content={
"uptime": time.time() - health_checker.start_time,
"requests_total": getattr(app.state, "requests_total", 0),
"errors_total": getattr(app.state, "errors_total", 0),
})
7.2 探针配置说明
八、优雅关闭
8.1 优雅关闭实现
import signal
import asyncio
from typing import Callable
class GracefulShutdown:
"""优雅关闭管理器"""
def __init__(self):
self._shutdown_handlers: list[Callable] = []
self._shutting_down = False
self._shutdown_event = asyncio.Event()
def register_handler(self, handler: Callable):
"""注册关闭处理器"""
self._shutdown_handlers.append(handler)
def setup_signal_handlers(self):
"""设置信号处理器"""
loop = asyncio.get_event_loop()
for sig in (signal.SIGTERM, signal.SIGINT):
loop.add_signal_handler(sig, self._handle_signal)
print("[GracefulShutdown] 信号处理器已注册")
def _handle_signal(self):
"""处理信号"""
if self._shutting_down:
print("[GracefulShutdown] 已在关闭中，请等待...")
return
self._shutting_down = True
print("[GracefulShutdown] 收到关闭信号，开始优雅关闭...")
self._shutdown_event.set()
async def wait_for_shutdown(self):
"""等待关闭信号"""
await self._shutdown_event.wait()
async def shutdown(self, timeout: int = 30):
"""执行优雅关闭"""
print(f"[GracefulShutdown] 开始执行关闭处理器（超时{timeout}秒）")
# 1. 停止接收新请求
print("[GracefulShutdown] 停止接收新请求")
# 2. 等待进行中的请求完成
print("[GracefulShutdown] 等待进行中的请求完成...")
await asyncio.sleep(5) # 给请求5秒完成时间
# 3. 执行关闭处理器
for handler in self._shutdown_handlers:
try:
if asyncio.iscoroutinefunction(handler):
await asyncio.wait_for(handler(), timeout=10)
else:
handler()
print(f"[GracefulShutdown] 处理器 {handler.__name__} 完成")
except asyncio.TimeoutError:
print(f"[GracefulShutdown] 处理器 {handler.__name__} 超时")
except Exception as e:
print(f"[GracefulShutdown] 处理器 {handler.__name__} 失败: {e}")
# 4. 关闭数据库连接
print("[GracefulShutdown] 关闭数据库连接")
# 5. 关闭Redis连接
print("[GracefulShutdown] 关闭Redis连接")
# 6. 清理资源
print("[GracefulShutdown] 清理资源")
print("[GracefulShutdown] 优雅关闭完成")
# 使用示例
async def main():
shutdown = GracefulShutdown()
shutdown.setup_signal_handlers()
# 注册关闭处理器
shutdown.register_handler(close_database)
shutdown.register_handler(close_redis)
shutdown.register_handler(flush_logs)
# 启动应用
app = AgentApp()
app_task = asyncio.create_task(app.start())
# 等待关闭信号
await shutdown.wait_for_shutdown()
# 执行关闭
await shutdown.shutdown(timeout=30)
# 取消应用任务
app_task.cancel()
try:
await app_task
except asyncio.CancelledError:
pass
print("[Main] 应用已关闭")
async def close_database():
"""关闭数据库"""
print(" → 关闭数据库连接池")
await asyncio.sleep(0.5)
async def close_redis():
"""关闭Redis"""
print(" → 关闭Redis连接")
await asyncio.sleep(0.3)
def flush_logs():
"""刷新日志"""
print(" → 刷新日志缓冲区")
8.2 优雅关闭流程
九、CI/CD 部署流水线
9.1 GitHub Actions 配置
# .github/workflows/deploy.yml
name: Deploy Agent Service
on:
push:
branches: [main, develop]
tags: ['v*']
pull_request:
branches: [main]
env:
REGISTRY: ghcr.io
IMAGE_NAME: ${{ github.repository }}
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.12'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install pytest pytest-asyncio pytest-cov
- name: Run tests
run: |
pytest --cov=src --cov-report=xml
- name: Upload coverage
uses: codecov/codecov-action@v4
build:
needs: test
runs-on: ubuntu-latest
if: github.event_name != 'pull_request'
permissions:
contents: read
packages: write
steps:
- uses: actions/checkout@v4
- name: Set environment
id: env
run: |
if [[ ${{ github.ref }} == refs/tags/v* ]]; then
echo "environment=production" >> $GITHUB_OUTPUT
echo "image_tag=${GITHUB_REF#refs/tags/}" >> $GITHUB_OUTPUT
elif [[ ${{ github.ref }} == refs/heads/main ]]; then
echo "environment=staging" >> $GITHUB_OUTPUT
echo "image_tag=staging-${{ github.sha }}" >> $GITHUB_OUTPUT
else
echo "environment=development" >> $GITHUB_OUTPUT
echo "image_tag=dev-${{ github.sha }}" >> $GITHUB_OUTPUT
fi
- name: Build Docker image
run: |
docker build -t ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ steps.env.outputs.image_tag }} .
- name: Log in to Registry
uses: docker/login-action@v3
with:
registry: ${{ env.REGISTRY }}
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Push image
run: |
docker push ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ steps.env.outputs.image_tag }}
outputs:
environment: ${{ steps.env.outputs.environment }}
image_tag: ${{ steps.env.outputs.image_tag }}
deploy:
needs: build
runs-on: ubuntu-latest
if: github.event_name != 'pull_request'
environment: ${{ needs.build.outputs.environment }}
steps:
- uses: actions/checkout@v4
- name: Set K8s context
uses: azure/k8s-set-context@v4
with:
kubeconfig: ${{ secrets.KUBE_CONFIG }}
- name: Deploy to K8s
run: |
env=${{ needs.build.outputs.environment }}
tag=${{ needs.build.outputs.image_tag }}
# 更新镜像标签
kubectl set image deployment/agent-service \
agent=${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${tag} \
-n agent-${env}
# 等待滚动更新完成
kubectl rollout status deployment/agent-service \
-n agent-${env} --timeout=300s
# 验证部署
kubectl get pods -n agent-${env} -l app=agent
- name: Run smoke tests
run: |
sleep 30 # 等待Pod就绪
SERVICE_URL=$(kubectl get svc agent-service -n agent-${{ needs.build.outputs.environment }} -o jsonpath='{.status.loadBalancer.ingress[0].ip}')
# 健康检查
curl -f http://${SERVICE_URL}/health
# 就绪检查
curl -f http://${SERVICE_URL}/ready
9.2 部署流程图
十、配置验证与安全
10.1 配置验证
class ConfigValidator:
"""配置验证器"""
@staticmethod
def validate(config: AppConfig) -> list[str]:
"""验证配置，返回错误列表"""
errors = []
# 基础验证
if config.port < 1 or config.port > 65535:
errors.append("端口必须在1-65535范围内")
# 数据库验证
if not config.database.url:
errors.append("数据库URL不能为空")
if config.database.pool_size < 1:
errors.append("数据库连接池大小必须大于0")
# LLM验证
if not config.llm.api_key:
errors.append("LLM API Key不能为空")
if config.llm.temperature < 0 or config.llm.temperature > 2:
errors.append("temperature必须在0-2范围内")
if config.llm.max_tokens < 1 or config.llm.max_tokens > 128000:
errors.append("max_tokens必须在1-128000范围内")
# Agent验证
if config.agent.max_steps < 1:
errors.append("max_steps必须大于0")
if config.agent.max_concurrent_requests < 1:
errors.append("max_concurrent_requests必须大于0")
# 安全验证
if config.environment == Environment.PRODUCTION:
if config.debug:
errors.append("[生产环境] 不应开启debug模式")
if config.security.secret_key == "change-me-in-production":
errors.append("[生产环境] 必须修改默认SECRET_KEY")
if config.security.jwt_secret == "change-me-in-production":
errors.append("[生产环境] 必须修改默认JWT_SECRET")
if "*" in config.security.cors_origins:
errors.append("[生产环境] 不应使用通配符CORS")
return errors
10.2 配置安全检查清单
| 检查项 | 开发环境 | 生产环境 | 说明 |
|---|---|---|---|
| API Key | 可用测试Key | 必须用真实Key | 不能硬编码 |
| SECRET_KEY | 默认值 | 必须修改 | 至少32字符 |
| CORS | 允许通配符 | 必须指定域名 | 防止CSRF |
| Debug | 开启 | 关闭 | 避免信息泄露 |
| 日志级别 | DEBUG | INFO | 避免敏感信息 |
| HTTPS | 可选 | 必须 | 防止中间人攻击 |
| Rate Limit | 宽松 | 严格 | 防止滥用 |
十一、总结
11.1 部署配置最佳实践
- 配置分离：代码与配置完全分离，不同环境不同配置文件
- 密钥安全：API Key等敏感信息不硬编码，用环境变量或Secret管理
- 分层覆盖：默认配置 → 环境配置 → 环境变量，逐层覆盖
- 热重载：支持配置变更不重启应用
- 健康检查：完善的liveness/readiness探针
- 优雅关闭：收到信号后先完成进行中的请求再退出
- CI/CD：自动化构建、测试、部署流程
- 配置验证：启动时验证配置完整性和安全性
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/lza_csdn2019/article/details/169158841
