FastAPI + LangChain 构建智能招聘系统实战：从简历解析到 AI 面试官的全链路设计

import PyPDF2
import io
from pydantic import BaseModel
from langchain.output_parsers import PydanticOutputParser

# 定义简历结构化 Schema
class ResumeSchema(BaseModel):
    name: str
    phone: str
    email: str
    skills: list[str]
    work_experience: list[dict]  # [{company, title, duration, achievements}]
    education: list[dict]
    summary: str

def parse_pdf_resume(file_bytes: bytes) -> str:
    """第一阶段：用传统 PDF 库提取纯文本（快速、低成本）"""
    reader = PyPDF2.PdfReader(io.BytesIO(file_bytes))
    text = ""
    for page in reader.pages:
        text += page.extract_text() or ""
    return text

def extract_resume_with_llm(raw_text: str) -> ResumeSchema:
    """第二阶段：LLM 结构化提取（只在文本提取成功后调用）"""
    parser = PydanticOutputParser(pydantic_object=ResumeSchema)
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一位资深 HR 助理。从简历文本中提取结构化信息。{format_instructions}"),
        ("human", "{raw_text}")
    ]).partial(format_instructions=parser.get_format_instructions())
    
    chain = prompt | ChatOpenAI(model="gpt-4o-mini", temperature=0) | parser
    return chain.invoke({"raw_text": raw_text})

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.tools import tool

# 初始化向量库（存储过往优秀简历/岗位描述）
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(collection_name="jd_vectors", embedding_function=embeddings)

@tool
def match_resume_to_jd(resume_summary: str, jd_requirements: str) -> dict:
    """
    计算简历与岗位匹配度，返回分数和理由。
    """
    # 1. 将 JD 转为向量，检索相似简历
    jd_vector = embeddings.embed_query(jd_requirements)
    similar_docs = vectorstore.similarity_search_by_vector(jd_vector, k=5)
    
    # 2. 让 LLM 评估匹配度
    eval_prompt = f"""
    岗位要求：{jd_requirements}
    候选人摘要：{resume_summary}
    相似成功案例：{similar_docs}
    
    请输出 JSON 格式：
    {{
        "match_score": 0-100,
        "strengths": ["候选人优势1", ...],
        "gaps": ["候选人不足1", ...],
        "recommendation": "建议进入下一轮/待定/不通过"
    }}
    """
    response = ChatOpenAI(model="gpt-4o-mini", temperature=0).invoke(eval_prompt)
    return json.loads(response.content)

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.prompts import MessagesPlaceholder

class InterviewerAgent:
    def __init__(self, session_id: str):
        self.session_id = session_id
        self.redis_client = redis.Redis()
        self.llm = ChatOpenAI(model="gpt-4-turbo", temperature=0.3)
    
    def build_prompt(self, jd: str, resume: str) -> ChatPromptTemplate:
        return ChatPromptTemplate.from_messages([
            ("system", f"""
            你是资深技术面试官。岗位 JD：{jd}，候选人背景：{resume}。
            规则：
            1. 每次只问一个问题。
            2. 如果候选人回答过于笼统（缺少数据、方法论），必须追问细节。
            3. 追问次数不超过 3 轮。
            4. 面试结束后，输出技术评级（S/A/B/C）。
            """),
            MessagesPlaceholder(variable_name="history"),
            ("human", "{input}")
        ])
    
    async def chat(self, user_message: str) -> str:
        # 从 Redis 获取对话历史
        history = self._get_history(self.session_id)
        
        prompt = self.build_prompt(...)
        chain = prompt | self.llm
        
        # 判断是否需要追问
        if self._needs_followup(user_message):
            followup = await self._generate_followup(user_message, history)
            response = followup
        else:
            response = await chain.ainvoke({"history": history, "input": user_message})
        
        # 存储历史
        self._save_history(self.session_id, user_message, response)
        return response
    
    def _needs_followup(self, text: str) -> bool:
        """用规则或小模型快速判断回答是否过于笼统"""
        vague_keywords = ["优化", "提升", "很多", "不错", "还行", "参与"]
        has_number = any(char.isdigit() for char in text)
        return any(kw in text for kw in vague_keywords) and not has_number

from fastapi import FastAPI, UploadFile, File, Form
from fastapi.responses import StreamingResponse
import json

app = FastAPI(title="AI 招聘系统", version="1.0")

@app.post("/interview/chat")
async def interview_chat(
    session_id: str = Form(...),
    user_message: str = Form(...)
):
    agent = InterviewerAgent(session_id)
    
    async def event_generator():
        # 使用 LangChain 的 astream 方法逐词生成
        async for chunk in agent.astream_chat(user_message):
            yield f"data: {json.dumps({'token': chunk, 'done': False})}\n\n"
        yield f"data: {json.dumps({'done': True})}\n\n"
    
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache"}
    )

@app.post("/resume/upload")
async def upload_resume(
    file: UploadFile = File(...),
    jd_id: str = Form(...)
):
    content = await file.read()
    raw_text = parse_pdf_resume(content)
    structured = extract_resume_with_llm(raw_text)
    
    # 自动触发匹配度评估
    match_result = match_resume_to_jd(structured.summary, get_jd_by_id(jd_id))
    
    return {
        "resume": structured.dict(),
        "match": match_result,
        "recommendation": match_result["recommendation"]
    }

@tool
def generate_evaluation_report(
    session_id: str,
    resume: ResumeSchema,
    interview_history: list
) -> str:
    """
    综合简历和面试表现，生成候选人评估报告。
    """
    prompt = f"""
    候选人简历：{resume.dict()}
    面试对话记录：{interview_history}
    
    请按以下维度输出报告：
    1. 技术能力（1-10 分）：基于面试中回答的技术深度
    2. 沟通表达（1-10 分）：逻辑清晰度、表达流畅度
    3. 岗位匹配度（%）：结合简历与 JD 的契合度
    4. 优势亮点：3-5 条
    5. 潜在风险：2-3 条
    6. 录用建议：强烈推荐/推荐/待定/不推荐
    
    输出 Markdown 格式。
    """
    response = ChatOpenAI(model="gpt-4-turbo", temperature=0.2).invoke(prompt)
    return response.content

import chardet
detected = chardet.detect(raw_bytes)
encoding = detected['encoding'] or 'utf-8'

def compress_history(history: list) -> str:
    summary_prompt = f"请将以下对话压缩为 100 字内的摘要：{history[:-4]}"
    return llm.invoke(summary_prompt).content