模版网站做支付功能,天津怎么建立企业网站,漳州正规网站建设哪家便宜,无锡滨湖住房与城乡建设局网站Nanbeige4.1-3B Chainlit日志审计#xff1a;用户行为追踪与响应质量分析 1. 引言#xff1a;为什么我们需要关注日志#xff1f; 想象一下#xff0c;你部署了一个智能对话模型#xff0c;用户每天都在和它聊天。但你真的知道用户问了什么吗#xff1f;模型回答得怎么…Nanbeige4.1-3B Chainlit日志审计用户行为追踪与响应质量分析1. 引言为什么我们需要关注日志想象一下你部署了一个智能对话模型用户每天都在和它聊天。但你真的知道用户问了什么吗模型回答得怎么样有没有什么奇怪的问题或者回答得不好的地方这就是日志审计要做的事情。它就像给对话系统装了一个“行车记录仪”把每一次对话都记录下来让我们可以回看、分析、改进。今天我们要聊的就是如何对基于Chainlit前端调用Nanbeige4.1-3B模型的系统进行日志审计。通过这个方法你可以追踪用户行为了解用户最常问什么问题什么时候问得最多分析响应质量看看模型回答得准不准有没有胡说八道发现潜在问题找到那些模型回答不好或者用户不满意的对话持续优化系统基于数据来改进提示词、调整模型参数听起来是不是很有用接下来我就带你一步步实现这个日志审计系统。2. 环境准备与日志基础2.1 系统架构回顾在开始之前我们先快速回顾一下整个系统的架构用户提问 → Chainlit前端 → Nanbeige4.1-3B模型 → 返回回答Chainlit是一个专门为AI应用设计的聊天界面框架它内置了对话管理、文件上传等功能。Nanbeige4.1-3B是一个3B参数的开源文本生成模型通过vLLM进行部署提供高效的推理服务。2.2 日志记录的基本原理日志记录的核心思想很简单在关键节点把重要信息保存下来。对于我们的对话系统关键节点包括用户输入时记录用户的问题、时间、用户ID模型响应时记录模型的回答、生成时间、token数量对话结束时记录整个对话的摘要信息Chainlit本身提供了一些日志功能但我们需要扩展它让它记录更多对我们分析有用的信息。3. 实现Chainlit日志记录3.1 基础日志配置首先我们需要在Chainlit应用中添加日志记录功能。创建一个新的Python文件比如叫chat_with_logging.pyimport chainlit as cl import json import datetime import os from typing import Dict, Any # 创建日志目录 LOG_DIR ./chat_logs os.makedirs(LOG_DIR, exist_okTrue) def save_conversation_log(user_input: str, model_response: str, metadata: Dict[str, Any]): 保存单次对话的日志 # 生成日志文件名按日期 date_str datetime.datetime.now().strftime(%Y-%m-%d) log_file os.path.join(LOG_DIR, fconversations_{date_str}.jsonl) # 构建日志条目 log_entry { timestamp: datetime.datetime.now().isoformat(), user_input: user_input, model_response: model_response, metadata: metadata } # 追加到日志文件 with open(log_file, a, encodingutf-8) as f: f.write(json.dumps(log_entry, ensure_asciiFalse) \n) return log_entry cl.on_chat_start async def start_chat(): 聊天开始时的初始化 await cl.Message( content你好我是基于Nanbeige4.1-3B模型的智能助手。我们的对话会被匿名记录用于分析改进请问有什么可以帮您 ).send() cl.on_message async def main(message: cl.Message): 处理用户消息 # 这里应该是调用Nanbeige4.1-3B模型的代码 # 为了演示我们先模拟一个响应 user_input message.content # 模拟模型响应实际使用时替换为真正的模型调用 model_response f这是对{user_input}的模拟回答。在实际部署中这里会调用Nanbeige4.1-3B模型。 # 记录元数据 metadata { session_id: cl.user_session.get(id), message_id: message.id, response_time: 0.5, # 模拟响应时间秒 token_count: len(model_response.split()), # 模拟token数量 model_name: Nanbeige4.1-3B } # 保存日志 save_conversation_log(user_input, model_response, metadata) # 发送响应 await cl.Message( contentmodel_response ).send() if __name__ __main__: # 启动Chainlit应用 cl.run()这个基础版本已经可以记录每次对话了。日志会按日期保存在chat_logs目录下格式是JSON Lines每行一个JSON对象方便后续处理。3.2 增强日志记录基础日志只能记录对话内容但为了更好的分析我们还需要更多信息。让我们增强一下日志功能import time from enum import Enum from dataclasses import dataclass, asdict import hashlib class ResponseQuality(Enum): 响应质量评级 EXCELLENT excellent # 优秀准确、完整、有帮助 GOOD good # 良好基本正确但可以更好 FAIR fair # 一般部分正确或有瑕疵 POOR poor # 差错误或不相关 dataclass class EnhancedLogEntry: 增强的日志条目 timestamp: str session_id: str message_id: str # 用户输入相关 user_input: str input_length: int input_category: str None # 问题分类技术、闲聊、咨询等 # 模型响应相关 model_response: str response_length: int response_time: float # 响应时间秒 token_count: int # 质量评估 quality_rating: str None confidence_score: float None # 置信度评分0-1 has_safety_issue: bool False # 系统信息 model_name: str model_version: str 4.1-3B def to_dict(self): return asdict(self) def analyze_input_category(user_input: str) - str: 分析用户输入的分类 input_lower user_input.lower() # 简单的关键词匹配实际中可以更复杂 tech_keywords [代码, 编程, bug, 错误, 部署, 安装, 配置] general_keywords [你好, 谢谢, 再见, 帮助, 介绍] creative_keywords [写诗, 故事, 创意, 想象, 如果] for keyword in tech_keywords: if keyword in input_lower: return technical for keyword in creative_keywords: if keyword in input_lower: return creative for keyword in general_keywords: if keyword in input_lower: return general return other def estimate_quality(user_input: str, response: str) - ResponseQuality: 估计响应质量简化版实际中可以用更复杂的逻辑 # 检查响应是否为空 if not response or len(response.strip()) 5: return ResponseQuality.POOR # 检查响应是否包含常见错误指示 error_indicators [我不知道, 无法回答, 抱歉, 对不起我] for indicator in error_indicators: if indicator in response: return ResponseQuality.FAIR # 检查响应长度是否合适 input_words len(user_input.split()) response_words len(response.split()) if response_words input_words * 0.5: # 响应太短 return ResponseQuality.FAIR elif response_words input_words * 5: # 响应太长 return ResponseQuality.GOOD return ResponseQuality.GOOD cl.on_message async def enhanced_message_handler(message: cl.Message): 增强的消息处理器 start_time time.time() user_input message.content # 1. 调用模型这里用模拟响应 # 实际部署时替换为response call_nanbeige_model(user_input) model_response f模拟响应{user_input} end_time time.time() response_time end_time - start_time # 2. 分析输入和响应 input_category analyze_input_category(user_input) quality estimate_quality(user_input, model_response) # 3. 创建增强日志条目 log_entry EnhancedLogEntry( timestampdatetime.datetime.now().isoformat(), session_idstr(cl.user_session.get(id, unknown)), message_idmessage.id, user_inputuser_input, input_lengthlen(user_input), input_categoryinput_category, model_responsemodel_response, response_lengthlen(model_response), response_timeresponse_time, token_countlen(model_response.split()), quality_ratingquality.value, confidence_score0.8, # 模拟置信度 has_safety_issueFalse, model_nameNanbeige4.1-3B ) # 4. 保存日志 save_enhanced_log(log_entry) # 5. 发送响应 await cl.Message(contentmodel_response).send() def save_enhanced_log(log_entry: EnhancedLogEntry): 保存增强日志 date_str datetime.datetime.now().strftime(%Y-%m-%d) log_file os.path.join(LOG_DIR, fenhanced_logs_{date_str}.jsonl) with open(log_file, a, encodingutf-8) as f: f.write(json.dumps(log_entry.to_dict(), ensure_asciiFalse) \n)现在我们的日志系统强大多了它可以记录用户输入的分类技术问题、闲聊等响应质量评级响应时间更多元数据4. 日志分析与可视化光有日志还不够我们还需要工具来分析这些日志。让我们创建一个分析脚本# analyze_logs.py import json import pandas as pd import matplotlib.pyplot as plt from datetime import datetime, timedelta import seaborn as sns from collections import Counter import os class LogAnalyzer: 日志分析器 def __init__(self, log_dir./chat_logs): self.log_dir log_dir self.df None def load_logs(self, days_back7): 加载最近N天的日志 all_logs [] for i in range(days_back): date datetime.now() - timedelta(daysi) date_str date.strftime(%Y-%m-%d) log_file os.path.join(self.log_dir, fenhanced_logs_{date_str}.jsonl) if os.path.exists(log_file): with open(log_file, r, encodingutf-8) as f: for line in f: try: log_entry json.loads(line.strip()) all_logs.append(log_entry) except json.JSONDecodeError: continue if all_logs: self.df pd.DataFrame(all_logs) print(f成功加载 {len(all_logs)} 条日志记录) else: print(未找到日志文件) def basic_statistics(self): 基础统计信息 if self.df is None or self.df.empty: print(没有可分析的数据) return print( 基础统计 ) print(f总对话数: {len(self.df)}) print(f时间范围: {self.df[timestamp].min()} 到 {self.df[timestamp].max()}) print(f平均响应时间: {self.df[response_time].mean():.2f}秒) print(f平均响应长度: {self.df[response_length].mean():.0f}字符) # 质量分布 if quality_rating in self.df.columns: quality_counts self.df[quality_rating].value_counts() print(\n响应质量分布:) for quality, count in quality_counts.items(): percentage count / len(self.df) * 100 print(f {quality}: {count}次 ({percentage:.1f}%)) def analyze_user_behavior(self): 分析用户行为 if self.df is None or self.df.empty: return print(\n 用户行为分析 ) # 1. 问题类型分布 if input_category in self.df.columns: category_counts self.df[input_category].value_counts() print(问题类型分布:) for category, count in category_counts.items(): print(f {category}: {count}次) # 2. 热门问题 print(\n最常被问的问题前10:) top_questions self.df[user_input].value_counts().head(10) for i, (question, count) in enumerate(top_questions.items(), 1): print(f {i}. {question[:50]}... ({count}次)) # 3. 对话时间分布 self.df[hour] pd.to_datetime(self.df[timestamp]).dt.hour hourly_counts self.df[hour].value_counts().sort_index() print(\n每小时对话量:) for hour in range(24): count hourly_counts.get(hour, 0) print(f {hour:02d}:00 - {count:3d}次) def analyze_response_quality(self): 分析响应质量 if self.df is None or self.df.empty: return print(\n 响应质量分析 ) # 1. 质量与响应时间的关系 if quality_rating in self.df.columns and response_time in self.df.columns: quality_time self.df.groupby(quality_rating)[response_time].agg([mean, std, count]) print(不同质量等级的响应时间:) print(quality_time) # 2. 质量与问题类型的关系 if quality_rating in self.df.columns and input_category in self.df.columns: quality_by_category pd.crosstab( self.df[input_category], self.df[quality_rating], normalizeindex ) print(\n各问题类型的质量分布:) print(quality_by_category) # 3. 找出低质量响应 if quality_rating in self.df.columns: poor_responses self.df[self.df[quality_rating] poor] if not poor_responses.empty: print(f\n发现 {len(poor_responses)} 条低质量响应:) for _, row in poor_responses.head(5).iterrows(): print(f 问题: {row[user_input][:50]}...) print(f 回答: {row[model_response][:50]}...) print() def generate_visualizations(self, output_dir./analysis_output): 生成可视化图表 if self.df is None or self.df.empty: return os.makedirs(output_dir, exist_okTrue) # 设置中文字体如果需要 plt.rcParams[font.sans-serif] [SimHei, DejaVu Sans] plt.rcParams[axes.unicode_minus] False # 1. 对话时间分布图 plt.figure(figsize(12, 8)) # 子图1每小时对话量 plt.subplot(2, 2, 1) hourly_counts self.df[hour].value_counts().sort_index() plt.bar(hourly_counts.index, hourly_counts.values) plt.xlabel(小时) plt.ylabel(对话数量) plt.title(24小时对话分布) plt.grid(True, alpha0.3) # 子图2响应质量分布 if quality_rating in self.df.columns: plt.subplot(2, 2, 2) quality_counts self.df[quality_rating].value_counts() colors [green, lightgreen, orange, red] plt.pie(quality_counts.values, labelsquality_counts.index, autopct%1.1f%%, colorscolors[:len(quality_counts)]) plt.title(响应质量分布) # 子图3响应时间分布 plt.subplot(2, 2, 3) plt.hist(self.df[response_time], bins30, alpha0.7, edgecolorblack) plt.xlabel(响应时间秒) plt.ylabel(频次) plt.title(响应时间分布) plt.grid(True, alpha0.3) # 子图4问题类型分布 if input_category in self.df.columns: plt.subplot(2, 2, 4) category_counts self.df[input_category].value_counts() plt.barh(range(len(category_counts)), category_counts.values) plt.yticks(range(len(category_counts)), category_counts.index) plt.xlabel(数量) plt.title(问题类型分布) plt.tight_layout() plt.savefig(os.path.join(output_dir, conversation_analysis.png), dpi150, bbox_inchestight) plt.close() print(f可视化图表已保存到: {output_dir}/conversation_analysis.png) def generate_daily_report(self): 生成日报 if self.df is None or self.df.empty: return today datetime.now().strftime(%Y-%m-%d) report_file f./analysis_output/daily_report_{today}.md with open(report_file, w, encodingutf-8) as f: f.write(f# 对话系统日报 - {today}\n\n) # 基础统计 f.write(## 1. 基础统计\n) f.write(f- 总对话数: {len(self.df)}\n) f.write(f- 平均响应时间: {self.df[response_time].mean():.2f}秒\n) f.write(f- 平均响应长度: {self.df[response_length].mean():.0f}字符\n\n) # 质量分析 if quality_rating in self.df.columns: f.write(## 2. 响应质量\n) quality_counts self.df[quality_rating].value_counts() for quality, count in quality_counts.items(): percentage count / len(self.df) * 100 f.write(f- {quality}: {count}次 ({percentage:.1f}%)\n) f.write(\n) # 热门问题 f.write(## 3. 热门问题\n) top_questions self.df[user_input].value_counts().head(5) for i, (question, count) in enumerate(top_questions.items(), 1): f.write(f{i}. {question[:60]}... ({count}次)\n) f.write(\n) # 建议 f.write(## 4. 改进建议\n) # 检查低质量响应 if quality_rating in self.df.columns: poor_count len(self.df[self.df[quality_rating] poor]) if poor_count 0: f.write(f- 发现{poor_count}条低质量响应建议检查相关对话\n) # 检查响应时间 slow_threshold 2.0 # 2秒阈值 slow_count len(self.df[self.df[response_time] slow_threshold]) if slow_count 0: percentage slow_count / len(self.df) * 100 f.write(f- {slow_count}条对话响应时间超过{slow_threshold}秒 ({percentage:.1f}%)\n) f.write(\n---\n) f.write(*报告生成时间: datetime.now().strftime(%Y-%m-%d %H:%M:%S) *\n) print(f日报已生成: {report_file}) # 使用示例 if __name__ __main__: analyzer LogAnalyzer() analyzer.load_logs(days_back3) # 加载最近3天的日志 if analyzer.df is not None and not analyzer.df.empty: analyzer.basic_statistics() analyzer.analyze_user_behavior() analyzer.analyze_response_quality() analyzer.generate_visualizations() analyzer.generate_daily_report() else: print(没有找到可分析的日志数据)这个分析工具可以帮我们查看基础统计对话数量、响应时间等分析用户行为什么类型的问题最多、什么时候最活跃评估响应质量找出回答不好的对话生成可视化图表直观看到数据分布自动生成日报每天的系统表现报告5. 高级功能实时监控与告警对于生产环境我们还需要实时监控。让我们添加一个简单的监控系统# monitor.py import time import threading from datetime import datetime import smtplib from email.mime.text import MIMEText from collections import deque class ConversationMonitor: 对话监控器 def __init__(self, alert_thresholdsNone): self.alert_thresholds alert_thresholds or { response_time: 5.0, # 响应时间超过5秒告警 error_rate: 0.1, # 错误率超过10%告警 poor_quality_rate: 0.2 # 低质量响应超过20%告警 } # 滑动窗口统计最近100条对话 self.recent_conversations deque(maxlen100) self.alerts_sent set() # 避免重复告警 def add_conversation(self, log_entry): 添加新的对话记录 self.recent_conversations.append(log_entry) # 检查是否需要告警 self.check_alerts() def check_alerts(self): 检查告警条件 if len(self.recent_conversations) 10: # 至少10条数据才检查 return # 计算统计指标 response_times [c.get(response_time, 0) for c in self.recent_conversations] avg_response_time sum(response_times) / len(response_times) # 检查响应时间 if avg_response_time self.alert_thresholds[response_time]: self.send_alert(f平均响应时间过高: {avg_response_time:.2f}秒) # 这里可以添加更多检查逻辑 # 比如错误率、低质量响应率等 def send_alert(self, message): 发送告警这里用打印模拟实际中可以发邮件、短信等 alert_key f{datetime.now().strftime(%Y-%m-%d %H)}:{message} if alert_key not in self.alerts_sent: print(f[ALERT] {datetime.now().strftime(%Y-%m-%d %H:%M:%S)} - {message}) self.alerts_sent.add(alert_key) # 实际部署中可以在这里发送邮件 # self.send_email_alert(message) # 在Chainlit应用中集成监控 monitor ConversationMonitor() cl.on_message async def monitored_message_handler(message: cl.Message): 带监控的消息处理器 start_time time.time() user_input message.content # 调用模型 model_response f响应: {user_input} end_time time.time() response_time end_time - start_time # 创建日志条目 log_entry { timestamp: datetime.now().isoformat(), user_input: user_input, model_response: model_response, response_time: response_time, quality: good # 这里可以添加质量评估 } # 保存日志 save_conversation_log(user_input, model_response, {response_time: response_time}) # 添加到监控 monitor.add_conversation(log_entry) # 发送响应 await cl.Message(contentmodel_response).send() # 后台监控线程 def background_monitor(): 后台监控线程 while True: # 定期检查系统状态 time.sleep(60) # 每分钟检查一次 # 这里可以添加定期检查逻辑 # 比如检查日志文件大小、磁盘空间等 # 启动后台监控 monitor_thread threading.Thread(targetbackground_monitor, daemonTrue) monitor_thread.start()6. 总结通过上面的实现我们为Nanbeige4.1-3B Chainlit系统构建了一个完整的日志审计和分析系统。让我们回顾一下关键点6.1 我们实现了什么完整的日志记录不仅记录对话内容还记录响应时间、质量评级、问题分类等元数据强大的分析工具可以分析用户行为模式、评估响应质量、发现系统问题可视化展示自动生成图表直观展示系统运行状况实时监控及时发现异常情况并告警自动报告每天生成系统运行报告6.2 这个系统能帮你做什么了解用户需求知道用户最关心什么问题什么时间段最活跃评估模型表现客观评估Nanbeige4.1-3B在不同类型问题上的表现发现改进点找出模型回答不好的地方针对性优化监控系统健康实时了解系统运行状态及时发现问题数据驱动决策基于真实数据做优化决策而不是凭感觉6.3 下一步可以做什么如果你想让这个系统更强大可以考虑添加更智能的质量评估使用另一个AI模型来自动评估响应质量集成用户反馈让用户给回答打分收集直接反馈A/B测试支持对比不同提示词或模型版本的效果异常检测自动检测异常对话模式比如恶意攻击个性化分析分析不同用户群体的使用模式日志审计可能听起来有点技术性但它其实是了解用户、改进系统的最直接方式。通过数据你可以真正知道用户需要什么模型表现如何以及哪里可以做得更好。希望这个指南对你有所帮助如果你在实际使用中遇到问题或者有更好的想法欢迎分享交流。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。