ChatGLM3-6B Streamlit定制API密钥管理速率限制使用统计1. 项目概述今天给大家分享一个超实用的本地AI对话系统定制方案。基于ChatGLM3-6B-32k模型和Streamlit框架我们打造了一个不仅速度快、稳定性好还具备企业级功能的管理系统。传统的AI对话工具往往缺少管理功能任何人都能随意使用既无法控制访问权限也无法统计使用情况。我们的方案解决了这些问题添加了API密钥管理、访问速率限制和使用统计三大核心功能。核心价值安全管控通过API密钥控制访问权限避免未授权使用资源保护设置速率限制防止系统被过度使用使用洞察详细统计使用情况了解AI使用模式和成本2. 环境准备与快速部署2.1 系统要求在开始之前请确保你的设备满足以下要求显卡RTX 4090D或同等级别显卡24GB显存以上内存32GB以上系统内存存储至少50GB可用空间系统Ubuntu 20.04或Windows 10/11 with WSL22.2 一键安装部署打开终端执行以下命令快速安装# 克隆项目仓库 git clone https://github.com/your-repo/chatglm3-streamlit.git cd chatglm3-streamlit # 创建虚拟环境 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装依赖版本已锁定确保稳定性 pip install -r requirements.txtrequirements.txt内容如下streamlit1.28.0 transformers4.40.2 torch2.1.0cu121 sentencepiece0.1.99 accelerate0.24.0 python-dotenv1.0.0 redis5.0.12.3 启动系统安装完成后一行命令启动服务streamlit run app.py --server.port 8501 --server.address 0.0.0.0启动成功后在浏览器访问http://localhost:8501即可看到管理界面。3. 核心功能实现详解3.1 API密钥管理系统API密钥是控制访问的第一道防线。我们实现了完整的密钥生命周期管理import secrets import json from datetime import datetime, timedelta class APIKeyManager: def __init__(self): self.keys_file api_keys.json self.load_keys() def generate_key(self, user_name, rate_limit10): 生成新的API密钥 api_key fsk-{secrets.token_urlsafe(32)} self.keys[api_key] { user_name: user_name, created_at: datetime.now().isoformat(), rate_limit: rate_limit, # 每分钟最大请求数 is_active: True, total_requests: 0 } self.save_keys() return api_key def validate_key(self, api_key): 验证API密钥有效性 if api_key in self.keys and self.keys[api_key][is_active]: return True return False def deactivate_key(self, api_key): 停用API密钥 if api_key in self.keys: self.keys[api_key][is_active] False self.save_keys() def load_keys(self): 加载密钥数据 try: with open(self.keys_file, r) as f: self.keys json.load(f) except FileNotFoundError: self.keys {} def save_keys(self): 保存密钥数据 with open(self.keys_file, w) as f: json.dump(self.keys, f, indent2)在实际使用中你可以这样管理密钥# 初始化密钥管理器 key_manager APIKeyManager() # 生成新密钥 new_key key_manager.generate_key(张三, rate_limit20) print(f新生成的API密钥: {new_key}) # 验证密钥 is_valid key_manager.validate_key(你的API密钥)3.2 智能速率限制系统为了防止系统被过度使用我们实现了基于令牌桶算法的速率限制import time from collections import defaultdict class RateLimiter: def __init__(self): self.token_buckets defaultdict(dict) def check_rate_limit(self, api_key, tokens_per_minute): 检查是否超过速率限制 now time.time() bucket self.token_buckets.get(api_key, {}) if not bucket: # 初始化令牌桶 self.token_buckets[api_key] { tokens: tokens_per_minute, last_refill: now } return True # 计算应补充的令牌数 time_passed now - bucket[last_refill] tokens_to_add (time_passed / 60) * tokens_per_minute new_tokens min(tokens_per_minute, bucket[tokens] tokens_to_add) if new_tokens 1: # 有可用令牌 self.token_buckets[api_key][tokens] new_tokens - 1 self.token_buckets[api_key][last_refill] now return True else: # 令牌不足 return False # 使用示例 limiter RateLimiter() api_key 你的API密钥 rate_limit 10 # 每分钟10次 if limiter.check_rate_limit(api_key, rate_limit): print(请求允许) else: print(请求过于频繁请稍后再试)3.3 使用统计与分析系统详细的使用统计能帮助你了解AI的使用情况和成本分布import pandas as pd from datetime import datetime class UsageTracker: def __init__(self): self.usage_file usage_stats.csv self.initialize_stats() def initialize_stats(self): 初始化统计文件 try: pd.read_csv(self.usage_file) except FileNotFoundError: df pd.DataFrame(columns[ timestamp, api_key, user_name, prompt_length, response_length, processing_time, model_used ]) df.to_csv(self.usage_file, indexFalse) def record_usage(self, api_key, user_name, prompt, response, processing_time): 记录使用情况 new_record { timestamp: datetime.now().isoformat(), api_key: api_key[:8] ..., # 只存储部分密钥保护隐私 user_name: user_name, prompt_length: len(prompt), response_length: len(response), processing_time: processing_time, model_used: ChatGLM3-6B-32k } # 追加到CSV文件 df pd.DataFrame([new_record]) df.to_csv(self.usage_file, modea, headerFalse, indexFalse) def get_usage_report(self, days7): 生成使用报告 df pd.read_csv(self.usage_file) df[timestamp] pd.to_datetime(df[timestamp]) # 过滤最近N天的数据 cutoff_date datetime.now() - timedelta(daysdays) recent_usage df[df[timestamp] cutoff_date] report { total_requests: len(recent_usage), total_tokens: recent_usage[response_length].sum() recent_usage[prompt_length].sum(), avg_processing_time: recent_usage[processing_time].mean(), top_users: recent_usage[user_name].value_counts().to_dict(), usage_by_hour: recent_usage.groupby(recent_usage[timestamp].dt.hour).size().to_dict() } return report4. 完整集成示例下面是如何将这些功能集成到Streamlit应用中的完整示例import streamlit as st import time from api_key_manager import APIKeyManager from rate_limiter import RateLimiter from usage_tracker import UsageTracker # 初始化组件 key_manager APIKeyManager() limiter RateLimiter() usage_tracker UsageTracker() # 设置页面标题 st.set_page_config(page_titleChatGLM3-6B 管理平台, layoutwide) # 侧边栏 - 管理功能 with st.sidebar: st.title( 管理面板) tab1, tab2, tab3 st.tabs([密钥管理, 速率限制, 使用统计]) with tab1: st.subheader(API密钥管理) user_name st.text_input(用户名) rate_limit st.slider(速率限制(次/分钟), 1, 100, 10) if st.button(生成新密钥): new_key key_manager.generate_key(user_name, rate_limit) st.success(f新密钥已生成: {new_key}) # 显示现有密钥 st.subheader(现有密钥) for key, info in key_manager.keys.items(): col1, col2 st.columns([3, 1]) with col1: st.text(f{info[user_name]}: {key[:10]}...{key[-10:]}) with col2: if st.button(停用, keyfdisable_{key}): key_manager.deactivate_key(key) st.rerun() with tab2: st.subheader(速率限制设置) # 这里可以显示当前的速率限制情况 with tab3: st.subheader(使用统计) if st.button(生成报告): report usage_tracker.get_usage_report(7) st.json(report) # 主界面 - 聊天功能 st.title( ChatGLM3-6B 智能对话) # API密钥输入 api_key st.text_input(请输入API密钥, typepassword) if api_key: if not key_manager.validate_key(api_key): st.error(无效的API密钥) st.stop() # 获取用户配置 user_config key_manager.keys[api_key] user_name user_config[user_name] rate_limit user_config[rate_limit] # 检查速率限制 if not limiter.check_rate_limit(api_key, rate_limit): st.warning(请求过于频繁请稍后再试) st.stop() # 聊天界面 if messages not in st.session_state: st.session_state.messages [] for message in st.session_state.messages: with st.chat_message(message[role]): st.markdown(message[content]) if prompt : st.chat_input(请输入您的问题): # 记录开始时间 start_time time.time() # 添加用户消息 st.session_state.messages.append({role: user, content: prompt}) with st.chat_message(user): st.markdown(prompt) # 调用模型生成回复 with st.chat_message(assistant): message_placeholder st.empty() full_response # 模拟流式输出 for chunk in generate_response(prompt): # 这是你的模型推理函数 full_response chunk message_placeholder.markdown(full_response ▌) message_placeholder.markdown(full_response) # 记录使用情况 processing_time time.time() - start_time usage_tracker.record_usage(api_key, user_name, prompt, full_response, processing_time) st.session_state.messages.append({role: assistant, content: full_response})5. 实际应用场景5.1 企业内部AI助手这个系统特别适合作为企业内部的AI助手平台部门协作为不同部门分配不同的API密钥和速率限制成本控制通过使用统计了解各部门的AI使用成本安全管理员工离职时只需停用其API密钥无需修改系统5.2 教育机构应用在教育场景中这个系统可以学生管理为每个学生分配独立的访问权限资源公平通过速率限制确保所有学生都能公平使用AI资源学习分析通过使用统计了解学生的学习模式和需求5.3 开发者API服务如果你希望将ChatGLM3-6B作为API服务提供给其他开发者多租户支持支持多个用户同时使用互不干扰用量计费基于使用统计实现按量计费服务保障通过速率限制保证服务稳定性6. 使用技巧与最佳实践6.1 密钥安全管理定期轮换建议每3-6个月更换一次API密钥最小权限根据实际需要设置速率限制不要过度授权访问日志记录所有API密钥的使用情况便于审计6.2 性能优化建议# 使用缓存提高性能 st.cache_resource def load_model(): 缓存模型加载避免重复初始化 from transformers import AutoModel, AutoTokenizer model AutoModel.from_pretrained(THUDM/chatglm3-6b-32k, trust_remote_codeTrue) tokenizer AutoTokenizer.from_pretrained(THUDM/chatglm3-6b-32k, trust_remote_codeTrue) return model, tokenizer # 在应用启动时加载模型 model, tokenizer load_model()6.3 监控与告警建议添加监控功能当出现异常时及时告警异常使用检测监控突然激增的使用量系统健康检查定期检查模型和服务状态自动备份定期备份密钥数据和使用统计7. 总结通过为ChatGLM3-6B Streamlit应用添加API密钥管理、速率限制和使用统计功能我们打造了一个真正企业级的AI对话平台。这个方案不仅保持了本地部署的隐私和安全优势还提供了云端服务级别的管理能力。主要收获完整权限控制通过API密钥实现精细化的访问管理资源保护机制智能速率限制防止系统被滥用使用洞察能力详细统计帮助了解AI使用情况和成本开箱即用所有代码都是可运行的可以直接集成到你的项目中这个方案的美妙之处在于它的灵活性——你可以根据实际需求调整速率限制策略、统计指标和管理功能。无论是小型团队还是大型企业都能找到合适的配置方案。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。