DeepSeek-R1-Distill-Qwen-1.5B日志聚合：ELK栈集成部署案例-平芜编程栈

DeepSeek-R1-Distill-Qwen-1.5B日志聚合：ELK栈集成部署案例

1. 引言

1.1 业务场景描述

随着AI模型在生产环境中的广泛应用，大语言模型服务的可观测性需求日益增长。以DeepSeek-R1-Distill-Qwen-1.5B为代表的高性能推理模型，在提供数学推理、代码生成和逻辑推导能力的同时，也产生了大量运行时日志数据。这些日志分散在Web服务进程、GPU资源监控、请求响应记录等多个维度，传统手动排查方式已无法满足快速定位问题的需求。

当前面临的核心痛点包括：

模型推理延迟波动难以追踪
用户输入异常与系统错误关联困难
多实例部署下日志分散，缺乏统一视图
缺乏历史请求的结构化存储与查询能力

为解决上述问题，本文提出将DeepSeek-R1-Distill-Qwen-1.5B服务与ELK（Elasticsearch + Logstash + Kibana）技术栈深度集成的完整方案，实现从日志采集、处理、存储到可视化分析的闭环管理。

1.2 技术方案预告

本实践将围绕以下核心环节展开：

基于Gradio构建的模型Web服务日志格式化输出
使用Filebeat采集应用日志并传输至Logstash
Logstash对JSON日志进行解析、增强与过滤
Elasticsearch存储结构化日志数据
Kibana构建实时监控仪表盘

最终实现对模型调用链路的全生命周期追踪，提升运维效率与服务质量。

2. 技术方案选型

2.1 架构设计对比

方案	优势	劣势	适用性
直接写入数据库	写入快，易查询	耦合度高，影响性能	小规模测试
Prometheus + Grafana	实时指标强	不适合文本日志	仅监控指标
ELK栈	全文检索强，可扩展性好，支持复杂分析	部署复杂，资源消耗较高	生产级日志聚合

选择ELK栈的主要原因在于其强大的文本处理能力和灵活的查询语法，特别适合分析自然语言输入/输出这类非结构化内容。

2.2 组件职责划分

Filebeat：轻量级日志收集器，部署在模型服务主机上，负责监听日志文件变化
Logstash：日志管道引擎，执行格式转换、字段提取、时间戳标准化等ETL操作
Elasticsearch：分布式搜索引擎，提供高性能的日志存储与检索能力
Kibana：可视化平台，用于构建交互式仪表板和告警规则

该架构具备良好的解耦性，各组件可独立横向扩展。

3. 实现步骤详解

3.1 日志格式改造：app.py增强

首先需修改原始app.py，使其输出结构化JSON日志。以下是关键代码片段：

import logging import json from datetime import datetime import gradio as gr import torch from transformers import AutoTokenizer, AutoModelForCausalLM # 配置结构化日志 class JSONFormatter(logging.Formatter): def format(self, record): log_entry = { "timestamp": datetime.utcnow().isoformat(), "level": record.levelname, "module": record.module, "function": record.funcName, "message": record.getMessage(), "service": "deepseek-r1-qwen-1.5b" } if hasattr(record, 'request_data'): log_entry['request'] = record.request_data if hasattr(record, 'response_data'): log_entry['response'] = record.response_data if hasattr(record, 'metrics'): log_entry['metrics'] = record.metrics return json.dumps(log_entry) logger = logging.getLogger("deepseek_web") handler = logging.FileHandler("/var/log/deepseek/app.log") handler.setFormatter(JSONFormatter()) logger.addHandler(handler) logger.setLevel(logging.INFO) # 加载模型 DEVICE = "cuda" if torch.cuda.is_available() else "cpu" MODEL_PATH = "/root/.cache/huggingface/deepseek-ai/DeepSeek-R1-Distill-Qwen-1___5B" tokenizer = AutoTokenizer.from_pretrained(MODEL_PATH) model = AutoModelForCausalLM.from_pretrained(MODEL_PATH).to(DEVICE) def generate_response(prompt, max_tokens=2048, temperature=0.6, top_p=0.95): start_time = datetime.now() inputs = tokenizer(prompt, return_tensors="pt").to(DEVICE) tokens_in = len(inputs["input_ids"][0]) with torch.no_grad(): outputs = model.generate( **inputs, max_new_tokens=max_tokens, temperature=temperature, top_p=top_p, do_sample=True ) response = tokenizer.decode(outputs[0], skip_special_tokens=True) tokens_out = len(tokenizer.encode(response)) duration = (datetime.now() - start_time).total_seconds() # 记录结构化日志 log_data = { "request_data": { "prompt": prompt[:500], # 截断过长输入 "max_tokens": max_tokens, "temperature": temperature, "top_p": top_p }, "response_data": { "response_length": len(response), "tokens_in": tokens_in, "tokens_out": tokens_out, "inference_time": round(duration, 2) }, "metrics": { "tgs": tokens_out / duration if duration > 0 else 0 } } logger.info("Query processed", extra=log_data) return response # Gradio界面 demo = gr.Interface( fn=generate_response, inputs=[ gr.Textbox(label="输入提示"), gr.Slider(128, 2048, value=2048, label="最大Token数"), gr.Slider(0.1, 1.0, value=0.6, label="温度"), gr.Slider(0.5, 1.0, value=0.95, label="Top-P") ], outputs="text", title="DeepSeek-R1-Distill-Qwen-1.5B 推理服务", description="支持数学推理、代码生成与逻辑推导" ) if __name__ == "__main__": demo.launch(server_name="0.0.0.0", server_port=7860, show_api=False)

核心改进点：
使用JSONFormatter输出标准JSON格式
在日志中嵌入请求参数、响应统计与性能指标
添加tokens_per_second (TGS)作为关键性能指标

3.2 Filebeat配置：日志采集

安装Filebeat后，创建配置文件/etc/filebeat/filebeat.yml：

filebeat.inputs: - type: log enabled: true paths: - /var/log/deepseek/app.log fields: service: deepseek-r1-qwen-1.5b env: production json.keys_under_root: true json.add_error_key: true json.overwrite_keys: true output.logstash: hosts: ["logstash-server:5044"]

启动命令：

sudo systemctl enable filebeat sudo systemctl start filebeat

3.3 Logstash处理管道

创建Logstash配置/etc/logstash/conf.d/deepseek-pipeline.conf：

input { beats { port => 5044 } } filter { # 确保时间戳正确解析 date { match => [ "timestamp", "ISO8601" ] target => "@timestamp" } # 分类请求类型（可根据关键词识别） if [request][prompt] =~ "def|class|import" { mutate { add_tag => [ "code_generation" ] } } else if [request][prompt] =~ "\d+\+|\*|\-|\/|=" { mutate { add_tag => [ "math_reasoning" ] } } else { mutate { add_tag => [ "general_conversation" ] } } # 性能分级 if [metrics][tgs] < 10 { mutate { add_tag => [ "low_performance" ] } } else if [metrics][tgs] >= 20 { mutate { add_tag => [ "high_performance" ] } } } output { elasticsearch { hosts => ["http://elasticsearch:9200"] index => "deepseek-logs-%{+YYYY.MM.dd}" } }

3.4 Kibana仪表板设计

在Kibana中创建索引模式deepseek-logs-*，并构建以下关键视图：

QPS趋势图：基于@timestamp的时间序列聚合
TGS分布热力图：展示不同时间段的吞吐表现
请求类型饼图：统计代码生成 vs 数学推理占比
慢查询TOP榜：按inference_time排序的长尾请求
错误日志过滤器：筛选level: ERROR条目

4. 实践问题与优化

4.1 常见问题及解决方案

问题现象	根本原因	解决方法
日志中文乱码	编码未指定UTF-8	在Python日志处理器中显式设置`encoding='utf-8'`
Logstash CPU过高	过滤规则低效	使用`dissect`替代正则，减少条件判断层级
Elasticsearch磁盘增长过快	保留策略缺失	启用ILM（Index Lifecycle Management），自动删除30天前数据
Filebeat丢失日志	文件权限不足	确保`/var/log/deepseek/`目录对`filebeat`用户可读