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(), key=lambda x: -x[1])[:20] top_city = max(city_counter.items(), key=lambda 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 + neg

neg_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(), key=lambda 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_size=0.2, random_state=42) 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, reference=dtrain) model = lgb.train(params, dtrain, num_boost_round=500, valid_sets=[dval], callbacks=[lgb.early_stopping(50)]) return model

learning_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(key=lambda 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(host="localhost", port=6379, db=0) class RecRequest(BaseModel): user_id: int city: str = "" top_n: int = 20 @app.post("/recommend") def recommend(req: RecRequest): profile = pickle.loads(r.get(f"profile:{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), key=lambda x: -x[1])[: req.top_n] return {"jobs": [jid for jid, _ in ranked]}

pickle.loads读 Redis 里的画像，生产环境建议换 JSON 或 MessagePack，pickle 有版本兼容风险。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(maxsize=10000) 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 只是门槛，还要看 Recall@K 和 NDCG。线上则看 CTR、投递转化率、人均投递数。离线好线上差是常态，因为离线样本有位置偏差。我的习惯是离线 AUC 涨 0.01 以上才考虑上线，上线后先跑 5% 流量一周，投递转化不涨就回滚。

表格里线上目标写「提升」是相对旧版，绝对值因业务而异。AB 实验要注意分流均匀，按 user_id 哈希分流比随机分流稳。指标观察期至少 7 天，覆盖工作日和周末，否则结论不可靠。

5. 避坑与排查：那些让我加班到凌晨的坑

5.1 技能词典不统一导致召回漏召

现象：用户画像里有「py」，职位要求写「Python」，倒排索引匹配不上，召回率骤降。原因：技能词没有做归一化，大小写、缩写、中英文混用各写各的。解决：建一个同义词映射表，入库前统一转小写并映射到标准词，比如 py/python 都映射到 python，k8s/kubernetes 映射到 kubernetes。这个表要持续维护，新技能词出现就补。

5.2 训练样本时间穿越

现象：离线 AUC 0.85，上线后 CTR 还不如旧规则。原因：构造样本时用了未来行为做特征，比如用「用户后来投递了」反推「当时该推荐」。解决：所有特征必须加时间戳约束，只能用预测时间点之前的数据。我一般把样本按时间切分，前 80% 训练后 20% 验证，绝不随机切。

5.3 Redis 大 key 拖垮接口

现象：推荐接口偶发超时，P99 从 80ms 飙到 2s。原因：用户画像存成一个大 JSON，单个 key 几 MB，Redis 单线程读取阻塞。解决：画像拆成多个小 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 selected

build_cooccurrence用组合遍历，职位量十万级时建议用稀疏矩阵存，别用 dict 硬扛。penalty0.3 是经验值，调大多样性更强但相关性会掉，调小则效果不明显。max_sim只算共享技能的两两共现，简化了计算但够用。这个重排放在排序层之后、返回之前，对 RT 影响很小，但用户翻页深度能提升 15% 左右。

验证多样性是否生效，别只看感觉，算一个列表内技能熵。熵越高说明技能分布越散，一般重排后熵能涨 0.2 以上。如果熵没涨，检查cooc是不是没更新，或者penalty设太小被相似度淹没了。我踩过的坑是共现矩阵用全量职位构建，新职位技能没进矩阵，导致新职位永远被惩罚，后来改成按周增量更新才解决。

这套系统从数据清洗到线上服务，最花时间的从来不是模型调参，而是数据管道和降级逻辑。我现在的习惯是每加一个特征，先问它线上挂了怎么办，答不上来就不加。希望帮到你。

本文还有配套的精品资源，点击获取