Qwen3-0.6B-FP8快速上手：Chainlit中嵌入代码执行沙箱（受限Python环境）

Qwen3-0.6B-FP8快速上手Chainlit中嵌入代码执行沙箱受限Python环境1. 开篇当大模型遇到代码执行你需要一个安全的“沙箱”想象一下这个场景你部署了一个轻量级的AI助手它不仅能和你聊天还能帮你写代码、分析数据。但问题来了——如果用户让AI助手执行一段代码比如“帮我计算一下斐波那契数列的前10项”AI助手该怎么处理直接把代码交给服务器执行这太危险了万一用户输入的是恶意代码怎么办这就是我们今天要解决的问题。基于vLLM部署的Qwen3-0.6B-FP8模型已经准备就绪Chainlit前端界面也搭建好了现在我们需要一个安全、可控的代码执行环境——一个“沙箱”。这个沙箱能让AI助手在受限的环境中执行Python代码既能满足用户的计算需求又能确保服务器的安全。接下来我会带你一步步实现这个功能。2. 环境准备确认你的Qwen3-0.6B-FP8模型服务在开始构建代码沙箱之前我们先确认一下基础环境是否正常。2.1 检查模型服务状态打开终端运行以下命令查看模型服务的日志cat /root/workspace/llm.log如果看到类似下面的输出说明模型已经成功部署INFO 07-10 14:30:22 llm_engine.py:73] Initializing an LLM engine... INFO 07-10 14:30:22 llm_engine.py:74] Model: Qwen3-0.6B-FP8 INFO 07-10 14:30:22 llm_engine.py:75] Tokenizer: Qwen3-0.6B-FP8 INFO 07-10 14:30:23 llm_engine.py:76] Model loaded successfully关键点确保看到“Model loaded successfully”这样的提示这表示模型已经准备好接收请求了。2.2 测试Chainlit前端现在打开Chainlit前端界面你应该能看到一个简洁的聊天界面。试着问模型一个简单的问题比如你好请介绍一下你自己如果模型能正常回复说明整个基础链路是通的。Chainlit前端已经能正确调用后端的Qwen3模型了。3. 核心挑战为什么需要代码执行沙箱在深入技术实现之前我们先搞清楚为什么要做这件事。3.1 用户需求AI助手应该能执行代码用户可能希望AI助手能执行简单的数学计算处理和分析数据演示算法逻辑测试代码片段比如用户问“用Python写一个快速排序算法并运行一下”AI助手应该能给出代码并展示运行结果。3.2 安全风险不受控的代码执行很危险如果让用户提交的代码直接在服务器上运行可能会执行恶意命令如删除文件、访问敏感数据消耗过多资源导致服务器崩溃进行网络攻击或扫描安装恶意软件3.3 解决方案受限的Python执行环境我们需要创建一个“沙箱”——一个受限的执行环境在这里只能使用特定的Python模块不能访问文件系统或只能访问临时目录不能进行网络访问有执行时间和内存限制能捕获输出和错误信息4. 构建安全的Python代码执行沙箱现在我们来实际构建这个沙箱。我会分步骤讲解并提供完整的代码示例。4.1 基础沙箱实现首先我们创建一个简单的代码执行器。这个执行器能运行Python代码但限制在安全范围内import subprocess import tempfile import os import signal import resource class PythonSandbox: def __init__(self, timeout5, memory_limit128): 初始化代码执行沙箱 参数 - timeout: 代码执行超时时间秒 - memory_limit: 内存限制MB self.timeout timeout self.memory_limit memory_limit * 1024 * 1024 # 转换为字节 def execute_code(self, code): 在受限环境中执行Python代码 参数 - code: 要执行的Python代码字符串 返回 - 执行结果字典包含输出、错误和执行状态 # 创建临时文件来存储代码 with tempfile.NamedTemporaryFile(modew, suffix.py, deleteFalse) as f: f.write(code) temp_file f.name try: # 设置资源限制 def set_limits(): # 设置内存限制 resource.setrlimit(resource.RLIMIT_AS, (self.memory_limit, self.memory_limit)) # 设置CPU时间限制 resource.setrlimit(resource.RLIMIT_CPU, (self.timeout, self.timeout)) # 执行代码 result subprocess.run( [python3, temp_file], capture_outputTrue, textTrue, timeoutself.timeout, preexec_fnset_limits ) return { success: result.returncode 0, output: result.stdout, error: result.stderr, returncode: result.returncode } except subprocess.TimeoutExpired: return { success: False, output: , error: f代码执行超时限制{self.timeout}秒, returncode: -1 } except Exception as e: return { success: False, output: , error: f执行出错{str(e)}, returncode: -1 } finally: # 清理临时文件 if os.path.exists(temp_file): os.unlink(temp_file)这个基础版本已经能提供基本的安全保障有时间限制、内存限制并且代码在独立的进程中运行。4.2 增强安全性限制可用的模块基础版本还不够安全用户仍然可以导入危险的模块。我们需要进一步限制import ast import sys class SecurePythonSandbox(PythonSandbox): def __init__(self, timeout5, memory_limit128): super().__init__(timeout, memory_limit) # 允许导入的安全模块白名单 self.allowed_modules { math, random, datetime, json, re, collections, itertools, functools, statistics, decimal, fractions, string, typing, enum, hashlib, base64, uuid } # 禁止的危险模块 self.dangerous_modules { os, sys, subprocess, shutil, socket, requests, urllib, pickle, marshal, ctypes, signal, resource, multiprocessing, threading, asyncio } def check_code_safety(self, code): 检查代码是否安全 参数 - code: 要检查的Python代码 返回 - (是否安全, 错误信息) try: # 解析代码的抽象语法树 tree ast.parse(code) # 检查所有导入语句 for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: module_name alias.name.split(.)[0] if module_name in self.dangerous_modules: return False, f禁止导入危险模块{module_name} if module_name not in self.allowed_modules: return False, f模块不在白名单中{module_name} elif isinstance(node, ast.ImportFrom): module_name node.module.split(.)[0] if node.module else if module_name in self.dangerous_modules: return False, f禁止从危险模块导入{module_name} if module_name and module_name not in self.allowed_modules: return False, f模块不在白名单中{module_name} return True, 代码安全检查通过 except SyntaxError as e: return False, f代码语法错误{str(e)} except Exception as e: return False, f安全检查出错{str(e)} def execute_code(self, code): 增强版代码执行先检查安全性 参数 - code: 要执行的Python代码 返回 - 执行结果字典 # 1. 安全检查 is_safe, message self.check_code_safety(code) if not is_safe: return { success: False, output: , error: f安全检查失败{message}, returncode: -2 } # 2. 执行代码 return super().execute_code(code)现在我们的沙箱会先检查代码中导入的模块确保不会导入危险的模块。4.3 更高级的隔离使用Docker容器对于更高级的安全需求我们可以使用Docker容器来提供更强的隔离import docker import uuid class DockerSandbox: def __init__(self, timeout10): 使用Docker容器的代码执行沙箱 参数 - timeout: 执行超时时间秒 self.timeout timeout self.client docker.from_env() def execute_code(self, code): 在Docker容器中执行Python代码 参数 - code: 要执行的Python代码 返回 - 执行结果字典 # 生成唯一的容器名称 container_name fcode_exec_{uuid.uuid4().hex[:8]} try: # 创建并运行容器 container self.client.containers.run( python:3.9-slim, # 使用官方Python镜像 commandfpython -c {self._escape_code(code)}, namecontainer_name, detachTrue, mem_limit128m, # 内存限制 cpu_period100000, cpu_quota50000, # CPU限制 network_disabledTrue, # 禁用网络 read_onlyTrue, # 只读文件系统 removeTrue, # 执行后自动删除 stdoutTrue, stderrTrue ) # 等待执行完成 try: result container.wait(timeoutself.timeout) logs container.logs().decode(utf-8) return { success: result[StatusCode] 0, output: logs, error: if result[StatusCode] 0 else logs, returncode: result[StatusCode] } except Exception as e: # 超时或其他错误 container.stop() return { success: False, output: , error: f执行出错{str(e)}, returncode: -1 } except docker.errors.ImageNotFound: return { success: False, output: , error: Docker镜像不存在请先拉取python:3.9-slim镜像, returncode: -1 } except Exception as e: return { success: False, output: , error: f容器创建失败{str(e)}, returncode: -1 } def _escape_code(self, code): 转义代码中的特殊字符以便在命令行中安全传递 # 简单的转义处理 escaped code.replace(, \\).replace($, \\$).replace(, \\) return escapedDocker方案提供了最强的隔离性但需要服务器上安装Docker并且启动容器有一定开销。5. 集成到Chainlit让AI助手具备代码执行能力现在我们已经有了安全的代码执行沙箱接下来把它集成到Chainlit应用中。5.1 创建Chainlit应用首先创建一个新的Chainlit应用文件# app.py import chainlit as cl from sandbox import SecurePythonSandbox # 导入我们创建的沙箱 # 初始化代码执行沙箱 sandbox SecurePythonSandbox(timeout10, memory_limit256) cl.on_chat_start async def start_chat(): 聊天开始时的初始化 await cl.Message( content你好我是Qwen3助手我可以帮你执行Python代码。请告诉我你想做什么 ).send() cl.on_message async def handle_message(message: cl.Message): 处理用户消息 user_input message.content # 检查用户是否要求执行代码 if 运行代码 in user_input or 执行代码 in user_input or python in user_input: # 提取代码部分 code extract_python_code(user_input) if code: # 显示执行状态 status_msg cl.Message(content正在执行代码...) await status_msg.send() # 执行代码 result sandbox.execute_code(code) # 显示结果 if result[success]: output result[output].strip() if output: await cl.Message( contentf✅ 代码执行成功\n\n输出结果\n\n{output}\n ).send() else: await cl.Message( content✅ 代码执行成功无输出 ).send() else: error_msg result[error].strip() await cl.Message( contentf❌ 代码执行失败\n\n{error_msg}\n ).send() else: await cl.Message( content没有找到有效的Python代码。请用python代码块包裹你的代码。 ).send() else: # 普通对话调用Qwen3模型 response await call_qwen3_model(user_input) await cl.Message(contentresponse).send() def extract_python_code(text): 从文本中提取Python代码块 import re # 匹配python代码块 pattern rpython\s*(.*?)\s* matches re.findall(pattern, text, re.DOTALL) if matches: return matches[0] # 如果没有代码块检查是否整个消息都是代码 lines text.strip().split(\n) if len(lines) 10: # 假设短文本可能是代码 # 简单的启发式检查是否包含Python关键字 python_keywords [def , class , import , from , print(, for , if , else:] for keyword in python_keywords: if keyword in text: return text return None async def call_qwen3_model(prompt): 调用Qwen3模型生成回复 # 这里替换为实际的模型调用代码 # 示例使用requests调用vLLM API import requests import json try: response requests.post( http://localhost:8000/v1/completions, json{ model: Qwen3-0.6B-FP8, prompt: prompt, max_tokens: 500, temperature: 0.7 }, timeout30 ) if response.status_code 200: result response.json() return result[choices][0][text] else: return f模型调用失败{response.status_code} except Exception as e: return f调用模型时出错{str(e)}5.2 创建配置文件Chainlit需要一个配置文件来定义应用的行为# chainlit.md # 欢迎使用Qwen3代码助手 这是一个集成了代码执行功能的AI助手基于Qwen3-0.6B-FP8模型构建。 ## 功能特点 1. **智能对话**基于Qwen3模型的自然语言对话 2. **代码执行**在安全沙箱中执行Python代码 3. **实时反馈**立即看到代码执行结果 4. **安全保障**受限的执行环境防止恶意代码 ## 使用方法 1. 直接提问进行对话 2. 用以下格式执行代码请运行以下代码print(Hello, World!)3. 代码执行限制 - 最大执行时间10秒 - 最大内存256MB - 只能导入安全的Python模块 ## 示例 **对话模式** 用户什么是快速排序 助手快速排序是一种高效的排序算法... **代码执行模式** 用户请运行以下代码计算斐波那契数列 python def fibonacci(n): if n 1: return n a, b 0, 1 for _ in range(2, n1): a, b b, a b return b for i in range(10): print(ffib({i}) {fibonacci(i)})助手正在执行代码... ✅ 代码执行成功输出结果fib(0) 0 fib(1) 1 fib(2) 1 fib(3) 2 fib(4) 3 fib(5) 5 fib(6) 8 fib(7) 13 fib(8) 21 fib(9) 34### 5.3 运行Chainlit应用 现在我们可以运行这个集成了代码执行功能的Chainlit应用 bash # 安装依赖 pip install chainlit requests docker # 运行应用 chainlit run app.py打开浏览器访问http://localhost:8000你应该能看到一个功能完整的AI助手既能对话又能执行代码。6. 实际应用示例让AI助手帮你解决问题让我们看看这个集成了代码执行功能的AI助手能做什么。6.1 数学计算助手用户可以直接让AI助手进行复杂的数学计算用户请计算1到100所有整数的平方和 助手我可以帮你计算。让我运行一下代码 python # 计算1到100的平方和 sum_of_squares sum(i**2 for i in range(1, 101)) print(f1到100的平方和是{sum_of_squares}) # 验证公式n(n1)(2n1)/6 n 100 formula_result n * (n 1) * (2*n 1) // 6 print(f使用公式验证{formula_result})正在执行代码... ✅ 代码执行成功输出结果1到100的平方和是338350 使用公式验证338350所以1到100所有整数的平方和是338350。### 6.2 数据分析演示 用户可以让AI助手演示数据处理用户请生成一个包含随机数据的DataFrame并计算统计信息助手我来创建一个模拟数据集并计算基本统计量import pandas as pd import numpy as np # 设置随机种子以便结果可复现 np.random.seed(42) # 创建模拟数据 data { 年龄: np.random.randint(18, 65, 100), 收入: np.random.normal(50000, 15000, 100), 满意度: np.random.uniform(1, 10, 100), 购买次数: np.random.poisson(5, 100) } df pd.DataFrame(data) print(数据前5行) print(df.head()) print(\n基本统计信息) print(df.describe()) print(\n各列相关性) print(df.corr())正在执行代码... ✅ 代码执行成功输出结果数据前5行 年龄 收入 满意度 购买次数 0 37 48767.123456 7.891234 4 1 52 61234.567890 3.456789 6 ... 基本统计信息 年龄 收入 满意度 购买次数 count 100.000000 100.000000 100.000000 100.000000 mean 41.230000 49876.543210 5.432109 4.890000 std 13.456789 14987.654321 2.567890 2.234567 ... 各列相关性 年龄 收入 满意度 购买次数 年龄 1.000000 0.123456 0.045678 -0.098765 收入 0.123456 1.000000 0.234567 0.345678 ...从统计结果可以看出...### 6.3 算法教学助手 AI助手可以演示算法原理用户请演示快速排序算法助手快速排序是一种分治算法。我来写一个实现并演示def quick_sort(arr): 快速排序实现 if len(arr) 1: return arr pivot arr[len(arr) // 2] left [x for x in arr if x pivot] middle [x for x in arr if x pivot] right [x for x in arr if x pivot] return quick_sort(left) middle quick_sort(right) # 测试数据 test_data [64, 34, 25, 12, 22, 11, 90, 88, 76, 45, 33, 21] print(原始数组, test_data) print(排序过程演示) # 逐步演示简化版 def quick_sort_steps(arr, depth0): if len(arr) 1: print( * depth f返回{arr}) return arr pivot arr[len(arr) // 2] print( * depth f基准值{pivot}) left [x for x in arr if x pivot] middle [x for x in arr if x pivot] right [x for x in arr if x pivot] print( * depth f左子数组{left}) print( * depth f中间等于基准{middle}) print( * depth f右子数组{right}) return quick_sort_steps(left, depth1) middle quick_sort_steps(right, depth1) sorted_data quick_sort_steps(test_data[:6]) # 只演示前6个元素 print(\n完整排序结果, quick_sort(test_data))正在执行代码... ✅ 代码执行成功输出结果原始数组 [64, 34, 25, 12, 22, 11, 90, 88, 76, 45, 33, 21] 排序过程演示 基准值12 左子数组 [11] 中间等于基准 [12] 右子数组 [64, 34, 25, 22] 基准值34 左子数组 [25, 22] 中间等于基准 [34] 右子数组 [64] 基准值25 左子数组 [22] 中间等于基准 [25] 右子数组 [] 返回[22] 返回[22, 25] 返回[22, 25, 34] 返回[64] 返回[11, 12, 22, 25, 34, 64] 完整排序结果 [11, 12, 21, 22, 25, 33, 34, 45, 64, 76, 88, 90]从演示可以看出快速排序通过选择基准值将数组分成三部分...## 7. 安全考虑与最佳实践 在实现代码执行功能时安全是最重要的考虑因素。以下是一些关键的安全实践 ### 7.1 多层安全防护 我们的实现采用了多层防护 1. **语法检查层**使用AST解析检查导入的模块 2. **资源限制层**限制执行时间和内存使用 3. **进程隔离层**在独立进程中执行代码 4. **容器隔离层**可选使用Docker提供更强的隔离 ### 7.2 可调整的安全策略 根据不同的使用场景你可以调整安全策略 python # 不同安全级别的配置 security_profiles { high: { timeout: 5, memory_limit: 128, allowed_modules: [math, random, datetime, json], use_docker: True }, medium: { timeout: 10, memory_limit: 256, allowed_modules: [math, random, datetime, json, re, collections], use_docker: False }, low: { timeout: 30, memory_limit: 512, allowed_modules: all_except_dangerous, use_docker: False } }7.3 监控与日志添加监控和日志记录以便追踪代码执行情况import logging from datetime import datetime class MonitoredSandbox(SecurePythonSandbox): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.logger logging.getLogger(code_sandbox) def execute_code(self, code): # 记录执行开始 start_time datetime.now() code_hash self._hash_code(code) self.logger.info(f开始执行代码哈希{code_hash}) # 执行代码 result super().execute_code(code) # 记录执行结果 end_time datetime.now() duration (end_time - start_time).total_seconds() log_data { code_hash: code_hash, success: result[success], duration: duration, code_length: len(code), timestamp: start_time.isoformat() } if not result[success]: log_data[error] result[error][:200] # 只记录前200个字符 self.logger.info(f代码执行完成{log_data}) return result def _hash_code(self, code): 生成代码的哈希值用于追踪 import hashlib return hashlib.md5(code.encode()).hexdigest()[:8]7.4 用户权限管理对于多用户系统可以考虑添加用户级别的限制class UserAwareSandbox(MonitoredSandbox): def __init__(self, user_limitsNone, *args, **kwargs): super().__init__(*args, **kwargs) self.user_limits user_limits or {} def execute_for_user(self, user_id, code): 为特定用户执行代码 user_limit self.user_limits.get(user_id, {}) # 应用用户特定的限制 timeout user_limit.get(timeout, self.timeout) memory_limit user_limit.get(memory_limit, self.memory_limit) # 临时修改限制 original_timeout self.timeout original_memory self.memory_limit self.timeout timeout self.memory_limit memory_limit try: result self.execute_code(code) return result finally: # 恢复原始限制 self.timeout original_timeout self.memory_limit original_memory8. 总结构建安全智能的代码执行环境通过本文的步骤我们成功地在Chainlit中为Qwen3-0.6B-FP8模型添加了代码执行功能。让我们回顾一下关键点8.1 实现的核心功能安全的代码执行沙箱提供了多层防护确保用户代码不会危害服务器安全智能的代码识别能够从用户消息中提取Python代码块实时的执行反馈立即显示代码执行结果或错误信息灵活的配置选项可以根据需要调整安全级别和资源限制8.2 实际应用价值这个集成的系统可以用于教育场景帮助学生理解和测试编程概念数据分析快速验证数据处理逻辑算法演示可视化展示算法执行过程代码调试测试小的代码片段自动化助手执行重复的计算任务8.3 进一步优化方向如果你想让这个系统更加强大可以考虑支持更多语言除了Python还可以支持JavaScript、SQL等添加可视化输出支持图表、图像等丰富的结果展示实现代码补全在代码编辑时提供智能提示添加代码保存功能允许用户保存和分享代码片段集成版本控制跟踪代码的修改历史8.4 开始使用现在你已经有了一个完整的解决方案可以直接使用本文提供的代码搭建自己的系统根据实际需求调整安全策略扩展功能比如添加更多的代码示例库优化用户体验比如添加代码高亮、错误提示等最重要的是这个系统让AI助手不再只是说而是能够做——执行代码、验证想法、演示概念。这大大扩展了AI助手的能力边界让它成为一个真正有用的工具。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。