青岛网站有限公司许昌注册公司

张

张建站

2026/4/11 12:46:58

10分钟阅读

青岛网站有限公司,许昌注册公司,wordpress建站视屏,php 缺少 wordpressGLM-4-9B-Chat-1M Chainlit企业定制#xff1a;SSO登录、审计日志、操作水印集成 1. 引言#xff1a;当大模型走进企业#xff0c;安全合规成为第一道门槛想象一下这个场景#xff1a;你的公司刚刚部署了最新的GLM-4-9B-Chat-1M大模型#xff0c;支持128K上下文#x…GLM-4-9B-Chat-1M Chainlit企业定制SSO登录、审计日志、操作水印集成1. 引言当大模型走进企业安全合规成为第一道门槛想象一下这个场景你的公司刚刚部署了最新的GLM-4-9B-Chat-1M大模型支持128K上下文甚至能处理1M的超长文本。技术团队兴奋地准备用它来提升工作效率——写代码、分析文档、生成报告。但IT安全部门的同事马上提出了几个问题“这个系统怎么管理用户登录能和我们公司的统一身份认证系统对接吗” “员工使用模型的所有操作有记录吗万一出了问题怎么追溯” “生成的内容会不会被截图外传能不能加上防泄漏的水印”这些问题不是杞人忧天而是企业级应用必须面对的“灵魂拷问”。今天我们就来聊聊如何给基于vLLM部署的GLM-4-9B-Chat-1M模型加上Chainlit前端然后深度定制企业级功能——SSO单点登录、完整的审计日志、还有防泄漏的操作水印。通过这篇文章你将学会如何把一个“裸奔”的大模型服务包装成符合企业安全规范的智能助手。无论你是企业开发者、IT管理员还是对AI应用安全感兴趣的工程师都能找到实用的解决方案。2. 基础环境搭建从零部署GLM-4-9B-Chat-1M在开始企业级定制之前我们先确保基础服务正常运行。这里我们使用vLLM来部署GLM-4-9B-Chat-1M模型然后用Chainlit构建一个简单的前端界面。2.1 模型部署与验证首先我们需要确认模型服务已经成功启动。通过WebShell连接到你的服务器检查部署状态# 查看模型服务日志 cat /root/workspace/llm.log如果看到类似下面的输出说明模型已经加载成功INFO 07-15 10:30:25 llm_engine.py:73] Initializing an LLM engine with config: modelTHUDM/glm-4-9b-chat-1m, tokenizerTHUDM/glm-4-9b-chat-1m, tokenizer_modeauto, trust_remote_codeTrue, dtypetorch.float16, ... INFO 07-15 10:32:10 llm_engine.py:145] GPU memory usage: 18.2 GB / 24.0 GB INFO 07-15 10:32:15 llm_engine.py:152] Model loaded successfully. Ready for inference.关键点GLM-4-9B-Chat-1M模型需要约18-20GB的GPU显存。如果你的显存不足可以考虑使用量化版本或者在vLLM配置中启用paged attention来优化内存使用。2.2 Chainlit前端基础配置Chainlit是一个专门为AI应用设计的聊天界面框架比直接调用API友好得多。基础配置很简单# chainlit_app.py import chainlit as cl from openai import OpenAI # 配置vLLM服务的API端点 client OpenAI( api_keytoken-abc123, # vLLM的默认token base_urlhttp://localhost:8000/v1 # vLLM默认端口 ) cl.on_message async def main(message: cl.Message): # 发送用户消息到GLM模型 response client.chat.completions.create( modelglm-4-9b-chat-1m, messages[ {role: system, content: 你是一个有帮助的AI助手。}, {role: user, content: message.content} ], temperature0.7, max_tokens2048 ) # 返回模型回复 await cl.Message( contentresponse.choices[0].message.content ).send()启动Chainlit服务chainlit run chainlit_app.py -w打开浏览器访问http://localhost:8000就能看到一个干净的聊天界面。输入问题模型会给出回复——这是最基础的功能。但问题来了任何人都能访问这个界面没有任何身份验证用户做了什么操作我们完全不知道生成的内容可以随意复制传播。这显然不符合企业使用标准。3. 企业级功能一SSO单点登录集成单点登录SSO是企业身份管理的基石。员工用一个账号密码通常是公司邮箱就能登录所有内部系统不用记一堆不同的密码。我们给Chainlit加上SSO支持。3.1 基于OAuth 2.0的SSO实现大多数企业的SSO都支持OAuth 2.0或SAML协议。这里我们以OAuth 2.0为例集成微软Azure AD或Google Workspace# auth_sso.py import os from authlib.integrations.starlette_client import OAuth from starlette.middleware.sessions import SessionMiddleware from starlette.config import Config import chainlit as cl # 配置OAuth客户端 config Config(.env) # 从环境变量读取配置 oauth OAuth(config) # 注册Azure AD提供商 oauth.register( nameazure, client_idos.getenv(AZURE_CLIENT_ID), client_secretos.getenv(AZURE_CLIENT_SECRET), server_metadata_urlhttps://login.microsoftonline.com/{tenant}/v2.0/.well-known/openid-configuration, client_kwargs{scope: openid email profile} ) cl.on_chat_start async def on_chat_start(): # 检查用户是否已登录 user cl.user_session.get(user) if not user: # 未登录重定向到SSO登录页面 redirect_uri http://localhost:8000/auth/callback return await oauth.azure.authorize_redirect(request, redirect_uri) cl.auth_callback async def auth_callback(access_token): # 使用access_token获取用户信息 userinfo await oauth.azure.parse_id_token(request, access_token) # 提取企业相关信息 user_email userinfo[email] user_name userinfo.get(name, ) department userinfo.get(department, 未知部门) # 检查用户是否有权限访问可选 allowed_domains [company.com, partner.com] if not any(user_email.endswith(domain) for domain in allowed_domains): raise cl.AuthError(该邮箱域名无权访问系统) # 将用户信息存入session cl.user_session.set(user, { email: user_email, name: user_name, department: department, login_time: datetime.now().isoformat() }) return True3.2 本地用户数据库同步对于没有标准SSO的小型企业或者需要额外权限管理的情况可以结合本地数据库# auth_local.py import sqlite3 from datetime import datetime import hashlib import chainlit as cl def init_user_db(): 初始化用户数据库 conn sqlite3.connect(users.db) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY, email TEXT UNIQUE, name TEXT, department TEXT, role TEXT DEFAULT user, created_at TIMESTAMP, last_login TIMESTAMP ) ) conn.commit() conn.close() cl.password_auth_callback def auth_callback(username: str, password: str): # 简单的密码验证生产环境请使用加盐哈希 conn sqlite3.connect(users.db) cursor conn.cursor() cursor.execute( SELECT email, name, department, role FROM users WHERE email ? AND password ?, (username, hashlib.sha256(password.encode()).hexdigest()) ) user cursor.fetchone() conn.close() if user: return cl.User( identifieruser[0], metadata{ name: user[1], department: user[2], role: user[3], login_time: datetime.now().isoformat() } ) return None实际效果用户首次访问Chainlit界面时会被重定向到公司的登录页面如微软、谷歌或自定义登录页。登录成功后自动跳回AI助手界面并且系统知道当前用户是谁、来自哪个部门。管理员可以在后台管理用户权限控制谁能访问系统。4. 企业级功能二完整的审计日志系统用户登录只是第一步。企业需要知道谁在什么时候问了什么问题模型回答了什幺有没有敏感信息泄露风险。这就是审计日志的作用。4.1 审计日志数据库设计我们需要记录每一次交互的完整信息# audit_logger.py import sqlite3 import json from datetime import datetime from typing import Dict, Any import chainlit as cl class AuditLogger: def __init__(self, db_pathaudit.db): self.db_path db_path self._init_db() def _init_db(self): 初始化审计日志数据库 conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS chat_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT, user_email TEXT, user_department TEXT, timestamp TIMESTAMP, user_input TEXT, model_response TEXT, input_tokens INTEGER, output_tokens INTEGER, total_tokens INTEGER, response_time_ms INTEGER, model_name TEXT, temperature REAL, max_tokens INTEGER, ip_address TEXT, user_agent TEXT ) ) cursor.execute( CREATE TABLE IF NOT EXISTS system_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TIMESTAMP, log_level TEXT, module TEXT, message TEXT, user_email TEXT, extra_data TEXT ) ) conn.commit() conn.close() def log_chat(self, session_id: str, user_info: Dict, user_input: str, model_response: str, token_usage: Dict, response_time: int, request_headers: Dict): 记录聊天交互日志 conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute( INSERT INTO chat_logs ( session_id, user_email, user_department, timestamp, user_input, model_response, input_tokens, output_tokens, total_tokens, response_time_ms, model_name, temperature, max_tokens, ip_address, user_agent ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) , ( session_id, user_info.get(email, unknown), user_info.get(department, unknown), datetime.now().isoformat(), user_input, model_response, token_usage.get(prompt_tokens, 0), token_usage.get(completion_tokens, 0), token_usage.get(total_tokens, 0), response_time, glm-4-9b-chat-1m, 0.7, # 实际使用的temperature 2048, # 实际使用的max_tokens request_headers.get(X-Forwarded-For, unknown), request_headers.get(User-Agent, unknown) )) conn.commit() conn.close() def log_system_event(self, level: str, module: str, message: str, user_email: str None, extra_data: Dict None): 记录系统事件日志 conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute( INSERT INTO system_logs ( timestamp, log_level, module, message, user_email, extra_data ) VALUES (?, ?, ?, ?, ?, ?) , ( datetime.now().isoformat(), level, module, message, user_email, json.dumps(extra_data) if extra_data else None )) conn.commit() conn.close() # 全局审计日志实例 audit_logger AuditLogger()4.2 集成到Chainlit消息处理修改之前的消息处理函数加入审计日志# chainlit_with_audit.py import chainlit as cl from openai import OpenAI import time from audit_logger import audit_logger client OpenAI( api_keytoken-abc123, base_urlhttp://localhost:8000/v1 ) cl.on_message async def main(message: cl.Message): # 获取用户信息 user_info cl.user_session.get(user, {}) session_id cl.user_session.get(id, unknown) # 记录请求开始时间 start_time time.time() try: # 调用GLM模型 response client.chat.completions.create( modelglm-4-9b-chat-1m, messages[ {role: system, content: 你是一个有帮助的AI助手。}, {role: user, content: message.content} ], temperature0.7, max_tokens2048 ) # 计算响应时间 response_time int((time.time() - start_time) * 1000) # 获取模型回复 model_response response.choices[0].message.content # 记录审计日志 audit_logger.log_chat( session_idsession_id, user_infouser_info, user_inputmessage.content, model_responsemodel_response, token_usageresponse.usage.dict() if response.usage else {}, response_timeresponse_time, request_headersdict(message.headers) if hasattr(message, headers) else {} ) # 记录系统日志 audit_logger.log_system_event( levelINFO, modulechat, messagef用户 {user_info.get(email, unknown)} 完成一次对话, user_emailuser_info.get(email), extra_data{ input_length: len(message.content), response_length: len(model_response), response_time_ms: response_time } ) # 返回回复 await cl.Message(contentmodel_response).send() except Exception as e: # 记录错误日志 audit_logger.log_system_event( levelERROR, modulechat, messagef模型调用失败: {str(e)}, user_emailuser_info.get(email), extra_data{error: str(e), user_input: message.content} ) await cl.Message(content抱歉处理您的请求时出现了错误。).send()4.3 审计日志查询与分析有了完整的日志数据我们可以构建管理界面来查询和分析# audit_query.py from flask import Flask, request, jsonify import sqlite3 from datetime import datetime, timedelta app Flask(__name__) app.route(/api/audit/logs, methods[GET]) def get_chat_logs(): 查询聊天日志 # 获取查询参数 user_email request.args.get(user_email) start_date request.args.get(start_date) end_date request.args.get(end_date) department request.args.get(department) limit int(request.args.get(limit, 100)) offset int(request.args.get(offset, 0)) conn sqlite3.connect(audit.db) conn.row_factory sqlite3.Row # 返回字典格式 cursor conn.cursor() # 构建查询条件 conditions [] params [] if user_email: conditions.append(user_email ?) params.append(user_email) if start_date: conditions.append(timestamp ?) params.append(start_date) if end_date: conditions.append(timestamp ?) params.append(end_date) if department: conditions.append(user_department ?) params.append(department) where_clause AND .join(conditions) if conditions else 11 # 执行查询 query f SELECT * FROM chat_logs WHERE {where_clause} ORDER BY timestamp DESC LIMIT ? OFFSET ? params.extend([limit, offset]) cursor.execute(query, params) logs [dict(row) for row in cursor.fetchall()] # 获取总数 count_query fSELECT COUNT(*) as total FROM chat_logs WHERE {where_clause} cursor.execute(count_query, params[:-2]) # 去掉LIMIT和OFFSET参数 total cursor.fetchone()[total] conn.close() return jsonify({ logs: logs, total: total, limit: limit, offset: offset }) app.route(/api/audit/stats, methods[GET]) def get_usage_stats(): 获取使用统计 conn sqlite3.connect(audit.db) cursor conn.cursor() # 今日使用统计 today datetime.now().date().isoformat() cursor.execute( SELECT COUNT(*) as today_sessions, SUM(total_tokens) as today_tokens, COUNT(DISTINCT user_email) as today_users FROM chat_logs WHERE DATE(timestamp) ? , (today,)) today_stats cursor.fetchone() # 部门使用排名 cursor.execute( SELECT user_department, COUNT(*) as session_count, SUM(total_tokens) as total_tokens, COUNT(DISTINCT user_email) as user_count FROM chat_logs WHERE timestamp DATE(now, -30 days) GROUP BY user_department ORDER BY session_count DESC ) dept_stats cursor.fetchall() # 热门问题分析 cursor.execute( SELECT user_input, COUNT(*) as ask_count FROM chat_logs WHERE timestamp DATE(now, -7 days) GROUP BY user_input ORDER BY ask_count DESC LIMIT 10 ) hot_questions cursor.fetchall() conn.close() return jsonify({ today: { sessions: today_stats[0] or 0, tokens: today_stats[1] or 0, users: today_stats[2] or 0 }, department_ranking: [ { department: row[0], sessions: row[1], tokens: row[2], users: row[3] } for row in dept_stats ], hot_questions: [ {question: row[0], count: row[1]} for row in hot_questions ] })实际效果现在管理员可以随时查看谁在使用系统、问了什么问题、模型回答了什么。如果发现敏感信息泄露比如有人问“公司的财务数据是什么”可以立即追溯并采取措施。统计功能还能帮助了解各部门的使用情况为资源分配提供依据。5. 企业级功能三防泄漏操作水印审计日志记录了行为但无法防止用户截图外传。操作水印就是在界面上叠加半透明的用户信息让截图也能追溯到源头。5.1 动态水印生成与叠加Chainlit本身不支持水印但我们可以通过自定义CSS和JavaScript来实现# watermark_integration.py import chainlit as cl import hashlib from datetime import datetime def generate_watermark_text(user_info: dict) - str: 生成水印文本 user_email user_info.get(email, unknownexample.com) user_name user_info.get(name, Unknown User) timestamp datetime.now().strftime(%Y%m%d%H%M) # 生成唯一标识避免太明显 unique_id hashlib.md5(f{user_email}{timestamp}.encode()).hexdigest()[:8] # 水印文本格式 return f{user_name} • {user_email} • {unique_id} cl.on_chat_start async def on_chat_start(): # 获取用户信息 user_info cl.user_session.get(user, {}) # 生成水印文本 watermark_text generate_watermark_text(user_info) # 注入自定义CSS和JavaScript custom_css f style .watermark-overlay {{ position: fixed; top: 0; left: 0; width: 100%; height: 100%; pointer-events: none; z-index: 9999; opacity: 0.03; font-size: 20px; color: #333; background: repeating-linear-gradient( 45deg, transparent, transparent 100px, rgba(0,0,0,0.05) 100px, rgba(0,0,0,0.05) 200px ); }} .watermark-text {{ position: absolute; transform: rotate(-30deg); white-space: nowrap; font-family: Arial, sans-serif; font-weight: bold; }} /style custom_js f script function createWatermark() {{ const overlay document.createElement(div); overlay.className watermark-overlay; // 创建重复的水印文本 const text {watermark_text}; const container document.querySelector(.chainlit-chat-page); if (container) {{ for (let i 0; i 50; i) {{ const watermark document.createElement(div); watermark.className watermark-text; watermark.textContent text; watermark.style.left (Math.random() * 100) %; watermark.style.top (Math.random() * 100) %; overlay.appendChild(watermark); }} container.appendChild(overlay); }} }} // 页面加载完成后创建水印 if (document.readyState loading) {{ document.addEventListener(DOMContentLoaded, createWatermark); }} else {{ createWatermark(); }} // 防止右键查看源码基础防护 document.addEventListener(contextmenu, function(e) {{ e.preventDefault(); return false; }}); // 防止F12开发者工具基础防护 document.onkeydown function(e) {{ if (e.keyCode 123) {{ // F12 return false; }} if (e.ctrlKey e.shiftKey e.keyCode I.charCodeAt(0)) {{ // CtrlShiftI return false; }} if (e.ctrlKey e.keyCode U.charCodeAt(0)) {{ // CtrlU return false; }} }}; /script # 发送自定义元素到前端 await cl.Message( content, elements[ cl.Html(namewatermark_css, htmlcustom_css, displayhidden), cl.Html(namewatermark_js, htmlcustom_js, displayhidden) ] ).send()5.2 响应内容水印除了界面水印我们还可以在模型回复的内容中嵌入隐形水印# content_watermark.py import base64 import json def embed_content_watermark(text: str, user_info: dict) - str: 在文本内容中嵌入隐形水印 watermark_data { user_id: user_info.get(email, unknown), timestamp: datetime.now().isoformat(), session_id: cl.user_session.get(id, unknown) } # 将水印数据编码为Base64 watermark_json json.dumps(watermark_data) watermark_b64 base64.b64encode(watermark_json.encode()).decode() # 将水印嵌入到文本中使用零宽字符 # 零宽字符不可见但会保留在文本中 zero_width_joiner \u200d zero_width_non_joiner \u200c # 将Base64字符串转换为零宽字符序列 watermark_chars [] for char in watermark_b64: if char.isalpha(): # 使用不同的零宽字符表示不同字符 watermark_chars.append(zero_width_joiner if ord(char) % 2 0 else zero_width_non_joiner) else: watermark_chars.append(\u200b) # 零宽空格 watermark_sequence .join(watermark_chars) # 在文本开头和结尾嵌入水印 watermarked_text f{watermark_sequence}{text}{watermark_sequence} return watermarked_text def extract_content_watermark(text: str) - dict: 从文本中提取水印信息 # 查找零宽字符 zero_width_chars [\u200b, \u200c, \u200d] # 提取开头的水印序列 watermark_start 0 for i, char in enumerate(text): if char not in zero_width_chars: watermark_start i break # 提取结尾的水印序列 watermark_end len(text) for i in range(len(text)-1, -1, -1): if text[i] not in zero_width_chars: watermark_end i 1 break if watermark_start 0 or watermark_end len(text): return None # 解码水印 watermark_sequence text[:watermark_start] # 将零宽字符转换回Base64 base64_chars [] for char in watermark_sequence: if char \u200d: base64_chars.append(A) # 简化示例实际需要完整映射 elif char \u200c: base64_chars.append(B) elif char \u200b: base64_chars.append(C) watermark_b64 .join(base64_chars) try: watermark_json base64.b64decode(watermark_b64).decode() return json.loads(watermark_json) except: return None5.3 集成水印到消息处理更新消息处理函数加入内容水印cl.on_message async def main_with_watermark(message: cl.Message): # 获取用户信息 user_info cl.user_session.get(user, {}) # 调用模型获取回复 response client.chat.completions.create( modelglm-4-9b-chat-1m, messages[ {role: system, content: 你是一个有帮助的AI助手。}, {role: user, content: message.content} ], temperature0.7, max_tokens2048 ) model_response response.choices[0].message.content # 嵌入内容水印 watermarked_response embed_content_watermark(model_response, user_info) # 发送带水印的回复 await cl.Message(contentwatermarked_response).send() # 记录审计日志包含水印信息 audit_logger.log_chat( session_idcl.user_session.get(id), user_infouser_info, user_inputmessage.content, model_responsemodel_response, # 原始回复 token_usageresponse.usage.dict() if response.usage else {}, response_timeint((time.time() - start_time) * 1000), request_headersdict(message.headers) if hasattr(message, headers) else {} )实际效果现在用户界面上布满了半透明的用户信息水印无论怎么截图都能看到是谁的屏幕。即使有人复制文本内容里面也嵌入了隐形水印可以通过专用工具提取溯源。这大大降低了敏感信息泄露的风险。6. 完整集成方案与部署建议我们把SSO登录、审计日志、操作水印三个功能整合到一起形成一个完整的企业级解决方案。6.1 配置文件管理创建统一的配置文件方便管理# config.yaml app: name: 企业AI助手-GLM4定制版 version: 1.0.0 debug: false auth: sso_enabled: true sso_provider: azure # azure, google, okta, custom local_auth_fallback: true allowed_domains: - company.com - partner.com azure: tenant_id: ${AZURE_TENANT_ID} client_id: ${AZURE_CLIENT_ID} client_secret: ${AZURE_CLIENT_SECRET} google: client_id: ${GOOGLE_CLIENT_ID} client_secret: ${GOOGLE_CLIENT_SECRET} audit: enabled: true database_path: /data/audit.db retention_days: 365 log_user_input: true log_model_response: true alerts: enabled: true sensitive_keywords: - 密码 - 密钥 - 财务数据 - 客户信息 alert_emails: - securitycompany.com watermark: enabled: true ui_watermark: true content_watermark: true opacity: 0.03 text_size: 20 protection: disable_right_click: true disable_devtools: true disable_print_screen: false model: name: glm-4-9b-chat-1m base_url: http://localhost:8000/v1 api_key: token-abc123 max_tokens: 2048 temperature: 0.7 timeout: 306.2 Docker部署配置使用Docker容器化部署确保环境一致性# Dockerfile FROM python:3.10-slim WORKDIR /app # 安装系统依赖 RUN apt-get update apt-get install -y \ gcc \ g \ rm -rf /var/lib/apt/lists/* # 复制依赖文件 COPY requirements.txt . # 安装Python依赖 RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 创建数据目录 RUN mkdir -p /data # 设置环境变量 ENV PYTHONPATH/app ENV CONFIG_PATH/app/config.yaml # 暴露端口 EXPOSE 8000 # 启动命令 CMD [chainlit, run, app/main.py, --host, 0.0.0.0, --port, 8000]# docker-compose.yaml version: 3.8 services: glm-ai-assistant: build: . container_name: glm-enterprise-assistant ports: - 8000:8000 volumes: - ./data:/data - ./logs:/app/logs environment: - AZURE_TENANT_ID${AZURE_TENANT_ID} - AZURE_CLIENT_ID${AZURE_CLIENT_ID} - AZURE_CLIENT_SECRET${AZURE_CLIENT_SECRET} - NODE_ENVproduction restart: unless-stopped networks: - ai-network vllm-backend: image: vllm/vllm-openai:latest container_name: vllm-glm4 ports: - 8001:8000 command: [ --model, THUDM/glm-4-9b-chat-1m, --tensor-parallel-size, 1, --gpu-memory-utilization, 0.9, --max-model-len, 131072, --served-model-name, glm-4-9b-chat-1m ] deploy: resources: reservations: devices: - driver: nvidia count: 1 capabilities: [gpu] restart: unless-stopped networks: - ai-network networks: ai-network: driver: bridge6.3 监控与告警添加监控指标确保系统稳定运行# monitoring.py from prometheus_client import start_http_server, Counter, Histogram, Gauge import time # 定义监控指标 REQUEST_COUNT Counter(chat_requests_total, Total chat requests) REQUEST_LATENCY Histogram(chat_request_latency_seconds, Request latency in seconds) ACTIVE_USERS Gauge(active_users_current, Number of currently active users) TOKEN_USAGE Counter(tokens_used_total, Total tokens used, [type]) def monitor_request(func): 监控装饰器 async def wrapper(*args, **kwargs): start_time time.time() # 增加请求计数 REQUEST_COUNT.inc() try: result await func(*args, **kwargs) return result finally: # 记录延迟 latency time.time() - start_time REQUEST_LATENCY.observe(latency) return wrapper # 在消息处理函数上使用装饰器 monitor_request cl.on_message async def monitored_main(message: cl.Message): # ... 原有逻辑 ... # 记录token使用 if response.usage: TOKEN_USAGE.labels(typeinput).inc(response.usage.prompt_tokens) TOKEN_USAGE.labels(typeoutput).inc(response.usage.completion_tokens) return response # 启动Prometheus指标服务器 start_http_server(9090)7. 总结通过今天的分享我们完成了一个GLM-4-9B-Chat-1M模型的企业级定制方案。从最基础的模型部署到SSO登录集成再到完整的审计日志系统最后加上防泄漏的操作水印——我们一步步把一个“裸奔”的AI服务包装成了符合企业安全规范的智能助手。7.1 关键收获身份安全是基础通过SSO集成确保只有授权员工能访问系统并且登录状态可管理行为可追溯是必须完整的审计日志记录了谁、在什么时候、问了什么、得到了什么回答满足合规要求内容防泄漏是保障界面水印和内容水印双重防护大大降低敏感信息泄露风险监控运维是支撑完善的监控告警系统确保服务稳定可用7.2 实际部署建议如果你要在自己的企业部署这套方案我建议第一阶段快速验证先部署基础的vLLM GLM-4-9B-Chat-1M用Chainlit搭建最简单的前端让技术团队试用收集反馈第二阶段安全加固加入SSO登录控制访问权限实现基础审计日志记录关键操作添加界面水印防止随意截图第三阶段全面合规完善审计日志加入敏感词检测实现内容水印文本级溯源建立监控告警确保服务稳定制定使用规范培训员工7.3 技术选型思考这套方案有几个关键选择值得思考为什么用Chainlit而不是自己写前端Chainlit专为AI对话设计开箱即用节省开发时间。而且它基于Python和我们后端的语言栈一致。为什么用SQLite而不是专业数据库对于中小型企业SQLite完全够用。如果数据量大可以轻松迁移到PostgreSQL或MySQL。水印安全级别够吗本文展示的是基础水印方案。对于金融、法律等超高安全要求的场景可以考虑更复杂的数字水印技术。GLM-4-9B-Chat-1M的替代选择如果对中文支持要求高GLM是很好的选择。如果需要更强的代码能力可以考虑CodeLlama如果需要多模态可以考虑Qwen-VL。企业级AI应用技术能力只是基础安全合规才是能否落地的关键。希望这套方案能帮助你在享受大模型带来的效率提升的同时守住安全的底线。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。