reachflow-cli 0.1.2

LLM-based candidate follow-up flow engine

ReachFlow

基于 LLM 的候选人跟进流程引擎。自动完成 简历筛选（9 步多维度评估）→ 邮件生成（个性化招聘邮件）→ 质量审核（三级检查 + 重试），将一整天的手工筛选压缩为一条命令。

快速开始

# 1. 安装（要求 Python >= 3.10）
pip install -e .

# 2. 配置 LLM 服务（在项目根目录创建 .env）
echo 'LLM_API_KEY=sk-your-key
LLM_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
LLM_MODEL=qwen-plus' > .env

# 3. 运行
reachflow run -c demo/candidate.json -j demo/job.json --config config/flow.yaml

架构概览

┌─────────┐     ┌───────────┐     ┌─────────────────┐
│ cli.py  │────▶│ engine.py │────▶│ steps/           │
│ (Typer) │     │ (流程引擎) │     │ (9步筛选+邮件)   │
└─────────┘     └─────┬─────┘     └───────┬─────────┘
                      │                   │
              ┌───────┴───────┐   ┌──────┴──────┐
              │ checkpoint.py │   │ llm.py      │
              │ recorder.py   │   │ prompts.py  │
              └───────────────┘   └─────────────┘
                      │                   │
              .reachflow/           config/*.yaml
              (断点/日志/结果)       (5个配置文件)

常用命令

reachflow run -c candidate.json -j job.json   # 执行完整流程
reachflow show-steps                          # 查看步骤配置
reachflow list                                # 列出断点
reachflow resume <flow_id>                    # 从断点恢复

---

快速上手指南

核心模块关系

文件	职责
cli.py	Typer CLI 入口，解析参数、加载数据、调用引擎、展示结果表格
engine.py	流程引擎核心：逐步执行、断点保存、异常处理、结果文件输出
step.py	Step 抽象基类 + StepResult 模型 + __init_subclass__ 自动注册
steps/	内置步骤实现（screening / company_check / product_experience_check / email / review / multi_level_review / email_with_retry）
config.py	Pydantic 配置模型（FlowConfig / RecruiterFileConfig）+ YAML 加载器
llm.py	LLM 客户端（OpenAI 兼容 API），含重试、JSON 解析、asyncio.to_thread
checkpoint.py	断点管理器：save / save_error / load / clear
recorder.py	JSON Lines 日志记录器，每步写入 .reachflow/logs/{flow_id}.jsonl
prompts.py	Prompt 模板加载 + {variable} 填充 + 模块级缓存
review_rules.py	审核规则检查器（正则/函数/LLM 三种检查方式）
utils.py	公共工具函数（find_config / extract_school）

九步流水线

#	步骤名	type	方式	作用
1	华人姓名判断	chinese_name_check	LLM	确认候选人为华人
2	职位相关性	job_title_check	LLM	确认从事软件工程
3	跳槽频率	job_hopping_check	LLM	平均在职 ≥ 12 月
4	毕业状态	graduation_status_check	LLM	已毕业 + 有全职经历
5	本科院校	university_check	LLM	211/985/QS500
6	毕业年份	graduation_year_check	函数	年份在配置区间内
7	岗位匹配	skill_job_match	LLM	技能与岗位匹配度 ≥ 60
8	目标公司	company_check	函数 + 可选 LLM	候选人在目标公司工作过
9	产品经验	product_experience_check	函数 + 可选 LLM	候选人做过目标产品/项目
10	�(邮件生成与审核)	email_generation_with_retry	LLM	生成邮件 + 三级审核 + 最多重试 3 次

短路终止与断点续跑

短路终止： 任何步骤返回 StepResult(success=False) → 引擎保存错误断点 → 流程立即终止，后续步骤不执行。

断点续跑：

reachflow list                # 查看可恢复的流程
reachflow resume <flow_id>    # 从断点继续

预期输出

文件产物：

路径	内容	生命周期
.reachflow/output/{flow_id}_result.json	结构化结果（含候选人、职位、各步骤结果、最终邮件）	永久保留
.reachflow/logs/{flow_id}.jsonl	JSON Lines 执行日志	永久保留
.reachflow/checkpoints/{flow_id}.json	断点文件	流程完成后自动清除

---

设计动机与理念

解决什么问题

日常招聘中，HR 需要逐个打开简历判断"是否华人？是否做软件的？跳槽频不频繁？学校行不行？"，筛完 200 人后还要手写个性化邮件并自检质量——一天 6 小时筛简历 + 3 小时写邮件。

ReachFlow 把 200 份简历丢进去，系统自动筛完、写好邮件、检查好质量，HR 只需最后点"发送"。

为什么用流程引擎而非大脚本

一个脚本跑 200 人时，第 87 个人 LLM 超时——整个脚本崩了，前 86 人结果全丢。流程引擎把"一个大脚本"拆成独立小步骤（流水线工位），每步自动存档，某步失败可从断点恢复，改某步规则只改那个工位。

为什么用 LLM

有些判断无法用 if/else 写出来：跳槽频率的时间格式多样（2020.01-2022.03 / Jan 2020 to Present / 两年经验），院校是否 QS 前 500 需要全球排名知识。LLM 做"人类凭直觉就能判断，但写规则写不出来"的事；数字比较（毕业年份 ≥ 2010）用代码，精确可靠。

核心设计决策

决策	说明
YAML 配置驱动	所有阈值放在 YAML 文件里，改个数字保存就行，不用碰代码
Step 自动注册	写 class MyStep(Step, name="my_step") 即自动录入"花名册"，无需手动登记
短路终止	任何一步不通过，后面全部跳过，节省 LLM 调用成本
模板直出优先	配好完整邮件模板 → 直接填变量输出（0.01 秒，不花钱）；模板不完整才调 LLM
断点续跑	每完成一步自动"存档"，reachflow resume 从存档点继续
多级审核 + 重试	LLM 写完邮件 → 三级审核（格式→内容→个性化）→ 没通过带反馈重写，最多 3 次

四条设计原则

• 简单 — 每个步骤就是一个类、一个 execute() 方法，从上往下读就懂
• 可靠 — 每次 LLM 调用都有 try-except 兜底 + 指数退避重试（1s、2s、4s）
• 健壮 — 跑到一半挂了，能从断点继续，而不是从头再来
• 可扩展 — 写一个新类 → 导入 → YAML 加一行，老代码一个字不用动

---

自定义 Step 扩展开发

自动注册原理

ReachFlow 使用 Python 的 __init_subclass__ 钩子实现步骤自动注册：

class Step(ABC):
    _registry: ClassVar[Dict[str, type]] = {}  # 全局注册表

    def __init_subclass__(cls, name: str = None, **kwargs):
        super().__init_subclass__(**kwargs)
        if name:
            Step._registry[name] = cls  # 类定义瞬间自动注册
            cls.name = name

写 class MyStep(Step, name="my_step") 时，Python 在类创建时调用 __init_subclass__，该类被自动注册到 Step._registry["my_step"]。流程引擎通过 Step.create("my_step", config) 工厂方法实例化步骤。

开发步骤（三步）

1. 新建步骤文件（src/reachflow/steps/my_check.py）：

from typing import Dict, Any
from ..step import Step, StepResult

class MyCheckStep(Step, name="my_check"):
    async def execute(self, context: Dict[str, Any]) -> StepResult:
        candidate = context.get("candidate", {})
        # 业务逻辑...
        return StepResult(success=True, data={"reason": "通过"})

2. 在 steps/__init__.py 中$导入（触发自动注册）A

from .my_check import MyCheckStep

3. 在 config/flow.yaml 中添加配置：

  - name: "我的检查"
    type: "my_check"
    enabled: true
    params: {}

context 字段参考

字段	写入时机	类型	说明
candidate	流程启动时（CLI 加载 JSON）	Dict	候选人简历完整数据
job	流程启动时（CLI 加载 JSON）	Dict	职位描述数据
flow_id	流程启动时（引擎自动写入）	str	当前流程唯一标识
results	每步完成后（引擎自动写入）	Dict[str, Dict]	按步骤 name 索引的历史结果
email	邮件生成步骤写入	Dict	最终邮件 {subject, body, tone}
match_result	岗位匹配步骤写入	Dict	技能匹配结果
review_feedback	审核失败时写入（重试循环内）	Dict	审核反馈，供重新生成参考

最佳实践

• 防御性读取：始终用 context.get("field", 默认值)，不假设前序步骤一定执行过
• LLM 调用必须 try-except：await self.llm_client.invoke_json(prompt) 必须包裹异常处理
• 数据缺失返回 success=True（跳过），仅在明确不满足硬性条件时返回 False
• 使用 structlog 结构化日志：logger.info("检查完成", step="my_check", passed=True)

---

配置体系架构

配置文件职责边界

文件	职责	加载方	消费方
flow.yaml	流程编排（步骤顺序、开关、参数传递）	config.py → FlowConfig.from_yaml()	engine.py
prompts.yaml	LLM 提示词模板集中管理	prompts.py → load_prompts()	所有 LLM 步骤
screening_rules.yaml	筛选阈值与区间规则	screening.py → load_screening_rules()	Step 3/5/6/8/9
review_rules.yaml	邮件审核规则（三级分层）	multi_level_review.py	邮件审核子流程
recruiter.yaml	招聘人员身份 + 邮件模板结构	config.py → load_recruiter_config()	邮件生成步骤

核心特征： flow.yaml 是唯一的"入口配置"，其他文件均通过 flow.yaml 中各步骤的 params 字段间接引用。步骤代码不硬编码任何配置文件路径。

配置加载优先级

所有子配置文件共享统一的三级搜索链：

优先级 1：用户显式指定路径（flow.yaml params 中的路径字符串）
    ↓ 不存在
优先级 2：CWD/config/<filename>（当前工作目录）
    ↓ 不存在
优先级 3：项目根/config/<filename>（包目录向上三级）
    ↓ 全部失败
安全回退：返回空字典/默认配置（不抛异常，记录 error 日志）

邮件生成双模式设计

IF subject 已配置 AND 所有 section.type == "template"
THEN → 模板直出（变量替换，0 LLM 调用，~0.01s，输出稳定）
ELSE → LLM 生成（组装 prompt，调用 API，个性化强）

切换方式：修改 recruiter.yaml 中 structure 各 section 的 type 字段即可，无需改代码。

三级审核分层

级别	配置键	检查方式	检查项
Level 1	level_1_format	正则 + 函数	no_markdown / no_placeholder / proper_length / recruiter_identity
Level 2	level_2_content	LLM	content_quality（内容质量）
Level 3	level_3_personalization	LLM	personalization（个性化程度）

执行策略：Level 1 → 2 → 3 !顺序执行，每级内所有 check 必须全部通过。

常见配置修改场景

修改跳槽频率阈值 — 编辑 config/screening_rules.yaml：

job_hopping:
  max_avg_tenure_months: 18  # 从 12 改为 18

禁用某步骤 — 编辑 config/flow.yaml，将 enabled 改为 false。

切换为纯模板直出邮件 — 编辑 config/recruiter.yaml，将所有 section 的 type 改为 "template" 并提供完整模板文本。

配置目标公司/产品筛选 — 编辑 config/screening_rules.yaml：

target_companies:       # 未配置或空时跳过本步骤
  - "Microsoft"
  - "腾讯"

target_products:        # 未配置或空时跳过本步骤
  - "英雄联盟"
  - "GPT"

---

项目结构

reachflow/
├── config/          # YAML 配置（流程编排/Prompt/规则/模板）
│   ├── flow.yaml               # 流程编排：步骤顺序/开关/参数
│   ├── prompts.yaml            # LLM 提示词模板
│   ├── screening_rules.yaml    # 筛选阈值 + 目标公司/产品列表
│   ├── review_rules.yaml       # 邮件审核规则
│   └── recruiter.yaml          # 招聘人员身份 + 邮件模板
├── demo/            # 演示数据（candidate.json / job.json）
├── docs/            # 项目文档
├── src/reachflow/   # 主包（src layout）
│   ├── cli.py       # Typer CLI 入口
│   ├── engine.py    # 流程引擎核心
│   ├── step.py      # Step 基类 + 自动注册
│   ├── steps/       # 内置步骤实现
│   │   ├── screening.py              # 7 步简历筛选
│   │   ├── company_check.py          # 目标公司判断
│   │   ├── product_experience_check.py # 产品/项目经验判断
│   │   ├── email.py                  # 邮件生成
│   │   ├── review.py                 # 单次审核
│   │   ├── multi_level_review.py     # 三级审核
│   │   └── email_with_retry.py       # 生成+审核+重试
│   └── ...          # llm / config / checkpoint / prompts 等
├── tests/           # 测试套件
└── pyproject.toml   # 包配置（hatchling）

License

MIT