GLM-4-9B-Chat-1M Chainlit企业定制：SSO登录、审计日志、操作水印集成
·
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: model='THUDM/glm-4-9b-chat-1m', tokenizer='THUDM/glm-4-9b-chat-1m', tokenizer_mode=auto, trust_remote_code=True, dtype=torch.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_key="token-abc123", # vLLM的默认token
base_url="http://localhost:8000/v1" # vLLM默认端口
)
@cl.on_message
async def main(message: cl.Message):
# 发送用户消息到GLM模型
response = client.chat.completions.create(
model="glm-4-9b-chat-1m",
messages=[
{"role": "system", "content": "你是一个有帮助的AI助手。"},
{"role": "user", "content": message.content}
],
temperature=0.7,
max_tokens=2048
)
# 返回模型回复
await cl.Message(
content=response.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(
name='azure',
client_id=os.getenv('AZURE_CLIENT_ID'),
client_secret=os.getenv('AZURE_CLIENT_SECRET'),
server_metadata_url='https://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 True
3.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(
identifier=user[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_path='audit.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_key="token-abc123",
base_url="http://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(
model="glm-4-9b-chat-1m",
messages=[
{"role": "system", "content": "你是一个有帮助的AI助手。"},
{"role": "user", "content": message.content}
],
temperature=0.7,
max_tokens=2048
)
# 计算响应时间
response_time = int((time.time() - start_time) * 1000)
# 获取模型回复
model_response = response.choices[0].message.content
# 记录审计日志
audit_logger.log_chat(
session_id=session_id,
user_info=user_info,
user_input=message.content,
model_response=model_response,
token_usage=response.usage.dict() if response.usage else {},
response_time=response_time,
request_headers=dict(message.headers) if hasattr(message, 'headers') else {}
)
# 记录系统日志
audit_logger.log_system_event(
level="INFO",
module="chat",
message=f"用户 {user_info.get('email', 'unknown')} 完成一次对话",
user_email=user_info.get('email'),
extra_data={
"input_length": len(message.content),
"response_length": len(model_response),
"response_time_ms": response_time
}
)
# 返回回复
await cl.Message(content=model_response).send()
except Exception as e:
# 记录错误日志
audit_logger.log_system_event(
level="ERROR",
module="chat",
message=f"模型调用失败: {str(e)}",
user_email=user_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 "1=1"
# 执行查询
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 = f"SELECT 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', 'unknown@example.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)) {{ // Ctrl+Shift+I
return false;
}}
if (e.ctrlKey && e.keyCode == 'U'.charCodeAt(0)) {{ // Ctrl+U
return false;
}}
}};
</script>
"""
# 发送自定义元素到前端
await cl.Message(
content="",
elements=[
cl.Html(name="watermark_css", html=custom_css, display="hidden"),
cl.Html(name="watermark_js", html=custom_js, display="hidden")
]
).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 None
5.3 集成水印到消息处理
更新消息处理函数，加入内容水印：
@cl.on_message
async def main_with_watermark(message: cl.Message):
# 获取用户信息
user_info = cl.user_session.get("user", {})
# 调用模型获取回复
response = client.chat.completions.create(
model="glm-4-9b-chat-1m",
messages=[
{"role": "system", "content": "你是一个有帮助的AI助手。"},
{"role": "user", "content": message.content}
],
temperature=0.7,
max_tokens=2048
)
model_response = response.choices[0].message.content
# 嵌入内容水印
watermarked_response = embed_content_watermark(model_response, user_info)
# 发送带水印的回复
await cl.Message(content=watermarked_response).send()
# 记录审计日志（包含水印信息）
audit_logger.log_chat(
session_id=cl.user_session.get("id"),
user_info=user_info,
user_input=message.content,
model_response=model_response, # 原始回复
token_usage=response.usage.dict() if response.usage else {},
response_time=int((time.time() - start_time) * 1000),
request_headers=dict(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:
- "security@company.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: 30
6.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_ENV=production
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: bridge
6.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(type='input').inc(response.usage.prompt_tokens)
TOKEN_USAGE.labels(type='output').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星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
5
0- 0
已为社区贡献27条内容
所有评论(0)