Metadata-Version: 2.4
Name: resume-flow
Version: 0.1.1
Summary: LangGraph 简历筛选应用 — 基于 LLM + 规则引擎的自动化候选人评估工具
Author-email: Chandler <275737875@qq.com>
License: MIT
Keywords: langgraph,llm,recruiting,resume,screening
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.10
Requires-Dist: langchain-core>=0.2.0
Requires-Dist: langgraph-checkpoint-sqlite>=2.0.0
Requires-Dist: langgraph>=0.2.0
Requires-Dist: openai>=1.0.0
Requires-Dist: openpyxl>=3.1.0
Requires-Dist: rich>=13.0.0
Requires-Dist: typer>=0.9.0
Provides-Extra: dev
Requires-Dist: mypy>=1.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.21; extra == 'dev'
Requires-Dist: pytest>=7.0; extra == 'dev'
Requires-Dist: ruff>=0.4.0; extra == 'dev'
Description-Content-Type: text/markdown

# Resume Flow

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

## 核心特性

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

---

## 目录

- [一、项目概述](#一项目概述)
- [二、架构说明](#二架构说明)
  - [2.1 模块结构](#21-模块结构)
  - [2.2 LangGraph 工作流](#22-langgraph-工作流)
  - [2.3 状态流转机制](#23-状态流转机制)
  - [2.4 双重验证机制](#24-双重验证机制)
  - [2.5 技术方向重试循环](#25-技术方向重试循环)
- [三、开发指南](#三开发指南)
  - [3.1 环境搭建](#31-环境搭建)
  - [3.2 常见修改场景](#32-常见修改场景)
  - [3.3 配置文件说明](#33-配置文件说明)
  - [3.4 CLI 参数说明](#34-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 自动合并。

```python
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 环境搭建

```bash
# 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`：

```ini
# 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 参数说明

```bash
# 查看所有帮助
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` 中添加新字段

```python
# 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`

```python
"""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` 中导出

```python
# resume_flow/nodes/__init__.py

from resume_flow.nodes.check_language import check_language, verify_language

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

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

```python
# 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` 中添加初始值

```python
# 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` 判定逻辑

```python
# 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()` 中添加新维度的结果数据：

```python
# 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 的表头和数据行。

### 测试新节点

**单独测试**：

```python
# 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"}
```

**集成测试**：

```bash
# 准备一份测试简历 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 终端多格式输出

## 安装

```bash
pip install -e .