开源模型应用落地-CodeQwen模型小试-探索更多使用场景（三）

一、前言

代码专家模型是基于人工智能的先进技术，它能够自动分析和理解大量的代码库，并从中学习常见的编码模式和最佳实践。这种模型可以提供准确而高效的代码建议，帮助开发人员在编写代码时避免常见的错误和陷阱。

通过学习代码专家模型，开发人员可以获得高效、准确和个性化的代码支持。这不仅可以提高工作效率，还可以在不同的技术环境中简化软件开发工作流程。代码专家模型的引入将为开发人员带来更多的机会去关注创造性的编程任务，从而推动软件开发的创新和进步。

开源模型应用落地-CodeQwen模型小试-小试牛刀（一）

开源模型应用落地-CodeQwen模型小试-SQL专家测试（二）

二、术语

2.1.CodeQwen1.5

基于 Qwen 语言模型初始化，拥有 7B 参数的模型，其拥有 GQA 架构，经过了 ~3T tokens 代码相关的数据进行预训练，共计支持 92 种编程语言、且最长支持 64K 的上下文输入。效果方面，CodeQwen1.5 展现出了非凡的代码生成、长序列建模、代码修改、SQL 能力等,该模型可以大大提高开发人员的工作效率，并在不同的技术环境中简化软件开发工作流程。

CodeQwen 是基础的 Coder

代码生成是大语言模型的关键能力之一，期待模型将自然语言指令转换为具有精确的、可执行的代码。仅拥有 70 亿参数的 CodeQwen1.5 在基础代码生成能力上已经超过了更尺寸的模型，进一步缩小了开源 CodeLLM 和 GPT-4 之间编码能力的差距。

CodeQwen 是长序列 Coder

长序列能力对于代码模型来说至关重要，是理解仓库级别代码、成为 Code Agent 的核心能力。而当前的代码模型对于长度的支持仍然非常有限，阻碍了其实际应用的潜力。CodeQwen1.5 希望进一步推进开源代码模型在长序列建模上的进展，我们收集并构造了仓库级别的长序列代码数据进行预训练，通过精细的数据配比和组织方式，使其最终可以最长支持 64K 的输入长度。

CodeQwen 是优秀的代码修改者

一个好的代码助手不仅可以根据指令生成代码，还能够针对已有代码或者新的需求进行修改或错误修复。

CodeQwen 是出色的 SQL 专家

CodeQwen1.5 可以作为一个智能的 SQL 专家，弥合了非编程专业人士与高效数据交互之间的差距。它通过自然语言使无编程专业知识的用户能够查询数据库，从而缓解了与SQL相关的陡峭学习曲线。

2.2.CodeQwen1.5-7B-Chat

CodeQwen1.5 is the Code-Specific version of Qwen1.5. It is a transformer-based decoder-only language model pretrained on a large amount of data of codes.

Strong code generation capabilities and competitve performance across a series of benchmarks;
Supporting long context understanding and generation with the context length of 64K tokens;
Supporting 92 coding languages
Excellent performance in text-to-SQL, bug fix, etc.

三、前置条件

3.1.基础环境

操作系统：centos7

Tesla V100-SXM2-32GB CUDA Version: 12.2

3.2.下载模型

huggingface：

https://huggingface.co/Qwen/CodeQwen1.5-7B-Chat/tree/main

ModelScope：

git clone https://www.modelscope.cn/qwen/CodeQwen1.5-7B-Chat.git

PS：

1. 根据实际情况选择不同规格的模型

3.3.更新transformers库

pip install --upgrade transformers==4.38.1

四、使用方式

4.1.基础代码

# -*-  coding = utf-8 -*-
from transformers import AutoModelForCausalLM, AutoTokenizer, GenerationConfigdevice = "cuda"modelPath='/model/CodeQwen1.5-7B-Chat'def loadTokenizer():# print("loadTokenizer: ", modelPath)tokenizer = AutoTokenizer.from_pretrained(modelPath)return tokenizerdef loadModel(config):# print("loadModel: ",modelPath)model = AutoModelForCausalLM.from_pretrained(modelPath,torch_dtype="auto",device_map="auto")model.generation_config = configreturn modelif __name__ == '__main__':prompt = '需要替换哦'messages = [{"role": "system", "content": "You are a helpful assistant."},{"role": "user", "content": prompt},]config = GenerationConfig.from_pretrained(modelPath, top_p=0.85, temperature=0.1, repetition_penalty=1.1,do_sample=True, max_new_tokens=8192)tokenizer = loadTokenizer()model = loadModel(config)text = tokenizer.apply_chat_template(messages,tokenize=False,add_generation_prompt=True)model_inputs = tokenizer([text], return_tensors="pt").to(device)generated_ids = model.generate(model_inputs.input_ids)generated_ids = [output_ids[len(input_ids):] for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)]response = tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]print(response)

4.2.安全漏洞检测

把基础代码的prompt替换成以下文本

'''
请分析下面代码，是否存在安全漏洞，具体代码如下：
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;public class UserLogin {public static void main(String[] args) {String username = "admin";String password = "password";try {// 建立数据库连接Connection connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/mydatabase", "root", "password");// 执行查询Statement statement = connection.createStatement();String query = "SELECT * FROM users WHERE username='" + username + "' AND password='" + password + "'";ResultSet resultSet = statement.executeQuery(query);// 处理查询结果if (resultSet.next()) {System.out.println("登录成功");} else {System.out.println("登录失败");}// 关闭连接resultSet.close();statement.close();connection.close();} catch (SQLException e) {e.printStackTrace();}}
}
'''

调用结果：

结论：

模型能分析出代码漏洞并给出优化建议

4.3.优化代码性能

把基础代码的prompt替换成以下文本

'''
请优化下面代码，提升运行性能，具体代码如下：
public class StringConcatenation {public static void main(String[] args) {int n = 10000;String result = "";for (int i = 0; i < n; i++) {result += " " + i;}System.out.println(result);}
}
'''

调用结果：

结论：

模型能分析出循环拼接字符串带来的性能问题，并给出更优的实现方式

4.4.代码翻译

把基础代码的prompt替换成以下文本

'''
请用Java语言改写下面代码，并实现一致的功能，具体代码如下：
def bubble_sort(arr):n = len(arr)# 遍历所有数组元素for i in range(n):# Last i elements are already in placefor j in range(0, n-i-1):# 遍历数组从 0 到 n-i-1# 交换如果元素发现大于下一个元素if arr[j] > arr[j+1] :arr[j], arr[j+1] = arr[j+1], arr[j]# 示例
arr = [64, 34, 25, 12, 22, 11, 90]
bubble_sort(arr)
print("排序后的数组为:")
for i in range(len(arr)):print("%d" %arr[i], end=" ")
'''

调用结果：

结论：

模型能理解基于python语言实现的冒泡算法，并转换成Java语言实现类似功能

4.5.生成代码注释

把基础代码的prompt替换成以下文本

'''
请分析下面代码，给出每行的代码注释，具体代码如下：
def train_step(self, batch: Dict, batch_n_steps: int, step: int, loader_start_time: float) -> Tuple[Dict, Dict]:self.callbacks.on_train_step_start(self)# format databatch = self.format_batch(batch)loader_time = time.time() - loader_start_time# conteainers to hold model outputs and losses for each optimizer.outputs_per_optimizer = Noneloss_dict = {}# OPTIMIZATIONif isimplemented(self.model, "optimize"):  # pylint: disable=too-many-nested-blocks# custom optimize for the modelstep_time = time.time()device, dtype = self._get_autocast_args(self.config.mixed_precision, self.config.precision)with torch.autocast(device_type=device, dtype=dtype, enabled=self.config.mixed_precision):outputs, loss_dict_new = self.model.optimize(batch,self,)step_time = time.time() - step_time# If None, skip the stepif outputs is None:return None, None# TODO: find a way to log grad_norm for custom optimizeloss_dict_new = self.detach_loss_dict(loss_dict_new, True, None, None)loss_dict.update(loss_dict_new)else:# gradient accumulation# TODO: grad accumulation for each optimizerstep_optimizer = Trueif ((step + 1) % self.grad_accum_steps != 0) and (step + 1 != batch_n_steps):step_optimizer = Falseif not isinstance(self.optimizer, list):# auto training with a single optimizeroutputs, loss_dict_new, step_time = self.optimize(batch,self.model,self.optimizer,self.scaler,self.criterion,self.scheduler,self.config,step_optimizer=step_optimizer,num_optimizers=1,)loss_dict.update(loss_dict_new)else:# auto training with multiple optimizers (e.g. GAN)outputs_per_optimizer = [None] * len(self.optimizer)total_step_time = 0for idx, optimizer in enumerate(self.optimizer):criterion = self.criterion# scaler = self.scaler[idx] if self.use_amp_scaler else Nonescaler = self.scalerscheduler = Noneif self.scheduler is not None:scheduler = self.scheduler[idx]outputs, loss_dict_new, step_time = self.optimize(batch,self.model,optimizer,scaler,criterion,scheduler,self.config,idx,step_optimizer=step_optimizer,num_optimizers=len(self.optimizer),)# skip the rest if the model returns Nonetotal_step_time += step_timeoutputs_per_optimizer[idx] = outputs# merge loss_dicts from each optimizer# rename duplicates with the optimizer idx# if None, model skipped this optimizerif loss_dict_new is not None:for k, v in loss_dict_new.items():if k in loss_dict:loss_dict[f"{k}-{idx}"] = velse:loss_dict[k] = vstep_time = total_step_timeoutputs = outputs_per_optimizer# clear any pesky gradients after gradient accumulationif step_optimizer:self.model.zero_grad(set_to_none=True)# update avg runtime statskeep_avg_update = {}keep_avg_update["avg_loader_time"] = loader_timekeep_avg_update["avg_step_time"] = step_timeself.keep_avg_train.update_values(keep_avg_update)# update avg loss statsupdate_eval_values = {}for key, value in loss_dict.items():update_eval_values["avg_" + key] = valueself.keep_avg_train.update_values(update_eval_values)# print training progressif self.total_steps_done % self.config.print_step == 0:# log learning rateslrs = {}if isinstance(self.optimizer, list):for idx, optimizer in enumerate(self.optimizer):current_lr = self.optimizer[idx].param_groups[0]["lr"]lrs.update({f"current_lr_{idx}": current_lr})elif isinstance(self.optimizer, dict):for key, optimizer in self.optimizer.items():current_lr = self.optimizer[key].param_groups[0]["lr"]lrs.update({f"current_lr_{key}": current_lr})else:current_lr = self.optimizer.param_groups[0]["lr"]lrs = {"current_lr": current_lr}# log run-time statsloss_dict.update(lrs)loss_dict.update({"step_time": round(step_time, 4),"loader_time": round(loader_time, 4),})self.c_logger.print_train_step(batch_n_steps,step,self.total_steps_done,loss_dict,self.keep_avg_train.avg_values,)if self.args.rank == 0:# Plot Training Iter Stats# reduce TB load and don't log every stepif self.total_steps_done % self.config.plot_step == 0:self.dashboard_logger.train_step_stats(self.total_steps_done, loss_dict)if self.total_steps_done % self.config.save_step == 0 and self.total_steps_done != 0:if self.config.save_checkpoints:# checkpoint the modelself.save_checkpoint()if self.total_steps_done % self.config.log_model_step == 0:# log checkpoint as artifactself.update_training_dashboard_logger(batch=batch, outputs=outputs)self.dashboard_logger.flush()self.total_steps_done += 1self.callbacks.on_train_step_end(self)return outputs, loss_dict
'''

调用结果：