Ollama远程调用避坑指南：Python API+安全加固全流程
·
Ollama远程调用避坑指南：Python API+安全加固全流程
如果你正在尝试将Ollama的大模型能力集成到现有系统中，那么远程调用这个环节绝对是个技术活。我见过不少开发者在本地测试时一切顺利，一旦部署到生产环境，各种问题就接踵而至：连接超时、响应缓慢、安全漏洞、内存泄漏……这些问题往往不是Ollama本身的问题，而是远程调用配置不当导致的。
作为全栈工程师，我们需要的不只是让API能跑起来，而是要构建一个稳定、高效、安全的远程大模型服务。这篇文章就是我在多个项目中踩坑后总结的实战经验，从Python API的流式处理到Nginx反向代理的精细配置，再到生产环境的常见问题排查，我会带你走完整个流程。
1. Python API调用：从基础到高级
1.1 基础调用与参数调优
很多人以为调用Ollama API就是简单的POST请求，但实际上参数配置直接影响响应质量和速度。先看一个最基本的非流式调用：
import requests
import json
def generate_text(prompt, model="llama3:8b", temperature=0.7, top_p=0.9):
"""
基础文本生成函数
temperature: 控制随机性，0-1之间，越高越有创意
top_p: 核采样参数，控制词汇选择范围
"""
url = "http://your-server:11434/api/generate"
payload = {
"model": model,
"prompt": prompt,
"stream": False,
"options": {
"temperature": temperature,
"top_p": top_p,
"num_predict": 512, # 最大生成token数
"repeat_penalty": 1.1, # 重复惩罚因子
"stop": ["\n\n", "Human:", "Assistant:"] # 停止词
}
}
try:
response = requests.post(url, json=payload, timeout=60)
response.raise_for_status()
result = response.json()
# 提取有用信息
return {
"response": result.get("response", ""),
"total_duration": result.get("total_duration", 0),
"eval_count": result.get("eval_count", 0),
"load_duration": result.get("load_duration", 0)
}
except requests.exceptions.Timeout:
return {"error": "请求超时，请检查网络或服务器状态"}
except requests.exceptions.RequestException as e:
return {"error": f"请求失败: {str(e)}"}
注意：
num_predict参数需要根据你的应用场景合理设置。对于对话场景，512-1024通常足够；对于长文本生成，可能需要2048或更高。但要注意，值越大，响应时间越长，内存消耗也越大。
1.2 流式响应的正确姿势
流式响应对于需要实时显示生成结果的场景至关重要，比如聊天应用或代码生成工具。但很多人在处理流式响应时会遇到数据不完整或连接中断的问题。
import requests
import json
def stream_generation(prompt, model="llama3:8b", callback=None):
"""
流式生成函数
callback: 每收到一个chunk时调用的函数，用于实时显示
"""
url = "http://your-server:11434/api/generate"
payload = {
"model": model,
"prompt": prompt,
"stream": True,
"options": {
"temperature": 0.7,
"num_predict": 1024
}
}
try:
# 关键：设置stream=True和合适的超时时间
response = requests.post(
url,
json=payload,
stream=True,
timeout=300 # 流式响应需要更长超时
)
response.raise_for_status()
full_response = ""
for line in response.iter_lines():
if line:
try:
# 解码并解析JSON
decoded_line = line.decode('utf-8')
data = json.loads(decoded_line)
# 检查是否完成
if data.get("done", False):
break
# 提取响应内容
chunk = data.get("response", "")
if chunk:
full_response += chunk
# 如果有回调函数，实时处理
if callback and callable(callback):
callback(chunk)
except json.JSONDecodeError:
print(f"JSON解析失败: {line}")
continue
except UnicodeDecodeError:
print(f"编码错误: {line}")
continue
return full_response
except requests.exceptions.ChunkedEncodingError:
return {"error": "流式响应中断，可能是网络问题或服务器超时"}
except Exception as e:
return {"error": f"流式请求失败: {str(e)}"}
# 使用示例
def print_chunk(chunk):
"""简单的回调函数，实时打印chunk"""
print(chunk, end='', flush=True)
# 调用
result = stream_generation(
"用Python实现快速排序算法，并添加详细注释",
callback=print_chunk
)
流式处理中有几个关键点需要注意：
- 超时设置：流式响应通常需要更长的超时时间，我一般设置为300秒
- 错误处理：必须处理JSON解析错误和编码错误
- 内存管理：对于超长响应，要考虑分块处理，避免内存溢出
- 连接保持：确保网络稳定，避免中途断开
1.3 批量处理与并发控制
在生产环境中，我们经常需要处理批量请求。直接并发调用可能会导致服务器过载，需要合理的并发控制。
import asyncio
import aiohttp
from typing import List, Dict
import time
class OllamaBatchProcessor:
"""批量处理Ollama请求的类"""
def __init__(self, base_url: str, max_concurrent: int = 3):
self.base_url = base_url
self.max_concurrent = max_concurrent
self.semaphore = asyncio.Semaphore(max_concurrent)
async def single_request(self, session: aiohttp.ClientSession,
prompt: str, model: str = "llama3:8b") -> Dict:
"""单个请求的异步实现"""
url = f"{self.base_url}/api/generate"
payload = {
"model": model,
"prompt": prompt,
"stream": False
}
async with self.semaphore: # 控制并发数
try:
async with session.post(url, json=payload, timeout=60) as response:
if response.status == 200:
result = await response.json()
return {
"success": True,
"response": result.get("response", ""),
"prompt": prompt
}
else:
return {
"success": False,
"error": f"HTTP {response.status}",
"prompt": prompt
}
except asyncio.TimeoutError:
return {
"success": False,
"error": "请求超时",
"prompt": prompt
}
except Exception as e:
return {
"success": False,
"error": str(e),
"prompt": prompt
}
async def process_batch(self, prompts: List[str], model: str = "llama3:8b") -> List[Dict]:
"""批量处理多个提示"""
connector = aiohttp.TCPConnector(limit=self.max_concurrent)
timeout = aiohttp.ClientTimeout(total=300)
async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
tasks = [self.single_request(session, prompt, model) for prompt in prompts]
results = await asyncio.gather(*tasks, return_exceptions=True)
# 处理异常结果
processed_results = []
for i, result in enumerate(results):
if isinstance(result, Exception):
processed_results.append({
"success": False,
"error": str(result),
"prompt": prompts[i]
})
else:
processed_results.append(result)
return processed_results
# 使用示例
async def main():
processor = OllamaBatchProcessor("http://your-server:11434", max_concurrent=3)
prompts = [
"解释量子计算的基本原理",
"写一个Python函数计算斐波那契数列",
"总结机器学习的主要类型",
"描述HTTP和HTTPS的区别",
"解释区块链技术的工作原理"
]
start_time = time.time()
results = await processor.process_batch(prompts)
elapsed = time.time() - start_time
print(f"处理 {len(prompts)} 个请求耗时: {elapsed:.2f}秒")
# 统计结果
success_count = sum(1 for r in results if r["success"])
print(f"成功: {success_count}, 失败: {len(prompts) - success_count}")
# 运行
if __name__ == "__main__":
asyncio.run(main())
批量处理时需要注意的几个参数：
| 参数 | 推荐值 | 说明 |
|---|---|---|
| max_concurrent | 2-5 | 根据服务器配置调整，GPU内存越大可并发数越高 |
| timeout | 60-300秒 | 根据请求复杂度调整 |
| TCP连接数 | 与并发数一致 | 避免连接数过多导致端口耗尽 |
2. 安全加固：不只是防火墙
2.1 Nginx反向代理的精细配置
很多教程只教了基本的Nginx配置，但在生产环境中，我们需要更细致的控制。下面是一个企业级的配置示例：
# /etc/nginx/nginx.conf 主配置
user nginx;
worker_processes auto;
error_log /var/log/nginx/error.log warn;
pid /var/run/nginx.pid;
events {
worker_connections 1024;
use epoll;
multi_accept on;
}
http {
include /etc/nginx/mime.types;
default_type application/octet-stream;
# 日志格式
log_format main '$remote_addr - $remote_user [$time_local] "$request" '
'$status $body_bytes_sent "$http_referer" '
'"$http_user_agent" "$http_x_forwarded_for" '
'rt=$request_time uct="$upstream_connect_time" '
'uht="$upstream_header_time" urt="$upstream_response_time"';
access_log /var/log/nginx/access.log main;
# 基础优化
sendfile on;
tcp_nopush on;
tcp_nodelay on;
keepalive_timeout 65;
types_hash_max_size 2048;
client_max_body_size 100M;
# Gzip压缩
gzip on;
gzip_vary on;
gzip_min_length 1024;
gzip_types text/plain text/css text/xml text/javascript
application/json application/javascript application/xml+rss;
# Ollama服务配置
upstream ollama_backend {
server 127.0.0.1:11434;
keepalive 32; # 保持连接池
}
include /etc/nginx/conf.d/*.conf;
}
# /etc/nginx/conf.d/ollama.conf 服务配置
server {
listen 443 ssl http2;
server_name ai.yourdomain.com;
# SSL配置
ssl_certificate /etc/ssl/certs/yourdomain.crt;
ssl_certificate_key /etc/ssl/private/yourdomain.key;
ssl_protocols TLSv1.2 TLSv1.3;
ssl_ciphers ECDHE-RSA-AES256-GCM-SHA512:DHE-RSA-AES256-GCM-SHA512;
ssl_prefer_server_ciphers off;
ssl_session_cache shared:SSL:10m;
ssl_session_timeout 10m;
# 安全头部
add_header X-Frame-Options "SAMEORIGIN" always;
add_header X-Content-Type-Options "nosniff" always;
add_header X-XSS-Protection "1; mode=block" always;
add_header Referrer-Policy "strict-origin-when-cross-origin" always;
add_header Content-Security-Policy "default-src 'self' https: data: 'unsafe-inline' 'unsafe-eval';" always;
# 限流配置
limit_req_zone $binary_remote_addr zone=ollama_limit:10m rate=10r/s;
limit_req zone=ollama_limit burst=20 nodelay;
# 连接限制
limit_conn_zone $binary_remote_addr zone=addr:10m;
limit_conn addr 20;
location / {
# IP白名单控制
allow 192.168.1.0/24;
allow 10.0.0.0/8;
deny all;
# 反向代理配置
proxy_pass http://ollama_backend;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
# 超时设置
proxy_connect_timeout 300s;
proxy_send_timeout 300s;
proxy_read_timeout 300s;
# 缓冲区优化
proxy_buffering on;
proxy_buffer_size 4k;
proxy_buffers 8 4k;
proxy_busy_buffers_size 8k;
# 禁用缓存
proxy_no_cache 1;
proxy_cache_bypass 1;
}
# 健康检查端点
location /health {
access_log off;
allow 127.0.0.1;
deny all;
proxy_pass http://ollama_backend;
proxy_set_header Host $host;
# 只返回状态码
return 200;
}
# 访问日志单独记录
access_log /var/log/nginx/ollama_access.log main;
error_log /var/log/nginx/ollama_error.log warn;
}
这个配置包含了几个关键的安全和性能优化：
- SSL/TLS强化：使用现代加密协议和算法
- 安全头部：防止XSS、点击劫持等攻击
- 限流限速：防止API被滥用
- IP白名单：只允许内网或特定IP访问
- 连接池：提高连接复用率
- 缓冲区优化：平衡内存使用和性能
2.2 API密钥认证中间件
对于需要对外提供服务的场景，仅靠IP白名单不够，还需要API密钥认证。下面是一个基于FastAPI的中间件示例：
from fastapi import FastAPI, Request, HTTPException, Depends
from fastapi.security import APIKeyHeader
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
import hashlib
import time
import json
from typing import Dict, List, Optional
from pydantic import BaseModel
import redis
import pickle
app = FastAPI(title="Ollama API Gateway", version="1.0.0")
# Redis连接池（用于存储API密钥和限流信息）
redis_pool = redis.ConnectionPool(
host='localhost',
port=6379,
db=0,
decode_responses=True,
max_connections=20
)
# API密钥验证模型
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
class RateLimiter:
"""基于Redis的限流器"""
def __init__(self, redis_conn, limit: int = 100, window: int = 3600):
self.redis = redis_conn
self.limit = limit
self.window = window
def is_allowed(self, key: str) -> bool:
"""检查是否允许请求"""
current = int(time.time())
window_start = current - self.window
# 使用Redis管道提高性能
pipe = self.redis.pipeline()
pipe.zremrangebyscore(key, 0, window_start)
pipe.zcard(key)
pipe.zadd(key, {str(current): current})
pipe.expire(key, self.window)
results = pipe.execute()
return results[1] <= self.limit
class APIKeyManager:
"""API密钥管理器"""
def __init__(self, redis_conn):
self.redis = redis_conn
self.key_prefix = "apikey:"
def validate_key(self, api_key: str) -> Optional[Dict]:
"""验证API密钥并返回用户信息"""
if not api_key:
return None
key_data = self.redis.get(f"{self.key_prefix}{api_key}")
if not key_data:
return None
try:
user_info = json.loads(key_data)
# 检查密钥是否过期
if user_info.get("expires_at") and user_info["expires_at"] < time.time():
self.redis.delete(f"{self.key_prefix}{api_key}")
return None
# 更新最后使用时间
user_info["last_used"] = time.time()
self.redis.setex(
f"{self.key_prefix}{api_key}",
86400 * 30, # 30天过期
json.dumps(user_info)
)
return user_info
except (json.JSONDecodeError, KeyError):
return None
def create_key(self, user_id: str, permissions: List[str],
expires_in_days: int = 30) -> str:
"""创建新的API密钥"""
# 生成随机密钥
raw_key = f"{user_id}:{time.time()}:{hashlib.sha256(os.urandom(32)).hexdigest()}"
api_key = hashlib.sha256(raw_key.encode()).hexdigest()
user_info = {
"user_id": user_id,
"permissions": permissions,
"created_at": time.time(),
"expires_at": time.time() + (expires_in_days * 86400),
"last_used": None,
"request_count": 0
}
# 存储到Redis
self.redis.setex(
f"{self.key_prefix}{api_key}",
expires_in_days * 86400,
json.dumps(user_info)
)
return api_key
# 初始化管理器
redis_conn = redis.Redis(connection_pool=redis_pool)
key_manager = APIKeyManager(redis_conn)
rate_limiter = RateLimiter(redis_conn, limit=1000, window=3600) # 每小时1000次
@app.middleware("http")
async def api_key_middleware(request: Request, call_next):
"""API密钥验证中间件"""
# 跳过健康检查
if request.url.path == "/health":
return await call_next(request)
# 获取API密钥
api_key = request.headers.get("X-API-Key")
# 验证密钥
user_info = key_manager.validate_key(api_key)
if not user_info:
raise HTTPException(
status_code=401,
detail="无效的API密钥或密钥已过期",
headers={"WWW-Authenticate": "API-Key"}
)
# 限流检查
user_id = user_info["user_id"]
if not rate_limiter.is_allowed(f"ratelimit:{user_id}"):
raise HTTPException(
status_code=429,
detail="请求过于频繁，请稍后再试",
headers={"Retry-After": "3600"}
)
# 权限检查（示例）
required_permission = get_required_permission(request)
if required_permission and required_permission not in user_info["permissions"]:
raise HTTPException(
status_code=403,
detail="权限不足"
)
# 将用户信息添加到请求状态
request.state.user = user_info
# 继续处理请求
response = await call_next(request)
# 添加请求ID到响应头
response.headers["X-Request-ID"] = request.state.get("request_id", "")
return response
# 添加其他中间件
app.add_middleware(
CORSMiddleware,
allow_origins=["https://yourdomain.com"], # 生产环境指定具体域名
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE"],
allow_headers=["X-API-Key", "Content-Type", "Authorization"],
max_age=3600
)
app.add_middleware(
TrustedHostMiddleware,
allowed_hosts=["ai.yourdomain.com", "localhost"]
)
# 请求/响应模型
class GenerationRequest(BaseModel):
prompt: str
model: str = "llama3:8b"
stream: bool = False
temperature: float = 0.7
max_tokens: int = 512
class GenerationResponse(BaseModel):
response: str
model: str
tokens_used: int
processing_time: float
request_id: str
@app.post("/api/generate", response_model=GenerationResponse)
async def generate_text(request: GenerationRequest, req: Request):
"""文本生成端点"""
# 这里添加实际的Ollama调用逻辑
# 可以使用前面介绍的批量处理器
return GenerationResponse(
response="生成的文本内容",
model=request.model,
tokens_used=100,
processing_time=1.5,
request_id=req.state.get("request_id", "")
)
@app.get("/health")
async def health_check():
"""健康检查端点"""
return {"status": "healthy", "timestamp": time.time()}
def get_required_permission(request: Request) -> Optional[str]:
"""根据请求路径确定所需权限"""
path = request.url.path
method = request.method
# 权限映射表
permission_map = {
("/api/generate", "POST"): "generate",
("/api/models", "GET"): "read_models",
("/api/models", "POST"): "manage_models",
}
return permission_map.get((path, method))
这个API网关提供了：
- API密钥管理：支持密钥创建、验证、过期控制
- 细粒度权限控制：不同接口可以设置不同权限
- 限流保护：防止API被滥用
- 请求追踪：每个请求都有唯一ID
- CORS控制：严格限制跨域请求
3. 生产环境部署与优化
3.1 系统级优化配置
在CentOS 7上部署Ollama时，系统配置对性能影响很大。以下是我在实际项目中总结的优化配置：
#!/bin/bash
# ollama-system-optimize.sh
# Ollama生产环境系统优化脚本
set -e
echo "开始优化系统配置..."
# 1. 内核参数优化
cat > /etc/sysctl.d/99-ollama.conf << EOF
# 网络相关
net.core.somaxconn = 65535
net.core.netdev_max_backlog = 65535
net.ipv4.tcp_max_syn_backlog = 65535
net.ipv4.tcp_syncookies = 1
net.ipv4.tcp_tw_reuse = 1
net.ipv4.tcp_tw_recycle = 0
net.ipv4.tcp_fin_timeout = 30
net.ipv4.tcp_keepalive_time = 1200
net.ipv4.tcp_keepalive_probes = 5
net.ipv4.tcp_keepalive_intvl = 15
# 内存相关
vm.swappiness = 10
vm.vfs_cache_pressure = 50
vm.dirty_ratio = 10
vm.dirty_background_ratio = 5
vm.overcommit_memory = 1
vm.overcommit_ratio = 80
# 文件系统
fs.file-max = 2097152
fs.nr_open = 2097152
fs.inotify.max_user_watches = 524288
EOF
sysctl -p /etc/sysctl.d/99-ollama.conf
# 2. 限制优化
cat > /etc/security/limits.d/99-ollama.conf << EOF
* soft nofile 65535
* hard nofile 65535
* soft nproc 65535
* hard nproc 65535
ollama soft nofile 1048576
ollama hard nofile 1048576
ollama soft nproc unlimited
ollama hard nproc unlimited
ollama soft memlock unlimited
ollama hard memlock unlimited
EOF
# 3. 创建专用的ollama用户和组
if ! id -u ollama >/dev/null 2>&1; then
groupadd -r ollama
useradd -r -g ollama -s /bin/false -d /opt/ollama ollama
echo "创建ollama用户完成"
fi
# 4. 创建数据目录并设置权限
mkdir -p /opt/ollama/{models,logs,tmp}
chown -R ollama:ollama /opt/ollama
chmod 755 /opt/ollama
# 5. 配置systemd服务（优化版）
cat > /etc/systemd/system/ollama.service << EOF
[Unit]
Description=Ollama Service
After=network-online.target
Wants=network-online.target
RequiresMountsFor=/opt/ollama
[Service]
Type=exec
User=ollama
Group=ollama
Environment="OLLAMA_HOST=0.0.0.0:11434"
Environment="OLLAMA_MODELS=/opt/ollama/models"
Environment="OLLAMA_KEEP_ALIVE=24h"
Environment="OLLAMA_MAX_LOADED_MODELS=3"
Environment="HOME=/opt/ollama"
Environment="TMPDIR=/opt/ollama/tmp"
# 资源限制
LimitNOFILE=1048576
LimitNPROC=unlimited
LimitMEMLOCK=infinity
# 安全配置
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ProtectHome=true
ReadWritePaths=/opt/ollama
ReadOnlyPaths=/usr/bin/ollama
# 重启策略
Restart=on-failure
RestartSec=5
StartLimitInterval=100
StartLimitBurst=10
# 日志配置
StandardOutput=journal
StandardError=journal
SyslogIdentifier=ollama
# 执行命令
ExecStart=/usr/bin/ollama serve
ExecReload=/bin/kill -HUP \$MAINPID
KillSignal=SIGTERM
TimeoutStopSec=30
[Install]
WantedBy=multi-user.target
EOF
# 6. 配置日志轮转
cat > /etc/logrotate.d/ollama << EOF
/opt/ollama/logs/*.log {
daily
rotate 30
compress
delaycompress
missingok
notifempty
create 0640 ollama ollama
sharedscripts
postrotate
systemctl kill -s HUP ollama.service >/dev/null 2>&1 || true
endscript
}
EOF
# 7. 配置防火墙（如果需要）
if command -v firewall-cmd &> /dev/null; then
firewall-cmd --permanent --add-port=11434/tcp
firewall-cmd --reload
echo "防火墙规则已添加"
fi
# 8. 创建监控脚本
cat > /opt/ollama/scripts/monitor.sh << 'EOF'
#!/bin/bash
# Ollama服务监控脚本
LOG_FILE="/opt/ollama/logs/monitor.log"
MAX_RETRIES=3
RETRY_DELAY=5
log_message() {
echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1" >> "$LOG_FILE"
}
check_service() {
if systemctl is-active --quiet ollama; then
# 检查端口是否监听
if ss -tln | grep -q ':11434 '; then
# 检查API是否响应
if curl -s http://localhost:11434/api/tags > /dev/null; then
return 0
fi
fi
fi
return 1
}
restart_service() {
log_message "尝试重启Ollama服务..."
systemctl restart ollama
# 等待服务启动
sleep 10
for i in $(seq 1 $MAX_RETRIES); do
if check_service; then
log_message "服务重启成功"
return 0
fi
sleep $RETRY_DELAY
done
log_message "服务重启失败"
return 1
}
# 主监控逻辑
if ! check_service; then
log_message "检测到服务异常"
if ! restart_service; then
# 发送告警（这里可以集成邮件、钉钉、企业微信等）
log_message "发送告警通知"
# 这里添加告警逻辑
fi
else
# 记录资源使用情况
MEM_USAGE=$(ps aux | grep ollama | grep -v grep | awk '{sum+=$6} END {print sum/1024}')
log_message "服务正常，内存使用: ${MEM_USAGE}MB"
fi
EOF
chmod +x /opt/ollama/scripts/monitor.sh
chown ollama:ollama /opt/ollama/scripts/monitor.sh
# 9. 添加定时监控任务
(crontab -l 2>/dev/null; echo "*/5 * * * * /opt/ollama/scripts/monitor.sh") | crontab -
echo "系统优化完成！"
echo "请执行以下命令启动服务："
echo "systemctl daemon-reload"
echo "systemctl enable ollama"
echo "systemctl start ollama"
这个优化脚本涵盖了：
- 内核参数调优：针对高并发场景优化
- 资源限制调整：提高文件描述符和进程数限制
- 服务配置优化：合理的systemd配置
- 日志管理：自动轮转和压缩
- 监控脚本：自动检测和恢复服务
- 安全加固：限制服务权限
3.2 模型管理与内存优化
大模型服务最头疼的就是内存管理。不同的模型、不同的并发数，内存使用差异很大。下面是一个智能模型加载器：
import psutil
import GPUtil
import time
import threading
import json
from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ModelPriority(Enum):
HIGH = 1
MEDIUM = 2
LOW = 3
@dataclass
class ModelInfo:
name: str
size_gb: float
memory_required_gb: float
priority: ModelPriority
last_used: float
load_count: int = 0
class SmartModelManager:
"""智能模型管理器"""
def __init__(self, max_memory_gb: float = 32, max_models: int = 3):
self.max_memory_gb = max_memory_gb
self.max_models = max_models
self.loaded_models: Dict[str, ModelInfo] = {}
self.lock = threading.RLock()
self.monitor_thread = None
self.running = False
# 加载模型元数据
self.model_metadata = self._load_model_metadata()
def _load_model_metadata(self) -> Dict[str, Dict]:
"""加载模型元数据"""
# 这里可以从配置文件或数据库加载
return {
"llama3:8b": {
"size_gb": 4.7,
"memory_required_gb": 8.0,
"description": "Meta的8B参数模型"
},
"deepseek-r1:7b": {
"size_gb": 4.7,
"memory_required_gb": 9.0,
"description": "深度求索的7B推理模型"
},
"deepseek-r1:32b": {
"size_gb": 19.0,
"memory_required_gb": 35.0,
"description": "深度求索的32B模型"
},
"codellama:7b": {
"size_gb": 3.8,
"memory_required_gb": 7.0,
"description": "代码生成专用模型"
}
}
def get_available_memory(self) -> float:
"""获取可用内存（GB）"""
try:
# 尝试获取GPU内存
gpus = GPUtil.getGPUs()
if gpus:
gpu = gpus[0]
free_memory = gpu.memoryFree
return free_memory / 1024.0
except:
pass
# 回退到系统内存
memory = psutil.virtual_memory()
return memory.available / (1024 ** 3)
def get_current_usage(self) -> float:
"""获取当前已用内存（GB）"""
total_used = 0.0
for model_info in self.loaded_models.values():
total_used += model_info.memory_required_gb
return total_used
def can_load_model(self, model_name: str, priority: ModelPriority) -> bool:
"""检查是否可以加载模型"""
if model_name not in self.model_metadata:
logger.warning(f"未知模型: {model_name}")
return False
model_data = self.model_metadata[model_name]
required_memory = model_data["memory_required_gb"]
available_memory = self.get_available_memory()
current_usage = self.get_current_usage()
# 检查内存是否足够
if current_usage + required_memory > self.max_memory_gb:
logger.info(f"内存不足，需要{required_memory}GB，当前使用{current_usage}GB")
# 尝试卸载低优先级模型
if self._make_space(required_memory, priority):
return True
return False
# 检查模型数量限制
if len(self.loaded_models) >= self.max_models:
logger.info(f"达到模型数量限制{self.max_models}")
# 尝试卸载一个模型
if self._unload_lowest_priority(priority):
return True
return False
return True
def _make_space(self, required_memory: float, new_priority: ModelPriority) -> bool:
"""尝试释放内存空间"""
with self.lock:
# 按优先级和最后使用时间排序
sorted_models = sorted(
self.loaded_models.items(),
key=lambda x: (
x[1].priority.value, # 优先级高的先保留
x[1].last_used # 最近使用的先保留
)
)
# 计算需要释放的内存
memory_to_free = required_memory - (self.max_memory_gb - self.get_current_usage())
freed_memory = 0.0
models_to_unload = []
for model_name, model_info in sorted_models:
if model_info.priority.value > new_priority.value: # 优先级低于新模型
freed_memory += model_info.memory_required_gb
models_to_unload.append(model_name)
if freed_memory >= memory_to_free:
break
# 执行卸载
for model_name in models_to_unload:
self._unload_model_internal(model_name)
return freed_memory >= memory_to_free
def _unload_lowest_priority(self, new_priority: ModelPriority) -> bool:
"""卸载优先级最低的模型"""
with self.lock:
if not self.loaded_models:
return False
# 找到优先级最低且最久未使用的模型
lowest_priority = None
lowest_model = None
for model_name, model_info in self.loaded_models.items():
if (lowest_priority is None or
model_info.priority.value > lowest_priority or
(model_info.priority.value == lowest_priority and
model_info.last_used < self.loaded_models[lowest_model].last_used)):
lowest_priority = model_info.priority.value
lowest_model = model_name
if lowest_model and self.loaded_models[lowest_model].priority.value > new_priority.value:
self._unload_model_internal(lowest_model)
return True
return False
def load_model(self, model_name: str, priority: ModelPriority = ModelPriority.MEDIUM) -> bool:
"""加载模型"""
with self.lock:
if model_name in self.loaded_models:
# 更新使用时间
self.loaded_models[model_name].last_used = time.time()
self.loaded_models[model_name].load_count += 1
logger.info(f"模型{model_name}已加载，更新使用时间")
return True
if not self.can_load_model(model_name, priority):
logger.error(f"无法加载模型{model_name}，资源不足")
return False
# 执行实际加载（这里调用Ollama API）
logger.info(f"开始加载模型: {model_name}")
# 模拟加载过程
time.sleep(2) # 实际应该调用Ollama的加载API
# 记录模型信息
model_data = self.model_metadata[model_name]
model_info = ModelInfo(
name=model_name,
size_gb=model_data["size_gb"],
memory_required_gb=model_data["memory_required_gb"],
priority=priority,
last_used=time.time(),
load_count=1
)
self.loaded_models[model_name] = model_info
logger.info(f"模型{model_name}加载完成")
return True
def _unload_model_internal(self, model_name: str):
"""内部卸载模型方法"""
if model_name in self.loaded_models:
logger.info(f"卸载模型: {model_name}")
# 实际应该调用Ollama的卸载API
del self.loaded_models[model_name]
def unload_model(self, model_name: str):
"""卸载模型"""
with self.lock:
self._unload_model_internal(model_name)
def get_model_stats(self) -> Dict:
"""获取模型统计信息"""
with self.lock:
stats = {
"loaded_models": len(self.loaded_models),
"total_memory_used_gb": self.get_current_usage(),
"available_memory_gb": self.get_available_memory(),
"models": []
}
for model_name, model_info in self.loaded_models.items():
stats["models"].append({
"name": model_name,
"memory_gb": model_info.memory_required_gb,
"priority": model_info.priority.name,
"last_used": time.strftime(
"%Y-%m-%d %H:%M:%S",
time.localtime(model_info.last_used)
),
"load_count": model_info.load_count
})
return stats
def start_monitoring(self, interval: int = 60):
"""启动监控线程"""
self.running = True
self.monitor_thread = threading.Thread(
target=self._monitor_loop,
args=(interval,),
daemon=True
)
self.monitor_thread.start()
logger.info(f"监控线程已启动，间隔{interval}秒")
def _monitor_loop(self, interval: int):
"""监控循环"""
while self.running:
try:
stats = self.get_model_stats()
logger.info(f"模型管理器状态: {json.dumps(stats, indent=2)}")
# 检查内存使用
available_memory = self.get_available_memory()
if available_memory < 2.0: # 小于2GB
logger.warning(f"可用内存不足: {available_memory:.2f}GB")
self._handle_low_memory()
except Exception as e:
logger.error(f"监控出错: {e}")
time.sleep(interval)
def _handle_low_memory(self):
"""处理低内存情况"""
with self.lock:
# 卸载最久未使用的低优先级模型
if self.loaded_models:
sorted_models = sorted(
self.loaded_models.items(),
key=lambda x: (x[1].priority.value, x[1].last_used)
)
for model_name, model_info in sorted_models:
if model_info.priority == ModelPriority.LOW:
self._unload_model_internal(model_name)
logger.info(f"因内存不足卸载模型: {model_name}")
break
def stop(self):
"""停止管理器"""
self.running = False
if self.monitor_thread:
self.monitor_thread.join(timeout=5)
# 卸载所有模型
with self.lock:
for model_name in list(self.loaded_models.keys()):
self._unload_model_internal(model_name)
# 使用示例
if __name__ == "__main__":
manager = SmartModelManager(max_memory_gb=32, max_models=3)
# 启动监控
manager.start_monitoring(interval=30)
try:
# 加载模型
manager.load_model("llama3:8b", ModelPriority.HIGH)
manager.load_model("deepseek-r1:7b", ModelPriority.MEDIUM)
# 获取状态
stats = manager.get_model_stats()
print("当前状态:")
print(json.dumps(stats, indent=2))
# 模拟使用
time.sleep(10)
# 尝试加载第三个模型
if manager.load_model("codellama:7b", ModelPriority.LOW):
print("第三个模型加载成功")
else:
print("第三个模型加载失败，资源不足")
# 等待一段时间
time.sleep(60)
finally:
manager.stop()
这个模型管理器提供了：
- 智能加载决策：根据内存和优先级决定是否加载
- 自动内存管理：低内存时自动卸载不重要的模型
- 使用统计：记录模型使用频率和时间
- 优先级系统：确保重要模型常驻内存
- 实时监控：定期检查系统状态
4. 故障排查与性能调优
4.1 常见问题诊断手册
在实际运维中，你会遇到各种奇怪的问题。下面是我整理的常见问题排查指南：
#!/bin/bash
# ollama-troubleshoot.sh
# Ollama故障排查脚本
set -e
LOG_FILE="/tmp/ollama_troubleshoot_$(date +%Y%m%d_%H%M%S).log"
log() {
echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1" | tee -a "$LOG_FILE"
}
check_service_status() {
log "检查Ollama服务状态..."
if systemctl is-active ollama >/dev/null 2>&1; then
log "✓ 服务正在运行"
return 0
else
log "✗ 服务未运行"
# 尝试查看状态
systemctl status ollama --no-pager | tee -a "$LOG_FILE"
return 1
fi
}
check_port_listening() {
log "检查端口监听..."
if ss -tln | grep -q ':11434 '; then
log "✓ 端口11434正在监听"
# 检查监听地址
LISTEN_ADDR=$(ss -tln | grep ':11434 ' | awk '{print $4}')
log "监听地址: $LISTEN_ADDR"
if [[ "$LISTEN_ADDR" == "*:11434" ]] || [[ "$LISTEN_ADDR" == "0.0.0.0:11434" ]]; then
log "✓ 服务监听在所有接口"
else
log "⚠ 服务只监听在特定接口: $LISTEN_ADDR"
fi
return 0
else
log "✗ 端口11434未监听"
return 1
fi
}
check_api_connectivity() {
log "检查API连通性..."
local max_retries=3
local retry_delay=2
for i in $(seq 1 $max_retries); do
if curl -s http://localhost:11434/api/tags >/dev/null 2>&1; then
log "✓ API连接正常"
# 获取模型列表
MODEL_LIST=$(curl -s http://localhost:11434/api/tags | jq -r '.models[].name' 2>/dev/null || echo "无法解析响应")
log "可用模型: $MODEL_LIST"
return 0
else
log "尝试 $i/$max_retries: API连接失败"
sleep $retry_delay
fi
done
log "✗ API连接失败"
return 1
}
check_system_resources() {
log "检查系统资源..."
# 内存
MEM_TOTAL=$(free -h | awk '/^Mem:/ {print $2}')
MEM_USED=$(free -h | awk '/^Mem:/ {print $3}')
MEM_FREE=$(free -h | awk '/^Mem:/ {print $4}')
log "内存: 总量=$MEM_TOTAL, 已用=$MEM_USED, 空闲=$MEM_FREE"
# 交换空间
SWAP_TOTAL=$(free -h | awk '/^Swap:/ {print $2}')
SWAP_USED=$(free -h | awk '/^Swap:/ {print $3}')
log "交换空间: 总量=$SWAP_TOTAL, 已用=$SWAP_USED"
# 磁盘空间
DISK_USAGE=$(df -h / | awk 'NR==2 {print $5}')
log "根分区使用率: $DISK_USAGE"
# 检查内存是否充足（至少8GB）
MEM_TOTAL_KB=$(free | awk '/^Mem:/ {print $2}')
if [ "$MEM_TOTAL_KB" -lt 8000000 ]; then
log "⚠ 警告: 内存可能不足（小于8GB）"
fi
}
check_gpu_status() {
log "检查GPU状态..."
if command -v nvidia-smi >/dev/null 2>&1; then
log "检测到NVIDIA GPU"
# 获取GPU信息
GPU_INFO=$(nvidia-smi --query-gpu=name,memory.total,memory.used,memory.free,temperature.gpu,utilization.gpu --format=csv,noheader,nounits 2>/dev/null || echo "无法获取GPU信息")
if [ -n "$GPU_INFO" ]; then
IFS=',' read -r NAME MEM_TOTAL MEM_USED MEM_FREE TEMP UTIL <<< "$GPU_INFO"
log "GPU: $NAME"
log "显存: 总量=${MEM_TOTAL}MB, 已用=${MEM_USED}MB, 空闲=${MEM_FREE}MB"
log "温度: ${TEMP}°C, 使用率: ${UTIL}%"
# 检查显存是否充足
if [ "$MEM_FREE" -lt 2000 ]; then
log "⚠ 警告: 显存空闲不足2GB"
fi
fi
elif command -v rocm-smi >/dev/null 2>&1; then
log "检测到AMD GPU"
rocm-smi | tee -a "$LOG_FILE"
else
log "未检测到GPU，使用CPU模式"
fi
}
check_firewall() {
log "检查防火墙..."
if command -v firewall-cmd >/dev/null 2>&1; then
if firewall-cmd --list-ports | grep -q '11434/tcp'; then
log "✓ 防火墙已放行11434端口"
else
log "⚠ 防火墙未放行11434端口"
fi
fi
# 检查SELinux
if command -v sestatus >/dev/null 2>&1; then
SELINUX_STATUS=$(sestatus | grep "SELinux status" | awk '{print $3}')
SELINUX_MODE=$(sestatus | grep "Current mode" | awk '{print $3}')
log "SELinux状态: $SELINUX_STATUS, 模式: $SELINUX_MODE"
if [ "$SELINUX_STATUS" = "enabled" ] && [ "$SELINUX_MODE" = "enforcing" ]; then
log "⚠ SELinux处于强制模式，可能影响Ollama运行"
# 检查相关策略
if getenforce | grep -q "Enforcing"; then
log "建议检查SELinux日志: ausearch -m avc -ts recent"
fi
fi
fi
}
check_logs() {
log "检查系统日志..."
local log_lines=20
# 检查journalctl日志
log "最近${log_lines}条Ollama日志:"
journalctl -u ollama -n $log_lines --no-pager | tee -a "$LOG_FILE"
# 检查Ollama自己的日志
if [ -f "/opt/ollama/logs/ollama.log" ]; then
log "Ollama应用日志尾部:"
tail -n $log_lines /opt/ollama/logs/ollama.log | tee -a "$LOG_FILE"
fi
}
check_performance() {
log "性能测试..."
# 简单的响应时间测试
local test_prompt="Hello, how are you?"
local start_time
local end_time
start_time=$(date +%s%N)
if curl -s -X POST http://localhost:11434/api/generate \
-H "Content-Type: application/json" \
-d "{\"model\":\"llama3:8b\",\"prompt\":\"$test_prompt\",\"stream\":false}" \
>/dev/null 2>&1; then
end_time=$(date +%s%N)
local duration=$(( (end_time - start_time) / 1000000 ))
log "简单请求响应时间: ${duration}ms"
if [ "$duration" -gt 5000 ]; then
log "⚠ 响应时间较慢（超过5秒）"
fi
else
log "✗ 性能测试请求失败"
fi
}
check_dependencies() {
log "检查依赖..."
local missing_deps=()
# 检查必要命令
for cmd in curl jq ss systemctl; do
if ! command -v $cmd >/dev/null 2>&1; then
missing_deps+=("$cmd")
fi
done
if [ ${#missing_deps[@]} -eq 0 ]; then
log "✓ 所有依赖已安装"
else
log "✗ 缺少依赖: ${missing_deps[*]}"
# 提供安装建议
if command -v yum >/dev/null 2>&1; then
log "安装建议: sudo yum install -y ${missing_deps[*]}"
elif command -v apt-get >/dev/null 2>&1; then
log "安装建议: sudo apt-get install -y ${missing_deps[*]}"
fi
fi
}
generate_report() {
log "生成诊断报告..."
echo ""
echo "========================================"
echo " Ollama诊断报告"
echo "========================================"
echo "生成时间: $(date)"
echo "日志文件: $LOG_FILE"
echo ""
# 汇总检查结果
local checks=(
"服务状态" "$(if check_service_status; then echo '正常'; else echo '异常'; fi)"
"端口监听" "$(if check_port_listening; then echo '正常'; else echo '异常'; fi)"
"API连通" "$(if check_api_connectivity; then echo '正常'; else echo '异常'; fi)"
"系统资源" "详见日志"
"GPU状态" "详见日志"
"防火墙" "详见日志"
"依赖检查" "$(if check_dependencies; then echo '正常'; else echo '异常'; fi)"
)
for ((i=0; i<${#checks[@]}; i+=2)); do
printf "%-15s: %s\n" "${checks[$i]}" "${checks[$i+1]}"
done
echo ""
echo "建议操作:"
if ! check_service_status; then
echo "1. 启动服务: sudo systemctl start ollama"
fi
if ! check_port_listening; then
echo "2. 检查OLLAMA_HOST环境变量配置"
fi
if ! check_api_connectivity; then
echo "3. 检查模型是否已下载: ollama list"
echo "4. 查看详细错误: journalctl -u ollama -f"
fi
echo ""
echo "详细日志请查看: $LOG_FILE"
}
main() {
log "开始Ollama故障诊断..."
# 执行所有检查
check_service_status
check_port_listening
check_api_connectivity
check_system_resources
check_gpu_status
check_firewall
check_logs
check_performance
check_dependencies
# 生成报告
generate_report
log "诊断完成"
}
# 执行主函数
main
这个诊断脚本可以帮你快速定位问题：
- 服务状态检查：是否运行、是否开机启动
- 网络检查：端口监听、防火墙、SELinux
- 资源检查：内存、磁盘、GPU
- API测试：连通性、响应时间
- 日志分析：系统日志和应用日志
- 依赖检查：必要工具是否安装
4.2 性能监控与告警
对于生产环境，监控是必不可少的。下面是一个完整的监控方案：
# monitor/ollama_monitor.py
import time
import json
import requests
import psutil
import GPUtil
from datetime import datetime
from typing import Dict, List, Optional
import logging
from logging.handlers import RotatingFileHandler
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import threading
from dataclasses import dataclass, asdict
from enum import Enum
import sqlite3
from contextlib import contextmanager
# 配置日志
log_handler = RotatingFileHandler(
'/var/log/ollama_monitor.log',
maxBytes=10*1024*1024, # 10MB
backupCount=5
)
log_handler.setFormatter(logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
))
logger = logging.getLogger('OllamaMonitor')
logger.setLevel(logging.INFO)
logger.addHandler(log_handler)
class AlertLevel(Enum):
INFO = "INFO"
WARNING = "WARNING"
CRITICAL = "CRITICAL"
@dataclass
class Metric:
timestamp: float
service: str
metric_name: str
value: float
tags: Dict[str, str]
@dataclass
class Alert:
timestamp: float
level: AlertLevel
service: str
message: str
metric_name: Optional[str] = None
metric_value: Optional[float] = None
resolved: bool = False
class DatabaseManager:
"""数据库管理器"""
def __init__(self, db_path: str = "/opt/ollama/monitor.db"):
self.db_path = db_path
self._init_db()
def _init_db(self):
"""初始化数据库"""
with self._get_connection() as conn:
cursor = conn.cursor()
# 创建指标表
cursor.execute('''
CREATE TABLE IF NOT EXISTS metrics (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp REAL NOT NULL,
service TEXT NOT NULL,
metric_name TEXT NOT NULL,
value REAL NOT NULL,
tags TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# 创建告警表
cursor.execute('''
CREATE TABLE IF NOT EXISTS alerts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp REAL NOT NULL,
level TEXT NOT NULL,
service TEXT NOT NULL,
message TEXT NOT NULL,
metric_name TEXT,
metric_value REAL,
resolved BOOLEAN DEFAULT 0,
resolved_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# 创建索引
cursor.execute('CREATE INDEX IF NOT EXISTS idx_metrics_timestamp ON metrics(timestamp)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_metrics_service ON metrics(service)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_alerts_resolved ON alerts(resolved)')
conn.commit()
@contextmanager
def _get_connection(self):
"""获取数据库连接"""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
try:
yield conn
finally:
conn.close()
def save_metric(self, metric: Metric):
"""保存指标"""
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute('''
INSERT INTO metrics (timestamp, service, metric_name, value, tags)
VALUES (?, ?, ?, ?, ?)
''', (
metric.timestamp,
metric.service,
metric.metric_name,
metric.value,
json.dumps(metric.tags) if metric.tags else None
))
conn.commit()
def save_alert(self, alert: Alert):
"""保存告警"""
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute('''
INSERT INTO alerts (timestamp, level, service, message, metric_name, metric_value, resolved)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', (
alert.timestamp,
alert.level.value,
alert.service,
alert.message,
alert.metric_name,
alert.metric_value,
1 if alert.resolved else 0
))
conn.commit()
def get_recent_metrics(self, service: str, metric_name: str,
hours: int = 24, limit: int = 1000) -> List[Dict]:
"""获取最近指标"""
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute('''
SELECT timestamp, value, tags
FROM metrics
WHERE service = ? AND metric_name = ?
AND timestamp >= ?
ORDER BY timestamp DESC
LIMIT ?
''', (
service,
metric_name,
time.time() - (hours * 3600),
limit
))
results = []
for row in cursor.fetchall():
results.append({
'timestamp': row['timestamp'],
'value': row['value'],
'tags': json.loads(row['tags']) if row['tags'] else {}
})
return results
def get_active_alerts(self, service: Optional[str] = None) -> List[Dict]:
"""获取活跃告警"""
with self._get_connection() as conn:
cursor = conn.cursor()
if service:
cursor.execute('''
SELECT * FROM alerts
WHERE resolved = 0 AND service = ?
ORDER BY timestamp DESC
''', (service,))
else:
cursor.execute('''
SELECT * FROM alerts
WHERE resolved = 0
ORDER BY timestamp DESC
''')
return [dict(row) for row in cursor.fetchall()]
class AlertManager:
"""告警管理器"""
def __init__(self, db_manager: DatabaseManager):
self.db = db_manager
self.alert_rules = self._load_alert_rules()
self.active_alerts = {} # 存储活跃告警的key
def _load_alert_rules(self) -> List[Dict]:
"""加载告警规则"""
return [
{
"metric_name": "api_response_time",
"condition": ">",
"threshold": 5000, # 5秒
"level": AlertLevel.WARNING,
"message": "API响应时间过长",
"service": "ollama"
},
{
"metric_name": "memory_usage_percent",
"condition": ">",
"threshold": 90, # 90%
"level": AlertLevel.CRITICAL,
"message": "内存使用率过高",
"service": "system"
},
{
"metric_name": "gpu_memory_usage_percent",
"condition": ">",
"threshold": 95, # 95%
"level": AlertLevel.CRITICAL,
"message": "GPU显存使用率过高",
"service": "gpu"
},
{
"metric_name": "api_error_rate",
"condition": ">",
"threshold": 10, # 10%
"level": AlertLevel.WARNING,
"message": "API错误率过高",
"service": "ollama"
},
{
"metric_name": "service_status",
"condition": "==",
"threshold": 0, # 服务停止
"level": AlertLevel.CRITICAL,
"message": "服务停止运行",
"service": "ollama"
}
]
def check_metric(self, metric: Metric):
"""检查指标是否触发告警"""
for rule in self.alert_rules:
if (rule["service"] == metric.service and
rule["metric_name"] == metric.metric_name):
triggered = False
if rule["condition"] == ">":
triggered = metric.value > rule["threshold"]
elif rule["condition"] == "<":
triggered = metric.value < rule["threshold"]
elif rule["condition"] == "==":
triggered = metric.value == rule["threshold"]
elif rule["condition"] == "!=":
triggered = metric.value != rule["threshold"]
if triggered:
self._trigger_alert(metric, rule)
else:
self._resolve_alert(metric, rule)
def _trigger_alert(self, metric: Metric, rule: Dict):
"""触发告警"""
alert_key = f"{metric.service}:{metric.metric_name}"
if alert_key not in self.active_alerts:
alert = Alert(
timestamp=time.time(),
level=rule["level"],
service=metric.service,
message=rule["message"],
metric_name=metric.metric_name,
metric_value=metric.value
)
self.db.save_alert(alert)
self.active_alerts[alert_key] = alert.timestamp
# 发送告警通知
self._send_alert_notification(alert)
logger.warning(f"触发告警: {alert.message} (值: {metric.value})")
def _resolve_alert(self, metric: Metric, rule: Dict):
"""解决告警"""
alert_key = f"{metric.service}:{metric.metric_name}"
if alert_key in self.active_alerts:
# 标记为已解决
logger.info(f"告警已解决: {rule['message']}")
del self.active_alerts[alert_key]
def _send_alert_notification(self, alert: Alert):
"""发送告警通知"""
# 这里可以实现邮件、钉钉、企业微信等通知方式
# 示例：发送邮件
try:
# 配置邮件参数
smtp_server = "smtp.yourdomain.com"
smtp_port = 587
sender_email = "monitor@yourdomain.com"
receiver_email = "admin@yourdomain.com"
password = "your_password"
# 创建邮件
message = MIMEMultipart("alternative")
message["Subject"] = f"[{alert.level.value}] Ollama告警: {alert.message}"
message["From"] = sender_email
message["To"] = receiver_email
# 邮件正文
text = f"""
告警时间: {datetime.fromtimestamp(alert.timestamp)}
告警级别: {alert.level.value}
服务: {alert.service}
消息: {alert.message}
指标: {alert.metric_name}
值: {alert.metric_value}
"""
html = f"""
<html>
<body>
<h2>Ollama服务告警</h2>
<table border="1">
<tr><td><b>告警时间</b></td><td>{datetime.fromtimestamp(alert.timestamp)}</td></tr>
<tr><td><b>告警级别</b></td><td><span style="color: {'red' if alert.level == AlertLevel.CRITICAL else 'orange'}">{alert.level.value}</span></td></tr>
<tr><td><b>服务</b></td><td>{alert.service}</td></tr>
<tr><td><b>消息</b></td><td>{alert.message}</td></tr>
<tr><td><b>指标</b></td><td>{alert.metric_name}</td></tr>
<tr><td><b>值</b></td><td>{alert.metric_value}</td></tr>
</table>
</body>
</html>
"""
part1 = MIMEText(text, "plain")
part2 = MIMEText(html, "html")
message.attach(part1)
message.attach(part2)
# 发送邮件
with smtplib.SMTP(smtp_server, smtp_port) as server:
server.starttls()
server.login(sender_email, password)
server.sendmail(sender_email, receiver_email, message.as_string())
logger.info(f"告警邮件已发送: {alert.message}")
except Exception as e:
logger.error(f"发送告警邮件失败: {e}")
class OllamaMonitor:
"""Ollama监控主类"""
def __init__(self, ollama_url: str = "http://localhost:11434"):
self.ollama_url = ollama_url
self.db_manager = DatabaseManager()
self.alert_manager = AlertManager(self.db_manager)
self.running = False
self.monitor_thread = None
# 指标收集器
self.metric_collectors = [
self._collect_system_metrics,
self._collect_ollama_metrics,
self._collect_gpu_metrics,
self._collect_api_metrics
]
def _collect_system_metrics(self) -> List[Metric]:
"""收集系统指标"""
metrics = []
timestamp = time.time()
# CPU使用率
cpu_percent = psutil.cpu_percent(interval=1)
metrics.append(Metric(
timestamp=timestamp,
service="system",
metric_name="cpu_usage_percent",
value=cpu_percent,
tags={"host": "localhost"}
))
# 内存使用率
memory = psutil.virtual_memory()
metrics.append(Metric(
timestamp=timestamp,
service="system",
metric_name="memory_usage_percent",
value=memory.percent,
tags={"host": "localhost"}
))
# 内存使用量(GB)
metrics.append(Metric(
timestamp=timestamp,
service="system",
metric_name="memory_used_gb",
value=memory.used / (1024**3),
tags={"host": "localhost"}
))
# 磁盘使用率
disk = psutil.disk_usage('/')
metrics.append(Metric(
timestamp=timestamp,
service="system",
metric_name="disk_usage_percent",
value=disk.percent,
tags={"mount": "/", "host": "localhost"}
))
# 网络IO
net_io = psutil.net_io_counters()
metrics.append(Metric(
timestamp=timestamp,
service="system",
metric_name="network_bytes_sent",
value=net_io.bytes_sent,
tags={"host": "localhost"}
))
metrics.append(Metric(
timestamp=timestamp,
service="system",
metric_name="network_bytes_recv",
value=net_io.bytes_recv,
tags={"host": "localhost"}
))
return metrics
def _collect_gpu_metrics(self) -> List[Metric]:
"""收集GPU指标"""
metrics = []
timestamp = time.time()
try:
gpus = GPUtil.getGPUs()
for i, gpu in enumerate(gpus):
# GPU使用率
metrics.append(Metric(
timestamp=timestamp,
service="gpu",
metric_name="gpu_usage_percent",
value=gpu.load * 100,
tags={"gpu_id": str(i), "name": gpu.name}
))
# GPU显存使用率
memory_percent = (gpu.memoryUsed / gpu.memoryTotal) * 100
metrics.append(Metric(
timestamp=timestamp,
service="gpu",
metric_name="gpu_memory_usage_percent",
value=memory_percent,
tags={"gpu_id": str(i), "name": gpu.name}
))
# GPU显存使用量(GB)
metrics.append(Metric(
timestamp=timestamp,
service="gpu",
metric_name="gpu_memory_used_gb",
value=gpu.memoryUsed / 1024,
tags={"gpu_id": str(i), "name": gpu.name}
))
# GPU温度
metrics.append(Metric(
timestamp=timestamp,
service="gpu",
metric_name="gpu_temperature",
value=gpu.temperature,
tags={"gpu_id": str(i), "name": gpu.name}
北京人形旗下天工造物具身智能开源社区，聚焦具身天工与慧思开物两大平台
更多推荐
3
0- 0
已为社区贡献3条内容
所有评论(0)