使用Java语言手搓LLM大语言模型“代码解释器”

手搓“代码解释器”

从ChatGLM-3到GLM-4再到MiniCPM3-4B，官方都提到了“代码解释器”这样的一个进阶功能。所谓代码解释器就是使用大语言模型生成一段代码，通过执行代码来完成某项任务的功能。

以下是MiniCPM3-4B官方提供的代码解释器的示例：

cd demo/minicpm3/code_interpreter
pip install -r requirements.txt
python code_interpreter.py openbmb/MiniCPM3-4B

通过简单查看示例代码，我们可以知道，代码解释器的功能和模型无关，而是要看官方提供的仓库代码是否支持该功能。

下面我将使用Java语言给我的Qwen模型手搓一个代码解释器。

需求

实现类似讯飞星火“数据分析助手”的功能：上传一个excel文件，使用python代码分析excel文件内容，返回我们想要的数据。

实现步骤

本文档仅介绍实现思路和主要步骤，不分享全部的实现代码。

创建一个CodeAgent对象，实现readSchema、parseSchema、parseData三个方法；

    /**
     * 读取excel文件
     *
     * @param sessionId 会话ID
     * @param filePath  文件路径
     */
publicFluxreadSchema(String sessionId, String filePath){
StringuserPrompt="nYour task is: 步骤1使用python读取Excel的所有列名以及基于所有列的前3条数据，注意要获取所有的列。";
StringdatePrompt="nCurrent date is: "+DateUtil.currentDate();
StringfilePrompt="nThe file you need to read is: "+filePath;
return baseLLM.streamChat(sessionId, SYSTEM_PROMPT+datePrompt+filePrompt+userPrompt);
}

/**
     * 分析excel结构
     * @param sessionId 会话ID
     * @return 结构
     */
publicFluxparseSchema(String sessionId){
StringuserPrompt="nYour task is: step 2 基于样例数据和各列信息，解析数据结构，分析其各列的含义。";
return baseLLM.streamChat(sessionId,userPrompt);
}

/**
     * 分析excel数据
     * @param sessionId 会话ID
     * @param filePath 文件路径
     * @param prompt 提示词
     * @return 分析结果
     */
publicFluxparseData(String sessionId, String filePath, String prompt){
StringuserPrompt="nStep 3 编写满足用户需求的python代码: "+prompt;
StringfilePrompt="nThe file you need to read is: "+filePath;
return baseLLM.streamChat(sessionId, filePrompt+userPrompt);
}

Service层使用Flux流式接收大模型响应;

    @Override
publicFluxrunPython(String sessionId, MultipartFile file, String prompt){
...省略...
...省略...
...省略...
StringfilePath="xxx";
CodeAgentcodeAgent=CodeAgent.fromLlm(openAIChat);
returnFlux.create(emitter ->{
            codeAgent.readSchema(sessionId, filePath).doOnComplete(()->{
                codeAgent.parseSchema(sessionId).doOnComplete(()->{
                    codeAgent.parseData(sessionId, filePath, prompt).doOnComplete(()->{
                        emitter.complete();
}).subscribe(s->{
                        emitter.next(s);
});
}).subscribe(s->{
                    emitter.next(s);
});
}).subscribe(s->{
                emitter.next(s);
});
});
    }

设置监听，当监听到模型响应完毕，立即执行completed方法。

completed方法将会从文本中提取代码，并执行返回结果。

extractCode会返回代码块列表，每个代码块存在language和code两个属性值。

我们拿到代码块列表后，就可以去执行代码了。

在excuteCode方法中，我们需要将代码写到文件中，并判断是否需要在docker容器中执行代码。

    // 将代码字符串写入 filename 指定的文件。
    writeCodeToFile(workDir, filename, code);
    // 本地执行代码或者在docker容器中执行代码
    CodeExecutionResult executionResult = StringUtils.isEmpty(config.getDocker())
                ? executeCodeLocally(language, workDir, filename, config.getTimeout())
                : executeCodeInDocker(language,config.getDocker(),filename);

如果要在docker容器中执行，我们需要提前创建docker容器，并设置与本地目录的映射。如我本地的目录是当前项目下的data文件夹，coding下是大模型生成的python脚本，file下是上传的excel文件。docker容器的目录则需要和我们本地的目录保持一致。

    /**
     * 在 Docker 中执行代码。
     * @param language 语言
     * @param containerName Docker容器
     * @param filename 文件名
     * @return CodeExecutionResult
     */
publicstaticCodeExecutionResultexecuteCodeInDocker(String language,String containerName,String filename){
DockerExecutorClientdockerExecutorClient=DockerExecutorClient.builder().build().init();
StringcontainerId= dockerExecutorClient.getContainerIdByName(containerName);
Stringresult= dockerExecutorClient.executeScript(language, containerId,  filename);
returnnewCodeExecutionResult(0, result.trim(),containerName);
    }

package com.zhbr.openai.agent.code.docker;

import com.github.dockerjava.api.DockerClient;
import com.github.dockerjava.api.command.*;
import com.github.dockerjava.api.exception.DockerClientException;
import com.github.dockerjava.api.model.Bind;
import com.github.dockerjava.api.model.HostConfig;
import com.github.dockerjava.api.model.Volume;
import com.github.dockerjava.core.DefaultDockerClientConfig;
import com.github.dockerjava.core.DockerClientImpl;
import com.github.dockerjava.core.command.ExecStartResultCallback;
import com.github.dockerjava.httpclient5.ApacheDockerHttpClient;
import com.github.dockerjava.transport.DockerHttpClient;
import com.zhbr.openai.exception.DockerContainerException;
import lombok.Builder;
import lombok.experimental.SuperBuilder;
import com.github.dockerjava.api.command.InspectContainerCmd;
import org.apache.commons.lang3.StringUtils;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.time.Duration;

/**
 * @ClassName DockerExecutorClient
 * @Description docker客户端
 * @Date 2024/9/19 10:19
 * @Version 1.0
 */
@SuperBuilder
publicclassDockerExecutorClient{

/**
     * docker服务器的ip
     */
@Builder.Default
privateStringclientHost="xxx.xxx.xxx.xxx";

/**
     * docker服务器的端口
     */
@Builder.Default
privateStringclientPort="xxxx";

/**
     * docker客户端
     */
privateDockerClient dockerClient;

/**
     * 工作目录
     */
privatestaticfinalStringworkDir="data";

/**
     * 初始化docker客户端
     * @return DockerExecutorClient
     */
publicDockerExecutorClientinit(){
Stringhost="tcp://"+ clientHost +":"+ clientPort;
DefaultDockerClientConfigconfig=DefaultDockerClientConfig.createDefaultConfigBuilder()
.withDockerHost(host)
.build();
DockerHttpClienthttpClient=newApacheDockerHttpClient.Builder()
.dockerHost(config.getDockerHost())
.maxConnections(100)
.connectionTimeout(Duration.ofSeconds(30))
.responseTimeout(Duration.ofSeconds(180))
.build();
this.dockerClient =DockerClientImpl.getInstance(config, httpClient);
returnthis;
}

/**
     * 创建容器并启动
     * @param imageName 镜像名称
     * @return 容器id
     */
publicStringcreateContainer(String imageName,String containerName){
StringbasePath=System.getProperty("user.dir");
StringfullDirectoryPath= basePath +File.separator + workDir ;
//创建容器
CreateContainerResponsecontainer= dockerClient.createContainerCmd(imageName)
.withName(containerName)
// 容器持久化运行，这样可以多次使用某个容器调用不同的python脚本
.withCmd("tail","-f","/dev/null")
// 卷挂载，将宿主机文件夹与容器文件夹绑定起来
.withHostConfig(newHostConfig().withBinds(newBind(fullDirectoryPath,newVolume(fullDirectoryPath))))
.exec();
//启动容器
        dockerClient.startContainerCmd(container.getId()).exec();
return container.getId();
}

/**
     * 执行脚本
     * @param containerId 容器id
     * @param scriptName 脚本名称
     * @return 执行结果
     */
publicStringexecuteScript(String language,String containerId,String scriptName){
StringbasePath=System.getProperty("user.dir");
StringfullDirectoryPath= basePath +File.separator + workDir ;
ExecCreateCmdResponseexecCreateCmdResponse= dockerClient.execCreateCmd(containerId)
.withCmd(language, fullDirectoryPath+"/coding/"+ scriptName)
.withAttachStdout(true)
.withAttachStderr(true)
.exec();
ByteArrayOutputStreamoutputStream=newByteArrayOutputStream();
try{
            dockerClient.execStartCmd(execCreateCmdResponse.getId())
.exec(newExecStartResultCallback(outputStream,System.err))
.awaitCompletion();
}catch(InterruptedException e){
thrownewDockerContainerException("Error executing code.",e);
}
return outputStream.toString().trim();
}

/**
     * 获取docker容器的ID
     * @param containerName 容器名称
     * @return 容器id
     */
publicStringgetContainerIdByName(String containerName){
InspectContainerCmdcmd= dockerClient.inspectContainerCmd(containerName);
return cmd.exec().getId();
}
}

实现效果

大模型响应内容（隐藏部分敏感数据）：

‍‍```python
# filename: read_excel_data.py
import pandas as pd

#读取Excel文件
excel_file ="/home/chatFile/cxyds_service/data/file/20240919164355-tb_dlyc_area_prediction_result.xlsx"
data = pd.read_excel(excel_file)

#输出所有列名
print("所有列名：", data.columns)

#输出所有列的前3条数据
print("前3条数据：")
print(data.head())
‍‍```
请执行上述代码，它将读取Excel文件并打印出所有列名以及基于所有列的前3条数据。执行结果将帮助我们理解数据的结构和内容，以便进行下一步分析。

请执行`read_excel_data.py`并提供执行结果。
Python代码执行结果：所有列名：Index(['pid','org_name','org_code','expect_year','expect_month',
'expect_type','expect_num','expect_rate','create_time',
'update_time'],
      dtype='object')
前3条数据：
                                pid  ...         update_time
0022f9fa72d8bec0cb24bfc9f5b1529f0...2024-06-1111:35:22
1036c628f728183802c56fc765d493f1b...2024-06-1111:35:26
2044ed07229e78dce94911cab22d09c6e...2024-06-1111:35:24
3063220035e14cf9fd9d965e8cd57d8d0...2024-06-1111:35:26
40668ee99b678cd666117211cf6ded710...2024-06-1111:35:22

根据提供的样例数据，我们可以推测以下列的含义：

1.`pid`:这可能是一个唯一的标识符，用于识别每一行数据。
2.`org_name`:这可能表示组织或部门的名称。
3.`org_code`:可能是组织或部门的代码，用于内部分类或识别。
4.`expect_year`:预期的年份，可能是预测或计划的时间范围。
5.`expect_month`:预期的月份，进一步细化预测或计划的时间范围。
6.`expect_type`:可能是预测或计划的类型，例如销售预测、产量预测等。
7.`expect_num`:预期数值，可能是预测的数值结果，比如销售额、产量等。
8.`expect_rate`:预期增长率，可能是预测的增长率或变化率。
9.`create_time`:记录数据创建的时间。
10.`update_time`:记录数据最后更新的时间。

请注意，这些仅是根据列名的猜测。要获得更准确的含义，可能需要查看数据的具体内容或查询相关文档。这些列名通常与业务预测或计划系统相关，例如销售预测、生产计划等。

接下来，根据用户的具体任务要求，我们可以进一步处理这些数据。请提供下一步的任务说明。

‍‍```python
# filename: calculate_electricity_sales.py
import pandas as pd

#读取Excel文件
excel_file ="/home/chatFile/cxyds_service/data/file/20240919164355-tb_dlyc_area_prediction_result.xlsx"
data = pd.read_excel(excel_file)

#确保列名和预期的匹配，如果没有匹配，需要根据实际情况调整列名
data = data.rename(columns={"expect_year":"year","expect_month":"month","expect_num":"sales","org_name":"area"})

#筛选2024年5月xxxx
filtered_data= data[(data['year']==2024)&(data['month']==5)&(data['area']=='xxx')]

#计算xxx总和
electricity_sales = filtered_data['sales'].sum()

print(f"2024年5月xxxx是：{electricity_sales} ")
‍‍```
请执行上述代码，它将计算并打印出2024年5月xxx。执行结果将帮助我们确认计算是否正确。

请执行`calculate_electricity_sales.py`并提供执行结果。
Python代码执行结果：2024年5月xxxx是：4.10992431640625 

我们可以复制代码到pycharm中验证一下，可以看到代码是没有问题的，而且执行的结果也是一样的。

参考：

部分代码参考了autogen4j项目

https://github.com/HamaWhiteGG/autogen4j