@2 PK-Platform

项目介绍

面试官版本

我参与了公司自动化评估平台项目的开发，这个项目的背景源于公司为外部企业客户提供大模型相关服务，包括 API 识别、自然语言转 SQL、知识库问答等多种场景。然而，当时各解决方案团队的服务效果缺乏系统化的评估流程，导致管理层难以准确衡量各团队的服务成效。为了解决这一问题，公司决定搭建一个自动化评估平台，旨在高效评估各团队的服务质量，并为管理层提供精准的决策依据。

这个平台主要分为两个核心部分：平台侧和评估程序侧。平台侧的功能是支持各团队提交服务策略，并能够自动获取这些策略在特定数据集上的评估结果。当某个团队提交策略后，系统会自动触发对应场景的评估程序，完成对策略的评估。而评估程序侧则采用了设计模式中的模板模式，将整个评估流程抽象为三个关键阶段：解析数据集数据、发送请求并接收策略返回结果、以及结果指标的计算。这种设计不仅提高了代码的复用性，还使得评估流程更加清晰和易于维护。

在项目中，我主要负责针对不同的评估场景，设计评估程序侧这三个阶段的逻辑。这一过程中，我与业务解决方案团队紧密合作，确保接口格式既能满足当前需求，又具备足够的灵活性以兼容未来的评估场景。同时，我还与产品团队深入探讨，不断迭代优化评价指标的选择，确保评估结果能够真实反映服务策略的效果。

最终，项目取得了显著的成果。我们成功构建了 Copilot 场景下的多个业务评估程序，并为解决方案团队提供了超过 xxx 次的评估测试。这些评估不仅帮助团队快速迭代和优化服务策略，还为管理层提供了有力的数据支持，推动了公司整体服务质量的提升。

简历版本

自动化评估平台项目

参与公司自动化评估平台的开发，旨在为大模型相关服务（如 API 识别、nl2sql、知识库问答等）提供高效评估流程，支持管理层决策。
负责设计评估程序侧的核心逻辑，采用模板模式抽象评估流程，包括数据解析、请求处理、结果计算三个阶段，提升代码复用性与可维护性。
与业务团队协作定义接口格式，确保兼容性与扩展性；与产品团队优化评价指标，提升评估结果的准确性。
完成 Copilot 场景下多个评估程序的构建，累计为解决方案团队提供 xxx 次评估测试，显著提升服务策略迭代效率与服务质量。

HR 版本

我参与了公司自动化评估平台的项目，这个平台的背景是公司为外部企业客户提供大模型相关服务，比如 API 识别、自然语言转 SQL、知识库问答等。由于当时缺乏系统化的评估流程，管理层难以准确衡量各团队的服务效果，因此公司决定搭建这个平台来高效评估服务质量，并为决策提供数据支持。

在这个项目中，我主要负责设计评估程序的核心逻辑。我们将评估流程抽象为三个阶段：数据解析、请求处理和结果计算，这样不仅提高了代码的复用性，也让整个评估过程更加清晰和高效。同时，我与业务团队紧密合作，确保接口设计既能满足当前需求，又能兼容未来的扩展；还与产品团队一起优化了评价指标，确保评估结果能够真实反映服务策略的效果。

最终，我们成功构建了 Copilot 场景下的多个评估程序，累计为解决方案团队提供了 xxx 次评估测试。这些测试帮助团队快速评估服务策略，也为管理层提供了有力的数据支持，推动了公司整体服务质量的提升。

技术实现

模板模式的实现方式

request_and_parse_response = __import__("benchmark." + self.benchmark, fromlist=True).request_and_parse_response

首先，规定了评估处理的 python 文件名字为对应的竞赛名称
然后，使用 python 的 __import__ 方法来导入对应的评估文件
之后，就是调用评估文件中对应的各个方法，如解析数据集、发送请求与接受结果等。

数据集

每个比赛都有对应的 config 文件，这个 config 文件中包含着数据集等信息。

数据集可以是本地或者说主程序所在的服务器上，也可以配置为内网 ftp 服务器上。

- local_file: dataset/sougou.719_721_724.1070case.cm.newform
  name: sougou_cm
  owner: system
  remote_uri: ftp://ftp.4pd.io/pub/pico/scene_lab/datasets/confident_matching/sougou_cm/sougou.719_721_724.1070case.cm.newform
- local_file: dataset/similar_queries.0731.0801.0802.1986case.cm.newform
  name: similar_queries_cm
  owner: system
  remote_uri: ftp://ftp.4pd.io/pub/pico/scene_lab/datasets/confident_matching/similar_queries_cm/similar_queries.0731.0801.0802.1986case.cm.newform

多个数据集处理

会针对每个数据集计算结果，在评估指标时，将全部数据集的结果根据需求进行综合计算。

启动被测服务

服务器上配置了 docker 服务
在启动被测服务容器的时候，会随机生成一个端口号，然后尝试 connect，若没有成功，则说明端口号没有被占用
进而调用 docker 库去基于端口号和提供的镜像等信息启动容器

docker_host: 172.27.231.64
docker_port: 2375

docker_sevice = "tcp://"+docker_host+":"+str(docker_port)
 
#todo 分配一个端口给这个服务
port=apply_port(docker_host)
#todo 如果没起来，需要处理，而不是直接挂了
container = start_strategy_service(strategy_config["docker_image"],strategy_config["parameters"],port,docker_service)
 
def apply_port(docker_host):
    #从规定范围内，找没被占用的端口
    while True:
        port = random.randint(25000,30000)
        if not socketconnect(docker_host,port):
            return port
 
def start_strategy_service(docker_image,parameters,port,docker_service):
    # client = docker.DockerClient(base_url='tcp://172.27.231.36:2375')
    env_list = get_env_list(parameters)
    if "-add-host=" in parameters:
        add_host = get_add_host(parameters)
    else:
        add_host = dict()
    client = docker.DockerClient(base_url=docker_service)
    docker_image_repo = docker_image[:docker_image.rindex(":")]
    docker_image_tag = docker_image[docker_image.rindex(":")+1:]
    client.images.pull(docker_image_repo,docker_image_tag)
    container = client.containers.run(
        image=docker_image,
        ports={'80/tcp': port},
        extra_hosts=add_host,
        # command="",
        # #记得带上端口映射
        # volumes=[""],
        detach=True,
        environment=env_list
    )
    return container
    
def socketconnect(addr, port):
    failed = False
    try:
        sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sk.settimeout(2)
        sk.connect((addr, int(port)))
    except:
        failed = True
    return not failed

排序的展示方式

每个比赛会有多个指标，一个是排序指标，其他的是参考指标。类似于线上 AB 测试的时候，也会有很多指标进行展示。
服务器有一个定时的脚本，会周期性的获取全量的提交结果，根据配置的排序指标对结果进行排序。
排序的结果会输出到一个文件中，通过 git push 到代码仓库中进行展示。
属于是一个非常简陋的方案。

def generate_md_report_and_push(benchmark,tables):
    report_content = "# Benchmark {} 榜单\n\n".format(benchmark)
    report_content += "榜单更新时间: {}\n\n".format(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    for table_name in tables:
        markdown = Tomark.table(tables[table_name])
        report_content += "{} *按全数据集下的{}指标降序排列\n\n".format(table_name, ranking_description_map[benchmark])
        report_content += markdown
        report_content += "\n\n"
    report_filepath = os.path.join(os.path.dirname(os.path.abspath(__file__)), "{}.md".format(benchmark))
    create_dir_for_filepath(report_filepath)
    f=open(report_filepath,"w")
    f.write(report_content)
    f.close()
    os.system("git commit -a -m update_report")
    os.system("git push")
 
 
if __name__ == '__main__':
    for benchmark_name in get_benchmark_list():
        benchmark_result = generate_ranking_list(benchmark_name)
        report_content = change_to_md(benchmark_result)
        if len(report_content)>0:
            generate_md_report_and_push(benchmark_name,report_content)

弊端

数据集处理方面

最初场景数据集格式简单，是单个 JSON 文件。初始化时，读取 JSON 文件后，传入各竞赛的自定义数据解析函数。
那么导致的问题就是，扩展性很差，比如对于图像识别场景，将图片 base64 编码到 JSON 文件中并不合适。
没有重构时，在数据集初始化阶段，增加了具体竞赛相关的耦合逻辑，比如 A 场景下，更改为解压缩一个压缩包文件。

for ds in self.dataset_list:
	ds_name = ds['name']
	f_ds = open(ds['local_path'])
	self.data_content_map[ds_name] = json.load(f_ds)
	f_ds.close()
	self.dataset_index_map[ds_name] = 0
	init_data = __import__("benchmark." + self.benchmark, fromlist=True).init_data
	if init_data != None:
		self.cached_data = init_data(ds_name,self.data_content_map[ds_name],self.strategy_config,self.cached_data)

被测服务形式

最开始，假定被测服务只会使用一个 docker 容器，容器内对解决方案团队的服务进行接口的数据转换处理等。
但是，后面竞赛类别增加、禁止调用外部服务接口后，被测服务就存在不止一个 docker 容器，也可能存在多个执行阶段。
这个的后果是，最初的评估主程序不再适合，几乎竞赛都有一个独立的评估程序。

耦合严重

除了上面关于数据集扩展性差导致的逻辑耦合外
平台的代码与不同的竞赛评估程序逻辑放在一个代码仓库中本身也是一种耦合
当竞赛类别比较少时，这种耦合并不会太大影响
当竞赛类别比较多时，平台内的会包含很多评估程序自己的各种 util 包，config 文件等等，导致项目文件杂乱

程序隔离和管控性差

当其他团队提交策略后，将会直接启动 nohup python 的方案在服务器上运行评估程序进行评估。这就导致后续若同时运行的比赛多 + 提交策略多后，服务器承受不住那么多的程序。
由于评估程序是直接 python 命令行运行的，因此无法做到单个评估程序的资源限制和隔离。可能存在着某个竞赛程序出了 bug 导致占用内存满了，进而导致服务器卡死只能重新启动，进而其他正在运行的评估程序均被中断的情况。
直接监控服务器上的 python 程序属于缺失状态。再次，当前程序在运行的时候，是每有一次策略提交，就会直接启动一个评估程序。

🪴 Obsidian Publish

探索