from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

from pydantic import BaseModel, Field

class User(BaseModel):
    id: int
    name: str = Field(..., min_length=1, max_length=50)
    email: str
    age: int = Field(ge=0, le=150)  # ge: greater or equal, le: less or equal

from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import PydanticOutputParser

# 1. 初始化模型
llm = init_chat_model(model="qwen-plus", model_provider="openai")

# 2. 定义提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个助手"),
    ("human", "{input}")
])

# 3. 定义输出解析器
parser = PydanticOutputParser(pydantic_object=User)

# 4. 创建链
chain = prompt | llm | parser

# 5. 执行
result = chain.invoke({"input": "你好"})

from langgraph.graph import StateGraph, END
from typing import TypedDict

# 1. 定义状态
class State(TypedDict):
    input: str
    output: str

# 2. 定义节点函数
def node_a(state):
    return {"output": state["input"] + " - processed by A"}

def node_b(state):
    return {"output": state["output"] + " - processed by B"}

# 3. 构建图
graph = StateGraph(State)
graph.add_node("node_a", node_a)
graph.add_node("node_b", node_b)
graph.set_entry_point("node_a")
graph.add_edge("node_a", "node_b")
graph.add_edge("node_b", END)

# 4. 编译并运行
compiled_graph = graph.compile()
result = compiled_graph.invoke({"input": "Hello"})

import chromadb

# 1. 创建客户端
client = chromadb.PersistentClient(path="./chroma_db")

# 2. 创建集合
collection = client.create_collection(name="my_collection")

# 3. 添加向量
collection.add(
    embeddings=[[0.1, 0.2, 0.3, ...]],  # 嵌入向量
    ids=["doc1"],
    documents=["这是文档内容"]
)

# 4. 查询相似向量
results = collection.query(
    query_embeddings=[[0.15, 0.25, 0.35, ...]],
    n_results=5
)

from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String(50))

const { createApp, ref, computed } = Vue

createApp({
    setup() {
        const count = ref(0)
        const double = computed(() => count.value * 2)

        function increment() {
            count.value++
        }

        return { count, double, increment }
    }
}).mount('#app')

<div class="flex items-center justify-center p-4 bg-white rounded-lg shadow-md">
    <span class="text-gray-700 font-bold">Hello World</span>
</div>

# 1. 下载 Python 3.11 安装包
# 访问 https://www.python.org/downloads/windows/

# 2. 运行安装程序，勾选"Add Python to PATH"

# 3. 验证安装
python --version  # 应显示 Python 3.11.x

# Ubuntu/Debian
sudo apt update
sudo apt install python3.11 python3.11-venv python3.11-dev

# CentOS/RHEL
sudo yum install python3.11

# Windows: 下载 Docker Desktop for Windows
# https://www.docker.com/products/docker-desktop

# Linux
curl -fsSL https://get.docker.com | sh
sudo systemctl start docker
sudo systemctl enable docker

# 验证安装
docker --version
docker compose version

QQ 邮箱：imap.qq.com
163 邮箱：imap.163.com
Gmail：imap.gmail.com
Outlook：imap-mail.outlook.com

# 1. 创建项目目录
mkdir resume_screen_agent
cd resume_screen_agent

# 2. 创建子目录
mkdir -p db utils workflow/nodes static fliter_resumes resumes chroma_db

# 3. 创建__init__.py 文件（使目录成为 Python 包）
touch db/__init__.py utils/__init__.py workflow/__init__.py workflow/nodes/__init__.py

# 1. 创建虚拟环境
python -m venv venv

# 2. 激活虚拟环境
# Windows
venv\Scripts\activate
# Linux/Mac
source venv/bin/activate

# 3. 验证
which python  # 应指向虚拟环境中的 Python

fastapi==0.135.1
uvicorn==0.34.0
python-dotenv==1.2.2
sqlalchemy==2.0.23
pymysql==1.1.2
langchain==1.2.12
langgraph==1.1.2
langchain-openai==1.1.11
chromadb==1.5.5
dashscope==1.25.14
pypdf2==3.0.1
python-docx==1.2.0
pydantic==2.12.5
numpy==2.4.3

pip install -r requirements.txt

# 邮箱配置
EMAIL_USER=your_email@example.com
EMAIL_PWD=your_email_password_or_auth_code
EMAIL_SERVER=imap.example.com

# 通义千问 API
DASHSCOPE_API_KEY=your_api_key
DASHSCOPE_API_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
DASHSCOPE_MODEL=qwen-plus

# 数据库配置
DB_USER=root
DB_PASS=your_password
DB_HOST=localhost
DB_NAME=resume_db

# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
venv/
env/
ENV/

# 环境变量
.env
.env.local

# IDE
.idea/
.vscode/
*.swp
*.swo

# 数据库
*.db
*.sqlite

# 日志
*.log

# 上传文件
resumes/
fliter_resumes/

# ChromaDB
chroma_db/

# 使用 Python 3.11 作为基础镜像
FROM python:3.11-slim

# 设置工作目录
WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    build-essential \
    curl \
    && rm -rf /var/lib/apt/lists/*

# 复制依赖文件
COPY requirements.txt .

# 安装 Python 依赖
RUN pip install --no-cache-dir -r requirements.txt

# 复制项目代码
COPY . .

# 暴露端口
EXPOSE 8000

# 设置环境变量
ENV PYTHONUNBUFFERED=1

# 启动命令
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

version: '3.8'

services:
  app:
    build: .
    ports:
      - "8000:8000"
    depends_on:
      - db
    environment:
      - DB_USER=root
      - DB_PASS=password
      - DB_HOST=db
      - DB_NAME=resume_db
    volumes:
      - ./fliter_resumes:/app/fliter_resumes
      - ./static:/app/static

  db:
    image: mysql:8.0
    environment:
      - MYSQL_ROOT_PASSWORD=password
      - MYSQL_DATABASE=resume_db
    ports:
      - "3306:3306"
    volumes:
      - mysql_data:/var/lib/mysql

volumes:
  mysql_data:

# 1. 构建并启动所有服务
docker compose up -d

# 2. 查看日志
docker compose logs -f

# 3. 停止服务
docker compose down

# 4. 停止并删除数据卷
docker compose down -v

# 1. 启动 MySQL（需要预先安装）
# Windows: 启动 MySQL 服务
# Linux: sudo systemctl start mysql

# 2. 创建数据库
mysql -u root -p -e "CREATE DATABASE resume_db;"

# 3. 运行应用
uvicorn main:app --reload --host 0.0.0.0 --port 8000

# --reload: 代码变化自动重启（开发环境）
# --host: 监听地址
# --port: 监听端口

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from dotenv import load_dotenv
import os

load_dotenv()

# 从环境变量读取配置
DB_USER = os.getenv("DB_USER")
DB_PASS = os.getenv("DB_PASS")
DB_HOST = os.getenv("DB_HOST")
DB_NAME = os.getenv("DB_NAME")

# 构建数据库 URL
DATABASE_URL = f"mysql+pymysql://{DB_USER}:{DB_PASS}@{DB_HOST}/{DB_NAME}"

# 创建数据库引擎
engine = create_engine(
    DATABASE_URL,
    pool_size=10,        # 连接池大小
    max_overflow=20,     # 最大溢出连接数
    pool_recycle=3600,   # 连接回收时间（秒）
    echo=False          # 是否打印 SQL 日志
)

# 创建会话工厂
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

from sqlalchemy import Column, Integer, String, ForeignKey, Text
from sqlalchemy.orm import relationship, declarative_base

Base = declarative_base()

# 部门模型
class Department(Base):
    __tablename__ = "departments"
    id = Column(String(100), primary_key=True)
    name = Column(String(255), nullable=False)
    # 一对多关系：一个部门有多个岗位
    positions = relationship("Position", back_populates="department", cascade="all, delete-orphan")

# 岗位模型
class Position(Base):
    __tablename__ = "positions"
    id = Column(String(100), primary_key=True)
    name = Column(String(255), nullable=False)
    jd = Column(Text, nullable=True)  # 岗位描述
    department_id = Column(String(100), ForeignKey("departments.id"))
    # 多对一关系：多个岗位属于一个部门
    department = relationship("Department", back_populates="positions")

# 系统配置模型
class SystemConfig(Base):
    __tablename__ = "system_configs"
    id = Column(Integer, primary_key=True)
    email = Column(String(255), nullable=False)
    email_password = Column(String(255), nullable=False)
    email_server = Column(String(255), nullable=False)
    api_key = Column(String(255), nullable=False)
    api_model = Column(String(255), nullable=False)
    api_url = Column(String(255), nullable=False)

# 筛选简历模型
class FilteredResume(Base):
    __tablename__ = "filtered_resumes"
    id = Column(Integer, primary_key=True)
    name = Column(String(255))
    school = Column(String(255))
    age = Column(Integer)
    education = Column(String(255))
    phone = Column(String(100))
    email = Column(String(255))
    final_score = Column(Integer)
    date = Column(String(100))
    email_path = Column(String(500))
    position = Column(String(255))

import imaplib
import email
from email.header import decode_header

# 1. 连接 IMAP 服务器
server = imaplib.IMAP4_SSL("imap.qq.com")

# 2. 登录
server.login("user@qq.com", "password")

# 3. 选择邮箱文件夹
server.select("INBOX")

# 4. 搜索邮件
status, messages = server.search(None, 'ALL')

# 5. 获取邮件内容
status, msg_data = server.fetch("1", "(RFC822)")

# 6. 解析邮件
email_obj = email.message_from_bytes(msg_data[0][1])

# 7. 解码邮件头
subject = decode_header(email_obj["Subject"])

import PyPDF2

def extract_pdf_text(filepath):
    with open(filepath, 'rb') as f:
        reader = PyPDF2.PdfReader(f)
        text = '\n'.join([page.extract_text() for page in reader.pages])
    return text

from docx import Document

def extract_docx_text(filepath):
    doc = Document(filepath)
    text = '\n'.join([para.text for para in doc.paragraphs])
    return text

def read_resume_emails(state: ResumeProcessState):
    """
    从邮箱读取简历附件
    1. 查询数据库获取最新邮件时间
    2. 登录 IMAP 服务器
    3. 搜索新邮件
    4. 下载附件
    5. 保存信息到数据库
    """
    # 获取最新邮件时间（避免重复处理）
    latest_date = get_latest_email_from_db()

    # 连接邮箱
    server = imaplib.IMAP4_SSL(EMAIL_SERVER)
    server.login(EMAIL_USER, EMAIL_PWD)
    server.select("INBOX", readonly=True)

    # 搜索邮件
    if latest_date:
        imap_date = format_imap_date(latest_date)
        status, messages = server.search(None, f'SINCE {imap_date}')
    else:
        status, messages = server.search(None, 'ALL')

    # 处理每封邮件
    resume_info_list = []
    for msg_id in messages[0].split():
        email_data = process_email(server, msg_id)
        if email_data:
            resume_info_list.append(email_data)

    # 保存到数据库
    save_to_db(resume_info_list)

    return {"email_raw": {"raw_email_resumes": resume_info_list}}

def parse_resume(state: ResumeProcessState):
    """
    使用 LLM 解析简历为结构化数据
    """
    # 初始化 LLM
    llm = init_chat_model(
        model=os.getenv("DASHSCOPE_MODEL"),
        model_provider="openai",
        api_key=os.getenv("DASHSCOPE_API_KEY"),
        base_url=os.getenv("DASHSCOPE_API_URL"),
    )

    # 定义输出格式
    parser = PydanticOutputParser(pydantic_object=Resume)

    # 创建提示模板
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你必须按照以下格式输出：{format_instructions}"),
        ("human", "请提取简历信息：{resume_text}")
    ])

    # 创建处理链
    chain = prompt | llm | parser

    # 处理每个简历
    structured_resumes = []
    for item in state.resume_text.resume_texts:
        result = chain.invoke({
            "resume_text": item["text"],
            "format_instructions": parser.get_format_instructions()
        })
        structured_resumes.append(result)

    return {"structured": {"structured_resumes": structured_resumes}}

from pydantic import BaseModel, Field
from typing import List, Dict, Optional

# 子状态：原始邮件信息
class EmailRawState(BaseModel):
    raw_email_resumes: List[Dict[str, str]] = Field(default_factory=list)

# 子状态：简历文本
class ResumeTextState(BaseModel):
    resume_texts: List[Dict[str, str]] = Field(default_factory=list)

# 子状态：结构化简历
class StructuredResumeState(BaseModel):
    structured_resumes: List[Resume] = Field(default_factory=list)

# 子状态：评分结果
class ScoreState(BaseModel):
    llm_scores: List[ResumeLLMScore] = Field(default_factory=list)
    final_scores: List[Dict[str, Any]] = Field(default_factory=list)
    filtered_resumes: List[Dict[str, Any]] = Field(default_factory=list)

# 子状态：JD 信息
class JdState(BaseModel):
    jd_text: Optional[str] = None

# 主状态：整合所有子状态
class ResumeProcessState(BaseModel):
    email_raw: EmailRawState = Field(default_factory=EmailRawState)
    resume_text: ResumeTextState = Field(default_factory=ResumeTextState)
    structured: StructuredResumeState = Field(default_factory=StructuredResumeState)
    score: ScoreState = Field(default_factory=ScoreState)
    jd: JdState = Field(default_factory=JdState)

初始状态 → [read_resume_emails] → 包含原始邮件信息
         → [read_resume_file] → 包含简历文本
         → [parse_resume] → 包含结构化简历
         → [llm_evaluate_resume] → 包含 LLM 评分
         → [get_filter_resume] → 包含最终筛选结果

from langgraph.graph import StateGraph, END

# 1. 初始化图
graph_builder = StateGraph(ResumeProcessState)

# 2. 添加节点
graph_builder.add_node("read_resume_emails", read_resume_emails)
graph_builder.add_node("read_resume_file", read_resume_file)
graph_builder.add_node("parse_resume", parse_resume)
graph_builder.add_node("llm_evaluate_resume", llm_evaluate_resume)
graph_builder.add_node("get_filter_resume", get_filter_resume)

# 3. 设置边（执行顺序）
graph_builder.set_entry_point("read_resume_emails")
graph_builder.add_edge("read_resume_emails", "read_resume_file")
graph_builder.add_edge("read_resume_file", "parse_resume")
graph_builder.add_edge("parse_resume", "llm_evaluate_resume")
graph_builder.add_edge("llm_evaluate_resume", "get_filter_resume")
graph_builder.add_edge("get_filter_resume", END)

# 4. 编译图
graph = graph_builder.compile()

from langgraph.graph import START, END

# 添加条件分支
def should_continue(state):
    if state["score"] >= 60:
        return "high_score_branch"
    else:
        return "low_score_branch"

graph.add_conditional_edges(
    "evaluate",
    should_continue,
    {
        "high_score_branch": "accept",
        "low_score_branch": "reject"
    }
)

sys_prompt = """
你是专业的简历匹配度评估专家，需要完成以下任务：
1. 分析 JD 要求：{jd_text}
2. 分析简历信息：{resume_info}
3. 从技能匹配度、经验匹配度、学历匹配度、岗位契合度、行业适配度 5 个维度综合评估
4. 严格要求以下格式输出：{format_instructions}
"""

examples = """
示例 1：
JD：招聘 Python 工程师，要求 3 年经验，熟悉 FastAPI、SQLAlchemy
简历：张三，5 年 Python 经验，精通 FastAPI、Django、MySQL
评分：{"score": 85}

示例 2：
JD：招聘数据科学家，要求硕士学历，精通机器学习
简历：李四，本科学历，1 年数据分析经验
评分：{"score": 45}
"""

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import PydanticOutputParser

# 1. 提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", sys_prompt),
    ("human", "请评估上述 JD 和简历的综合匹配度")
])

# 2. 输出解析器
parser = PydanticOutputParser(pydantic_object=ResumeLLMScore)

# 3. 创建链
chain = prompt | llm | parser

# 4. 执行
result = chain.invoke({
    "jd_text": jd_text,
    "resume_info": resume_text,
    "format_instructions": parser.get_format_instructions()
})

import dashscope

def get_embedding(text):
    """
    使用通义千问 API 生成文本嵌入向量
    """
    dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")

    res = dashscope.TextEmbedding.call(
        model="text-embedding-v1",
        input=text,
    )

    if res and "output" in res and "embeddings" in res["output"]:
        embedding = res["output"]["embeddings"][0]["embedding"]
        return embedding  # 返回向量列表，如 [0.1, 0.2, 0.3, ...]

    return None

import numpy as np

def get_cosine_similarity(vec1, vec2):
    """
    计算两个向量的余弦相似度
    公式：cos(θ) = (A·B) / (||A|| × ||B||)
    """
    vec1 = np.array(vec1)
    vec2 = np.array(vec2)

    # 点积
    dot_product = np.dot(vec1, vec2)

    # L2 范数（模长）
    norm1 = np.linalg.norm(vec1)
    norm2 = np.linalg.norm(vec2)

    # 余弦相似度
    similarity = dot_product / (norm1 * norm2)

    # 限制在 [-1, 1] 范围（浮点精度问题）
    similarity = np.clip(similarity, -1.0, 1.0)

    return float(similarity)

import chromadb
import uuid

# 创建客户端
chroma_client = chromadb.PersistentClient(path="../chroma_db")

def add_jd_to_chroma(jd_text: str):
    """
    将 JD 添加到 Chroma 数据库
    """
    # 获取集合
    collection = chroma_client.get_or_create_collection(name="jd_collection")

    # 生成嵌入向量
    jd_embedding = get_embedding(jd_text)

    # 检查是否已存在相似 JD
    results = collection.query(
        query_embeddings=[jd_embedding],
        n_results=1
    )

    # 如果距离很小（相似度高），复用已有向量
    if results['distances'] and results['distances'][0][0] < 0.01:
        print("找到相似 JD，复用向量")
        return results['embeddings'][0][0]

    # 添加新向量
    doc_id = str(uuid.uuid4())
    collection.add(
        embeddings=[jd_embedding],
        ids=[doc_id],
        documents=[jd_text]
    )

    return jd_embedding

def calculate_final_score(rag_score, llm_score):
    """
    加权计算最终评分
    RAG 评分（余弦相似度）× 100 × 0.6 + LLM 评分 × 0.4
    """
    final_score = round(rag_score * 100 * 0.6 + llm_score * 0.4)
    return final_score

┌─────────────────┐       ┌─────────────────┐
│   departments   │       │  system_configs │
├─────────────────┤       ├─────────────────┤
│ id (PK)         │       │ id (PK)         │
│ name            │       │ email           │
└────────┬────────┘       │ email_password  │
         │                │ email_server    │
         │ 1:N            │ api_key         │
         │                │ api_model       │
         ▼                │ api_url         │
┌─────────────────┐       └─────────────────┘
│    positions    │
├─────────────────┤
│ id (PK)         │       ┌─────────────────┐
│ name            │       │ filtered_resumes│
│ jd (TEXT)       │       ├─────────────────┤
│ department_id   │───┐   │ id (PK)         │
└─────────────────┘   │   │ name            │
                      │   │ school          │
                      │   │ age             │
                      │   │ education       │
                      │   │ phone           │
                      │   │ email           │
                      │   │ final_score     │
                      │   │ date            │
                      │   │ position        │
                      │   └─────────────────┘
                      │
                      │   ┌─────────────────┐
                      └──▶│     resumes     │
                          ├─────────────────┤
                          │ id (PK)         │
                          │ sender          │
                          │ subject         │
                          │ date            │
                          │ filepath        │
                          │ position        │
                          └─────────────────┘

CREATE TABLE departments (
    id VARCHAR(100) PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    INDEX idx_name (name)
);

CREATE TABLE positions (
    id VARCHAR(100) PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    jd TEXT,
    department_id VARCHAR(100) NOT NULL,
    FOREIGN KEY (department_id) REFERENCES departments(id)
        ON DELETE CASCADE,
    INDEX idx_department (department_id)
);

CREATE TABLE filtered_resumes (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(255),
    school VARCHAR(255),
    age INT,
    education VARCHAR(255),
    phone VARCHAR(100),
    email VARCHAR(255),
    final_score INT,
    date VARCHAR(100),
    email_path VARCHAR(500),
    position VARCHAR(255),
    INDEX idx_score (final_score DESC),
    INDEX idx_position (position)
);

from sqlalchemy.orm import Session

# 依赖注入：获取数据库会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 在 API 中使用
@app.get("/api/departments")
async def get_departments(db: Session = Depends(get_db)):
    departments = db.query(Department).all()
    return departments

# 正确
GET /api/departments      # 获取部门列表
GET /api/departments/123  # 获取单个部门
POST /api/departments     # 创建部门

# 错误
GET /api/getDepartments   # 动词不应出现在 URL 中
POST /api/createDepartment

@app.get("/api/run-workflow")
async def run_workflow(position_id: str, db: Session = Depends(get_db)):
    """
    运行简历筛选工作流
    """
    # 1. 获取岗位 JD
    position = db.query(Position).filter(Position.id == position_id).first()
    if not position:
        raise HTTPException(status_code=404, detail="岗位不存在")

    # 2. 创建初始状态
    initial_state = create_initial_state(position.jd)

    # 3. 执行工作流
    final_state = graph.invoke(initial_state)

    # 4. 返回结果
    return {
        "status": "success",
        "message": f"处理了 {len(final_state['email_raw']['raw_email_resumes'])} 个简历",
        "data": {
            "filtered_resumes": final_state["score"]["filtered_resumes"],
            "final_scores": final_state["score"]["final_scores"]
        }
    }

@app.get("/api/candidates")
async def get_candidates(
    page: int = 1,
    page_size: int = 10,
    position_id: str = None,
    db: Session = Depends(get_db)
):
    """
    获取筛选后的候选人列表（支持分页和岗位筛选）
    """
    # 计算偏移量
    offset = (page - 1) * page_size

    # 构建查询
    query = db.query(FilteredResume)
    if position_id:
        position = db.query(Position).filter(Position.id == position_id).first()
        query = query.filter(FilteredResume.position == position.name)

    # 获取总数
    total = query.count()

    # 分页查询
    resumes = query.offset(offset).limit(page_size).all()

    # 转换为响应格式
    items = []
    for resume in resumes:
        items.append({
            "id": resume.id,
            "name": resume.name,
            "school": resume.school,
            "education": resume.education,
            "final_score": resume.final_score,
            "rating": resume.final_score / 100 * 5,  # 转换为 5 星制
            "info": f"{resume.education} / {resume.school}"
        })

    return {
        "items": items,
        "total": total,
        "page": page,
        "page_size": page_size
    }

# 成功响应
{
    "status": "success",
    "message": "操作成功",
    "data": {...}
}

# 错误响应
{
    "status": "error",
    "message": "错误描述",
    "code": "ERROR_CODE"
}

<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <title>HR 简历管理系统</title>
    <script src="https://cdn.jsdelivr.net/npm/vue@3/dist/vue.global.prod.js"></script>
    <script src="https://cdn.tailwindcss.com"></script>
</head>
<body>
    <div id="app">
        <!-- 侧边栏 -->
        <aside class="w-64 bg-slate-900">
            <!-- 部门列表 -->
            <div v-for="dept in departments" :key="dept.id">
                <h3>{{ dept.name }}</h3>
                <a v-for="position in dept.positions" 
                   @click="selectPosition(position.id)">
                    {{ position.name }}
                </a>
            </div>
        </aside>

        <!-- 主内容区 -->
        <main class="flex-1">
            <!-- 候选人列表 -->
            <div v-for="candidate in candidates" :key="candidate.id">
                <h4>{{ candidate.name }}</h4>
                <p>{{ candidate.school }}</p>
                <span>评分：{{ candidate.final_score }}</span>
            </div>
        </main>
    </div>

    <script>
        const { createApp, ref, onMounted } = Vue

        createApp({
            setup() {
                const departments = ref([])
                const candidates = ref([])
                const activePosition = ref(null)

                // 加载部门列表
                async function loadDepartments() {
                    const res = await fetch('/api/departments')
                    departments.value = await res.json()
                }

                // 选择岗位
                async function selectPosition(positionId) {
                    activePosition.value = positionId
                    const res = await fetch(`/api/candidates?position_id=${positionId}`)
                    const data = await res.json()
                    candidates.value = data.items
                }

                onMounted(() => {
                    loadDepartments()
                })

                return {
                    departments,
                    candidates,
                    selectPosition
                }
            }
        }).mount('#app')
    </script>
</body>
</html>

<!-- 布局 -->
<div class="flex h-screen">              <!-- Flex 布局，全屏高度 -->
    <div class="w-64 flex-shrink-0">    <!-- 固定宽度 256px -->

    <!-- 间距 -->
    <div class="p-4">                   <!-- 内边距 1rem -->
    <div class="m-2">                   <!-- 外边距 0.5rem -->
    <div class="space-y-2">             <!-- 子元素垂直间距 -->

    <!-- 颜色 -->
    <div class="bg-blue-500">           <!-- 蓝色背景 -->
    <div class="text-white">            <!-- 白色文字 -->

    <!-- 响应式 -->
    <div class="md:w-1/2 lg:w-1/4">     <!-- 中屏 50%，大屏 25% -->

# 阶段 1：构建依赖
FROM python:3.11-slim as builder

WORKDIR /app
COPY requirements.txt .
RUN pip install --user --no-cache-dir -r requirements.txt

# 阶段 2：运行环境
FROM python:3.11-slim

WORKDIR /app
COPY --from=builder /root/.local /root/.local
COPY . .

ENV PATH=/root/.local/bin:$PATH
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

version: '3.8'

services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DB_USER=root
      - DB_PASS=${DB_PASSWORD}
      - DB_HOST=db
      - DB_NAME=resume_db
    depends_on:
      db:
        condition: service_healthy
    restart: always
    deploy:
      resources:
        limits:
          cpus: '2'
          memory: 2G

  db:
    image: mysql:8.0
    environment:
      - MYSQL_ROOT_PASSWORD=${DB_PASSWORD}
      - MYSQL_DATABASE=resume_db
    volumes:
      - mysql_data:/var/lib/mysql
    healthcheck:
      test: ["CMD", "mysqladmin", "ping", "-h", "localhost"]
      interval: 10s
      timeout: 5s
      retries: 5
    restart: always

volumes:
  mysql_data:

# 1. 准备生产环境变量
cp .env.example .env
# 编辑.env 文件，设置生产配置

# 2. 构建并启动
docker compose -f docker-compose.prod.yml up -d --build

# 3. 查看日志
docker compose logs -f app

# 4. 数据库迁移（如果需要）
docker compose exec app alembic upgrade head

# 5. 健康检查
curl http://localhost:8000/health

# 6. 停止服务
docker compose -f docker-compose.prod.yml down

Error: DASHSCOPE_API_KEY is not set

# 确保在文件开头加载环境变量
from dotenv import load_dotenv
load_dotenv()  # 加载.env 文件

# 验证环境变量
import os
print(os.getenv("DASHSCOPE_API_KEY"))

sqlalchemy.exc.OperationalError: (2003, "Can't connect to MySQL server")

# 1. 检查 MySQL 容器是否运行
docker compose ps

# 2. 查看 MySQL 日志
docker compose logs db

# 3. 测试连接
docker compose exec db mysql -u root -p

# 4. 检查网络
docker compose exec app ping db

dashscope.api_error.APIError: Invalid API key

# 1. 验证 API Key
import dashscope
dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")
print(f"API Key: {dashscope.api_key[:10]}...")  # 打印前 10 位

# 2. 测试 API
res = dashscope.TextEmbedding.call(
    model="text-embedding-v1",
    input="测试"
)
print(res)

# 3. 检查余额
# 登录阿里云控制台 -> DashScope -> 用量查询

# 1. 添加日志
def node_a(state):
    print("=== 开始执行 node_a ===")
    print(f"输入状态：{state}")
    # ... 处理逻辑
    print("=== node_a 执行完成 ===")
    return {"output": "..."}

# 2. 单步调试
final_state = graph.invoke(initial_state, {"debug": True})

# 3. 检查节点输出
intermediate_states = graph.invoke(
    initial_state,
    return_intermediate_steps=True
)

# 1. 批量处理（使用 LangChain batch）
results = chain.batch(resume_texts)

# 2. 异步处理
async def process_resume(text):
    return await chain.ainvoke({"resume_text": text})

# 3. 并发限制
import asyncio
semaphore = asyncio.Semaphore(5)  # 最多 5 个并发

# 4. 缓存嵌入向量
from functools import lru_cache

@lru_cache(maxsize=1000)
def get_embedding_cached(text):
    return get_embedding(text)

# 支持同时处理多个岗位
@app.post("/api/run-workflow-batch")
async def run_workflow_batch(position_ids: list[str]):
    tasks = []
    for position_id in position_ids:
        task = process_position(position_id)
        tasks.append(task)

    results = await asyncio.gather(*tasks)
    return {"results": results}

def is_duplicate_resume(email, phone, db):
    """
    检查简历是否重复
    """
    existing = db.query(FilteredResume).filter(
        (FilteredResume.email == email) | 
        (FilteredResume.phone == phone)
    ).first()
    return existing is not None

def generate_tags(resume_dict, jd_text):
    """
    基于简历和 JD 自动生成标签
    """
    tags = []

    # 学历标签
    if resume_dict.get("education") in ["博士", "硕士"]:
        tags.append("高学历")

    # 经验标签
    work_years = int(resume_dict.get("work_years", 0))
    if work_years >= 5:
        tags.append("资深")
    elif work_years >= 3:
        tags.append("中级")
    else:
        tags.append("初级")

    # 技能标签（基于关键词匹配）
    skills = ["Python", "Java", "机器学习", "深度学习"]
    for skill in skills:
        if skill.lower() in jd_text.lower():
            tags.append(skill)

    return tags

import redis
import json

redis_client = redis.Redis(host='redis', port=6379, db=0)

def get_cached_embedding(text):
    """
    从 Redis 获取缓存的嵌入向量
    """
    cache_key = f"embedding:{hash(text)}"
    cached = redis_client.get(cache_key)

    if cached:
        return json.loads(cached)

    # 未命中，生成新向量并缓存
    embedding = get_embedding(text)
    redis_client.setex(cache_key, 3600, json.dumps(embedding))  # 缓存 1 小时
    return embedding

# 使用 Celery 处理异步任务
from celery import Celery

celery = Celery('tasks', broker='redis://redis:6379/0')

@celery.task
def process_resume_async(resume_id):
    """
    异步处理简历
    """
    # 处理逻辑...
    return {"status": "completed"}

# API 中调用
@app.post("/api/process-resume-async")
async def process_resume_api(resume_id: int):
    task = process_resume_async.delay(resume_id)
    return {"task_id": task.id}

from prometheus_fastapi_instrumentator import Instrumentator

app = FastAPI()
Instrumentator().instrument(app).expose(app)

# 访问 /metrics 查看指标

import logging
import json

# 配置 JSON 日志
class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            "level": record.levelname,
            "message": record.getMessage(),
            "timestamp": self.formatTime(record),
            "module": record.module
        }
        return json.dumps(log_data)

logger = logging.getLogger()
logger.addHandler(logging.StreamHandler())
logger.handlers[0].setFormatter(JSONFormatter())

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    """
    验证 Bearer Token
    """
    token = credentials.credentials
    if token != os.getenv("API_TOKEN"):
        raise HTTPException(status_code=401, detail="Invalid token")
    return token

@app.get("/api/protected")
async def protected_route(token: str = Depends(verify_token)):
    return {"message": "访问成功"}

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.get("/api/run-workflow")
@limiter.limit("5/minute")  # 每分钟最多 5 次
async def run_workflow_limited(request):
    # ...

resume_screen_agent/
├── db/
│   ├── __init__.py
│   └── database.py         # 数据库模型和会话管理
├── utils/
│   ├── __init__.py
│   ├── cosine_similarity.py  # 余弦相似度计算
│   ├── get_embedding.py      # 嵌入向量生成
│   ├── jd_chroma.py          # JD 向量存储
│   ├── message_pydantic.py   # 简历数据模型
│   └── resume_core_text.py   # 简历文本提取
├── workflow/
│   ├── __init__.py
│   ├── graph_desin.py        # 工作流图设计
│   ├── graph_state.py        # 工作流状态定义
│   └── nodes/
│       ├── __init__.py
│       ├── filter_resume_node.py    # 简历筛选节点
│       ├── llm_score_node.py        # LLM 评分节点
│       ├── meta_email_node.py       # 邮件处理节点
│       ├── resume_text_node.py      # 简历文本读取节点
│       └── structured_resume_node.py # 简历结构化节点
├── static/
│   └── index.html          # Web 前端界面
├── fliter_resumes/         # 筛选后的简历存储
├── resumes/                # 原始简历存储
├── chroma_db/              # ChromaDB 数据（.gitignore）
├── .env                    # 环境变量（.gitignore）
├── .env.example            # 环境变量模板
├── .gitignore
├── Dockerfile
├── docker-compose.yml
├── docker-compose.prod.yml
├── main.py                 # 应用入口
├── requirements.txt
└── README.md

# 开发环境
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install -r requirements.txt
uvicorn main:app --reload

# Docker 部署
docker compose up -d
docker compose logs -f
docker compose down

# 数据库操作
docker compose exec db mysql -u root -p
docker compose exec app python -c "from db.database import Base, engine; Base.metadata.create_all(bind=engine)"

# 测试 API
curl http://localhost:8000/api/departments
curl http://localhost:8000/docs