Python职位推荐系统实战：从数据清洗到FastAPI服务化落地

简介这份资源面向具备一定Python基础、希望了解推荐系统落地流程的开发者与在校学生围绕职位推荐场景提供一套可运行的完整项目代码与配套说明。压缩包共79个文件约942KB其中47个py脚本承担数据读取、协同过滤、冷启动与评估等核心逻辑17张png图表用于展示实验结果另有html页面、md说明、csv推荐结果、docx流程文档与ini配置文件结构清晰、便于按模块查阅。项目包含itemCF_IUF与userCF_IIF两类推荐算法实现并配有user_cold_start冷启动处理、evaluate评估脚本及plot可视化脚本可帮助读者理解从数据获取、过滤、推荐计算到结果展示的完整链路。目前已有1130人学习下载适合作为课程设计、毕业设计或推荐算法入门实践的参考案例也可在此基础上替换数据集进行二次开发。1. 从一份简历石沉大海说起职位推荐系统到底在解决什么投了三十份简历没有回音问题往往不在候选人而在匹配环节根本没跑起来。基于 Python 的职位推荐系统设计与实现要解决的就是把「人找岗位」和「岗位找人」这两件事从人工翻页变成可计算、可复现的排序问题。它适合两类人一类是手里有招聘数据、想做冷启动推荐的后端或数据工程师另一类是想拿一个完整项目练手 Python 工程能力的学生和转行者。这个系统的核心不是算法多花哨而是把职位文本、用户画像、行为日志三路数据接进一条能跑通的流水线最后输出一个带解释的推荐列表。下面按「数据怎么来、模型怎么选、服务怎么搭、坑在哪」四段推进每一步都给可抄的代码和参数。2. 职位推荐系统的数据层从原始 JD 到可训练样本2.1 职位文本清洗与结构化字段抽取招聘网站的 JD 通常是一大段混杂文本包含岗位职责、任职要求、公司介绍、福利标签。直接丢给模型效果很差因为噪声词占比高。常见做法是先做字段切分再对职责和要求两段分别处理。我一般用正则加关键词词典做粗切再用 jieba 做细粒度分词最后把技能词映射到统一标签体系。import re import jieba import jieba.posseg as pseg # 技能词典实际项目从配置文件加载 SKILL_DICT {python, java, sql, redis, docker, k8s, pytorch, tensorflow} def clean_jd(raw_text: str) - dict: # 按常见小标题切分保留职责和要求 sections re.split(r(岗位职责|任职要求|职位要求|工作内容)[:]?, raw_text) duty, require , for i, seg in enumerate(sections): if seg in (岗位职责, 工作内容): duty sections[i 1] if i 1 len(sections) else if seg in (任职要求, 职位要求): require sections[i 1] if i 1 len(sections) else # 抽取技能词统一小写 words [w.word.lower() for w in pseg.cut(require) if w.flag in (eng, n)] skills sorted({w for w in words if w in SKILL_DICT}) return {duty: duty.strip(), require: require.strip(), skills: skills} sample 岗位职责负责推荐算法开发。任职要求熟悉Python、SQL了解Docker。 print(clean_jd(sample))这段代码的关键在re.split的分组捕获它让分隔符本身留在结果里方便定位后续段落。pseg.cut的词性过滤eng和n能挡掉大量虚词。参数上SKILL_DICT建议单独放 YAML 文件方便运营同学维护不要硬编码在函数里。清洗后的skills字段是后续做倒排索引和向量化的基础如果这一步漏抽后面召回率会直接掉一截。2.2 用户画像与行为日志的拼接用户侧数据分显式和隐式。显式是简历里的技能、期望城市、期望薪资隐式是点击、收藏、投递、停留时长。很多教程只讲显式结果冷启动用户一多系统就废了。我的做法是给隐式行为算一个加权分投递权重最高收藏次之点击最低停留时长做对数压缩防止长尾主导。import math from collections import defaultdict BEHAVIOR_WEIGHT {apply: 5.0, collect: 3.0, click: 1.0} def build_user_profile(user_id: int, behaviors: list) - dict: skill_score defaultdict(float) city_counter defaultdict(int) for b in behaviors: w BEHAVIOR_WEIGHT.get(b[type], 0.5) # 停留时长做 log 压缩避免个别超长浏览主导 dwell math.log1p(b.get(dwell_sec, 0)) * 0.1 for s in b.get(job_skills, []): skill_score[s] w dwell city_counter[b.get(city, )] 1 top_skills sorted(skill_score.items(), keylambda x: -x[1])[:20] top_city max(city_counter.items(), keylambda x: x[1])[0] if city_counter else return {user_id: user_id, skills: top_skills, city: top_city}BEHAVIOR_WEIGHT是业务可调的投递给 5 分是因为它代表强意图。math.log1p处理停留时长避免某个用户挂机两小时把画像带偏。top_skills截断到 20 个是为了控制后续向量维度太多会让相似度计算被长尾技能稀释。这一步产出的字典结构直接喂给下一章的召回模块。2.3 样本构造正负样本怎么划才不偏推荐系统训练样本的核心是正负样本定义。职位推荐里投递和收藏算正样本曝光未点击算负样本。但直接这么划会有位置偏差——排在前面的职位天然点击高。常见做法是对负样本做降采样并给曝光位置加一个纠偏权重。我一般按 1:4 的正负比构造负样本从曝光未点击里随机抽同时记录曝光位置用于后续加权。import random def build_samples(impressions: list, positives: set, neg_ratio: int 4): pos, neg_pool [], [] for imp in impressions: pair (imp[user_id], imp[job_id]) if pair in positives: pos.append({**imp, label: 1}) else: neg_pool.append({**imp, label: 0}) random.shuffle(neg_pool) neg neg_pool[: len(pos) * neg_ratio] return pos negneg_ratio设 4 是经验值太小模型学不到区分度太大正样本被淹没。random.shuffle前建议先按时间排序再抽避免用未来数据预测过去。这个样本集直接决定模型上限如果正样本定义太宽比如把「查看详情」也算正模型会学出一堆弱意图线上 CTR 看着高但投递转化差。3. 召回与排序两阶段推荐怎么落地3.1 基于技能倒排的召回层实现召回层目标是从几十万职位里快速捞出几百个候选要求快且别漏。最稳的方案是技能倒排加城市过滤。把每个技能映射到职位 ID 列表用户画像里的技能逐个查倒排表合并去重后按技能命中数排序。这个方案不需要 GPU单机 Redis 就能扛。from collections import defaultdict def build_inverted_index(jobs: list) - dict: index defaultdict(set) for job in jobs: for skill in job[skills]: index[skill].add(job[job_id]) return {k: list(v) for k, v in index.items()} def recall_by_skills(user_profile: dict, index: dict, city: str, top_k: int 300): hit_count defaultdict(int) for skill, _ in user_profile[skills]: for job_id in index.get(skill, []): hit_count[job_id] 1 # 城市过滤放在排序后避免过早截断 ranked sorted(hit_count.items(), keylambda x: -x[1])[:top_k] return [jid for jid, _ in ranked]build_inverted_index用set去重防止同一职位重复技能把命中数刷高。recall_by_skills里城市过滤我故意放在最后因为如果先过滤城市跨城投递的候选人会被误杀。top_k设 300 是给排序层留足空间太小会漏掉长尾好职位太大排序层压力大。这个召回层召回率通常能到 70% 以上剩下的靠向量召回补。3.2 排序层LightGBM 特征工程与参数排序层用 LightGBM 是性价比最高的选择特征可解释、训练快、调参直观。特征分四组用户侧技能匹配数、期望薪资差、职位侧薪资、公司规模、发布时间、交叉特征技能重合度、城市是否一致、行为特征历史点击率。下面是一个最小可跑的训练脚本。import lightgbm as lgb import pandas as pd from sklearn.model_selection import train_test_split FEATURES [skill_match, salary_gap, city_match, company_size, post_age_days, user_ctr] def train_ranker(df: pd.DataFrame): X df[FEATURES] y df[label] X_train, X_val, y_train, y_val train_test_split(X, y, test_size0.2, random_state42) params { objective: binary, metric: auc, learning_rate: 0.05, num_leaves: 31, min_data_in_leaf: 50, feature_fraction: 0.8, bagging_fraction: 0.8, bagging_freq: 5, verbose: -1, } dtrain lgb.Dataset(X_train, y_train) dval lgb.Dataset(X_val, y_val, referencedtrain) model lgb.train(params, dtrain, num_boost_round500, valid_sets[dval], callbacks[lgb.early_stopping(50)]) return modellearning_rate0.05 配 500 轮是稳的组合想快可以调到 0.1 但 AUC 会掉一点。num_leaves31 是防止过拟合的保守值数据量上百万可以加到 63。min_data_in_leaf50 很关键职位推荐里很多特征稀疏叶子太小会记住噪声。early_stopping(50)是后悔药验证集 50 轮不涨就停省得手动盯。训练完记得用model.feature_importance()看特征贡献如果post_age_days权重异常高说明模型在偷懒用发布时间得做时间衰减处理。3.3 冷启动新用户和新职位怎么推新用户没行为新职位没曝光这是推荐系统的经典难题。我的做法分两路新用户走热门加规则按城市和学历筛一批高投递率职位新职位给一个探索流量池按技能标签匹配给相似用户曝光同时记录反馈快速更新画像。规则层代码很短但能兜住 80% 的冷启动场景。def cold_start_recall(city: str, degree: str, hot_jobs: list, top_k: int 50): filtered [j for j in hot_jobs if j[city] city and j.get(degree, ) degree] filtered.sort(keylambda x: -x.get(apply_rate, 0)) return [j[job_id] for j in filtered[:top_k]]degree比较用字符串排序是简化写法实际项目建议映射成数值等级。apply_rate是职位历史投递率没有就用曝光点击率代替。新职位探索建议给 10% 到 20% 的流量太少学不出来太多伤用户体验。这个模块上线后要盯两个指标新用户首日投递率、新职位 7 日曝光量掉任何一个都得回查规则阈值。4. 服务化与工程落地从脚本到能扛流量的接口4.1 FastAPI 封装推荐接口模型训完不能只躺在 notebook 里得包成接口。FastAPI 是当前 Python 服务化的主流选择异步支持好、自动生成文档。下面是一个最小推荐接口包含画像查询、召回、排序、返回四步。from fastapi import FastAPI from pydantic import BaseModel import redis import pickle app FastAPI() r redis.Redis(hostlocalhost, port6379, db0) class RecRequest(BaseModel): user_id: int city: str top_n: int 20 app.post(/recommend) def recommend(req: RecRequest): profile pickle.loads(r.get(fprofile:{req.user_id}) or b{}) if not profile: return {jobs: cold_start_recall(req.city, 本科, load_hot_jobs(), req.top_n)} candidates recall_by_skills(profile, load_index(), req.city, 300) features build_features(profile, candidates) scores model.predict(features) ranked sorted(zip(candidates, scores), keylambda x: -x[1])[: req.top_n] return {jobs: [jid for jid, _ in ranked]}pickle.loads读 Redis 里的画像生产环境建议换 JSON 或 MessagePackpickle 有版本兼容风险。load_index和load_hot_jobs应该做进程内缓存别每次请求都读文件。top_n默认 20 是移动端一屏的量PC 端可以调到 50。接口返回只给 job_id 列表职位详情走单独的缓存接口这样推荐接口的 RT 能压到 50ms 以内。4.2 缓存策略与降级方案推荐接口最怕两件事Redis 挂了、模型加载慢。缓存策略上用户画像缓存 1 小时召回结果缓存 10 分钟排序结果不缓存因为个性化太强。降级方案要提前写好Redis 不可用时走本地 LRU 缓存模型不可用时走规则排序。这些分支不写线上出问题就是黑匣子。from functools import lru_cache lru_cache(maxsize10000) def load_hot_jobs_cached(): return load_hot_jobs() def safe_recommend(req): try: return recommend(req) except redis.ConnectionError: # 降级到本地热门 return {jobs: cold_start_recall(req.city, 本科, load_hot_jobs_cached(), req.top_n)} except Exception: # 兜底返回空列表前端展示默认推荐位 return {jobs: []}lru_cache的maxsize10000 是内存和命中率的折中太小频繁淘汰太大吃内存。降级分支里except Exception要记日志别静默吞掉否则排查时两眼一抹黑。这套降级逻辑上线前一定要做混沌测试手动 kill Redis 看接口是否还返回 200。4.3 离线评估与线上 AB 指标模型上线前必须过离线评估AUC 只是门槛还要看 RecallK 和 NDCG。线上则看 CTR、投递转化率、人均投递数。离线好线上差是常态因为离线样本有位置偏差。我的习惯是离线 AUC 涨 0.01 以上才考虑上线上线后先跑 5% 流量一周投递转化不涨就回滚。指标离线目标线上目标说明AUC 0.75—排序区分度Recall100 0.6—召回覆盖CTR—提升 5%点击率投递转化率—提升 3%核心指标接口 P99— 200ms性能底线表格里线上目标写「提升」是相对旧版绝对值因业务而异。AB 实验要注意分流均匀按 user_id 哈希分流比随机分流稳。指标观察期至少 7 天覆盖工作日和周末否则结论不可靠。5. 避坑与排查那些让我加班到凌晨的坑5.1 技能词典不统一导致召回漏召现象用户画像里有「py」职位要求写「Python」倒排索引匹配不上召回率骤降。原因技能词没有做归一化大小写、缩写、中英文混用各写各的。解决建一个同义词映射表入库前统一转小写并映射到标准词比如 py/python 都映射到 pythonk8s/kubernetes 映射到 kubernetes。这个表要持续维护新技能词出现就补。5.2 训练样本时间穿越现象离线 AUC 0.85上线后 CTR 还不如旧规则。原因构造样本时用了未来行为做特征比如用「用户后来投递了」反推「当时该推荐」。解决所有特征必须加时间戳约束只能用预测时间点之前的数据。我一般把样本按时间切分前 80% 训练后 20% 验证绝不随机切。5.3 Redis 大 key 拖垮接口现象推荐接口偶发超时P99 从 80ms 飙到 2s。原因用户画像存成一个大 JSON单个 key 几 MBRedis 单线程读取阻塞。解决画像拆成多个小 key技能、城市、行为分开放或者改用 Hash 结构按字段读。单个 key 控制在 10KB 以内超过就拆。5.4 模型文件加载阻塞服务启动现象服务重启后前几分钟接口全超时。原因模型文件几百 MB启动时同步加载期间请求全排队。解决模型加载放后台线程加载完成前走降级规则或者用模型服务单独部署推荐接口通过 RPC 调用。我倾向后者模型更新不影响接口进程。5.5 冷启动流量给太多伤体验现象新职位探索流量给到 30%老用户投诉推荐不准。原因探索流量挤占了精准推荐的位置。解决探索流量控制在 10% 以内且只对活跃度低的用户开放。同时给探索职位加一个快速反馈通道曝光后没点击就降权别让烂职位一直占坑。6. 进阶技巧用技能共现矩阵提升召回多样性召回层只按技能命中数排序容易推出一堆同质化职位用户翻两页就腻。我的解法是加一个技能共现矩阵衡量职位之间的技能相似度对召回结果做多样性重排。思路很简单两个职位共享的技能越多相似度越高重排时对高相似职位做惩罚让列表里技能组合更分散。import numpy as np from itertools import combinations def build_cooccurrence(jobs: list) - dict: cooc {} for job in jobs: skills job[skills] for a, b in combinations(sorted(skills), 2): cooc[(a, b)] cooc.get((a, b), 0) 1 return cooc def diversity_rerank(candidates: list, jobs_map: dict, cooc: dict, top_k: int 20, penalty: float 0.3): selected [] for jid in candidates: if len(selected) top_k: break skills set(jobs_map[jid][skills]) # 计算与已选职位的最大相似度 max_sim 0.0 for sid in selected: s_skills set(jobs_map[sid][skills]) shared skills s_skills sim sum(cooc.get(tuple(sorted(p)), 0) for p in combinations(shared, 2)) if len(shared) 1 else 0 max_sim max(max_sim, sim) # 相似度越高得分惩罚越大 if max_sim * penalty 1.0: selected.append(jid) return selectedbuild_cooccurrence用组合遍历职位量十万级时建议用稀疏矩阵存别用 dict 硬扛。penalty0.3 是经验值调大多样性更强但相关性会掉调小则效果不明显。max_sim只算共享技能的两两共现简化了计算但够用。这个重排放在排序层之后、返回之前对 RT 影响很小但用户翻页深度能提升 15% 左右。验证多样性是否生效别只看感觉算一个列表内技能熵。熵越高说明技能分布越散一般重排后熵能涨 0.2 以上。如果熵没涨检查cooc是不是没更新或者penalty设太小被相似度淹没了。我踩过的坑是共现矩阵用全量职位构建新职位技能没进矩阵导致新职位永远被惩罚后来改成按周增量更新才解决。这套系统从数据清洗到线上服务最花时间的从来不是模型调参而是数据管道和降级逻辑。我现在的习惯是每加一个特征先问它线上挂了怎么办答不上来就不加。希望帮到你。本文还有配套的精品资源点击获取