resume-flow 0.1.1

LangGraph 简历筛选应用 — 基于 LLM + 规则引擎的自动化候选人评估工具

Resume Flow

基于 LangGraph 的简历筛选应用，使用 LLM + 规则引擎双重验证机制自动化评估候选人。

核心特性

• 五维评估体系：姓名有效性、FAANG 公司经历、学历毕业年份、软件技术岗位、技术方向
• 双重验证机制：每个 LLM 判断后紧跟规则引擎交叉验证，冲突时以规则为准
• 技术方向重试循环：第二个 LLM 独立审核，不一致时自动重试（最多 3 次）
• 断点续传：基于 SqliteSaver 的 Checkpoint 持久化，中断后可从断点恢复
• 多格式输出：JSON 单文件结果 + _summary.json 汇总 + Rich 终端美化 + Excel 报表
• 交互式配置：resume-flow init 一键生成 .env 配置文件
• 全量评估：五个维度全部判断，不因某项失败而短路，便于全面了解候选人

---

目录

• 一、项目概述
• 二、架构说明
  • 2.1 模块结构
  • 2.2 LangGraph 工作流
  • 2.3 状态流转机制
  • 2.4 双重验证机制
  • 2.5 技术方向重试循环
• 三、开发指南
  • 3.1 环境搭建
  • 3.2 常见修改场景
  • 3.3 配置文件说明
  • 3.4 CLI 参数说明
• 四、扩展教程：新增评估节点

---

一、项目概述

Resume Flow 是一个自动化简历筛选工具，面向技术招聘场景。它读取 JSON 格式的简历文件，通过 LangGraph 构建的状态图工作流，依次对候选人进行五个维度的评估，最终判定是否符合条件并为合格候选人生成招聘邮件。

设计理念：LLM 擅长理解语义但可能产生幻觉，规则引擎确定性强但无法处理模糊表述。两者结合，取各自所长，实现高准确率的自动化筛选。

输入：resume_json/ 目录下的 JSON 简历文件（包含 name、education、experience、skills 等字段）

输出：

• resume_json_eval/<filename>.json — 每份简历的详细评估结果
• resume_json_eval/_summary.json — 汇总统计
• resume_json_eval/eval_resume.xlsx — Excel 报表（含符合条件/不符合/出错三个 Sheet）

---

二、架构说明

2.1 模块结构

resume_flow/
├── __init__.py          # 包初始化，导出 ResumeState / build_graph / LLMClient
├── __main__.py          # python -m resume_flow 入口
├── cli.py               # Typer CLI 入口 (init / run 子命令)
├── config.py            # 配置加载、常量定义、日志初始化
├── state.py             # LangGraph State 定义 (ResumeState)
├── graph.py             # LangGraph StateGraph 构建
├── llm.py               # LLM 客户端封装 (OpenAI 兼容)
├── output.py            # JSON / Excel / Rich 终端输出
└── nodes/               # 工作流节点
    ├── __init__.py      # 节点函数导出
    ├── load_resume.py   # 加载简历 JSON
    ├── check_name.py    # 姓名有效性判断 + 验证
    ├── check_company.py # 公司判断 + 验证 (含全局 LLM 工具函数)
    ├── check_grad.py    # 毕业年份判断 + 验证
    ├── check_title.py   # 职位名称判断 + 验证
    ├── check_tech.py    # 技术方向判断 + 验证 (含重试循环)
    ├── finalize.py      # 汇总结果，判定 qualified / disqualified
    └── generate_email.py # 为合格候选人生成招聘邮件

各模块职责：

模块	职责
config.py	日志初始化（Rich + 文件）；加载 .env 配置到 env_config 字典；定义常量（目标公司、年份范围、输出路径）
state.py	定义 ResumeState(TypedDict)，包含 30+ 字段，描述工作流中所有中间状态
graph.py	构建 LangGraph StateGraph，注册 13 个节点，定义边和条件分支
llm.py	封装 OpenAI 兼容客户端，支持同步/异步 chat / chat_json，内置指数退避重试和 markdown 代码块 JSON 提取
output.py	单结果保存 (save_result)、汇总保存 (save_summary)、断点续传检查 (is_already_processed)、Rich 终端打印 (print_summary)、Excel 输出 (save_excel)
cli.py	Typer CLI 入口，init 子命令生成 .env，run 子命令执行主流程
nodes/check_company.py	除自身节点逻辑外，提供 call_llm_json / call_llm_text 全局工具函数，被其他节点模块引用

2.2 LangGraph 工作流

graph.py 中的 build_graph() 构建了一个 13 节点的线性状态图，含 2 个条件分支：

load_resume
    │
    ▼
check_name ──→ verify_name
    │
    ▼
check_company ──→ verify_company
    │
    ▼
check_grad_year ──→ verify_grad_year
    │
    ▼
check_title ──→ verify_title
    │
    ▼
check_tech ──→ verify_tech ──┬── confirmed ──────────→ finalize
                              │── conflict_need_retry ─→ check_tech (重试)
                              │── failed_after_retry ──→ finalize
                              │
                              ▼
                          finalize ──────┬── 五项全 pass ──→ generate_email ──→ END
                                         │── 任一 fail ────→ END

节点执行顺序：

#	节点	函数	说明
1	load_resume	load_resume()	读取 JSON 简历文件，提取 name
2	check_name	check_name()	LLM 判断姓名是否为有效人名
3	verify_name	verify_name()	规则引擎验证（占位符/纯数字/重复字符检测）
4	check_company	check_company()	LLM 判断是否有 FAANG 工作经历
5	verify_company	verify_company()	规则引擎验证（公司名字符串匹配）
6	check_grad_year	check_grad_year()	LLM 按学历层级判断毕业年份
7	verify_grad_year	verify_grad_year()	规则引擎验证（解析教育经历日期）
8	check_title	check_title()	LLM 判断当前职位是否为软件技术岗
9	verify_title	verify_title()	规则引擎验证（职位关键词匹配）
10	check_tech	check_tech()	LLM 综合判断是否技术方向
11	verify_tech	verify_tech()	第二个 LLM 独立审核，不一致触发重试
12	finalize	finalize()	汇总五项结果，保存 JSON，判定合格/不合格
13	generate_email	generate_email()	仅为合格候选人生成招聘邮件，更新结果文件

条件分支：

• after_verify_tech：verify_tech 返回后，若 tech_check_verified == "conflict_need_retry" 且未超过 3 次，回到 check_tech 重试；否则进入 finalize
• after_finalize：finalize 后，若五项全 pass 进入 generate_email；否则直接 END

2.3 状态流转机制

所有节点共享一个 ResumeState（TypedDict），每个节点接收当前 state 并返回一个 partial dict（仅包含需要更新的字段），LangGraph 自动合并。

class ResumeState(TypedDict):
    # 输入
    file_path: str              # 简历 JSON 文件路径
    resume_data: dict           # 解析后的简历数据
    name: str                   # 候选人姓名

    # Step 1: 姓名
    name_check_result: str      # "pass" | "fail"
    name_check_reason: str
    name_check_verified: str    # "confirmed" | "conflict"
    detected_name: str

    # Step 2: 公司
    company_check_result: str
    company_check_reason: str
    company_check_verified: str
    company_matched: str

    # Step 3: 毕业年份
    grad_year_check_result: str # "pass" | "fail" | "unknown"
    grad_year_check_reason: str
    grad_year_check_verified: str
    matched_degree_type: str    # "bachelor" | "master" | "phd" | "unknown"
    matched_grad_year: int
    matched_school: str

    # Step 4: 职位
    title_check_result: str
    title_check_reason: str
    title_check_verified: str
    title_matched_role: str

    # Step 5: 技术方向
    tech_check_result: str
    tech_check_reason: str
    tech_check_verified: str
    tech_category: str
    tech_check_retry_count: int

    # 邮件
    email_subject: str
    email_greetings: str
    email_content: str
    email_signature: str

    # 结果
    final_result: str           # "qualified" | "disqualified"
    disqualify_reason: str
    error: str

典型流转示例（以 check_name 为例）：

1. load_resume 返回 {"resume_data": {...}, "name": "Alex Yu"}
2. check_name 读取 state["name"]，调用 LLM，返回 {"name_check_result": "pass", "name_check_reason": "...", "detected_name": "Alex Yu"}
3. verify_name 读取 state["name"] 和 state["name_check_result"]，用规则引擎验证，返回 {"name_check_verified": "confirmed"}

2.4 双重验证机制

每个评估维度由一对节点完成：check（LLM 判断）→ verify（规则引擎验证）。

设计意图：

• LLM 擅长理解模糊表述（如 "Bachelor of Science in Computer Science, 2015-2019"），但可能产生幻觉
• 规则引擎基于确定性逻辑（字符串匹配、正则提取），不会幻觉但无法处理复杂语义
• 两者交叉验证，提高准确率

冲突处理策略：

• 一致 → verified = "confirmed"，继续流程
• 不一致 → verified = "conflict"，以规则引擎为准，覆盖 LLM 结果
• 规则引擎修正结果时，会在 reason 字段标注 "规则修正: ..." 便于追溯

2.5 技术方向重试循环

技术方向（check_tech / verify_tech）是唯一使用 LLM 交叉验证 而非规则引擎的维度，因为技术方向的判断高度依赖语义理解。

流程：

1. check_tech：第一个 LLM 判断候选人是否技术方向
2. verify_tech：第二个 LLM（使用相同模型但独立调用）审核第一个 LLM 的结论
3. 如果两个 LLM 结论一致 → confirmed
4. 如果不一致 → conflict_need_retry，回到 check_tech 重新判断
5. 最多重试 3 次（MAX_TECH_RETRY = 3），超限后默认判定 fail

---

三、开发指南

3.1 环境搭建

# 1. 创建 conda 虚拟环境
conda create -n resume_flow python=3.10 -y
conda activate resume_flow

# 2. 可编辑安装项目
pip install -e .

# 3. 初始化配置文件（交互式）
resume-flow init

# 4. 编辑 .env，填入真实的 API Key
#    配置文件位置: <项目根目录>/.env

# 5. 准备简历 JSON 文件，放入 resume_json/ 目录

# 6. 运行
resume-flow run

3.2 常见修改场景

需求	修改文件	具体位置
修改目标公司列表	config.py	TARGET_COMPANIES
修改本科/硕士/博士毕业年份范围	config.py	TARGET_BACHELOR_YEARS / TARGET_MASTER_YEARS / TARGET_PHD_YEARS
修改默认输出目录	config.py	OUTPUT_DIR
修改 LLM prompt	nodes/check_*.py	各节点函数中的 system 字符串
修改规则引擎逻辑	nodes/check_*.py	各 verify_* 函数
修改技术重试次数	nodes/check_tech.py	MAX_TECH_RETRY
修改 Excel 输出格式	output.py	save_excel()
修改邮件生成 prompt	nodes/generate_email.py	generate_email() 中的 system 字符串
添加新的评估维度	多个文件	参见第四章

3.3 配置文件说明

.env 文件位于项目根目录，由 resume-flow init 生成，格式为 KEY: value：

# API 密钥 (必填)
LLM_API_KEY: sk-your-real-api-key

# API 地址 (必填)
# 支持完整请求地址（自动去除 /chat/completions 后缀）
# 也支持只写到 /v1
LLM_API_URL: https://api.openai.com/v1/chat/completions

# 模型名称 (必填)
MODEL: gpt-4o

配置项	说明	示例
LLM_API_KEY	LLM API 密钥	sk-xxx、key-xxx
LLM_API_URL	OpenAI 兼容接口的完整请求地址。系统自动去除 /chat/completions 后缀转换为 base_url	https://api.openai.com/v1/chat/completions、http://localhost:11434/v1/chat/completions
MODEL	使用的聊天模型名称	gpt-4o、deepseek-chat、qwen-plus

配置读取优先级（从高到低）：

1. 构造函数显式传入
2. .env 文件（env_config）
3. 系统环境变量（LLM_API_KEY / LLM_BASE_URL / LLM_MODEL）
4. 内置默认值

3.4 CLI 参数说明

# 查看所有帮助
resume-flow --help

# 初始化配置
resume-flow init

# 运行（使用默认参数）
resume-flow run

# 运行（自定义参数）
resume-flow run -i /data/resumes -o /data/results -e /data/results/report.xlsx

# 清除旧结果重新运行
resume-flow run --fresh

# 不使用 checkpoint 持久化
resume-flow run --no-checkpoint

resume-flow run 参数：

长选项	短选项	默认值	说明
--input	-i	<项目根>/resume_json	输入目录：简历 JSON 文件所在路径
--output	-o	<项目根>/resume_json_eval	输出目录：JSON 结果、汇总、Excel 的存放目录
--excel	-e	<输出目录>/eval_resume.xlsx	Excel 文件完整路径，仅在需要自定义文件名时使用
--fresh	—	False	清除输出目录和 checkpoint 数据库，从头重新运行
--no-checkpoint	—	False	不使用 SqliteSaver，改用内存模式（不保留断点续传）

---

四、扩展教程：新增评估节点

以新增一个 "语言能力评估" 维度为例，演示从零到一的完整步骤。

步骤 1：在 state.py 中添加新字段

# resume_flow/state.py

class ResumeState(TypedDict):
    # ... 现有字段 ...

    # Step 6: 语言能力评估 (新增)
    language_check_result: str      # "pass" | "fail"
    language_check_reason: str
    language_check_verified: str    # "confirmed" | "conflict"
    language_matched: str           # 匹配到的语言能力描述

步骤 2：创建节点文件 nodes/check_language.py

"""Node: 语言能力评估 + 验证"""

from __future__ import annotations

from resume_flow.config import logger
from resume_flow.state import ResumeState
from resume_flow.nodes.check_company import call_llm_json


def check_language(state: ResumeState) -> dict:
    """LLM 判断候选人是否具备目标语言能力"""
    resume = state["resume_data"]
    name = state["name"]
    logger.info("-" * 40)
    logger.info(f"[语言评估] 分析 {name} ...")

    languages = resume.get("languages", [])
    bio_summary = resume.get("bio_summary", "")

    system = """你是语言能力评估专家。判断候选人是否具备流利的中文或英文沟通能力。
返回JSON:
{
    "is_language_qualified": true/false,
    "matched_languages": ["匹配到的语言"],
    "reason": "判断理由"
}"""

    user = f"""姓名: {name}
已知语言: {', '.join(languages)}
简介: {bio_summary}
请判断该候选人是否具备流利的中文或英文沟通能力。"""

    result = call_llm_json(system, user)
    is_pass = result.get("is_language_qualified", False)
    matched = result.get("matched_languages", [])
    reason = result.get("reason", "")

    return {
        "language_check_result": "pass" if is_pass else "fail",
        "language_check_reason": reason,
        "language_matched": ", ".join(matched),
    }


def verify_language(state: ResumeState) -> dict:
    """规则引擎验证语言判断结果"""
    resume = state["resume_data"]
    name = state["name"]
    llm_result = state.get("language_check_result", "")
    logger.info(f"[语言验证] 验证 {name} ...")

    # 规则引擎：检查 languages 字段中是否包含中文/英文关键词
    languages = [lang.lower() for lang in resume.get("languages", [])]
    target_keywords = ["chinese", "mandarin", "english", "中文", "英文", "汉语"]
    rule_matched = [lang for lang in languages if any(kw in lang for kw in target_keywords)]
    rule_pass = len(rule_matched) > 0

    if (rule_pass and llm_result == "pass") or (not rule_pass and llm_result == "fail"):
        return {"language_check_verified": "confirmed"}
    else:
        logger.warning(f"[语言验证] 冲突! 以规则为准")
        return {
            "language_check_result": "pass" if rule_pass else "fail",
            "language_check_reason": f"规则修正: 匹配={rule_matched}",
            "language_check_verified": "conflict",
            "language_matched": ", ".join(rule_matched),
        }

步骤 3：在 nodes/__init__.py 中导出

# resume_flow/nodes/__init__.py

from resume_flow.nodes.check_language import check_language, verify_language

__all__ = [
    # ... 现有导出 ...
    "check_language",
    "verify_language",
]

步骤 4：在 graph.py 中注册节点和边

# resume_flow/graph.py

from resume_flow.nodes import (
    # ... 现有导入 ...
    check_language,
    verify_language,
)

def build_graph(checkpointer) -> StateGraph:
    workflow = StateGraph(ResumeState)

    # ... 现有节点注册 ...
    workflow.add_node("check_language", check_language)
    workflow.add_node("verify_language", verify_language)

    # 修改边：在 verify_title 之后插入语言评估
    # 原来: verify_title → check_tech
    # 现在: verify_title → check_language → verify_language → check_tech
    workflow.add_edge("verify_title", "check_language")       # 新增
    workflow.add_edge("check_language", "verify_language")     # 新增
    workflow.add_edge("verify_language", "check_tech")         # 新增（替代原来的 verify_title → check_tech）

    # 删除原来的:
    # workflow.add_edge("verify_title", "check_tech")

    # ... 其余不变 ...

步骤 5：在 cli.py 的 initial_state 中添加初始值

# resume_flow/cli.py — run() 函数中的 initial_state dict

initial_state = {
    # ... 现有字段 ...

    # Step 6: 语言能力评估 (新增)
    "language_check_result": "",
    "language_check_reason": "",
    "language_check_verified": "",
    "language_matched": "",
}

步骤 6：更新 finalize.py 判定逻辑

# resume_flow/nodes/finalize.py

def finalize(state: ResumeState) -> dict:
    # ... 现有判断 ...
    language_pass = state.get("language_check_result") == "pass"  # 新增

    # 修改 all_pass 条件
    if name_pass and company_pass and grad_pass and title_pass and tech_pass and language_pass:
        final = "qualified"
    else:
        final = "disqualified"
        reasons = []
        # ... 现有原因 ...
        if not language_pass:                                    # 新增
            reasons.append("语言能力不符合要求")

    # 更新 disqualify_reason 格式
    return {
        "final_result": final,
        "disqualify_reason": "" if final == "qualified" else
            f"姓名:{'PASS' if name_pass else 'FAIL'}, ...语言:{'PASS' if language_pass else 'FAIL'}",
    }

步骤 7：更新 output.py（如需 Excel 输出）

在 _build_result_data() 中添加新维度的结果数据：

# resume_flow/output.py — _build_result_data()

return {
    # ... 现有字段 ...
    "language_check": {                                          # 新增
        "result": state.get("language_check_result", ""),
        "reason": state.get("language_check_reason", ""),
        "verified": state.get("language_check_verified", ""),
        "matched": state.get("language_matched", ""),
    },
}

如需在 Excel 中显示，修改 save_excel() 中 "符合条件" Sheet 的表头和数据行。

测试新节点

单独测试：

# test_check_language.py
from resume_flow.nodes.check_language import check_language, verify_language

mock_state = {
    "name": "Test User",
    "resume_data": {
        "languages": ["English", "Mandarin"],
        "bio_summary": "Software engineer with 5 years experience",
    },
    "language_check_result": "",
    "language_check_reason": "",
    "language_check_verified": "",
    "language_matched": "",
}

result = check_language(mock_state)
print(result)
# 预期: {"language_check_result": "pass", ...}

# 模拟 verify
mock_state.update(result)
verify_result = verify_language(mock_state)
print(verify_result)
# 预期: {"language_check_verified": "confirmed"}

集成测试：

# 准备一份测试简历 JSON
echo '{"name":"Test User","languages":["English","Mandarin"],"education":[],"experience":[]}' > resume_json/test_lang.json

# 运行（使用 --no-checkpoint 避免影响已有数据）
resume-flow run -i resume_json --no-checkpoint

Resume Flow

基于 LangGraph 的简历筛选应用，使用 LLM + 规则引擎双重验证机制自动化评估候选人。

功能

• 姓名有效性判断
• 目标公司 (FAANG) 工作经历匹配
• 毕业年份按学历层级判断 (本科/硕士/博士)
• 软件技术岗位识别
• 技术方向综合判断 (含 LLM 交叉验证 + 重试机制)
• 合格候选人自动生招聘邮件
• 断点续传 + Checkpoint 持久化
• JSON / Excel / Rich 终端多格式输出

安装

pip install -e .