# resume_flow — Python 文件汇总

> **生成时间**：2026-07-10 09:10:48 UTC  
> **根目录**：`C:\Users\s84336076\IDEProjects\check_lk_json\resume_flow`

## 路径：`__init__.py`

```python
"""Resume Flow - LangGraph 简历筛选应用"""

__version__ = "0.1.0"

from resume_flow.state import ResumeState
from resume_flow.graph import build_graph
from resume_flow.llm import LLMClient

__all__ = ["ResumeState", "build_graph", "LLMClient", "__version__"]

```

---

## 路径：`__main__.py`

```python
"""python -m resume_flow 入口"""

from resume_flow.cli import app

if __name__ == "__main__":
    app()

```

---

## 路径：`cli.py`

```python
"""typer CLI 入口"""

from __future__ import annotations

import os
from pathlib import Path

import typer
from rich.panel import Panel

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver

from resume_flow.config import (
    logger, console, env_config,
    TARGET_COMPANIES, TARGET_BACHELOR_YEARS, TARGET_MASTER_YEARS, TARGET_PHD_YEARS,
    OUTPUT_DIR, CHECKPOINT_DB, BASE_DIR,
)
from resume_flow.graph import build_graph
from resume_flow.output import save_summary, print_summary, save_excel, is_already_processed

app = typer.Typer(help="LangGraph 简历筛选应用", add_completion=False)


@app.command()
def main(
    fresh: bool = typer.Option(False, "--fresh", help="清除 output 目录和 checkpoint 数据库，从头重新运行所有简历"),
    no_checkpoint: bool = typer.Option(False, "--no-checkpoint", help="不使用 SqliteSaver 持久化 checkpoint，改用内存模式"),
    output: str = typer.Option("", "--output", "-o", help="Excel 输出文件路径 (默认: eval_resume_output/eval_resume.xlsx)"),
):
    # --fresh: 清除旧结果和 checkpoint
    if fresh:
        import shutil
        if os.path.exists(OUTPUT_DIR):
            shutil.rmtree(OUTPUT_DIR)
            console.print(f"[yellow]--fresh: 已清除输出目录 {OUTPUT_DIR}[/yellow]")
        if os.path.exists(CHECKPOINT_DB):
            os.remove(CHECKPOINT_DB)
            console.print(f"[yellow]--fresh: 已清除 checkpoint 数据库 {CHECKPOINT_DB}[/yellow]")

    # 启动信息面板
    checkpoint_info = "MemorySaver (内存模式)" if no_checkpoint else f"SqliteSaver ({CHECKPOINT_DB})"
    startup_text = (
        f"[bold]目标公司:[/bold] {', '.join(TARGET_COMPANIES)}\n"
        f"[bold]本科毕业年份:[/bold] {', '.join(str(y) for y in TARGET_BACHELOR_YEARS)}\n"
        f"[bold]硕士毕业年份:[/bold] {', '.join(str(y) for y in TARGET_MASTER_YEARS)}\n"
        f"[bold]博士毕业年份:[/bold] {', '.join(str(y) for y in TARGET_PHD_YEARS)}\n"
        f"[bold]使用模型:[/bold] {env_config.get('model', 'N/A')}\n"
        f"[bold]输出目录:[/bold] {OUTPUT_DIR}\n"
        f"[bold]Checkpoint:[/bold] {checkpoint_info}"
    )
    console.print(Panel(startup_text, title="[bold blue]Resume Flow - LangGraph 简历筛选[/bold blue]", border_style="blue"))

    # 根据 --no-checkpoint 选择 checkpointer
    if no_checkpoint:
        checkpointer = MemorySaver()
        graph = build_graph(checkpointer)
        logger.info("[主流程] LangGraph 工作流构建完成 (MemorySaver 内存模式)")
    else:
        _sqlite_ctx = SqliteSaver.from_conn_string(CHECKPOINT_DB)
        checkpointer = _sqlite_ctx.__enter__()
        graph = build_graph(checkpointer)
        logger.info("[主流程] LangGraph 工作流构建完成 (SqliteSaver 持久化 checkpoint)")

    resume_dir = os.path.join(BASE_DIR, "resume_json")
    json_files = sorted(Path(resume_dir).glob("*.json"))

    if not json_files:
        logger.error(f"[主流程] 未找到简历文件! 目录: {resume_dir}")
        return

    logger.info(f"[主流程] 找到 {len(json_files)} 个简历文件")

    # 检查已完成的简历，实现断点续传
    pending_files = []
    skipped_files = []
    for jf in json_files:
        if is_already_processed(jf.name):
            skipped_files.append(jf)
        else:
            pending_files.append(jf)

    if skipped_files:
        logger.info(f"[主流程] 跳过已完成的 {len(skipped_files)} 份: {', '.join(f.name for f in skipped_files)}")
    if not pending_files:
        logger.info("[主流程] 所有简历已处理完成，无需重新运行")
        print_summary()
        excel_path = output if output else os.path.join(OUTPUT_DIR, "eval_resume.xlsx")
        save_excel(excel_path)
        return

    logger.info(f"[主流程] 待处理 {len(pending_files)} 份简历")

    qualified_list = []
    disqualified_list = []
    error_list = []

    for json_file in pending_files:
        logger.info(f"\n{'#' * 60}")
        logger.info(f"[主流程] 开始处理: {json_file.name}")
        logger.info(f"{'#' * 60}")

        thread_id = f"resume_{json_file.stem}"
        config = {"configurable": {"thread_id": thread_id}}

        initial_state = {
            "file_path": str(json_file),
            "resume_data": {},
            "name": "",
            "name_check_result": "",
            "name_check_reason": "",
            "name_check_verified": "",
            "detected_name": "",
            "company_check_result": "",
            "company_check_reason": "",
            "company_check_verified": "",
            "company_matched": "",
            "grad_year_check_result": "",
            "grad_year_check_reason": "",
            "grad_year_check_verified": "",
            "matched_degree_type": "",
            "matched_grad_year": 0,
            "matched_school": "",
            "title_check_result": "",
            "title_check_reason": "",
            "title_check_verified": "",
            "title_matched_role": "",
            "tech_check_result": "",
            "tech_check_reason": "",
            "tech_check_verified": "",
            "tech_category": "",
            "tech_check_retry_count": 0,
            "email_subject": "",
            "email_greetings": "",
            "email_content": "",
            "email_signature": "",
            "final_result": "",
            "disqualify_reason": "",
            "error": "",
        }

        try:
            result = graph.invoke(initial_state, config)

            if result.get("final_result") == "qualified":
                qualified_list.append({
                    "name": result.get("name"),
                    "file": json_file.name,
                    "email": {
                        "subject": result.get("email_subject", ""),
                        "greetings": result.get("email_greetings", ""),
                        "content": result.get("email_content", ""),
                        "signature": result.get("email_signature", ""),
                    },
                    "company_matched": result.get("company_matched", ""),
                    "matched_degree_type": result.get("matched_degree_type", ""),
                    "matched_grad_year": result.get("matched_grad_year", 0),
                    "matched_school": result.get("matched_school", ""),
                    "tech_category": result.get("tech_category", ""),
                })
            else:
                disqualified_list.append({
                    "name": result.get("name"),
                    "file": json_file.name,
                    "reason": result.get("disqualify_reason", ""),
                })

            save_summary(len(json_files), len(skipped_files), qualified_list, disqualified_list, error_list)

        except Exception as e:
            logger.error(f"[主流程] 处理 {json_file.name} 出错: {e}")
            error_list.append({"file": json_file.name, "error": str(e)})
            save_summary(len(json_files), len(skipped_files), qualified_list, disqualified_list, error_list)

    # ============ 最终汇总 ============
    print_summary()

    # ============ 输出 Excel ============
    excel_path = output if output else os.path.join(OUTPUT_DIR, "eval_resume.xlsx")
    save_excel(excel_path)

    # 关闭 SqliteSaver
    if not no_checkpoint:
        try:
            _sqlite_ctx.__exit__(None, None, None)
        except Exception:
            pass

```

---

## 路径：`config.py`

```python
"""配置加载、常量定义、日志初始化"""

from __future__ import annotations

import os
import logging

from rich.console import Console
from rich.logging import RichHandler

# ============ 日志配置 ============
console = Console()

# 文件日志
file_handler = logging.FileHandler("resume_check.log", encoding="utf-8", mode="a")
file_handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s"))

# 终端日志：Rich 美化
rich_handler = RichHandler(console=console, show_time=True, show_path=False,
                           markup=True, rich_tracebacks=True)

logging.basicConfig(
    level=logging.INFO,
    format="%(message)s",
    datefmt="[%X]",
    handlers=[rich_handler, file_handler],
)
logger = logging.getLogger(__name__)


# ============ 读取 env 配置 ============
def load_env_config(env_path: str) -> dict:
    """读取 env 文件中的 API 配置"""
    config = {}
    with open(env_path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            if ":" in line:
                key, value = line.split(":", 1)
                config[key.strip()] = value.strip()
    return config


# ============ 常量 ============
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
ENV_PATH = os.path.join(BASE_DIR, ".env")
env_config = load_env_config(ENV_PATH)

TARGET_COMPANIES = ["Apple", "Amazon", "Google", "Microsoft", "Meta"]
TARGET_BACHELOR_YEARS = [2012, 2013, 2014, 2015, 2016, 2017, 2018]
TARGET_MASTER_YEARS = [2013, 2014, 2015, 2016, 2017, 2018, 2019, 2020]
TARGET_PHD_YEARS = [2017, 2018, 2019, 2020, 2021, 2022, 2023, 2024]

OUTPUT_DIR = os.path.join(BASE_DIR, "resume_flow_eval")
CHECKPOINT_DB = os.path.join(BASE_DIR, "checkpoints.db")

```

---

## 路径：`graph.py`

```python
"""构建 LangGraph 工作流"""

from __future__ import annotations

from langgraph.graph import StateGraph, END

from resume_flow.state import ResumeState
from resume_flow.nodes import (
    load_resume,
    check_name,
    verify_name,
    check_company,
    verify_company,
    check_grad_year,
    verify_grad_year,
    check_title,
    verify_title,
    check_tech,
    verify_tech,
    generate_email,
    finalize,
)


def after_verify_tech(state: ResumeState) -> str:
    """技术验证后：一致→汇总，不一致且未超限→重试check_tech，超限→汇总"""
    verified = state.get("tech_check_verified", "")
    if verified == "confirmed":
        return "finalize"
    elif verified == "conflict_need_retry":
        return "check_tech"
    else:
        return "finalize"


def after_finalize(state: ResumeState) -> str:
    """汇总后判断：五项全通过则生成邮件，否则结束"""
    name_pass = state.get("name_check_result") == "pass"
    company_pass = state.get("company_check_result") == "pass"
    grad_pass = state.get("grad_year_check_result") == "pass"
    title_pass = state.get("title_check_result") == "pass"
    tech_pass = state.get("tech_check_result") == "pass"
    if name_pass and company_pass and grad_pass and title_pass and tech_pass:
        return "generate_email"
    return END


def build_graph(checkpointer) -> StateGraph:
    """构建 LangGraph 简历筛选工作流

    流程: load_resume → check_name → verify_name → check_company → verify_company
          → check_grad_year → verify_grad_year → check_title → verify_title
          → check_tech → verify_tech → finalize → generate_email
    """
    workflow = StateGraph(ResumeState)

    workflow.add_node("load_resume", load_resume)
    workflow.add_node("check_name", check_name)
    workflow.add_node("verify_name", verify_name)
    workflow.add_node("check_company", check_company)
    workflow.add_node("verify_company", verify_company)
    workflow.add_node("check_grad_year", check_grad_year)
    workflow.add_node("verify_grad_year", verify_grad_year)
    workflow.add_node("check_title", check_title)
    workflow.add_node("verify_title", verify_title)
    workflow.add_node("check_tech", check_tech)
    workflow.add_node("verify_tech", verify_tech)
    workflow.add_node("finalize", finalize)
    workflow.add_node("generate_email", generate_email)

    workflow.set_entry_point("load_resume")

    # 全量评估：五个条件都判断，不短路
    workflow.add_edge("load_resume", "check_name")
    workflow.add_edge("check_name", "verify_name")
    workflow.add_edge("verify_name", "check_company")
    workflow.add_edge("check_company", "verify_company")
    workflow.add_edge("verify_company", "check_grad_year")
    workflow.add_edge("check_grad_year", "verify_grad_year")
    workflow.add_edge("verify_grad_year", "check_title")
    workflow.add_edge("check_title", "verify_title")
    workflow.add_edge("verify_title", "check_tech")
    workflow.add_edge("check_tech", "verify_tech")
    # 技术验证后条件跳转
    workflow.add_conditional_edges("verify_tech", after_verify_tech,
                                   {"finalize": "finalize", "check_tech": "check_tech"})
    # 汇总后：全通过才生成邮件，否则结束
    workflow.add_conditional_edges("finalize", after_finalize,
                                   {"generate_email": "generate_email", END: END})
    workflow.add_edge("generate_email", END)

    return workflow.compile(checkpointer=checkpointer)

```

---

## 路径：`llm.py`

```python
"""轻量级 LLM 客户端，基于 openai 官方库封装"""

from __future__ import annotations

import os
import json
import time
from typing import Optional, Union, Dict, List
from openai import OpenAI
from openai._exceptions import RateLimitError, APITimeoutError

# 设置代理环境变量 (针对华为内网地址)
os.environ['NO_PROXY'] = 'api.openai.rnd.huawei.com'
os.environ['no_proxy'] = 'api.openai.rnd.huawei.com'


class LLMClient:
    """
    轻量级 LLM 客户端，基于 openai 官方库封装。
    支持 OpenAI / DeepSeek / 本地 Ollama 等任何兼容 OpenAI Chat API 的服务。
    """

    def __init__(
        self,
        api_key: Optional[str] = None,
        base_url: Optional[str] = None,
        model: Optional[str] = None,
        timeout: Optional[float] = None,
        max_retries: Optional[int] = None,
    ):
        self.api_key = api_key or os.getenv("LLM_API_KEY", "sk-1234")
        self.base_url = (base_url or os.getenv("LLM_BASE_URL", "http://api.openai.rnd.huawei.com/v1")).rstrip("/")
        self.model = model or os.getenv("LLM_MODEL", "deepseek-v3.1-terminus-chat")
        self.timeout = timeout or float(os.getenv("LLM_TIMEOUT", "120"))
        self.max_retries = max_retries or int(os.getenv("LLM_MAX_RETRIES", "3"))

        self.client = OpenAI(
            api_key=self.api_key,
            base_url=self.base_url,
            timeout=self.timeout,
            max_retries=0
        )

    def chat(self, system: str, user: str, temperature: float = 0.3) -> str:
        """同步调用 Chat Completion API，返回助手回复文本。内置指数退避重试机制。"""
        last_error = None
        messages = [
            {"role": "system", "content": system},
            {"role": "user", "content": user},
        ]

        for attempt in range(self.max_retries):
            try:
                stream = self.client.chat.completions.create(
                    model=self.model,
                    messages=messages,
                    stream=True,
                    temperature=temperature,
                    timeout=self.timeout
                )

                response_content = ""
                for chunk in stream:
                    if chunk.choices and chunk.choices[0].delta.content:
                        response_content += chunk.choices[0].delta.content

                return response_content.strip()

            except (RateLimitError, APITimeoutError) as e:
                last_error = e
                wait = 2 ** attempt
                print(f"  [LLM] 请求失败 (HTTP 429/Timeout)，第 {attempt + 1}/{self.max_retries} 次重试，等待 {wait}s...")
                time.sleep(wait)
                continue

            except Exception as e:
                raise e

        raise ConnectionError(
            f"LLM API 调用失败，已重试 {self.max_retries} 次。最后错误: {last_error}"
        )

    def chat_json(self, system: str, user: str, temperature: float = 0.1) -> Union[Dict, List]:
        """调用 Chat API 并解析 JSON 响应。自动从 markdown 代码块中提取 JSON。"""
        raw = self.chat(system, user, temperature)
        return self._extract_json(raw)

    @staticmethod
    def _extract_json(text: str) -> Union[Dict, List]:
        """从 LLM 回复中提取 JSON（支持 markdown 代码块包裹）"""
        text = text.strip()

        try:
            return json.loads(text)
        except json.JSONDecodeError:
            pass

        if "```json" in text:
            start = text.index("```json") + 7
            end = text.index("```", start)
            return json.loads(text[start:end].strip())

        if "```" in text:
            start = text.index("```") + 3
            end = text.index("```", start)
            return json.loads(text[start:end].strip())

        raise ValueError(f"无法从 LLM 回复中提取 JSON: {text[:200]}")

    async def achat(self, system: str, user: str, temperature: float = 0.3) -> str:
        """异步版本：在线程池中运行同步调用。"""
        import asyncio
        return await asyncio.to_thread(self.chat, system, user, temperature)

    async def achat_json(self, system: str, user: str, temperature: float = 0.1) -> Union[Dict, List]:
        """异步版本：调用并解析 JSON"""
        import asyncio
        return await asyncio.to_thread(self.chat_json, system, user, temperature)

```

---

## 路径：`output.py`

```python
"""JSON/Excel 输出、Rich 终端输出"""

from __future__ import annotations

import json
import os
import re

from rich.table import Table

from resume_flow.config import logger, console, OUTPUT_DIR
from resume_flow.state import ResumeState


# ============ JSON 输出 ============

def save_result(name: str, data: dict, source_file: str = ""):
    """保存单份简历的处理结果到 JSON 文件，文件名与输入文件同名"""
    os.makedirs(OUTPUT_DIR, exist_ok=True)
    if source_file:
        out_name = source_file
    else:
        out_name = re.sub(r'[^\w\-]', '_', name) + ".json"
    out_path = os.path.join(OUTPUT_DIR, out_name)
    with open(out_path, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
    logger.info(f"[输出] 结果已保存: {out_path}")


def _build_result_data(state: ResumeState) -> dict:
    """从 state 中构建完整的结果 dict"""
    name = state["name"]
    name_pass = state.get("name_check_result") == "pass"
    company_pass = state.get("company_check_result") == "pass"
    grad_pass = state.get("grad_year_check_result") == "pass"
    grad_unknown = state.get("grad_year_check_result") == "unknown"
    title_pass = state.get("title_check_result") == "pass"
    tech_pass = state.get("tech_check_result") == "pass"
    all_pass = name_pass and company_pass and grad_pass and title_pass and tech_pass

    degree_type = state.get("matched_degree_type", "unknown")
    degree_label = {"bachelor": "本科", "master": "硕士", "phd": "博士", "unknown": "未知"}.get(degree_type, "未知")

    reasons = []
    if not name_pass:
        reasons.append("姓名无效")
    if not company_pass:
        reasons.append("不在目标公司")
    if grad_unknown:
        reasons.append("毕业年份数据缺失，无法判断")
    elif not grad_pass:
        reasons.append(f"{degree_label}毕业年份({state.get('matched_grad_year', 'N/A')})不在目标范围")
    if not title_pass:
        reasons.append("非软件技术岗位")
    if not tech_pass:
        reasons.append("非技术方向")

    return {
        "name": name,
        "file": os.path.basename(state.get("file_path", "")),
        "name_check": {
            "result": state.get("name_check_result", ""),
            "reason": state.get("name_check_reason", ""),
            "verified": state.get("name_check_verified", ""),
            "detected_name": state.get("detected_name", ""),
        },
        "company_check": {
            "result": state.get("company_check_result", ""),
            "reason": state.get("company_check_reason", ""),
            "verified": state.get("company_check_verified", ""),
            "matched": state.get("company_matched", ""),
        },
        "grad_year_check": {
            "result": state.get("grad_year_check_result", ""),
            "reason": state.get("grad_year_check_reason", ""),
            "verified": state.get("grad_year_check_verified", ""),
            "matched_degree_type": state.get("matched_degree_type", ""),
            "matched_grad_year": state.get("matched_grad_year", 0),
            "matched_school": state.get("matched_school", ""),
        },
        "title_check": {
            "result": state.get("title_check_result", ""),
            "reason": state.get("title_check_reason", ""),
            "verified": state.get("title_check_verified", ""),
            "matched_role": state.get("title_matched_role", ""),
        },
        "tech_check": {
            "result": state.get("tech_check_result", ""),
            "reason": state.get("tech_check_reason", ""),
            "verified": state.get("tech_check_verified", ""),
            "tech_category": state.get("tech_category", ""),
            "retry_count": state.get("tech_check_retry_count", 0),
        },
        "email": {
            "subject": state.get("email_subject", ""),
            "greetings": state.get("email_greetings", ""),
            "content": state.get("email_content", ""),
            "signature": state.get("email_signature", ""),
        },
        "final_result": "qualified" if all_pass else "disqualified",
        "disqualify_reason": "" if all_pass else "; ".join(reasons),
    }


# ============ 汇总 JSON ============

def save_summary(total: int, skipped: int, qualified_list: list, disqualified_list: list, error_list: list):
    """即时保存汇总结果"""
    summary = {
        "total": total,
        "skipped": skipped,
        "processed": len(qualified_list) + len(disqualified_list) + len(error_list),
        "qualified": len(qualified_list),
        "disqualified": len(disqualified_list),
        "errors": len(error_list),
        "qualified_list": qualified_list,
        "disqualified_list": disqualified_list,
        "error_list": error_list,
    }
    os.makedirs(OUTPUT_DIR, exist_ok=True)
    summary_path = os.path.join(OUTPUT_DIR, "_summary.json")
    with open(summary_path, "w", encoding="utf-8") as f:
        json.dump(summary, f, ensure_ascii=False, indent=2)
    logger.info(f"[汇总] 即时保存已更新: {summary_path}")


# ============ 断点续传检查 ============

def is_already_processed(json_file_name: str) -> bool:
    """检查该简历是否已经处理完成（output 目录中存在同名结果文件且为终态）"""
    os.makedirs(OUTPUT_DIR, exist_ok=True)
    out_path = os.path.join(OUTPUT_DIR, json_file_name)
    if not os.path.exists(out_path):
        return False
    try:
        with open(out_path, "r", encoding="utf-8") as fp:
            data = json.load(fp)
        return data.get("final_result") in ("qualified", "disqualified")
    except (json.JSONDecodeError, KeyError):
        return False


# ============ Rich 终端输出 ============

def print_summary():
    """从 output 目录加载已有结果并用 Rich 美化打印汇总"""
    summary_path = os.path.join(OUTPUT_DIR, "_summary.json")
    if not os.path.exists(summary_path):
        console.print("[dim]暂无结果[/dim]")
        return

    with open(summary_path, "r", encoding="utf-8") as f:
        summary = json.load(f)

    # 统计概览
    stats_table = Table(title="简历筛选汇总", show_header=False, border_style="blue")
    stats_table.add_column("项目", style="bold cyan")
    stats_table.add_column("数量", style="bold white")
    stats_table.add_row("总计", str(summary.get("total", 0)))
    stats_table.add_row("跳过(已完成)", str(summary.get("skipped", 0)))
    stats_table.add_row("本次处理", str(summary.get("processed", 0)))
    stats_table.add_row("[green]符合条件[/green]", f"[green]{summary.get('qualified', 0)}[/green]")
    stats_table.add_row("[red]不符合[/red]", f"[red]{summary.get('disqualified', 0)}[/red]")
    stats_table.add_row("[yellow]出错[/yellow]", f"[yellow]{summary.get('errors', 0)}[/yellow]")
    console.print(stats_table)

    qualified_list = summary.get("qualified_list", [])
    disqualified_list = summary.get("disqualified_list", [])
    error_list = summary.get("error_list", [])

    if qualified_list:
        q_table = Table(title="符合条件候选人", border_style="green")
        q_table.add_column("#", style="dim", width=3)
        q_table.add_column("姓名", style="bold")
        q_table.add_column("文件")
        q_table.add_column("公司")
        q_table.add_column("学历/毕业")
        q_table.add_column("技术分类")
        q_table.add_column("邮件主题")
        for i, q in enumerate(qualified_list, 1):
            email_sub = q.get("email", {}).get("subject", "") if isinstance(q.get("email"), dict) else ""
            q_table.add_row(
                str(i),
                q.get("name", ""),
                q.get("file", ""),
                q.get("company_matched", ""),
                f"{q.get('matched_degree_type', '')} {q.get('matched_grad_year', '')} {q.get('matched_school', '')}",
                q.get("tech_category", ""),
                email_sub,
            )
        console.print(q_table)

    if disqualified_list:
        d_table = Table(title="不符合条件候选人", border_style="red")
        d_table.add_column("#", style="dim", width=3)
        d_table.add_column("姓名", style="bold")
        d_table.add_column("文件")
        d_table.add_column("原因")
        for i, d in enumerate(disqualified_list, 1):
            d_table.add_row(str(i), d.get("name", ""), d.get("file", ""), d.get("reason", ""))
        console.print(d_table)

    if error_list:
        e_table = Table(title="处理出错", border_style="yellow")
        e_table.add_column("文件")
        e_table.add_column("错误")
        for e in error_list:
            e_table.add_row(e.get("file", ""), e.get("error", "")[:100])
        console.print(e_table)

    console.print(f"\n[dim]结果已保存: {summary_path}[/dim]")


# ============ Excel 输出 ============

def save_excel(excel_path: str):
    """从 _summary.json 读取结果，输出为格式化的 Excel 文件"""
    from openpyxl import Workbook
    from openpyxl.styles import Font, Alignment, PatternFill, Border, Side

    summary_path = os.path.join(OUTPUT_DIR, "_summary.json")
    if not os.path.exists(summary_path):
        console.print("[yellow]暂无汇总结果，无法生成 Excel[/yellow]")
        return

    with open(summary_path, "r", encoding="utf-8") as f:
        summary = json.load(f)

    qualified_list = summary.get("qualified_list", [])
    disqualified_list = summary.get("disqualified_list", [])
    error_list = summary.get("error_list", [])

    wb = Workbook()

    # ---- 样式定义 ----
    header_font = Font(bold=True, color="FFFFFF", size=11)
    header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
    header_align = Alignment(horizontal="center", vertical="center", wrap_text=True)
    cell_align = Alignment(vertical="center", wrap_text=True)
    thin_border = Border(
        left=Side(style="thin"), right=Side(style="thin"),
        top=Side(style="thin"), bottom=Side(style="thin"),
    )
    green_fill = PatternFill(start_color="E2EFDA", end_color="E2EFDA", fill_type="solid")
    red_fill = PatternFill(start_color="FCE4EC", end_color="FCE4EC", fill_type="solid")
    yellow_fill = PatternFill(start_color="FFF9C4", end_color="FFF9C4", fill_type="solid")

    def _style_header(ws, col_count):
        for col in range(1, col_count + 1):
            cell = ws.cell(row=1, column=col)
            cell.font = header_font
            cell.fill = header_fill
            cell.alignment = header_align
            cell.border = thin_border

    def _style_data(ws, row, col_count, fill=None):
        for col in range(1, col_count + 1):
            cell = ws.cell(row=row, column=col)
            cell.alignment = cell_align
            cell.border = thin_border
            if fill:
                cell.fill = fill

    # ---- Sheet 1: 符合条件 ----
    ws_q = wb.active
    ws_q.title = "符合条件"
    q_headers = ["#", "姓名", "源文件", "匹配公司", "学历类型", "毕业年份", "毕业院校", "技术分类", "邮件主题", "邮件问候", "邮件正文", "邮件签名"]
    ws_q.append(q_headers)
    _style_header(ws_q, len(q_headers))

    for i, q in enumerate(qualified_list, 1):
        email = q.get("email", {}) if isinstance(q.get("email"), dict) else {}
        row_data = [
            i, q.get("name", ""), q.get("file", ""), q.get("company_matched", ""),
            q.get("matched_degree_type", ""), q.get("matched_grad_year", ""),
            q.get("matched_school", ""), q.get("tech_category", ""),
            email.get("subject", ""), email.get("greetings", ""),
            email.get("content", ""), email.get("signature", ""),
        ]
        ws_q.append(row_data)
        _style_data(ws_q, i + 1, len(q_headers), fill=green_fill)

    col_widths_q = [4, 12, 25, 20, 10, 10, 20, 15, 30, 20, 50, 20]
    for idx, w in enumerate(col_widths_q, 1):
        ws_q.column_dimensions[ws_q.cell(row=1, column=idx).column_letter].width = w

    # ---- Sheet 2: 不符合条件 ----
    ws_d = wb.create_sheet("不符合条件")
    d_headers = ["#", "姓名", "源文件", "原因"]
    ws_d.append(d_headers)
    _style_header(ws_d, len(d_headers))

    for i, d in enumerate(disqualified_list, 1):
        ws_d.append([i, d.get("name", ""), d.get("file", ""), d.get("reason", "")])
        _style_data(ws_d, i + 1, len(d_headers), fill=red_fill)

    col_widths_d = [4, 15, 30, 50]
    for idx, w in enumerate(col_widths_d, 1):
        ws_d.column_dimensions[ws_d.cell(row=1, column=idx).column_letter].width = w

    # ---- Sheet 3: 出错 ----
    ws_e = wb.create_sheet("出错")
    e_headers = ["#", "源文件", "错误信息"]
    ws_e.append(e_headers)
    _style_header(ws_e, len(e_headers))

    for i, e in enumerate(error_list, 1):
        ws_e.append([i, e.get("file", ""), e.get("error", "")])
        _style_data(ws_e, i + 1, len(e_headers), fill=yellow_fill)

    col_widths_e = [4, 30, 60]
    for idx, w in enumerate(col_widths_e, 1):
        ws_e.column_dimensions[ws_e.cell(row=1, column=idx).column_letter].width = w

    wb.save(excel_path)
    console.print(f"[green]Excel 已保存: {excel_path}[/green]")

```

---

## 路径：`state.py`

```python
"""LangGraph State 定义"""

from __future__ import annotations

from typing import TypedDict


class ResumeState(TypedDict):
    file_path: str
    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       # "pass" | "fail"
    company_check_reason: str
    company_check_verified: str     # "confirmed" | "conflict"
    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         # "pass" | "fail"
    title_check_reason: str
    title_check_verified: str
    title_matched_role: str         # 匹配到的技术角色分类
    # Step 5: 技术方向判断
    tech_check_result: str          # "pass" | "fail"
    tech_check_reason: str
    tech_check_verified: str
    tech_category: str
    tech_check_retry_count: int     # 重试次数
    # Step 6: 生成邮件
    email_subject: str              # 邮件主题
    email_greetings: str            # 邮件称呼，如 "您好Mr Wang，"
    email_content: str              # 邮件正文
    email_signature: str            # 邮件署名
    # 整体结果
    final_result: str               # "qualified" | "disqualified"
    disqualify_reason: str
    # 错误信息
    error: str

```

---

## 路径：`nodes\__init__.py`

```python
"""节点函数导出"""

from resume_flow.nodes.load_resume import load_resume
from resume_flow.nodes.check_name import check_name, verify_name
from resume_flow.nodes.check_company import check_company, verify_company
from resume_flow.nodes.check_grad import check_grad_year, verify_grad_year
from resume_flow.nodes.check_title import check_title, verify_title
from resume_flow.nodes.check_tech import check_tech, verify_tech
from resume_flow.nodes.generate_email import generate_email
from resume_flow.nodes.finalize import finalize

__all__ = [
    "load_resume",
    "check_name",
    "verify_name",
    "check_company",
    "verify_company",
    "check_grad_year",
    "verify_grad_year",
    "check_title",
    "verify_title",
    "check_tech",
    "verify_tech",
    "generate_email",
    "finalize",
]

```

---

## 路径：`nodes\check_company.py`

```python
"""Node 2: 公司判断 + 验证"""

from __future__ import annotations

from resume_flow.config import logger, TARGET_COMPANIES
from resume_flow.state import ResumeState
from resume_flow.llm import LLMClient

_llm_client: LLMClient | None = None


def _get_llm() -> LLMClient:
    global _llm_client
    if _llm_client is None:
        from resume_flow.config import env_config
        _llm_client = LLMClient(
            api_key=env_config.get("api_key", "sk-1234"),
            base_url=env_config.get("api_url", "").replace("/chat/completions", ""),
            model=env_config.get("model", "deepseek-v3.1-terminus-chat"),
        )
    return _llm_client


def call_llm_json(system: str, user: str, temperature: float = 0.1) -> dict:
    """调用大模型并解析 JSON 响应，失败时返回空 dict"""
    try:
        result = _get_llm().chat_json(system, user, temperature)
        if isinstance(result, dict):
            return result
        return {"raw": str(result)}
    except Exception as e:
        logger.error(f"[LLM] 调用失败: {e}")
        return {"error": str(e)}


def call_llm_text(system: str, user: str, temperature: float = 0.3) -> str:
    """调用大模型返回文本"""
    try:
        return _get_llm().chat(system, user, temperature)
    except Exception as e:
        logger.error(f"[LLM] 调用失败: {e}")
        return f"[LLM调用失败: {e}]"


def check_company(state: ResumeState) -> dict:
    resume = state["resume_data"]
    name = state["name"]
    logger.info("-" * 40)
    logger.info(f"[公司判断] 分析 {name} ...")

    companies = [exp.get("company", "") for exp in resume.get("experience", [])]
    current_company = resume.get("company", "")

    system = """你是简历分析专家。判断候选人是否在目标公司工作过。
目标公司: Apple, Amazon, Google, Microsoft, Meta
返回JSON:
{
    "is_target_company": true/false,
    "matched_companies": ["匹配的公司"],
    "reason": "判断理由"
}"""

    user = f"""姓名: {name}
当前公司: {current_company}
工作经历公司: {', '.join(companies)}
请判断是否在目标公司工作过（含当前和过去）。"""

    result = call_llm_json(system, user)
    is_match = result.get("is_target_company", False)
    matched = result.get("matched_companies", [])
    reason = result.get("reason", "")
    check = "pass" if is_match else "fail"

    logger.info(f"[公司判断] 结果: {check} | 匹配: {matched} | 理由: {reason}")
    return {
        "company_check_result": check,
        "company_check_reason": reason,
        "company_matched": ", ".join(matched) if matched else "",
    }


def verify_company(state: ResumeState) -> dict:
    resume = state["resume_data"]
    name = state["name"]
    llm_result = state["company_check_result"]
    llm_reason = state["company_check_reason"]
    logger.info(f"[公司验证] 验证 {name} ...")

    # 规则引擎：直接匹配公司名
    all_companies = [resume.get("company", "")] + [exp.get("company", "") for exp in resume.get("experience", [])]
    rule_matched = list({t for c in all_companies for t in TARGET_COMPANIES if t.lower() in c.lower()})
    rule_pass = len(rule_matched) > 0

    if rule_pass and llm_result == "pass":
        verified = "confirmed"
    elif not rule_pass and llm_result == "fail":
        verified = "confirmed"
    else:
        verified = "conflict"
        logger.warning(f"[公司验证] 冲突! LLM={llm_result}, 规则={'pass' if rule_pass else 'fail'}, 以规则为准")
        return {
            "company_check_result": "pass" if rule_pass else "fail",
            "company_check_reason": f"规则修正: 匹配={rule_matched}. 原LLM: {llm_reason}",
            "company_check_verified": verified,
            "company_matched": ", ".join(rule_matched),
        }

    logger.info(f"[公司验证] {verified} | 规则匹配: {rule_matched}")
    return {"company_check_verified": verified, "company_matched": ", ".join(rule_matched) if rule_matched else state.get("company_matched", "")}

```

---

## 路径：`nodes\check_grad.py`

```python
"""Node 3: 毕业年份判断 + 验证"""

from __future__ import annotations

import json
import re
from typing import Optional

from resume_flow.config import logger, TARGET_BACHELOR_YEARS, TARGET_MASTER_YEARS, TARGET_PHD_YEARS
from resume_flow.state import ResumeState
from resume_flow.nodes.check_company import call_llm_json


def check_grad_year(state: ResumeState) -> dict:
    """用大模型判断毕业年份，按学历层级递进：本科→硕士→博士→数据缺失"""
    resume = state["resume_data"]
    name = state["name"]
    logger.info("-" * 40)
    logger.info(f"[毕业年份] 分析 {name} ...")

    education_str = json.dumps(resume.get("education", []), ensure_ascii=False, indent=2)

    system = f"""你是简历分析专家。按以下优先级判断候选人毕业年份是否在目标范围内：

1. 优先判断本科学士(Bachelor)毕业年份，目标:{', '.join(str(y) for y in TARGET_BACHELOR_YEARS)}
2. 如果没有本科，判断硕士(Master/MS)毕业年份，目标:{', '.join(str(y) for y in TARGET_MASTER_YEARS)}
3. 如果没有硕士，判断博士(PhD/Doctor)毕业年份，目标: {', '.join(str(y) for y in TARGET_PHD_YEARS)}
4. 如果本科、硕士、博士毕业时间都没有，标记为数据缺失

注意："2012 - 2016" 格式中毕业年份是结束年份。
返回JSON:
{{
    "matched_degree_type": "bachelor"/"master"/"phd"/"unknown",
    "matched_school": "匹配到的学校名",
    "matched_grad_year": 2016,
    "is_target_year": true/false,
    "reason": "判断理由"
}}"""

    user = f"""姓名: {name}
教育背景: {education_str}
本科目标年份: {', '.join(str(y) for y in TARGET_BACHELOR_YEARS)}
硕士目标年份: {', '.join(str(y) for y in TARGET_MASTER_YEARS)}
博士目标年份: {', '.join(str(y) for y in TARGET_PHD_YEARS)}

请按优先级判断毕业年份是否在目标范围内。"""

    result = call_llm_json(system, user)
    is_match = result.get("is_target_year", False)
    degree_type = result.get("matched_degree_type", "unknown")
    grad_year = result.get("matched_grad_year", 0)
    school = result.get("matched_school", "")
    reason = result.get("reason", "")

    if degree_type == "unknown":
        check = "unknown"
    else:
        check = "pass" if is_match else "fail"

    logger.info(f"[毕业年份] 结果: {check} | 学历: {degree_type} | 学校: {school} {grad_year} | 理由: {reason}")
    return {
        "grad_year_check_result": check,
        "grad_year_check_reason": reason,
        "matched_degree_type": degree_type,
        "matched_grad_year": grad_year,
        "matched_school": school,
    }


def verify_grad_year(state: ResumeState) -> dict:
    """规则引擎验证毕业年份判断结果"""
    resume = state["resume_data"]
    name = state["name"]
    llm_result = state["grad_year_check_result"]
    llm_year = state.get("matched_grad_year", 0)
    llm_degree = state.get("matched_degree_type", "unknown")
    logger.info(f"[毕业验证] 验证 {name} ...")

    rule_info = _extract_grad_year_by_priority(resume.get("education", []))
    rule_degree = rule_info["degree_type"]
    rule_year = rule_info["grad_year"]
    rule_school = rule_info["school"]

    target_years = _get_target_years(rule_degree)
    rule_pass = rule_year in target_years if rule_year and rule_degree != "unknown" else False

    logger.info(f"[毕业验证] 规则: degree={rule_degree}, year={rule_year}, school={rule_school}, pass={rule_pass}")
    logger.info(f"[毕业验证] LLM:  degree={llm_degree}, year={llm_year}, result={llm_result}")

    if rule_degree == "unknown":
        verified = "confirmed" if llm_result == "unknown" else "conflict"
        if verified == "conflict":
            logger.warning(f"[毕业验证] 冲突! 规则=unknown, LLM={llm_result}, 以规则为准")
        return {
            "grad_year_check_result": "unknown",
            "grad_year_check_reason": "规则引擎: 无法提取任何学历的毕业时间，数据缺失",
            "grad_year_check_verified": verified,
            "matched_degree_type": "unknown",
            "matched_grad_year": 0,
            "matched_school": "",
        }

    if rule_pass and llm_result == "pass":
        verified = "confirmed"
    elif not rule_pass and llm_result == "fail":
        verified = "confirmed"
    else:
        verified = "conflict"
        logger.warning(f"[毕业验证] 冲突! 以规则为准")
        return {
            "grad_year_check_result": "pass" if rule_pass else "fail",
            "grad_year_check_reason": f"规则修正: {rule_degree} {rule_school} {rule_year}. 原LLM: {state.get('grad_year_check_reason', '')}",
            "grad_year_check_verified": verified,
            "matched_degree_type": rule_degree,
            "matched_grad_year": rule_year,
            "matched_school": rule_school,
        }

    return {
        "grad_year_check_verified": verified,
        "matched_degree_type": rule_degree,
        "matched_grad_year": rule_year or llm_year,
        "matched_school": rule_school or state.get("matched_school", ""),
    }


# ============ 辅助函数 ============

def _extract_grad_year_by_priority(education: list) -> dict:
    """按优先级从教育背景中提取毕业年份：本科→硕士→博士→unknown"""
    bachelor_info = _find_degree_info(education, ["bachelor", "学士", "本科"])
    if bachelor_info:
        bachelor_info["degree_type"] = "bachelor"
        return bachelor_info

    master_info = _find_degree_info(education, ["master", "ms", "m.s.", "m.sc", "硕士"])
    if master_info:
        master_info["degree_type"] = "master"
        return master_info

    phd_info = _find_degree_info(education, ["phd", "ph.d", "doctor", "博士"])
    if phd_info:
        phd_info["degree_type"] = "phd"
        return phd_info

    if education:
        first = _parse_edu_entry(education[0])
        if first["grad_year"]:
            first["degree_type"] = "bachelor"
            return first

    return {"degree_type": "unknown", "grad_year": None, "school": ""}


def _find_degree_info(education: list, keywords: list) -> Optional[dict]:
    for edu in education:
        degree = edu.get("degree", "").lower()
        if any(kw in degree for kw in keywords):
            return _parse_edu_entry(edu)
    return None


def _parse_edu_entry(edu: dict) -> dict:
    year = _parse_end_year(edu.get("dates", ""))
    return {"grad_year": year, "school": edu.get("school", "")}


def _get_target_years(degree_type: str) -> list:
    if degree_type == "bachelor":
        return TARGET_BACHELOR_YEARS
    elif degree_type == "master":
        return TARGET_MASTER_YEARS
    elif degree_type == "phd":
        return TARGET_PHD_YEARS
    return []


def _parse_end_year(dates: str) -> Optional[int]:
    m = re.search(r'(\d{4})\s*[-–]\s*(\d{4})', dates)
    if m:
        return int(m.group(2))
    m = re.search(r'(\d{4})\s*$', dates.strip())
    if m:
        return int(m.group(1))
    return None

```

---

## 路径：`nodes\check_name.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_name(state: ResumeState) -> dict:
    """用大模型判断候选人姓名是否为有效姓名"""
    resume = state["resume_data"]
    name = state["name"]
    logger.info("-" * 40)
    logger.info(f"[姓名判断] 分析 {name} ...")

    system = """你是简历分析专家。判断候选人的姓名是否为有效姓名。

有效姓名: 真实的人名，如 "Alex Yu"、"张三"、"John Smith"、"王明" 等。
无效姓名: 空值、占位符（如 "N/A"、"Unknown"、"Test"、"匿名"）、明显伪造（如 "AAA"、"123"）。

返回JSON:
{
    "is_valid_name": true/false,
    "detected_name": "检测到的姓名",
    "reason": "判断理由"
}"""

    user = f"""候选人姓名: {name}
简历中name字段: {resume.get('name', '')}

请判断该候选人的姓名是否为有效姓名。"""

    result = call_llm_json(system, user)
    is_valid = result.get("is_valid_name", False)
    detected_name = result.get("detected_name", name)
    reason = result.get("reason", "")
    check = "pass" if is_valid else "fail"

    logger.info(f"[姓名判断] 结果: {check} | 检测姓名: {detected_name} | 理由: {reason}")
    return {
        "name_check_result": check,
        "name_check_reason": reason,
        "detected_name": detected_name,
    }


def verify_name(state: ResumeState) -> dict:
    """规则引擎验证姓名判断结果"""
    name = state["name"]
    llm_result = state.get("name_check_result", "")
    logger.info(f"[姓名验证] 验证 {name} ...")

    # 规则引擎：基于姓名字符串特征判断
    name_stripped = name.strip()

    # 空值或极短
    if not name_stripped or len(name_stripped) < 1:
        rule_pass = False
        rule_reason = "姓名为空"
    # 常见占位符
    elif name_stripped.lower() in ("n/a", "na", "unknown", "test", "none", "null", "匿名", "未知", "无"):
        rule_pass = False
        rule_reason = f"姓名为占位符: {name_stripped}"
    # 纯数字
    elif name_stripped.isdigit():
        rule_pass = False
        rule_reason = f"姓名为纯数字: {name_stripped}"
    # 全部相同字符（如 "AAA"、"!!!"）
    elif len(set(name_stripped.lower())) == 1 and len(name_stripped) > 2:
        rule_pass = False
        rule_reason = f"姓名为重复字符: {name_stripped}"
    else:
        rule_pass = True
        rule_reason = ""

    logger.info(f"[姓名验证] 规则: {'pass' if rule_pass else 'fail'} ({rule_reason}) | LLM: {llm_result}")

    if (rule_pass and llm_result == "pass") or (not rule_pass and llm_result == "fail"):
        verified = "confirmed"
        return {
            "name_check_verified": verified,
            "detected_name": state.get("detected_name", name),
        }
    else:
        verified = "conflict"
        logger.warning(f"[姓名验证] 冲突! 以规则为准")
        return {
            "name_check_result": "pass" if rule_pass else "fail",
            "name_check_reason": f"规则修正: {rule_reason}. 原LLM: {state.get('name_check_reason', '')}" if not rule_pass else state.get("name_check_reason", ""),
            "name_check_verified": verified,
            "detected_name": name_stripped if rule_pass else state.get("detected_name", name),
        }

```

---

## 路径：`nodes\check_tech.py`

```python
"""Node 4: 技术方向判断 + 验证（大模型验证 + 循环重试）"""

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

MAX_TECH_RETRY = 3


def check_tech(state: ResumeState) -> dict:
    resume = state["resume_data"]
    name = state["name"]
    logger.info("-" * 40)
    logger.info(f"[技术判断] 分析 {name} ...")

    title = resume.get("title", "")
    position = resume.get("position", "")
    skills = resume.get("skills", [])
    suitable_roles = resume.get("suitable_job_roles", [])
    bio_summary = resume.get("bio_summary", "")
    job_titles = [exp.get("title", "") for exp in resume.get("experience", [])]

    system = """你是技术岗位分析专家。判断候选人是否从事技术方向。
技术方向: 软件工程、算法、ML/AI、数据科学、前后端开发、系统架构、DevOps、测试开发、硬件工程、芯片设计等。
非技术方向: 产品经理(纯管理)、项目管理、HR、销售、市场、公关、行政、财务、法务、纯质量管控等。
边界: 硬件/设计验证/测试工程师=技术; 技术Lead=技术; 纯管理/运营=非技术
返回JSON:
{
    "is_tech": true/false,
    "tech_category": "分类",
    "reason": "判断理由"
}"""

    user = f"""姓名: {name}
职位: {title} / {position}
技能: {', '.join(skills)}
适合职位: {', '.join(suitable_roles)}
摘要: {bio_summary}
历史职位: {', '.join(job_titles)}
请判断是否技术方向。"""

    result = call_llm_json(system, user)
    is_tech = result.get("is_tech", False)
    category = result.get("tech_category", "")
    reason = result.get("reason", "")
    check = "pass" if is_tech else "fail"

    logger.info(f"[技术判断] 结果: {check} | 分类: {category} | 理由: {reason}")
    return {
        "tech_check_result": check,
        "tech_check_reason": reason,
        "tech_category": category,
    }


def verify_tech(state: ResumeState) -> dict:
    """用大模型验证技术方向判断结果，不一致则触发重试 check_tech"""
    resume = state["resume_data"]
    name = state["name"]
    llm_result = state["tech_check_result"]
    llm_reason = state["tech_check_reason"]
    llm_category = state.get("tech_category", "")
    retry_count = state.get("tech_check_retry_count", 0)

    logger.info(f"[技术验证] 第 {retry_count + 1} 轮验证 {name} ...")
    logger.info(f"[技术验证] check_tech 结果: {llm_result} | 分类: {llm_category} | 理由: {llm_reason}")

    title = resume.get("title", "")
    position = resume.get("position", "")
    skills = resume.get("skills", [])
    suitable_roles = resume.get("suitable_job_roles", [])
    bio_summary = resume.get("bio_summary", "")
    job_titles = [exp.get("title", "") for exp in resume.get("experience", [])]

    system = """你是一个技术岗位审核专家。你的任务是审核另一个分析师对候选人技术方向的判断是否正确。

技术方向: 软件工程、算法、ML/AI、数据科学、前后端开发、系统架构、DevOps、测试开发、硬件工程、芯片设计等。
非技术方向: 产品经理(纯管理)、项目管理、HR、销售、市场、公关、行政、财务、法务、纯质量管控等。
边界: 硬件/设计验证/测试工程师=技术; 技术Lead=技术; 纯管理/运营=非技术

你需要独立判断，然后与原判断对比。返回JSON:
{
    "verify_is_tech": true/false,
    "verify_reason": "你的独立判断理由",
    "is_consistent": true/false,
    "final_verdict": "tech"/"non_tech"
}"""

    user = f"""候选人信息:
- 姓名: {name}
- 职位: {title} / {position}
- 技能: {', '.join(skills)}
- 适合职位: {', '.join(suitable_roles)}
- 摘要: {bio_summary}
- 历史职位: {', '.join(job_titles)}

原判断结果: {'技术方向' if llm_result == 'pass' else '非技术方向'}
原判断分类: {llm_category}
原判断理由: {llm_reason}

请独立审核该候选人是否属于技术方向，并判断你的结论与原判断是否一致。"""

    result = call_llm_json(system, user)
    verify_is_tech = result.get("verify_is_tech", None)
    verify_reason = result.get("verify_reason", "")
    is_consistent = result.get("is_consistent", None)

    if is_consistent is None and verify_is_tech is not None:
        verify_result = "pass" if verify_is_tech else "fail"
        is_consistent = (verify_result == llm_result)
    elif is_consistent is None:
        is_consistent = False

    logger.info(f"[技术验证] 验证模型判断: {'技术' if verify_is_tech else '非技术'} | 一致: {is_consistent} | 理由: {verify_reason}")

    if is_consistent:
        logger.info(f"[技术验证] 验证通过! 结果确认: {llm_result}")
        return {"tech_check_verified": "confirmed", "tech_check_retry_count": retry_count}
    else:
        retry_count += 1
        if retry_count >= MAX_TECH_RETRY:
            logger.warning(f"[技术验证] 已重试 {retry_count} 次，超过上限({MAX_TECH_RETRY})，默认判定为 FAIL")
            return {
                "tech_check_result": "fail",
                "tech_check_reason": f"验证不一致且重试{retry_count}次仍未收敛，默认失败。原判断: {llm_reason}; 验证判断: {verify_reason}",
                "tech_check_verified": "failed_after_retry",
                "tech_category": llm_category,
                "tech_check_retry_count": retry_count,
            }
        else:
            logger.warning(f"[技术验证] 验证不一致! 将重新执行 check_tech (第 {retry_count + 1} 次)")
            return {
                "tech_check_verified": "conflict_need_retry",
                "tech_check_retry_count": retry_count,
            }

```

---

## 路径：`nodes\check_title.py`

```python
"""Node: 职位名称判断 + 验证

识别候选人当前职位是否为软件技术岗位。
与 check_tech 的区别：check_title 仅判断当前职位名称，check_tech 综合判断整体技术方向。
"""

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_title(state: ResumeState) -> dict:
    """用大模型判断候选人当前职位是否为软件技术岗位"""
    resume = state["resume_data"]
    name = state["name"]
    logger.info("-" * 40)
    logger.info(f"[职位判断] 分析 {name} ...")

    title = resume.get("title", "")
    position = resume.get("position", "")

    system = """你是技术岗位分析专家。判断候选人的当前职位是否为软件技术岗位。

软件技术岗位: 软件工程师、开发工程师、架构师、技术负责人(Tech Lead)、DevOps工程师、
SRE、数据工程师、算法工程师、AI/ML工程师、测试开发工程师、前端工程师、后端工程师等。
关键特征: 日常工作以编写代码、设计系统、解决技术问题为主。

非软件技术岗位: 产品经理、项目经理、技术经理(纯管理)、销售、市场、HR、行政、
财务、法务、咨询顾问、设计(非前端)、运营等。
关键特征: 日常工作不涉及编写代码。

边界: 技术Lead(仍写代码)=软件技术; 纯管理(不写代码)=非软件技术; 硬件工程师=非软件技术

返回JSON:
{
    "is_software_tech": true/false,
    "matched_role": "匹配到的角色分类",
    "reason": "判断理由"
}"""

    user = f"""姓名: {name}
当前职位: {title}
当前岗位: {position}

请判断该候选人的当前职位是否为软件技术岗位。"""

    result = call_llm_json(system, user)
    is_match = result.get("is_software_tech", False)
    matched_role = result.get("matched_role", "")
    reason = result.get("reason", "")
    check = "pass" if is_match else "fail"

    logger.info(f"[职位判断] 结果: {check} | 角色: {matched_role} | 理由: {reason}")
    return {
        "title_check_result": check,
        "title_check_reason": reason,
        "title_matched_role": matched_role,
    }


def verify_title(state: ResumeState) -> dict:
    """规则引擎验证职位判断结果"""
    resume = state["resume_data"]
    name = state["name"]
    llm_result = state.get("title_check_result", "")
    llm_role = state.get("title_matched_role", "")
    logger.info(f"[职位验证] 验证 {name} ...")

    # 规则引擎：基于职位名称关键词匹配
    title = resume.get("title", "").lower()
    position = resume.get("position", "").lower()
    combined = f"{title} {position}"

    # 软件技术关键词
    tech_keywords = [
        "software", "developer", "engineer", "architect", "devops", "sre",
        "programmer", "coder", "frontend", "backend", "fullstack", "full-stack",
        "data engineer", "ml engineer", "algorithm", "test automation",
        "开发", "软件", "架构", "算法", "前端", "后端", "全栈",
    ]
    # 非技术关键词（优先级更高，出现则排除）
    non_tech_keywords = [
        "manager", "director", "vp", "head of", "chief", "president",
        "sales", "marketing", "hr", "recruiter", "accountant", "consultant",
        "product manager", "project manager", "program manager",
        "经理", "总监", "总裁", "销售", "市场", "人力", "招聘", "顾问",
    ]

    is_tech = False
    for kw in tech_keywords:
        if kw in combined:
            is_tech = True
            break

    # 非技术关键词优先级更高
    for kw in non_tech_keywords:
        if kw in combined:
            is_tech = False
            break

    rule_pass = is_tech
    logger.info(f"[职位验证] 规则: {'pass' if rule_pass else 'fail'} | LLM: {llm_result}")

    if (rule_pass and llm_result == "pass") or (not rule_pass and llm_result == "fail"):
        verified = "confirmed"
        return {
            "title_check_verified": verified,
            "title_matched_role": llm_role,
        }
    else:
        verified = "conflict"
        logger.warning(f"[职位验证] 冲突! 以规则为准")
        return {
            "title_check_result": "pass" if rule_pass else "fail",
            "title_check_reason": f"规则修正: {'软件技术岗位' if rule_pass else '非软件技术岗位'}. 原LLM: {state.get('title_check_reason', '')}",
            "title_check_verified": verified,
            "title_matched_role": "软件技术岗位" if rule_pass else "非软件技术岗位",
        }

```

---

## 路径：`nodes\finalize.py`

```python
"""Node: 汇总结果"""

from __future__ import annotations

import os

from resume_flow.config import logger
from resume_flow.state import ResumeState
from resume_flow.output import save_result, _build_result_data


def finalize(state: ResumeState) -> dict:
    name = state["name"]

    name_pass = state.get("name_check_result") == "pass"
    company_pass = state.get("company_check_result") == "pass"
    grad_pass = state.get("grad_year_check_result") == "pass"
    grad_unknown = state.get("grad_year_check_result") == "unknown"
    title_pass = state.get("title_check_result") == "pass"
    tech_pass = state.get("tech_check_result") == "pass"

    degree_type = state.get("matched_degree_type", "unknown")
    degree_label = {"bachelor": "本科", "master": "硕士", "phd": "博士", "unknown": "未知"}.get(degree_type, "未知")

    logger.info("=" * 60)
    logger.info(f"[最终结果] {name}")
    logger.info(f"  姓名: {'PASS' if name_pass else 'FAIL'} - {state.get('detected_name', '')} (验证: {state.get('name_check_verified', 'N/A')})")
    logger.info(f"  公司: {'PASS' if company_pass else 'FAIL'} - {state.get('company_check_reason', '')} (验证: {state.get('company_check_verified', 'N/A')})")
    if grad_unknown:
        logger.info(f"  毕业: UNKNOWN - 毕业年份数据缺失 (验证: {state.get('grad_year_check_verified', 'N/A')})")
    else:
        logger.info(f"  毕业: {'PASS' if grad_pass else 'FAIL'} - {degree_label} {state.get('matched_school', '')} {state.get('matched_grad_year', 'N/A')} (验证: {state.get('grad_year_check_verified', 'N/A')})")
    logger.info(f"  职位: {'PASS' if title_pass else 'FAIL'} - {state.get('title_matched_role', '')} (验证: {state.get('title_check_verified', 'N/A')})")
    logger.info(f"  技术: {'PASS' if tech_pass else 'FAIL'} - {state.get('tech_category', '')} (验证: {state.get('tech_check_verified', 'N/A')})")

    if name_pass and company_pass and grad_pass and title_pass and tech_pass:
        final = "qualified"
        logger.info(f"  >>> {name} 符合所有条件! 将生成邮件...")
    else:
        final = "disqualified"
        reasons = []
        if not name_pass:
            reasons.append("姓名无效")
        if not company_pass:
            reasons.append("不在目标公司")
        if grad_unknown:
            reasons.append("毕业年份数据缺失，无法判断")
        elif not grad_pass:
            reasons.append(f"{degree_label}毕业年份({state.get('matched_grad_year', 'N/A')})不在目标范围")
        if not title_pass:
            reasons.append("非软件技术岗位")
        if not tech_pass:
            reasons.append("非技术方向")
        logger.info(f"  >>> {name} 不符合: {'; '.join(reasons)}")

    logger.info("=" * 60)

    # 保存结果到文件（disqualified 的情况在此保存，qualified 会在 generate_email 后更新）
    result_data = _build_result_data(state)
    save_result(name, result_data, source_file=os.path.basename(state.get("file_path", "")))

    return {
        "final_result": final,
        "disqualify_reason": "" if final == "qualified" else
            f"姓名:{'PASS' if name_pass else 'FAIL'}, 公司:{'PASS' if company_pass else 'FAIL'}, 毕业:{'PASS' if grad_pass else 'FAIL'}, 职位:{'PASS' if title_pass else 'FAIL'}, 技术:{'PASS' if tech_pass else 'FAIL'}",
    }

```

---

## 路径：`nodes\generate_email.py`

```python
"""Node 5: 生成邮件"""

from __future__ import annotations

import os

from resume_flow.config import logger
from resume_flow.state import ResumeState
from resume_flow.nodes.check_company import call_llm_text
from resume_flow.output import save_result, _build_result_data


def generate_email(state: ResumeState) -> dict:
    resume = state["resume_data"]
    name = state["name"]
    logger.info("-" * 40)
    logger.info(f"[生成邮件] 为 {name} 生成招聘邮件...")

    company = resume.get("company", "")
    position = resume.get("position", "")
    title = resume.get("title", "")
    skills = resume.get("skills", [])
    suitable_roles = resume.get("suitable_job_roles", [])
    bio_summary = resume.get("bio_summary", "")
    linkedin = resume.get("linkedin", "")

    system = """你是一名专业的技术招聘人员。请根据候选人背景生成一封可以直接发送的招聘邮件。

严格格式要求:
1. 必须包含以下四个部分，每部分独占一行:
   Subject: 邮件主题
   (空行)
   称呼(如: 尊敬的XXX先生/女士，)
   (空行)
   邮件正文
   (空行)
   署名(如: 此致敬礼 / 顺颂商祺 + 换行 + 招聘团队)

2. 纯文本格式，禁止使用任何Markdown符号(如 * # - > ` 等)
3. 禁止使用任何占位符(如 [公司名] [职位] 等)，所有内容必须填写真实信息
4. 邮件正文要专业友好，突出候选人背景与机会的匹配度，包含明确的行动号召
5. 中文撰写
6. 邮件正文3-5段，长度适中"""

    user = f"""候选人:
姓名: {name}
公司: {company}
职位: {position} / {title}
技能: {', '.join(skills)}
适合职位: {', '.join(suitable_roles)}
简介: {bio_summary}
LinkedIn: {linkedin}

请生成一封可以直接发送的纯文本招聘邮件。"""

    email = call_llm_text(system, user, temperature=0.3)
    logger.info(f"[生成邮件] 邮件内容:\n{email}")

    email_subject, email_greetings, email_content, email_signature = _parse_email(email)

    # 生成邮件后更新保存的结果文件
    result_data = _build_result_data(state)
    result_data["email"] = {
        "subject": email_subject,
        "greetings": email_greetings,
        "content": email_content,
        "signature": email_signature,
    }
    result_data["final_result"] = "qualified"
    save_result(name, result_data, source_file=os.path.basename(state.get("file_path", "")))

    return {
        "email_subject": email_subject,
        "email_greetings": email_greetings,
        "email_content": email_content,
        "email_signature": email_signature,
    }


def _parse_email(email_text: str) -> tuple:
    """将 LLM 生成的邮件文本解析为 subject / greetings / content / signature 四个字段"""
    lines = email_text.strip().split("\n")
    subject = ""
    greetings = ""
    signature_lines = []
    in_signature = False

    # 提取 Subject 行
    for i, line in enumerate(lines):
        stripped = line.strip()
        if stripped.lower().startswith("subject:") or stripped.startswith("Subject:"):
            subject = stripped.split(":", 1)[1].strip()
            lines = lines[i + 1:]
            break

    # 提取称呼行
    greeting_prefixes = ["尊敬的", "您好", "亲爱的", "Dear ", "dear ", "Hi ", "hi "]
    while lines and not lines[0].strip():
        lines.pop(0)
    if lines:
        first_line = lines[0].strip()
        is_greeting = False
        for prefix in greeting_prefixes:
            if first_line.startswith(prefix):
                is_greeting = True
                break
        if not is_greeting:
            title_words = ["先生", "女士", "Mr.", "Mr ", "Ms.", "Ms ", "Mrs.", "Mrs "]
            for tw in title_words:
                if tw in first_line:
                    is_greeting = True
                    break
        if is_greeting:
            greetings = first_line
            lines = lines[1:]

    # 从剩余行中分离正文和署名
    sig_keywords = ["此致", "顺颂", "祝好", "商祺", "敬礼", "best regards", "sincerely", "regards"]
    sig_identity = ["招聘团队", "招聘顾问", "HR团队", "人力资源", "技术招聘", "recruiting team", "hiring team"]

    body_lines = []
    for line in lines:
        stripped = line.strip()
        if not stripped:
            if in_signature:
                signature_lines.append(line)
            else:
                body_lines.append(line)
            continue

        if not in_signature:
            is_sig_start = False
            for kw in sig_keywords:
                if stripped.startswith(kw) or stripped == kw:
                    is_sig_start = True
                    break
            if not is_sig_start:
                for kw in sig_identity:
                    if kw in stripped and len(stripped) < 30:
                        is_sig_start = True
                        break
            if is_sig_start:
                in_signature = True
                signature_lines.append(line)
            else:
                body_lines.append(line)
        else:
            signature_lines.append(line)

    content = "\n".join(body_lines).strip()
    signature = "\n".join(signature_lines).strip()

    return subject, greetings, content, signature

```

---

## 路径：`nodes\load_resume.py`

```python
"""Node 1: 加载简历"""

from __future__ import annotations

import json

from resume_flow.config import logger
from resume_flow.state import ResumeState


def load_resume(state: ResumeState) -> dict:
    file_path = state["file_path"]
    logger.info("=" * 60)
    logger.info(f"[加载简历] 读取: {file_path}")

    with open(file_path, "r", encoding="utf-8") as f:
        data = json.load(f)

    name = data.get("name", "Unknown")
    logger.info(f"[加载简历] 姓名: {name} | 公司: {data.get('company', 'N/A')} | 职位: {data.get('position', 'N/A')}")
    for edu in data.get("education", []):
        logger.info(f"[加载简历] 教育: {edu.get('school')} - {edu.get('degree')} {edu.get('field')} ({edu.get('dates')})")

    return {"resume_data": data, "name": name, "error": ""}

```

---